mathiasandClaude Opus 4.8
59050c4db6
Returns up to limit of a user's newest videos (published_at DESC, NULLS LAST) that have no summary yet. RLS-scoped via withUser — the test proves a second user's newer video never leaks. Drives the connect-time onboarding burst (Feature 1); the caller routes each through the shared rate gate. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
113 lines
3.7 KiB
Go
113 lines
3.7 KiB
Go
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)
|
|
VALUES ($1, $2, $3, $4, $5, $6)
|
|
ON CONFLICT (user_id, provider, provider_video_id) DO UPDATE SET
|
|
title = EXCLUDED.title,
|
|
url = EXCLUDED.url,
|
|
published_at = EXCLUDED.published_at
|
|
RETURNING id`,
|
|
v.UserID, provider, v.ProviderVideoID, v.Title, v.URL, nullTime(v.PublishedAt),
|
|
).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
|
|
}
|
|
|
|
// 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
|
|
}
|