package creator import ( "context" "database/sql" "fmt" "math" "strings" "time" ) func (s *Store) ListDueMetricWorks(ctx context.Context, now time.Time) ([]Work, error) { if now.IsZero() { return nil, ErrInvalid } rows, err := s.db.QueryContext(ctx, workSelect+` WHERE published_at_status='verified' AND next_metric_at IS NOT NULL AND next_metric_at <= $1 ORDER BY next_metric_at,id`, now.UTC()) if err != nil { return nil, databaseError(err) } works := make([]Work, 0) for rows.Next() { work, scanErr := scanWork(rows) if scanErr != nil { _ = rows.Close() return nil, scanErr } works = append(works, work) } if err := rows.Err(); err != nil { _ = rows.Close() return nil, err } if err := rows.Close(); err != nil { return nil, err } for index := range works { loadWorkSources(&works[index]) } return works, nil } func (s *Store) StopMetricPlan(ctx context.Context, workID, reason string) error { workID = strings.TrimSpace(workID) reason = strings.TrimSpace(reason) if workID == "" || reason == "" || len(reason) > 200 { return ErrInvalid } _, err := s.db.ExecContext(ctx, `UPDATE creator_work SET metric_plan_next_at=NULL, metric_plan_stopped=true, next_metric_at=NULL, metric_stop_reason=$2, updated_at=now() WHERE work_id=$1 AND NOT metric_plan_stopped`, workID, reason) return databaseError(err) } func (s *Store) EnsureMetricPlan(ctx context.Context, workID string, settings Settings) error { if err := ValidateSettings(SettingsUpdate{LookbackDays: settings.LookbackDays, NewWorkIntervalSeconds: settings.NewWorkIntervalSeconds, MetricInitialIntervalSeconds: settings.MetricInitialIntervalSeconds, MetricMultiplier: settings.MetricMultiplier, MetricMaxIntervalSeconds: settings.MetricMaxIntervalSeconds, MetricAgeSeconds: settings.MetricAgeSeconds}); err != nil { return err } work, err := s.GetWork(ctx, workID) if err != nil { return err } if work.PublishedAt == nil || work.PublishedAtStatus != "verified" { return nil } now := time.Now().UTC() publishedAt := work.PublishedAt.UTC() monitoringEnd := publishedAt.Add(time.Duration(settings.MetricAgeSeconds) * time.Second) if !publishedAt.Before(now) { _, err := s.db.ExecContext(ctx, `UPDATE creator_work SET next_metric_at=NULL, metric_stop_reason='published_at_pending_verification', updated_at=now() WHERE work_id=$1`, workID) return databaseError(err) } if !now.Before(monitoringEnd) { _, err := s.db.ExecContext(ctx, `UPDATE creator_work SET metric_plan_next_at=NULL, metric_plan_stopped=true, next_metric_at=NULL, metric_stop_reason='monitoring_age_reached', updated_at=now() WHERE work_id=$1 AND NOT metric_plan_stopped`, workID) return databaseError(err) } nextAt := publishedAt.Add(time.Duration(settings.MetricInitialIntervalSeconds) * time.Second) if !nextAt.Before(monitoringEnd) { _, err := s.db.ExecContext(ctx, `UPDATE creator_work SET next_metric_at=NULL, metric_stop_reason='monitoring_age_reached', updated_at=now() WHERE work_id=$1`, workID) return databaseError(err) } if _, err := s.db.ExecContext(ctx, ` UPDATE creator_work SET metric_plan_published_at = COALESCE(metric_plan_published_at, $2), metric_monitoring_end = COALESCE(metric_monitoring_end, $3), metric_plan_next_at = COALESCE(metric_plan_next_at, $4), metric_plan_interval_seconds = COALESCE(metric_plan_interval_seconds, $5), metric_plan_multiplier = COALESCE(metric_plan_multiplier, $6), metric_plan_max_interval_seconds = COALESCE(metric_plan_max_interval_seconds, $7), updated_at = now() WHERE work_id = $1`, workID, publishedAt, monitoringEnd, nextAt, settings.MetricInitialIntervalSeconds, settings.MetricMultiplier, settings.MetricMaxIntervalSeconds); err != nil { return databaseError(err) } _, err = s.db.ExecContext(ctx, ` UPDATE creator_work SET next_metric_at=COALESCE(CASE WHEN NOT metric_plan_stopped THEN metric_plan_next_at END, next_metric_at), metric_stop_reason=CASE WHEN metric_plan_stopped THEN metric_stop_reason ELSE '' END, updated_at=now() WHERE work_id=$1`, workID) return databaseError(err) } func nextMetricInterval(interval int64, multiplier float64, maximum int64) (int64, error) { if interval <= 0 || math.IsNaN(multiplier) || multiplier <= 1 || maximum < interval { return 0, ErrInvalid } if interval == maximum || math.IsInf(multiplier, 0) { return maximum, nil } candidate := float64(interval) * multiplier if math.IsNaN(candidate) { return 0, ErrInvalid } if math.IsInf(candidate, 0) || candidate >= float64(maximum) || candidate < float64(interval) { return maximum, nil } result := int64(math.Round(candidate)) if result <= interval { result = interval + 1 } if result > maximum { result = maximum } return result, nil } func (s *Store) recordMetricWithPlan(ctx context.Context, input MetricInput, settings Settings, now time.Time) (MetricPoint, error) { collectedAt := input.CollectedAt.UTC() now = now.UTC() if collectedAt.After(now) { return MetricPoint{}, ErrInvalid } tx, err := s.db.BeginTx(ctx, nil) if err != nil { return MetricPoint{}, fmt.Errorf("begin creator metric: %w", err) } defer tx.Rollback() var publishedAt sql.NullTime var publishedStatus string var latestAt sql.NullTime if err := tx.QueryRowContext(ctx, `SELECT published_at, published_at_status, latest_metrics_at FROM creator_work WHERE work_id=$1 FOR UPDATE`, input.WorkID).Scan(&publishedAt, &publishedStatus, &latestAt); err != nil { return MetricPoint{}, rowError(err) } _ = publishedStatus point := MetricPoint{CollectedAt: collectedAt, Likes: input.Likes, CommentsCount: input.CommentsCount, Shares: input.Shares, CollectCount: input.CollectCount, PlayCount: input.PlayCount} if !publishedAt.Valid || publishedStatus != "verified" { if _, err := tx.ExecContext(ctx, `UPDATE creator_work SET likes=$2, comments_count=$3, shares=$4, collect_count=$5, play_count=$6, latest_metrics_at=$7, next_metric_at=NULL, metric_stop_reason='published_at_pending_verification', updated_at=now() WHERE work_id=$1`, input.WorkID, input.Likes, input.CommentsCount, input.Shares, input.CollectCount, input.PlayCount, collectedAt); err != nil { return MetricPoint{}, databaseError(err) } if err := tx.Commit(); err != nil { return MetricPoint{}, fmt.Errorf("commit creator metric: %w", err) } return point, nil } published := publishedAt.Time.UTC() monitoringEnd := published.Add(time.Duration(settings.MetricAgeSeconds) * time.Second) if !published.Before(collectedAt) { return MetricPoint{}, ErrConflict } if latestAt.Valid && latestAt.Time.UTC().Equal(collectedAt) { return point, nil } var nextPlan sql.NullTime var interval int64 var multiplier float64 var maximum int64 var stopped bool if latestAt.Valid { if !now.Before(monitoringEnd) || !collectedAt.Before(monitoringEnd) { if _, err := tx.ExecContext(ctx, `UPDATE creator_work SET metric_plan_next_at=NULL, metric_plan_stopped=true, next_metric_at=NULL, metric_stop_reason='monitoring_age_reached', updated_at=now() WHERE work_id=$1 AND NOT metric_plan_stopped`, input.WorkID); err != nil { return MetricPoint{}, databaseError(err) } if err := tx.Commit(); err != nil { return MetricPoint{}, fmt.Errorf("commit stopped creator metric plan: %w", err) } return MetricPoint{}, ErrConflict } } if _, err := tx.ExecContext(ctx, ` UPDATE creator_work SET metric_plan_published_at = COALESCE(metric_plan_published_at, $2), metric_monitoring_end = COALESCE(metric_monitoring_end, $3), metric_plan_next_at = COALESCE(metric_plan_next_at, $4), metric_plan_interval_seconds = COALESCE(metric_plan_interval_seconds, $5), metric_plan_multiplier = COALESCE(metric_plan_multiplier, $6), metric_plan_max_interval_seconds = COALESCE(metric_plan_max_interval_seconds, $7), updated_at = now() WHERE work_id = $1`, input.WorkID, published, monitoringEnd, published.Add(time.Duration(settings.MetricInitialIntervalSeconds)*time.Second), settings.MetricInitialIntervalSeconds, settings.MetricMultiplier, settings.MetricMaxIntervalSeconds); err != nil { return MetricPoint{}, databaseError(err) } if err := tx.QueryRowContext(ctx, `SELECT metric_plan_next_at, metric_plan_interval_seconds, metric_plan_multiplier, metric_plan_max_interval_seconds, metric_plan_stopped FROM creator_work WHERE work_id=$1 FOR UPDATE`, input.WorkID).Scan(&nextPlan, &interval, &multiplier, &maximum, &stopped); err != nil { return MetricPoint{}, rowError(err) } if !latestAt.Valid { nextAt, nextReason := NextMetricAt(published, now, time.Duration(settings.MetricInitialIntervalSeconds)*time.Second, time.Duration(settings.MetricMaxIntervalSeconds)*time.Second, settings.MetricMultiplier, time.Duration(settings.MetricAgeSeconds)*time.Second) if _, err := tx.ExecContext(ctx, `INSERT INTO creator_work_metric (work_id,collected_at,likes,comments_count,shares,collect_count,play_count) SELECT w.id, $2, $3, $4, $5, $6, $7 FROM creator_work w WHERE w.work_id = $1 ON CONFLICT (work_id,collected_at) DO UPDATE SET likes=EXCLUDED.likes,comments_count=EXCLUDED.comments_count,shares=EXCLUDED.shares,collect_count=EXCLUDED.collect_count,play_count=EXCLUDED.play_count`, input.WorkID, collectedAt, input.Likes, input.CommentsCount, input.Shares, input.CollectCount, input.PlayCount); err != nil { return MetricPoint{}, databaseError(err) } nextInterval := time.Duration(settings.MetricMaxIntervalSeconds) * time.Second if !nextAt.IsZero() { following, _ := NextMetricAt(published, nextAt, time.Duration(settings.MetricInitialIntervalSeconds)*time.Second, time.Duration(settings.MetricMaxIntervalSeconds)*time.Second, settings.MetricMultiplier, time.Duration(settings.MetricAgeSeconds)*time.Second) if !following.IsZero() { nextInterval = following.Sub(nextAt) } } if nextAt.IsZero() || stopped || !nextPlan.Valid { nextAt = time.Time{} nextReason = coalesceReason(nextReason, "monitoring_age_reached") if _, err := tx.ExecContext(ctx, `UPDATE creator_work SET likes=$2,comments_count=$3,shares=$4,collect_count=$5,play_count=$6,latest_metrics_at=$7,next_metric_at=NULL,metric_stop_reason=$8,metric_plan_next_at=NULL,metric_plan_stopped=true,updated_at=now() WHERE work_id=$1`, input.WorkID, input.Likes, input.CommentsCount, input.Shares, input.CollectCount, input.PlayCount, collectedAt, nextReason); err != nil { return MetricPoint{}, databaseError(err) } } else { if _, err := tx.ExecContext(ctx, `UPDATE creator_work SET likes=$2,comments_count=$3,shares=$4,collect_count=$5,play_count=$6,latest_metrics_at=$7,next_metric_at=$8,metric_stop_reason='',metric_plan_next_at=$8,metric_plan_interval_seconds=$9,metric_plan_stopped=false,updated_at=now() WHERE work_id=$1`, input.WorkID, input.Likes, input.CommentsCount, input.Shares, input.CollectCount, input.PlayCount, collectedAt, nullableArg(nextAt), int64(nextInterval/time.Second)); err != nil { return MetricPoint{}, databaseError(err) } } if err := tx.Commit(); err != nil { return MetricPoint{}, fmt.Errorf("commit initial creator metric: %w", err) } return point, nil } if stopped || !nextPlan.Valid || collectedAt.Before(nextPlan.Time.UTC()) { return MetricPoint{}, ErrConflict } nextInterval, err := nextMetricInterval(interval, multiplier, maximum) if err != nil { return MetricPoint{}, err } nextAt := nextPlan.Time.UTC().Add(time.Duration(interval) * time.Second) steps := 1 for !nextAt.After(now) && nextAt.Before(monitoringEnd) { interval = nextInterval nextAt = nextAt.Add(time.Duration(interval) * time.Second) nextInterval, err = nextMetricInterval(interval, multiplier, maximum) if err != nil { return MetricPoint{}, err } steps++ } stopped = !nextAt.Before(monitoringEnd) stopReason := "" if stopped { nextAt = time.Time{} stopReason = "monitoring_age_reached" } if _, err := tx.ExecContext(ctx, `INSERT INTO creator_work_metric (work_id,collected_at,likes,comments_count,shares,collect_count,play_count) SELECT w.id, $2, $3, $4, $5, $6, $7 FROM creator_work w WHERE w.work_id = $1 ON CONFLICT (work_id,collected_at) DO UPDATE SET likes=EXCLUDED.likes,comments_count=EXCLUDED.comments_count,shares=EXCLUDED.shares,collect_count=EXCLUDED.collect_count,play_count=EXCLUDED.play_count`, input.WorkID, collectedAt, input.Likes, input.CommentsCount, input.Shares, input.CollectCount, input.PlayCount); err != nil { return MetricPoint{}, databaseError(err) } if _, err := tx.ExecContext(ctx, `UPDATE creator_work SET likes=$2,comments_count=$3,shares=$4,collect_count=$5,play_count=$6,latest_metrics_at=$7,next_metric_at=$8,metric_stop_reason=$9,metric_plan_next_at=$8,metric_plan_interval_seconds=$10,metric_plan_point_index=metric_plan_point_index+$11,metric_plan_stopped=$12,updated_at=now() WHERE work_id=$1`, input.WorkID, input.Likes, input.CommentsCount, input.Shares, input.CollectCount, input.PlayCount, collectedAt, nullableArg(nextAt), stopReason, nextInterval, steps, stopped); err != nil { return MetricPoint{}, databaseError(err) } if err := tx.Commit(); err != nil { return MetricPoint{}, fmt.Errorf("commit creator metric plan: %w", err) } return point, nil } func coalesceReason(value, fallback string) string { if value != "" { return value } return fallback }