765 lines
31 KiB
Go
765 lines
31 KiB
Go
package creator
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/url"
|
|
"strings"
|
|
"time"
|
|
"unicode/utf8"
|
|
)
|
|
|
|
const maxRawPayloadBytes = 4 << 20
|
|
|
|
func validateHomepage(value string) error {
|
|
parsed, err := url.Parse(strings.TrimSpace(value))
|
|
if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Host == "" || parsed.User != nil {
|
|
return ErrInvalid
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) CreateCompetitor(ctx context.Context, input CompetitorInput) (Competitor, error) {
|
|
input.Platform = strings.TrimSpace(input.Platform)
|
|
input.PlatformAccountKey = strings.TrimSpace(input.PlatformAccountKey)
|
|
input.Nickname = strings.TrimSpace(input.Nickname)
|
|
input.AvatarURL = strings.TrimSpace(input.AvatarURL)
|
|
input.HomepageURL = strings.TrimSpace(input.HomepageURL)
|
|
if !ValidatePlatform(input.Platform) || input.PlatformAccountKey == "" || input.HomepageURL == "" ||
|
|
utf8.RuneCountInString(input.PlatformAccountKey) > 255 || utf8.RuneCountInString(input.Nickname) > 255 ||
|
|
utf8.RuneCountInString(input.AvatarURL) > 1000 || validateHomepage(input.HomepageURL) != nil {
|
|
return Competitor{}, ErrInvalid
|
|
}
|
|
id := newID("competitor")
|
|
if _, err := s.db.ExecContext(ctx, `
|
|
INSERT INTO creator_competitor (id, platform, platform_account_key, nickname, avatar_url, homepage_url, next_sync_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, now())`,
|
|
id, input.Platform, input.PlatformAccountKey, input.Nickname, input.AvatarURL, input.HomepageURL); err != nil {
|
|
return Competitor{}, databaseError(err)
|
|
}
|
|
return s.GetCompetitor(ctx, id)
|
|
}
|
|
|
|
func scanCompetitor(scanner interface{ Scan(...any) error }) (Competitor, error) {
|
|
var result Competitor
|
|
var leaseUntil, lastSync, nextSync sql.NullTime
|
|
if err := scanner.Scan(&result.ID, &result.Platform, &result.PlatformAccountKey, &result.Nickname,
|
|
&result.AvatarURL, &result.HomepageURL, &result.Enabled, &result.SyncStatus, &result.SyncCursor,
|
|
&result.SyncError, &leaseUntil, &lastSync, &nextSync, &result.CreatedAt, &result.UpdatedAt); err != nil {
|
|
return Competitor{}, err
|
|
}
|
|
result.SyncLeaseUntil = nullableTime(leaseUntil)
|
|
result.LastSyncAt = nullableTime(lastSync)
|
|
result.NextSyncAt = nullableTime(nextSync)
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Store) GetCompetitor(ctx context.Context, id string) (Competitor, error) {
|
|
result, err := scanCompetitor(s.db.QueryRowContext(ctx, `
|
|
SELECT id, platform, platform_account_key, nickname, avatar_url, homepage_url,
|
|
enabled, sync_status, sync_cursor, sync_error, sync_lease_until, last_sync_at, next_sync_at, created_at, updated_at
|
|
FROM creator_competitor WHERE id = $1`, id))
|
|
return result, rowError(err)
|
|
}
|
|
|
|
func (s *Store) ListCompetitors(ctx context.Context, platform string) ([]Competitor, error) {
|
|
query := `SELECT id, platform, platform_account_key, nickname, avatar_url, homepage_url,
|
|
enabled, sync_status, sync_cursor, sync_error, sync_lease_until, last_sync_at, next_sync_at, created_at, updated_at
|
|
FROM creator_competitor`
|
|
args := []any{}
|
|
if platform != "" {
|
|
if !ValidatePlatform(platform) {
|
|
return nil, ErrInvalid
|
|
}
|
|
query += ` WHERE platform = $1`
|
|
args = append(args, platform)
|
|
}
|
|
query += ` ORDER BY created_at DESC, id`
|
|
rows, err := s.db.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, databaseError(err)
|
|
}
|
|
defer rows.Close()
|
|
result := make([]Competitor, 0)
|
|
for rows.Next() {
|
|
item, err := scanCompetitor(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, item)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (s *Store) SetCompetitorEnabled(ctx context.Context, id string, enabled bool) (Competitor, error) {
|
|
status := "paused"
|
|
if enabled {
|
|
status = "idle"
|
|
}
|
|
_, err := s.db.ExecContext(ctx, `
|
|
UPDATE creator_competitor SET enabled = $2, sync_status = $3, sync_lease_until = NULL,
|
|
next_sync_at = CASE WHEN $2 THEN now() ELSE NULL END, updated_at = now()
|
|
WHERE id = $1`, id, enabled, status)
|
|
if err != nil {
|
|
return Competitor{}, databaseError(err)
|
|
}
|
|
return s.GetCompetitor(ctx, id)
|
|
}
|
|
|
|
func (s *Store) MarkCompetitorSync(ctx context.Context, id, leaseToken, status, cursor, syncError string, nextAt *time.Time) error {
|
|
if id == "" || leaseToken == "" {
|
|
return ErrInvalid
|
|
}
|
|
if status != "idle" && status != "running" && status != "paused" && status != "failed" && status != "blocked" {
|
|
return ErrInvalid
|
|
}
|
|
if len(syncError) > 2000 || len(cursor) > 1000 {
|
|
return ErrInvalid
|
|
}
|
|
var next any
|
|
if nextAt != nil {
|
|
next = nextAt.UTC()
|
|
}
|
|
result, err := s.db.ExecContext(ctx, `
|
|
UPDATE creator_competitor
|
|
SET sync_status = $3, sync_cursor = $4, sync_error = $5,
|
|
sync_lease_token = CASE WHEN $3 = 'running' THEN $2 ELSE NULL END,
|
|
sync_lease_until = CASE WHEN $3 = 'running' THEN now() + interval '10 minutes' ELSE NULL END,
|
|
last_sync_at = CASE WHEN $3 IN ('idle', 'failed', 'blocked') THEN now() ELSE last_sync_at END,
|
|
next_sync_at = $6, updated_at = now()
|
|
WHERE id = $1 AND sync_lease_token = $2`, id, leaseToken, status, cursor, syncError, next)
|
|
if err != nil {
|
|
return databaseError(err)
|
|
}
|
|
if affected, err := result.RowsAffected(); err != nil {
|
|
return databaseError(err)
|
|
} else if affected != 1 {
|
|
return ErrConflict
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) ClaimCompetitorSync(ctx context.Context, id string, force bool, now time.Time) (string, bool, error) {
|
|
if id == "" || now.IsZero() {
|
|
return "", false, ErrInvalid
|
|
}
|
|
condition := `enabled AND (next_sync_at IS NULL OR next_sync_at <= $2)`
|
|
if force {
|
|
condition = `enabled`
|
|
}
|
|
token := newID("competitor-lease")
|
|
var claimed string
|
|
err := s.db.QueryRowContext(ctx, `UPDATE creator_competitor SET sync_status='running', sync_lease_token=$2, sync_lease_until=$3 + interval '10 minutes', sync_error='', updated_at=$3 WHERE id=$1 AND `+strings.ReplaceAll(condition, "$2", "$3")+` AND (sync_status <> 'running' OR sync_lease_until IS NULL OR sync_lease_until <= $3) RETURNING id`, id, token, now.UTC()).Scan(&claimed)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return "", false, nil
|
|
}
|
|
if err != nil {
|
|
return "", false, databaseError(err)
|
|
}
|
|
return token, claimed != "", nil
|
|
}
|
|
|
|
func (s *Store) ListDueCompetitors(ctx context.Context, now time.Time) ([]Competitor, error) {
|
|
if now.IsZero() {
|
|
return nil, ErrInvalid
|
|
}
|
|
rows, err := s.db.QueryContext(ctx, `SELECT id, platform, platform_account_key, nickname, avatar_url, homepage_url,
|
|
enabled, sync_status, sync_cursor, sync_error, sync_lease_until, last_sync_at, next_sync_at, created_at, updated_at
|
|
FROM creator_competitor
|
|
WHERE enabled AND next_sync_at IS NOT NULL AND next_sync_at <= $1
|
|
AND (sync_status <> 'running' OR sync_lease_until IS NULL OR sync_lease_until <= $1)
|
|
ORDER BY COALESCE(next_sync_at, created_at), id`, now.UTC())
|
|
if err != nil {
|
|
return nil, databaseError(err)
|
|
}
|
|
defer rows.Close()
|
|
result := make([]Competitor, 0)
|
|
for rows.Next() {
|
|
item, err := scanCompetitor(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, item)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (s *Store) validateWorkSource(ctx context.Context, platform, sourceType, sourceID string) error {
|
|
var query string
|
|
switch sourceType {
|
|
case SourceOwned:
|
|
query = `SELECT 1 FROM social_account WHERE id=$1 AND platform=$2`
|
|
case SourceCompetitor:
|
|
query = `SELECT 1 FROM creator_competitor WHERE id=$1 AND platform=$2`
|
|
default:
|
|
return ErrInvalid
|
|
}
|
|
var exists int
|
|
if err := s.db.QueryRowContext(ctx, query, sourceID, platform).Scan(&exists); err != nil {
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return ErrInvalid
|
|
}
|
|
return databaseError(err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) UpsertWork(ctx context.Context, input WorkInput, now time.Time) (Work, bool, error) {
|
|
input.Platform = strings.TrimSpace(input.Platform)
|
|
input.WorkKey = strings.TrimSpace(input.WorkKey)
|
|
input.SourceType = strings.TrimSpace(input.SourceType)
|
|
input.SourceID = strings.TrimSpace(input.SourceID)
|
|
input.AuthorName = strings.TrimSpace(input.AuthorName)
|
|
input.Title = strings.TrimSpace(input.Title)
|
|
input.OriginalURL = strings.TrimSpace(input.OriginalURL)
|
|
input.CoverURL = strings.TrimSpace(input.CoverURL)
|
|
if !ValidatePlatform(input.Platform) || input.WorkKey == "" || input.SourceID == "" ||
|
|
(input.SourceType != SourceOwned && input.SourceType != SourceCompetitor) ||
|
|
utf8.RuneCountInString(input.WorkKey) > 255 || utf8.RuneCountInString(input.Title) > 1000 ||
|
|
utf8.RuneCountInString(input.Body) > 100000 || utf8.RuneCountInString(input.OriginalURL) > 2000 ||
|
|
utf8.RuneCountInString(input.CoverURL) > 2000 || len(input.RawPayload) > maxRawPayloadBytes ||
|
|
input.RawPayload != "" && !json.Valid([]byte(input.RawPayload)) {
|
|
return Work{}, false, ErrInvalid
|
|
}
|
|
if err := s.validateWorkSource(ctx, input.Platform, input.SourceType, input.SourceID); err != nil {
|
|
return Work{}, false, err
|
|
}
|
|
if now.IsZero() {
|
|
now = time.Now().UTC()
|
|
}
|
|
status := input.PublishedAtStatus
|
|
if status == "" {
|
|
status = "verified"
|
|
if input.PublishedAt == nil {
|
|
status = "pending_verification"
|
|
}
|
|
}
|
|
if status != "verified" && status != "pending_verification" && status != "invalid" && status != "future" {
|
|
return Work{}, false, ErrInvalid
|
|
}
|
|
if input.PublishedAt != nil {
|
|
input.PublishedAt = ptrTime(input.PublishedAt.UTC())
|
|
if input.PublishedAt.After(now.UTC()) && status == "verified" {
|
|
status = "future"
|
|
}
|
|
}
|
|
if input.Likes != nil && *input.Likes < 0 || input.CommentsCount != nil && *input.CommentsCount < 0 || input.Shares != nil && *input.Shares < 0 {
|
|
return Work{}, false, ErrInvalid
|
|
}
|
|
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return Work{}, false, fmt.Errorf("begin creator work upsert: %w", err)
|
|
}
|
|
defer tx.Rollback()
|
|
id := newID("work")
|
|
var returnedID string
|
|
var inserted bool
|
|
err = tx.QueryRowContext(ctx, `
|
|
INSERT INTO creator_work (id, platform, work_key, source_type, source_id, author_name, title, body,
|
|
published_at, published_at_status, original_url, cover_url, raw_payload, likes, comments_count, shares)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
|
|
ON CONFLICT (platform, work_key) DO UPDATE SET
|
|
author_name = CASE WHEN EXCLUDED.author_name = '' THEN creator_work.author_name ELSE EXCLUDED.author_name END,
|
|
title = CASE WHEN EXCLUDED.title = '' THEN creator_work.title ELSE EXCLUDED.title END,
|
|
body = CASE WHEN EXCLUDED.body = '' THEN creator_work.body ELSE EXCLUDED.body END,
|
|
published_at = COALESCE(EXCLUDED.published_at, creator_work.published_at),
|
|
published_at_status = CASE WHEN EXCLUDED.published_at IS NULL THEN creator_work.published_at_status ELSE EXCLUDED.published_at_status END,
|
|
original_url = CASE WHEN EXCLUDED.original_url = '' THEN creator_work.original_url ELSE EXCLUDED.original_url END,
|
|
cover_url = CASE WHEN EXCLUDED.cover_url = '' THEN creator_work.cover_url ELSE EXCLUDED.cover_url END,
|
|
raw_payload = COALESCE(EXCLUDED.raw_payload, creator_work.raw_payload),
|
|
likes = COALESCE(EXCLUDED.likes, creator_work.likes),
|
|
comments_count = COALESCE(EXCLUDED.comments_count, creator_work.comments_count),
|
|
shares = COALESCE(EXCLUDED.shares, creator_work.shares), updated_at = now()
|
|
RETURNING id, (xmax = 0)`, id, input.Platform, input.WorkKey, input.SourceType, input.SourceID,
|
|
input.AuthorName, input.Title, input.Body, input.PublishedAt, status, input.OriginalURL, input.CoverURL,
|
|
nullableRawPayload(input.RawPayload), input.Likes, input.CommentsCount, input.Shares).Scan(&returnedID, &inserted)
|
|
if err != nil {
|
|
return Work{}, false, databaseError(err)
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
INSERT INTO creator_work_source (work_id, platform, source_type, source_id)
|
|
VALUES ($1, $2, $3, $4) ON CONFLICT (work_id, source_type, source_id) DO NOTHING`,
|
|
returnedID, input.Platform, input.SourceType, input.SourceID); err != nil {
|
|
return Work{}, false, databaseError(err)
|
|
}
|
|
result, err := scanWork(tx.QueryRowContext(ctx, workSelect+` WHERE id = $1`, returnedID))
|
|
if err != nil {
|
|
return Work{}, false, rowError(err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return Work{}, false, fmt.Errorf("commit creator work upsert: %w", err)
|
|
}
|
|
if err := s.loadWorkSources(ctx, &result); err != nil {
|
|
return Work{}, false, err
|
|
}
|
|
return result, inserted, nil
|
|
}
|
|
|
|
func ptrTime(value time.Time) *time.Time { return &value }
|
|
|
|
func nullableRawPayload(value string) any {
|
|
if value == "" {
|
|
return nil
|
|
}
|
|
return value
|
|
}
|
|
|
|
func scanWork(scanner interface{ Scan(...any) error }) (Work, error) {
|
|
var result Work
|
|
var publishedAt, latestAt, nextAt sql.NullTime
|
|
var likes, commentsCount, shares sql.NullInt64
|
|
var rawPayload sql.NullString
|
|
if err := scanner.Scan(&result.ID, &result.Platform, &result.WorkKey, &result.SourceType, &result.SourceID,
|
|
&result.AuthorName, &result.Title, &result.Body, &publishedAt, &result.PublishedAtStatus,
|
|
&result.OriginalURL, &result.CoverURL, &rawPayload, &likes, &commentsCount, &shares, &latestAt, &nextAt,
|
|
&result.MetricStopReason, &result.CreatedAt, &result.UpdatedAt); err != nil {
|
|
return Work{}, err
|
|
}
|
|
result.PublishedAt = nullableTime(publishedAt)
|
|
if rawPayload.Valid {
|
|
result.RawPayload = rawPayload.String
|
|
}
|
|
result.Likes, result.CommentsCount, result.Shares = nullableInt64(likes), nullableInt64(commentsCount), nullableInt64(shares)
|
|
result.LatestMetricsAt, result.NextMetricAt = nullableTime(latestAt), nullableTime(nextAt)
|
|
return result, nil
|
|
}
|
|
|
|
const workSelect = `SELECT id, platform, work_key, source_type, source_id, author_name, title, body,
|
|
published_at, published_at_status, original_url, cover_url, raw_payload, likes, comments_count, shares,
|
|
latest_metrics_at, next_metric_at, metric_stop_reason, created_at, updated_at FROM creator_work`
|
|
|
|
func (s *Store) loadWorkSources(ctx context.Context, work *Work) error {
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT platform, source_type, source_id
|
|
FROM creator_work_source WHERE work_id = $1
|
|
ORDER BY source_type, source_id`, work.ID)
|
|
if err != nil {
|
|
return databaseError(err)
|
|
}
|
|
defer rows.Close()
|
|
work.Sources = make([]WorkSource, 0)
|
|
for rows.Next() {
|
|
var source WorkSource
|
|
if err := rows.Scan(&source.Platform, &source.SourceType, &source.SourceID); err != nil {
|
|
return err
|
|
}
|
|
work.Sources = append(work.Sources, source)
|
|
}
|
|
return rows.Err()
|
|
}
|
|
|
|
func (s *Store) GetWork(ctx context.Context, id string) (Work, error) {
|
|
result, err := scanWork(s.db.QueryRowContext(ctx, workSelect+` WHERE id = $1`, id))
|
|
if err != nil {
|
|
return Work{}, rowError(err)
|
|
}
|
|
if err := s.loadWorkSources(ctx, &result); err != nil {
|
|
return Work{}, err
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Store) ListWorks(ctx context.Context, filter WorkFilter) ([]Work, error) {
|
|
query, args := workSelect, make([]any, 0, 8)
|
|
where := make([]string, 0, 7)
|
|
add := func(clause string, value any) {
|
|
args = append(args, value)
|
|
where = append(where, fmt.Sprintf(clause, len(args)))
|
|
}
|
|
if filter.Platform != "" {
|
|
if !ValidatePlatform(filter.Platform) {
|
|
return nil, ErrInvalid
|
|
}
|
|
add("platform = $%d", filter.Platform)
|
|
}
|
|
if filter.SourceType != "" {
|
|
if filter.SourceType != SourceOwned && filter.SourceType != SourceCompetitor {
|
|
return nil, ErrInvalid
|
|
}
|
|
args = append(args, filter.SourceType)
|
|
where = append(where, fmt.Sprintf(`EXISTS (SELECT 1 FROM creator_work_source ws WHERE ws.work_id = creator_work.id AND ws.source_type = $%d)`, len(args)))
|
|
}
|
|
if filter.SourceID != "" {
|
|
args = append(args, filter.SourceID)
|
|
where = append(where, fmt.Sprintf(`EXISTS (SELECT 1 FROM creator_work_source ws WHERE ws.work_id = creator_work.id AND ws.source_id = $%d)`, len(args)))
|
|
}
|
|
if filter.PublishedAfter != nil {
|
|
add("published_at >= $%d", filter.PublishedAfter.UTC())
|
|
}
|
|
if filter.PublishedBefore != nil {
|
|
add("published_at <= $%d", filter.PublishedBefore.UTC())
|
|
}
|
|
if filter.MinLikes != nil {
|
|
if *filter.MinLikes < 0 {
|
|
return nil, ErrInvalid
|
|
}
|
|
where = append(where, fmt.Sprintf("likes IS NOT NULL AND likes >= $%d", len(args)+1))
|
|
args = append(args, *filter.MinLikes)
|
|
}
|
|
if filter.MinComments != nil {
|
|
if *filter.MinComments < 0 {
|
|
return nil, ErrInvalid
|
|
}
|
|
where = append(where, fmt.Sprintf("comments_count IS NOT NULL AND comments_count >= $%d", len(args)+1))
|
|
args = append(args, *filter.MinComments)
|
|
}
|
|
if filter.MinShares != nil {
|
|
if *filter.MinShares < 0 {
|
|
return nil, ErrInvalid
|
|
}
|
|
where = append(where, fmt.Sprintf("shares IS NOT NULL AND shares >= $%d", len(args)+1))
|
|
args = append(args, *filter.MinShares)
|
|
}
|
|
if len(where) > 0 {
|
|
query += ` WHERE ` + strings.Join(where, ` AND `)
|
|
}
|
|
query += ` ORDER BY published_at DESC NULLS LAST, id`
|
|
rows, err := s.db.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, databaseError(err)
|
|
}
|
|
result := make([]Work, 0)
|
|
for rows.Next() {
|
|
item, err := scanWork(rows)
|
|
if err != nil {
|
|
_ = rows.Close()
|
|
return nil, err
|
|
}
|
|
result = append(result, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
_ = rows.Close()
|
|
return nil, err
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return nil, err
|
|
}
|
|
for index := range result {
|
|
if err := s.loadWorkSources(ctx, &result[index]); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Store) RecordMetric(ctx context.Context, input MetricInput, settings Settings, now time.Time) (MetricPoint, error) {
|
|
if input.WorkID == "" || input.CollectedAt.IsZero() || input.Likes != nil && *input.Likes < 0 || input.CommentsCount != nil && *input.CommentsCount < 0 || input.Shares != nil && *input.Shares < 0 {
|
|
return MetricPoint{}, ErrInvalid
|
|
}
|
|
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 MetricPoint{}, err
|
|
}
|
|
if now.IsZero() {
|
|
now = time.Now().UTC()
|
|
}
|
|
return s.recordMetricWithPlan(ctx, input, settings, now)
|
|
}
|
|
|
|
func nullableArg(value time.Time) any {
|
|
if value.IsZero() {
|
|
return nil
|
|
}
|
|
return value.UTC()
|
|
}
|
|
|
|
func (s *Store) ListMetrics(ctx context.Context, workID string) ([]MetricPoint, error) {
|
|
rows, err := s.db.QueryContext(ctx, `SELECT collected_at, likes, comments_count, shares FROM creator_work_metric WHERE work_id = $1 ORDER BY collected_at`, workID)
|
|
if err != nil {
|
|
return nil, databaseError(err)
|
|
}
|
|
defer rows.Close()
|
|
result := make([]MetricPoint, 0)
|
|
for rows.Next() {
|
|
var point MetricPoint
|
|
var likes, commentsCount, shares sql.NullInt64
|
|
if err := rows.Scan(&point.CollectedAt, &likes, &commentsCount, &shares); err != nil {
|
|
return nil, err
|
|
}
|
|
point.CollectedAt = point.CollectedAt.UTC()
|
|
point.Likes, point.CommentsCount, point.Shares = nullableInt64(likes), nullableInt64(commentsCount), nullableInt64(shares)
|
|
result = append(result, point)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (s *Store) GetMaterial(ctx context.Context, workID string) (MaterialJob, error) {
|
|
if _, err := s.db.ExecContext(ctx, `INSERT INTO creator_material_job (work_id) VALUES ($1) ON CONFLICT (work_id) DO NOTHING`, workID); err != nil {
|
|
return MaterialJob{}, databaseError(err)
|
|
}
|
|
return s.scanMaterial(s.db.QueryRowContext(ctx, `SELECT work_id, selected, select_confirmed_at, download_status, video_reference, audio_status, audio_reference, transcription_status, transcript, failed_step, failure_reason, rewrite_confirmed_at, rewrite_requirement, generated_title, generated_script, created_at, updated_at FROM creator_material_job WHERE work_id = $1`, workID))
|
|
}
|
|
|
|
func (s *Store) scanMaterial(scanner interface{ Scan(...any) error }) (MaterialJob, error) {
|
|
var result MaterialJob
|
|
var selectedAt, rewriteAt sql.NullTime
|
|
if err := scanner.Scan(&result.WorkID, &result.Selected, &selectedAt, &result.DownloadStatus, &result.VideoReference, &result.AudioStatus, &result.AudioReference, &result.TranscriptionStatus, &result.Transcript, &result.FailedStep, &result.FailureReason, &rewriteAt, &result.RewriteRequirement, &result.GeneratedTitle, &result.GeneratedScript, &result.CreatedAt, &result.UpdatedAt); err != nil {
|
|
return MaterialJob{}, rowError(err)
|
|
}
|
|
result.SelectConfirmedAt, result.RewriteConfirmedAt = nullableTime(selectedAt), nullableTime(rewriteAt)
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Store) SelectMaterial(ctx context.Context, workID string) (MaterialJob, bool, error) {
|
|
if _, err := s.GetWork(ctx, workID); err != nil {
|
|
return MaterialJob{}, false, err
|
|
}
|
|
var inserted bool
|
|
if err := s.db.QueryRowContext(ctx, `INSERT INTO creator_material_job (work_id, selected, select_confirmed_at) VALUES ($1, true, now()) ON CONFLICT (work_id) DO UPDATE SET selected = true, select_confirmed_at = COALESCE(creator_material_job.select_confirmed_at, now()), updated_at = now() RETURNING (xmax = 0)`, workID).Scan(&inserted); err != nil {
|
|
return MaterialJob{}, false, databaseError(err)
|
|
}
|
|
job, err := s.GetMaterial(ctx, workID)
|
|
return job, inserted, err
|
|
}
|
|
|
|
func (s *Store) SetMaterialStep(ctx context.Context, workID, step, status, reference, reason string) (MaterialJob, error) {
|
|
if step != "download" && step != "audio" && step != "transcription" {
|
|
return MaterialJob{}, ErrInvalid
|
|
}
|
|
valid := map[string]map[string]bool{
|
|
"download": {"not_started": true, "running": true, "succeeded": true, "failed": true},
|
|
"audio": {"not_started": true, "running": true, "succeeded": true, "no_audio": true, "failed": true},
|
|
"transcription": {"not_started": true, "running": true, "succeeded": true, "no_speech": true, "failed": true},
|
|
}
|
|
if !valid[step][status] || len(reference) > 2000 || len(reason) > 2000 {
|
|
return MaterialJob{}, ErrInvalid
|
|
}
|
|
job, err := s.GetMaterial(ctx, workID)
|
|
if err != nil {
|
|
return MaterialJob{}, err
|
|
}
|
|
if !job.Selected {
|
|
return MaterialJob{}, ErrConflict
|
|
}
|
|
var query string
|
|
var args []any
|
|
switch step {
|
|
case "download":
|
|
query = `UPDATE creator_material_job SET download_status = $2, video_reference = $3, failed_step = CASE WHEN $2 = 'failed' THEN 'download' ELSE failed_step END, failure_reason = CASE WHEN $2 = 'failed' THEN $4 ELSE failure_reason END, updated_at = now() WHERE work_id = $1`
|
|
args = []any{workID, status, reference, reason}
|
|
case "audio":
|
|
query = `UPDATE creator_material_job SET audio_status = $2, audio_reference = $3, failed_step = CASE WHEN $2 = 'failed' THEN 'audio' ELSE failed_step END, failure_reason = CASE WHEN $2 = 'failed' THEN $4 ELSE failure_reason END, updated_at = now() WHERE work_id = $1`
|
|
args = []any{workID, status, reference, reason}
|
|
case "transcription":
|
|
query = `UPDATE creator_material_job SET transcription_status = $2, transcript = $3, failed_step = CASE WHEN $2 = 'failed' THEN 'transcription' ELSE failed_step END, failure_reason = CASE WHEN $2 = 'failed' THEN $4 ELSE failure_reason END, updated_at = now() WHERE work_id = $1`
|
|
args = []any{workID, status, reference, reason}
|
|
}
|
|
if _, err := s.db.ExecContext(ctx, query, args...); err != nil {
|
|
return MaterialJob{}, databaseError(err)
|
|
}
|
|
return s.GetMaterial(ctx, workID)
|
|
}
|
|
|
|
func (s *Store) ConfirmRewrite(ctx context.Context, workID, requirement string) (MaterialJob, error) {
|
|
requirement = strings.TrimSpace(requirement)
|
|
if utf8.RuneCountInString(requirement) > 4000 {
|
|
return MaterialJob{}, ErrInvalid
|
|
}
|
|
job, err := s.GetMaterial(ctx, workID)
|
|
if err != nil {
|
|
return MaterialJob{}, err
|
|
}
|
|
if !job.Selected || job.DownloadStatus != "succeeded" || job.AudioStatus == "failed" || job.TranscriptionStatus == "failed" || job.AudioStatus == "running" || job.TranscriptionStatus == "running" || job.AudioStatus == "not_started" || job.TranscriptionStatus == "not_started" {
|
|
return MaterialJob{}, ErrConflict
|
|
}
|
|
if _, err := s.db.ExecContext(ctx, `UPDATE creator_material_job SET rewrite_confirmed_at = now(), rewrite_requirement = $2, updated_at = now() WHERE work_id = $1`, workID, requirement); err != nil {
|
|
return MaterialJob{}, databaseError(err)
|
|
}
|
|
return s.GetMaterial(ctx, workID)
|
|
}
|
|
|
|
func parseGeneratedRewrite(content string) (string, string, error) {
|
|
var result struct {
|
|
Title string `json:"title"`
|
|
Script string `json:"script"`
|
|
}
|
|
if err := json.Unmarshal([]byte(strings.TrimSpace(content)), &result); err != nil {
|
|
return "", "", fmt.Errorf("decode generated rewrite: %w", err)
|
|
}
|
|
result.Title, result.Script = strings.TrimSpace(result.Title), strings.TrimSpace(result.Script)
|
|
if result.Title == "" || result.Script == "" || utf8.RuneCountInString(result.Title) > 1000 || utf8.RuneCountInString(result.Script) > 100000 {
|
|
return "", "", ErrInvalid
|
|
}
|
|
return result.Title, result.Script, nil
|
|
}
|
|
|
|
func (s *Store) GenerateRewrite(ctx context.Context, workID string, generator TextGenerator) (MaterialJob, error) {
|
|
if generator == nil {
|
|
return MaterialJob{}, fmt.Errorf("%w: text generator is not configured", ErrUnavailable)
|
|
}
|
|
settings, err := s.GetSettings(ctx)
|
|
if err != nil {
|
|
return MaterialJob{}, err
|
|
}
|
|
if !settings.AIConfigured || strings.TrimSpace(settings.AIProvider) == "" || strings.TrimSpace(settings.AIModel) == "" {
|
|
return MaterialJob{}, fmt.Errorf("%w: AI settings are not approved", ErrUnavailable)
|
|
}
|
|
job, err := s.GetMaterial(ctx, workID)
|
|
if err != nil {
|
|
return MaterialJob{}, err
|
|
}
|
|
if job.RewriteConfirmedAt == nil {
|
|
return MaterialJob{}, ErrConflict
|
|
}
|
|
work, err := s.GetWork(ctx, workID)
|
|
if err != nil {
|
|
return MaterialJob{}, err
|
|
}
|
|
content, err := generator.Generate(ctx,
|
|
"根据原作品和转写内容生成仿写草稿。只返回 JSON，不要 Markdown 或额外文字，格式必须是 {\"title\":\"标题\",\"script\":\"脚本\"}。不得编造原文没有的事实。",
|
|
fmt.Sprintf("改写要求：%s\n原标题：%s\n原文案：%s\n转写：%s", job.RewriteRequirement, work.Title, work.Body, job.Transcript))
|
|
if err != nil {
|
|
return MaterialJob{}, err
|
|
}
|
|
title, script, err := parseGeneratedRewrite(content)
|
|
if err != nil {
|
|
return MaterialJob{}, err
|
|
}
|
|
if _, err := s.db.ExecContext(ctx, `UPDATE creator_material_job SET generated_title = $2, generated_script = $3, updated_at = now() WHERE work_id = $1`, workID, title, script); err != nil {
|
|
return MaterialJob{}, databaseError(err)
|
|
}
|
|
return s.GetMaterial(ctx, workID)
|
|
}
|
|
|
|
func (s *Store) SaveRewrite(ctx context.Context, workID, title, script string) (MaterialJob, error) {
|
|
title, script = strings.TrimSpace(title), strings.TrimSpace(script)
|
|
if title == "" || script == "" || utf8.RuneCountInString(title) > 1000 || utf8.RuneCountInString(script) > 100000 {
|
|
return MaterialJob{}, ErrInvalid
|
|
}
|
|
job, err := s.GetMaterial(ctx, workID)
|
|
if err != nil {
|
|
return MaterialJob{}, err
|
|
}
|
|
if job.RewriteConfirmedAt == nil {
|
|
return MaterialJob{}, ErrConflict
|
|
}
|
|
if _, err := s.db.ExecContext(ctx, `UPDATE creator_material_job SET generated_title = $2, generated_script = $3, updated_at = now() WHERE work_id = $1`, workID, title, script); err != nil {
|
|
return MaterialJob{}, databaseError(err)
|
|
}
|
|
return s.GetMaterial(ctx, workID)
|
|
}
|
|
|
|
func (s *Store) SaveComment(ctx context.Context, input CommentInput) (Comment, bool, error) {
|
|
input.Platform, input.CommentKey, input.WorkID, input.AuthorUID, input.AuthorName, input.CommentType = strings.TrimSpace(input.Platform), strings.TrimSpace(input.CommentKey), strings.TrimSpace(input.WorkID), strings.TrimSpace(input.AuthorUID), strings.TrimSpace(input.AuthorName), strings.TrimSpace(input.CommentType)
|
|
if !ValidatePlatform(input.Platform) || input.CommentKey == "" || input.WorkID == "" || strings.TrimSpace(input.Content) == "" || (input.CommentType != "top_level" && input.CommentType != "unknown") || utf8.RuneCountInString(input.Content) > 10000 || len(input.RawPayload) > maxRawPayloadBytes || input.RawPayload != "" && !json.Valid([]byte(input.RawPayload)) {
|
|
return Comment{}, false, ErrInvalid
|
|
}
|
|
work, err := s.GetWork(ctx, input.WorkID)
|
|
if err != nil {
|
|
return Comment{}, false, err
|
|
}
|
|
if work.Platform != input.Platform {
|
|
return Comment{}, false, ErrInvalid
|
|
}
|
|
id := newID("comment")
|
|
var returnedID string
|
|
var inserted bool
|
|
err = s.db.QueryRowContext(ctx, `INSERT INTO creator_comment (id, platform, comment_key, work_id, author_uid, author_name, content, raw_payload, published_at, comment_type) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) ON CONFLICT (platform, comment_key) DO UPDATE SET work_id = EXCLUDED.work_id, author_uid = EXCLUDED.author_uid, author_name = EXCLUDED.author_name, content = EXCLUDED.content, raw_payload = COALESCE(EXCLUDED.raw_payload, creator_comment.raw_payload), published_at = EXCLUDED.published_at, comment_type = EXCLUDED.comment_type RETURNING id, (xmax = 0)`, id, input.Platform, input.CommentKey, input.WorkID, input.AuthorUID, input.AuthorName, input.Content, nullableRawPayload(input.RawPayload), input.PublishedAt, input.CommentType).Scan(&returnedID, &inserted)
|
|
if err != nil {
|
|
return Comment{}, false, databaseError(err)
|
|
}
|
|
result, err := s.GetComment(ctx, returnedID)
|
|
return result, inserted, err
|
|
}
|
|
|
|
func scanComment(scanner interface{ Scan(...any) error }) (Comment, error) {
|
|
var result Comment
|
|
var publishedAt sql.NullTime
|
|
var rawPayload sql.NullString
|
|
if err := scanner.Scan(&result.ID, &result.Platform, &result.CommentKey, &result.WorkID, &result.AuthorUID, &result.AuthorName, &result.Content, &rawPayload, &publishedAt, &result.CollectedAt, &result.CommentType); err != nil {
|
|
return Comment{}, err
|
|
}
|
|
result.PublishedAt = nullableTime(publishedAt)
|
|
if rawPayload.Valid {
|
|
result.RawPayload = rawPayload.String
|
|
}
|
|
result.CollectedAt = result.CollectedAt.UTC()
|
|
return result, nil
|
|
}
|
|
|
|
const commentSelect = `SELECT id, platform, comment_key, work_id, author_uid, author_name, content, raw_payload, published_at, collected_at, comment_type FROM creator_comment`
|
|
|
|
func (s *Store) GetComment(ctx context.Context, id string) (Comment, error) {
|
|
result, err := scanComment(s.db.QueryRowContext(ctx, commentSelect+` WHERE id = $1`, id))
|
|
return result, rowError(err)
|
|
}
|
|
|
|
func pageBounds(page, pageSize int) (int, int, error) {
|
|
if page < 1 || pageSize < 1 || pageSize > 100 {
|
|
return 0, 0, ErrInvalid
|
|
}
|
|
if page > int(maxDurationSeconds) || pageSize > int(maxDurationSeconds)/page {
|
|
return 0, 0, ErrInvalid
|
|
}
|
|
return (page - 1) * pageSize, pageSize, nil
|
|
}
|
|
|
|
func slicePage[T any](items []T, page, pageSize int) (Page[T], error) {
|
|
offset, limit, err := pageBounds(page, pageSize)
|
|
if err != nil {
|
|
return Page[T]{}, err
|
|
}
|
|
if offset > len(items) {
|
|
offset = len(items)
|
|
}
|
|
end := offset + limit
|
|
if end > len(items) {
|
|
end = len(items)
|
|
}
|
|
data := append([]T(nil), items[offset:end]...)
|
|
return Page[T]{Data: data, Page: page, PageSize: pageSize, Total: len(items), HasNext: end < len(items)}, nil
|
|
}
|
|
|
|
func (s *Store) ListWorksPage(ctx context.Context, filter WorkFilter, page, pageSize int) (Page[Work], error) {
|
|
items, err := s.ListWorks(ctx, filter)
|
|
if err != nil {
|
|
return Page[Work]{}, err
|
|
}
|
|
return slicePage(items, page, pageSize)
|
|
}
|
|
|
|
func (s *Store) ListCommentsPage(ctx context.Context, platform, workID string, page, pageSize int) (Page[Comment], error) {
|
|
items, err := s.ListComments(ctx, platform, workID)
|
|
if err != nil {
|
|
return Page[Comment]{}, err
|
|
}
|
|
return slicePage(items, page, pageSize)
|
|
}
|
|
|
|
func (s *Store) ListComments(ctx context.Context, platform, workID string) ([]Comment, error) {
|
|
query, args := commentSelect, []any{}
|
|
where := []string{}
|
|
if platform != "" {
|
|
if !ValidatePlatform(platform) {
|
|
return nil, ErrInvalid
|
|
}
|
|
args = append(args, platform)
|
|
where = append(where, fmt.Sprintf("platform = $%d", len(args)))
|
|
}
|
|
if workID != "" {
|
|
args = append(args, workID)
|
|
where = append(where, fmt.Sprintf("work_id = $%d", len(args)))
|
|
}
|
|
if len(where) > 0 {
|
|
query += ` WHERE ` + strings.Join(where, ` AND `)
|
|
}
|
|
query += ` ORDER BY published_at DESC NULLS LAST, collected_at DESC, id`
|
|
rows, err := s.db.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, databaseError(err)
|
|
}
|
|
defer rows.Close()
|
|
result := make([]Comment, 0)
|
|
for rows.Next() {
|
|
item, err := scanComment(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, item)
|
|
}
|
|
return result, rows.Err()
|
|
}
|