package creator import ( "context" "database/sql" "fmt" "math" "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 { if err := s.loadWorkSources(ctx, &works[index]); err != nil { return nil, err } } return works, nil } 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 id=$1`, workID) return databaseError(err) } if !now.Before(monitoringEnd) { if _, err := s.db.ExecContext(ctx, `UPDATE creator_metric_plan SET next_plan_at=NULL, stopped=true, stop_reason='monitoring_age_reached', updated_at=now() WHERE work_id=$1 AND NOT stopped`, workID); err != nil { return databaseError(err) } _, err := s.db.ExecContext(ctx, `UPDATE creator_work SET next_metric_at=NULL, metric_stop_reason='monitoring_age_reached', updated_at=now() WHERE id=$1`, 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 id=$1`, workID) return databaseError(err) } if _, err := s.db.ExecContext(ctx, ` INSERT INTO creator_metric_plan (work_id, published_at, monitoring_end, next_plan_at, interval_seconds, multiplier, max_interval_seconds) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (work_id) DO NOTHING`, 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((SELECT next_plan_at FROM creator_metric_plan WHERE work_id=$1 AND NOT stopped), next_metric_at), metric_stop_reason=CASE WHEN EXISTS (SELECT 1 FROM creator_metric_plan WHERE work_id=$1 AND NOT stopped) THEN '' ELSE metric_stop_reason END, updated_at=now() WHERE 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 id=$1 FOR UPDATE`, input.WorkID).Scan(&publishedAt, &publishedStatus, &latestAt); err != nil { return MetricPoint{}, rowError(err) } point := MetricPoint{CollectedAt: collectedAt, Likes: input.Likes, CommentsCount: input.CommentsCount, Shares: input.Shares} if !publishedAt.Valid || publishedStatus != "verified" { if _, err := tx.ExecContext(ctx, `UPDATE creator_work SET likes=$2, comments_count=$3, shares=$4, latest_metrics_at=$5, next_metric_at=NULL, metric_stop_reason='published_at_pending_verification', updated_at=now() WHERE id=$1`, input.WorkID, input.Likes, input.CommentsCount, input.Shares, 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 pointIndex int var stopped bool var stopReason string if latestAt.Valid { if !now.Before(monitoringEnd) || !collectedAt.Before(monitoringEnd) { if _, err := tx.ExecContext(ctx, `UPDATE creator_metric_plan SET next_plan_at=NULL, stopped=true, stop_reason='monitoring_age_reached', updated_at=now() WHERE work_id=$1 AND NOT stopped`, input.WorkID); err != nil { return MetricPoint{}, databaseError(err) } if _, err := tx.ExecContext(ctx, `UPDATE creator_work SET next_metric_at=NULL, metric_stop_reason='monitoring_age_reached', updated_at=now() WHERE id=$1`, 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, `INSERT INTO creator_metric_plan (work_id,published_at,monitoring_end,next_plan_at,interval_seconds,multiplier,max_interval_seconds) VALUES ($1,$2,$3,$4,$5,$6,$7) ON CONFLICT (work_id) DO NOTHING`, 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 next_plan_at, interval_seconds, multiplier, max_interval_seconds, point_index, stopped, stop_reason FROM creator_metric_plan WHERE work_id=$1 FOR UPDATE`, input.WorkID).Scan(&nextPlan, &interval, &multiplier, &maximum, &pointIndex, &stopped, &stopReason); 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) VALUES ($1,$2,$3,$4,$5) ON CONFLICT (work_id,collected_at) DO UPDATE SET likes=EXCLUDED.likes,comments_count=EXCLUDED.comments_count,shares=EXCLUDED.shares`, input.WorkID, collectedAt, input.Likes, input.CommentsCount, input.Shares); err != nil { return MetricPoint{}, databaseError(err) } if nextAt.IsZero() || stopped || !nextPlan.Valid { if _, err := tx.ExecContext(ctx, `UPDATE creator_metric_plan SET next_plan_at=NULL, stopped=true, stop_reason=$2, updated_at=now() WHERE work_id=$1`, input.WorkID, coalesceReason(nextReason, "monitoring_age_reached")); err != nil { return MetricPoint{}, databaseError(err) } nextAt = time.Time{} nextReason = coalesceReason(nextReason, "monitoring_age_reached") } else if _, err := tx.ExecContext(ctx, `UPDATE creator_metric_plan SET next_plan_at=$2, stopped=false, stop_reason='', updated_at=now() WHERE work_id=$1`, input.WorkID, nextAt); err != nil { return MetricPoint{}, databaseError(err) } else { nextReason = "" } if _, err := tx.ExecContext(ctx, `UPDATE creator_work SET likes=$2,comments_count=$3,shares=$4,latest_metrics_at=$5,next_metric_at=$6,metric_stop_reason=$7,updated_at=now() WHERE id=$1`, input.WorkID, input.Likes, input.CommentsCount, input.Shares, collectedAt, nullableArg(nextAt), nextReason); 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 } newInterval, err := nextMetricInterval(interval, multiplier, maximum) if err != nil { return MetricPoint{}, err } nextAt := nextPlan.Time.UTC().Add(time.Duration(newInterval) * time.Second) steps := 1 for !nextAt.After(now) && nextAt.Before(monitoringEnd) { newInterval, err = nextMetricInterval(newInterval, multiplier, maximum) if err != nil { return MetricPoint{}, err } nextAt = nextAt.Add(time.Duration(newInterval) * time.Second) steps++ } stopped = !nextAt.Before(monitoringEnd) if stopped { nextAt = time.Time{} stopReason = "monitoring_age_reached" } else { stopReason = "" } if _, err := tx.ExecContext(ctx, `INSERT INTO creator_work_metric (work_id,collected_at,likes,comments_count,shares) VALUES ($1,$2,$3,$4,$5) ON CONFLICT (work_id,collected_at) DO UPDATE SET likes=EXCLUDED.likes,comments_count=EXCLUDED.comments_count,shares=EXCLUDED.shares`, input.WorkID, collectedAt, input.Likes, input.CommentsCount, input.Shares); err != nil { return MetricPoint{}, databaseError(err) } if _, err := tx.ExecContext(ctx, `UPDATE creator_metric_plan SET next_plan_at=$2, interval_seconds=$3, point_index=point_index+$4, stopped=$5, stop_reason=$6, updated_at=now() WHERE work_id=$1`, input.WorkID, nullableArg(nextAt), newInterval, steps, stopped, stopReason); err != nil { return MetricPoint{}, databaseError(err) } if _, err := tx.ExecContext(ctx, `UPDATE creator_work SET likes=$2,comments_count=$3,shares=$4,latest_metrics_at=$5,next_metric_at=$6,metric_stop_reason=$7,updated_at=now() WHERE id=$1`, input.WorkID, input.Likes, input.CommentsCount, input.Shares, collectedAt, nullableArg(nextAt), stopReason); 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 }