267 lines
13 KiB
Go
267 lines
13 KiB
Go
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 {
|
|
if err := s.loadWorkSources(ctx, &works[index]); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
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
|
|
}
|
|
if _, err := s.db.ExecContext(ctx, `UPDATE creator_metric_plan SET next_plan_at=NULL, stopped=true, stop_reason=$2, updated_at=now() WHERE work_id=$1 AND NOT stopped`, workID, reason); err != nil {
|
|
return databaseError(err)
|
|
}
|
|
_, err := s.db.ExecContext(ctx, `UPDATE creator_work SET next_metric_at=NULL, metric_stop_reason=$2, updated_at=now() WHERE id=$1`, 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 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)
|
|
}
|
|
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 {
|
|
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, interval_seconds=$3, stopped=false, stop_reason='', updated_at=now() WHERE work_id=$1`, input.WorkID, nextAt, int64(nextInterval/time.Second)); 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
|
|
}
|
|
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)
|
|
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), nextInterval, 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
|
|
}
|