package store import ( "context" "fmt" "time" "github.com/jackc/pgx/v5" "gitea.d-ma.be/mathias/tapir/internal/domain" ) // UpsertVideo persists a video's metadata and returns its durable store id (the // videos.id UUID). It is idempotent on (user_id, provider, provider_video_id): // the same provider video for a user always resolves to the same row and the // same returned id, so the run loop can use that id as the stable dedup key // across restarts (it matches summaries.video_id once a summary exists). // // This lives in a separate file from store.go on purpose: the Sink port only // carries a domain.Summary (no title/channel), so video metadata is persisted // here, out of the delivery path, to keep the reader's rows readable. // // subscription_id is intentionally left NULL at Stage 0: the YouTube // Subscription.ID is a provider resource id, not the UUID that column expects, // and the subscriptions table is not part of this slice (data-model.md). func (s *Store) UpsertVideo(ctx context.Context, v domain.Video) (string, error) { if v.UserID == "" { return "", fmt.Errorf("store: upsert video: empty user id") } if v.ProviderVideoID == "" { return "", fmt.Errorf("store: upsert video: empty provider video id") } provider := string(v.Provider) if provider == "" { provider = string(domain.ProviderYouTube) } var id string if err := s.withUser(ctx, v.UserID, func(tx pgx.Tx) error { // Ensure the owning user exists (FK target) — same as the Deliver path. if _, err := tx.Exec(ctx, `INSERT INTO users (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`, v.UserID); err != nil { return fmt.Errorf("store: upsert user: %w", err) } if err := tx.QueryRow(ctx, `INSERT INTO videos (user_id, provider, provider_video_id, title, url, published_at, channel_title, duration_s) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (user_id, provider, provider_video_id) DO UPDATE SET title = EXCLUDED.title, url = EXCLUDED.url, published_at = EXCLUDED.published_at, channel_title = COALESCE(NULLIF(EXCLUDED.channel_title, ''), videos.channel_title), duration_s = COALESCE(EXCLUDED.duration_s, videos.duration_s) RETURNING id`, v.UserID, provider, v.ProviderVideoID, v.Title, v.URL, nullTime(v.PublishedAt), v.ChannelTitle, nullDuration(v.DurationSeconds), ).Scan(&id); err != nil { return fmt.Errorf("store: upsert video: %w", err) } return nil }); err != nil { return "", err } return id, nil } // nullTime maps the zero time to NULL so an unknown published_at is stored as // SQL NULL rather than year 0001. func nullTime(t time.Time) *time.Time { if t.IsZero() { return nil } return &t } // nullDuration maps an unknown duration (0) to SQL NULL so the upsert's // COALESCE(EXCLUDED.duration_s, videos.duration_s) preserves a previously-known // value instead of clobbering it with 0 (ADR-028; the channel_title backfill // stance, migration 014). func nullDuration(seconds int) *int { if seconds <= 0 { return nil } return &seconds } // NewestUnsummarizedVideoIDs returns up to limit of the user's videos that have // no summary yet, newest first (published_at DESC, NULLS LAST). It caps the // connect-time onboarding burst (Feature 1) at a fixed count: the caller marks // these for summarization through the shared rate gate. RLS-scoped via withUser, // so it only ever sees the requesting user's rows. limit <= 0 returns nil. func (s *Store) NewestUnsummarizedVideoIDs(ctx context.Context, userID string, limit int) ([]string, error) { if limit <= 0 { return nil, nil } var ids []string if err := s.withUser(ctx, userID, func(tx pgx.Tx) error { rows, err := tx.Query(ctx, `SELECT v.id FROM videos v WHERE v.user_id = $1 AND NOT EXISTS ( SELECT 1 FROM summaries su WHERE su.user_id = v.user_id AND su.video_id = v.id) ORDER BY v.published_at DESC NULLS LAST, v.seen_at DESC LIMIT $2`, userID, limit) if err != nil { return fmt.Errorf("store: newest unsummarized: %w", err) } defer rows.Close() for rows.Next() { var id string if err := rows.Scan(&id); err != nil { return fmt.Errorf("store: scan newest unsummarized: %w", err) } ids = append(ids, id) } return rows.Err() }); err != nil { return nil, err } return ids, nil } // OnboardBurstVideoIDs returns up to limit of the user's unsummarized videos for // the connect-time onboarding burst (ADR-028), newest-first but quality-aware: a // video is excluded when its duration is KNOWN and outside [minSeconds, maxSeconds] // — dropping Shorts (below min) and multi-hour livestream VODs (above max) that // would waste a scarce caption fetch on a poor first impression. A NULL/unknown // duration is kept (degrade-open) but ranked AFTER known-good rows, so a freshly // enriched good pick wins when both exist. minSeconds<=0 / maxSeconds<=0 each // disable that bound (0/0 == pure newest-first, the reversibility lever). // RLS-scoped via withUser; limit <= 0 returns nil. func (s *Store) OnboardBurstVideoIDs(ctx context.Context, userID string, limit, minSeconds, maxSeconds int) ([]string, error) { if limit <= 0 { return nil, nil } var ids []string if err := s.withUser(ctx, userID, func(tx pgx.Tx) error { rows, err := tx.Query(ctx, `SELECT v.id FROM videos v WHERE v.user_id = $1 AND NOT EXISTS ( SELECT 1 FROM summaries su WHERE su.user_id = v.user_id AND su.video_id = v.id) AND NOT ( v.duration_s IS NOT NULL AND ( ($3 > 0 AND v.duration_s < $3) OR ($4 > 0 AND v.duration_s > $4) )) ORDER BY (v.duration_s IS NOT NULL) DESC, v.published_at DESC NULLS LAST, v.seen_at DESC LIMIT $2`, userID, limit, minSeconds, maxSeconds) if err != nil { return fmt.Errorf("store: onboard burst videos: %w", err) } defer rows.Close() for rows.Next() { var id string if err := rows.Scan(&id); err != nil { return fmt.Errorf("store: scan onboard burst video: %w", err) } ids = append(ids, id) } return rows.Err() }); err != nil { return nil, err } return ids, nil } // DistinctChannels returns the user's distinct, non-empty source channel titles // (the channels they have videos from), alphabetically — the option list for the // feed's channel filter. RLS-scoped via withUser. func (s *Store) DistinctChannels(ctx context.Context, userID string) ([]string, error) { var out []string if err := s.withUser(ctx, userID, func(tx pgx.Tx) error { rows, err := tx.Query(ctx, `SELECT DISTINCT channel_title FROM videos WHERE user_id = $1 AND channel_title IS NOT NULL AND channel_title <> '' ORDER BY channel_title`, userID) if err != nil { return fmt.Errorf("store: distinct channels: %w", err) } defer rows.Close() for rows.Next() { var c string if err := rows.Scan(&c); err != nil { return fmt.Errorf("store: scan channel: %w", err) } out = append(out, c) } return rows.Err() }); err != nil { return nil, err } return out, nil }