diff --git a/internal/adapters/store/videos.go b/internal/adapters/store/videos.go new file mode 100644 index 0000000..64e63a5 --- /dev/null +++ b/internal/adapters/store/videos.go @@ -0,0 +1,77 @@ +package store + +import ( + "context" + "fmt" + "time" + + "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") + } + + tx, err := s.pool.Begin(ctx) + if err != nil { + return "", fmt.Errorf("store: begin: %w", err) + } + defer tx.Rollback(ctx) //nolint:errcheck // no-op after Commit + + // 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) + } + + provider := string(v.Provider) + if provider == "" { + provider = string(domain.ProviderYouTube) + } + + var id string + 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) + } + + if err := tx.Commit(ctx); err != nil { + return "", fmt.Errorf("store: commit: %w", 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 +} diff --git a/internal/adapters/store/videos_test.go b/internal/adapters/store/videos_test.go new file mode 100644 index 0000000..a30d383 --- /dev/null +++ b/internal/adapters/store/videos_test.go @@ -0,0 +1,83 @@ +package store_test + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "gitea.d-ma.be/mathias/tapir/internal/domain" +) + +func ytVideo(userID, provVideoID, title string) domain.Video { + return domain.Video{ + UserID: userID, + Provider: domain.ProviderYouTube, + ProviderVideoID: provVideoID, + Title: title, + URL: "https://www.youtube.com/watch?v=" + provVideoID, + PublishedAt: time.Date(2026, 6, 1, 12, 0, 0, 0, time.UTC), + // SubscriptionID is a provider resource id, not a UUID — must not be + // written to the UUID column. Set it to prove UpsertVideo ignores it. + SubscriptionID: "yt-subscription-resource-id", + } +} + +func TestUpsertVideo_ReturnsStableID(t *testing.T) { + ctx := context.Background() + s := newStore(t) + resetDB(t, rawPool(t)) + + id1, err := s.UpsertVideo(ctx, ytVideo(userA, "dQw4w9WgXcQ", "first title")) + require.NoError(t, err) + require.NotEmpty(t, id1) + + // Same (user, provider, provider_video_id) -> same row, same id, updated meta. + id2, err := s.UpsertVideo(ctx, ytVideo(userA, "dQw4w9WgXcQ", "updated title")) + require.NoError(t, err) + require.Equal(t, id1, id2, "idempotent upsert must return the same durable id") + + p := rawPool(t) + var ( + title string + count int + ) + require.NoError(t, p.QueryRow(ctx, + `SELECT title FROM videos WHERE id = $1`, id1).Scan(&title)) + require.Equal(t, "updated title", title, "second upsert must update metadata in place") + + require.NoError(t, p.QueryRow(ctx, + `SELECT count(*) FROM videos WHERE user_id = $1`, userA).Scan(&count)) + require.Equal(t, 1, count, "must not duplicate the row") +} + +func TestUpsertVideo_IDMatchesSummaryDedup(t *testing.T) { + ctx := context.Background() + s := newStore(t) + resetDB(t, rawPool(t)) + + // Upsert assigns the durable video id; a summary delivered under that id + // must then show up in SeenVideoIDs — this is the cross-restart dedup chain. + id, err := s.UpsertVideo(ctx, ytVideo(userA, "abc123", "t")) + require.NoError(t, err) + + require.NoError(t, s.Deliver(ctx, summary(userA, id, "the summary"))) + + seen, err := s.SeenVideoIDs(ctx, userA) + require.NoError(t, err) + require.True(t, seen[id], "the upserted video id must match the summary dedup key") +} + +func TestUpsertVideo_PerUserIsolation(t *testing.T) { + ctx := context.Background() + s := newStore(t) + resetDB(t, rawPool(t)) + + idA, err := s.UpsertVideo(ctx, ytVideo(userA, "same-provider-id", "a")) + require.NoError(t, err) + idB, err := s.UpsertVideo(ctx, ytVideo(userB, "same-provider-id", "b")) + require.NoError(t, err) + + require.NotEqual(t, idA, idB, "same provider video for two users must be two distinct rows") +}