tapir/internal/adapters/store/reads.go

package store

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
)

// ErrNotFound is returned by GetSummaryByVideo when no summary exists for the
// (userID, videoID) pair scoped to that user.
var ErrNotFound = errors.New("store: summary not found")

// SummaryRow is a read-side view of a stored summary, enriched with the video's
// metadata when a matching videos row exists. It is deliberately separate from
// domain.Summary: it carries display fields (Title, Channel, URL, PublishedAt)
// that the engine never sees, sourced via a LEFT JOIN so a summary with no
// videos row still renders (Title/URL empty, PublishedAt zero).
//
// Channel currently mirrors the video provider ("youtube"): the per-channel
// channel_title lives on the subscriptions table (docs/data-model.md), which is
// not part of the Stage-0 store slice yet. When that table is migrated, swap the
// JOIN source — callers already fall back gracefully on an empty Channel.
type SummaryRow struct {
	VideoID         string
	ProviderVideoID string    // videos.provider_video_id; empty when no videos row
	Title           string    // videos.title; empty when no videos row
	Channel         string    // videos.provider for now; empty when no videos row
	URL             string    // videos.url; empty when no videos row
	PublishedAt     time.Time // videos.published_at; zero when absent
	Summary         string
	Highlights      []string
	Takeaways       []string
	AIProvider      string
	AIModel         string
	FallbackUsed    bool
	CreatedAt       time.Time
	Actions         []string // current active actions for this video; nil when none

	// Summarized reports whether a summary exists for this video. The summary-only
	// reads (ListSummaries/GetSummaryByVideo) always yield true; the all-videos
	// read (ListVideos) yields false for a discovered-but-unsummarized video, whose
	// Summary/Highlights/AIProvider fields are then empty.
	Summarized bool
	// SummarizeRequested reflects videos.summarize_requested: the manual queue flag
	// set by the web "Summarize" button and cleared by the next `tapir run`. Only
	// populated by ListVideos/GetVideoRow (summary-only reads leave it false).
	SummarizeRequested bool
	// TranscriptStatus mirrors videos.transcript_status (migration 007): "" (unset),
	// "none", "rate_limited", or "fetched". Drives the "Retrying later" list badge.
	// Only populated by ListVideos/GetVideoRow ("" on summary-only reads).
	TranscriptStatus string
}

// selectSummary is the shared projection for both reads. videos is LEFT JOINed
// on its UUID id (= summaries.video_id) and the same user_id, so the join never
// crosses users and a missing videos row yields nulls, not a dropped summary.
const selectSummary = `
	SELECT s.video_id,
	       COALESCE(v.provider_video_id, ''),
	       COALESCE(v.title, ''),
	       COALESCE(v.provider, ''),
	       COALESCE(v.url, ''),
	       v.published_at,
	       s.summary,
	       s.highlights,
	       s.takeaways,
	       COALESCE(s.ai_provider, ''),
	       COALESCE(s.ai_model, ''),
	       s.fallback_used,
	       s.created_at
	  FROM summaries s
	  LEFT JOIN videos v ON v.id = s.video_id AND v.user_id = s.user_id`

// ListSummaries returns the user's summaries, most recent first, capped at limit.
// A non-positive limit defaults to 50. Scoped by user_id: one user never sees
// another's summaries (per-user isolation, docs/data-model.md).
func (s *Store) ListSummaries(ctx context.Context, userID string, limit int) ([]SummaryRow, error) {
	if limit <= 0 {
		limit = 50
	}
	var out []SummaryRow
	if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
		rows, err := tx.Query(ctx,
			selectSummary+`
			 WHERE s.user_id = $1
			 ORDER BY s.created_at DESC
			 LIMIT $2`,
			userID, limit)
		if err != nil {
			return fmt.Errorf("store: list summaries: %w", err)
		}
		defer rows.Close()

		for rows.Next() {
			row, err := scanSummaryRow(rows)
			if err != nil {
				return err
			}
			out = append(out, row)
		}
		if err := rows.Err(); err != nil {
			return fmt.Errorf("store: iterate summaries: %w", err)
		}
		return nil
	}); err != nil {
		return nil, err
	}
	if err := s.attachActions(ctx, userID, out); err != nil {
		return nil, err
	}
	return out, nil
}

// selectVideo is the all-videos projection: it drives from the videos table and
// LEFT JOINs the (at most one) summary, so a discovered-but-unsummarized video
// still appears with empty summary fields. The column order mirrors selectSummary
// for the shared fields, then appends summarized + summarize_requested. created_at
// falls back to the video's seen_at when there is no summary, so the read-side row
// always carries a sortable timestamp.
const selectVideo = `
	SELECT v.id,
	       v.provider_video_id,
	       COALESCE(v.title, ''),
	       v.provider,
	       COALESCE(v.url, ''),
	       v.published_at,
	       COALESCE(s.summary, ''),
	       s.highlights,
	       s.takeaways,
	       COALESCE(s.ai_provider, ''),
	       COALESCE(s.ai_model, ''),
	       COALESCE(s.fallback_used, FALSE),
	       COALESCE(s.created_at, v.seen_at),
	       (s.id IS NOT NULL) AS summarized,
	       v.summarize_requested,
	       COALESCE(v.transcript_status, '')
	  FROM videos v
	  LEFT JOIN summaries s ON s.video_id = v.id AND s.user_id = v.user_id`

// ListVideos returns ALL of the user's videos — summarized first then most recent
// by seen_at — capped at limit (non-positive defaults to 500). Unsummarized
// videos come back with Summarized=false and empty summary fields, so the list
// view can render them with a "Summarize" affordance. Scoped by user_id.
func (s *Store) ListVideos(ctx context.Context, userID string, limit int) ([]SummaryRow, error) {
	if limit <= 0 {
		limit = 500
	}
	var out []SummaryRow
	if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
		rows, err := tx.Query(ctx,
			selectVideo+`
			 WHERE v.user_id = $1
			 ORDER BY (s.id IS NOT NULL) DESC, v.seen_at DESC
			 LIMIT $2`,
			userID, limit)
		if err != nil {
			return fmt.Errorf("store: list videos: %w", err)
		}
		defer rows.Close()

		for rows.Next() {
			row, err := scanVideoRow(rows)
			if err != nil {
				return err
			}
			out = append(out, row)
		}
		if err := rows.Err(); err != nil {
			return fmt.Errorf("store: iterate videos: %w", err)
		}
		return nil
	}); err != nil {
		return nil, err
	}
	if err := s.attachActions(ctx, userID, out); err != nil {
		return nil, err
	}
	return out, nil
}

// GetVideoRow returns a single video row (summarized or not) for (userID,
// videoID), used to re-render one card after queuing it. Returns ErrNotFound when
// the user has no such video. Scoped by user_id.
func (s *Store) GetVideoRow(ctx context.Context, userID, videoID string) (*SummaryRow, error) {
	var (
		row   SummaryRow
		found bool
	)
	if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
		rows, err := tx.Query(ctx,
			selectVideo+`
			 WHERE v.user_id = $1 AND v.id = $2`,
			userID, videoID)
		if err != nil {
			return fmt.Errorf("store: get video: %w", err)
		}
		defer rows.Close()

		if !rows.Next() {
			if err := rows.Err(); err != nil {
				return fmt.Errorf("store: get video: %w", err)
			}
			return nil
		}
		row, err = scanVideoRow(rows)
		if err != nil {
			return err
		}
		found = true
		return nil
	}); err != nil {
		return nil, err
	}
	if !found {
		return nil, ErrNotFound
	}
	holder := []SummaryRow{row}
	if err := s.attachActions(ctx, userID, holder); err != nil {
		return nil, err
	}
	return &holder[0], nil
}

// scanVideoRow reads one row in the selectVideo column order. published_at is
// nullable so it scans through a pointer.
func scanVideoRow(rows pgx.Row) (SummaryRow, error) {
	var (
		row         SummaryRow
		highlights  []byte
		takeaways   []byte
		publishedAt *time.Time
	)
	if err := rows.Scan(
		&row.VideoID,
		&row.ProviderVideoID,
		&row.Title,
		&row.Channel,
		&row.URL,
		&publishedAt,
		&row.Summary,
		&highlights,
		&takeaways,
		&row.AIProvider,
		&row.AIModel,
		&row.FallbackUsed,
		&row.CreatedAt,
		&row.Summarized,
		&row.SummarizeRequested,
		&row.TranscriptStatus,
	); err != nil {
		return SummaryRow{}, fmt.Errorf("store: scan video: %w", err)
	}
	if publishedAt != nil {
		row.PublishedAt = *publishedAt
	}
	var err error
	if row.Highlights, err = unmarshalList(highlights); err != nil {
		return SummaryRow{}, fmt.Errorf("store: unmarshal highlights: %w", err)
	}
	if row.Takeaways, err = unmarshalList(takeaways); err != nil {
		return SummaryRow{}, fmt.Errorf("store: unmarshal takeaways: %w", err)
	}
	return row, nil
}

// GetSummaryByVideo returns the full summary for (userID, videoID), including
// highlights and takeaways. Returns ErrNotFound when the user has no such
// summary. Scoped by user_id.
func (s *Store) GetSummaryByVideo(ctx context.Context, userID, videoID string) (*SummaryRow, error) {
	var (
		row   SummaryRow
		found bool
	)
	if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
		rows, err := tx.Query(ctx,
			selectSummary+`
			 WHERE s.user_id = $1 AND s.video_id = $2`,
			userID, videoID)
		if err != nil {
			return fmt.Errorf("store: get summary: %w", err)
		}
		defer rows.Close()

		if !rows.Next() {
			if err := rows.Err(); err != nil {
				return fmt.Errorf("store: get summary: %w", err)
			}
			return nil
		}
		row, err = scanSummaryRow(rows)
		if err != nil {
			return err
		}
		found = true
		return nil
	}); err != nil {
		return nil, err
	}
	if !found {
		return nil, ErrNotFound
	}
	holder := []SummaryRow{row}
	if err := s.attachActions(ctx, userID, holder); err != nil {
		return nil, err
	}
	return &holder[0], nil
}

// scanSummaryRow reads one row in the selectSummary column order. published_at is
// nullable (no videos row, or an unset date) so it scans through a pointer.
func scanSummaryRow(rows pgx.Row) (SummaryRow, error) {
	var (
		row         SummaryRow
		highlights  []byte
		takeaways   []byte
		publishedAt *time.Time
	)
	if err := rows.Scan(
		&row.VideoID,
		&row.ProviderVideoID,
		&row.Title,
		&row.Channel,
		&row.URL,
		&publishedAt,
		&row.Summary,
		&highlights,
		&takeaways,
		&row.AIProvider,
		&row.AIModel,
		&row.FallbackUsed,
		&row.CreatedAt,
	); err != nil {
		return SummaryRow{}, fmt.Errorf("store: scan summary: %w", err)
	}
	if publishedAt != nil {
		row.PublishedAt = *publishedAt
	}
	var err error
	if row.Highlights, err = unmarshalList(highlights); err != nil {
		return SummaryRow{}, fmt.Errorf("store: unmarshal highlights: %w", err)
	}
	if row.Takeaways, err = unmarshalList(takeaways); err != nil {
		return SummaryRow{}, fmt.Errorf("store: unmarshal takeaways: %w", err)
	}
	return row, nil
}

// unmarshalList decodes a jsonb array column into a string slice, mirroring
// marshalList in store.go. Empty/NULL bytes decode to a nil slice.
func unmarshalList(b []byte) ([]string, error) {
	if len(b) == 0 {
		return nil, nil
	}
	var xs []string
	if err := json.Unmarshal(b, &xs); err != nil {
		return nil, err
	}
	return xs, nil
}