tapir/internal/adapters/store/store.go

// Package store is the Postgres-backed implementation of ports.Sink: the user's
// own durable store of summaries. It is the primary sink (ADR-003) and the source
// of the engine's durable, cross-restart dedup (the in-engine map is process-
// lifetime only). Connection is pgx/v5 + pgxpool with the DSN from the caller;
// schema is applied via golang-migrate from embedded migrations.
//
// Per-user isolation (docs/data-model.md) is a Stage-0 promise: every row carries
// user_id and every read is scoped by it, even though the demo has one user.
package store

import (
	"context"
	"database/sql"
	"embed"
	"encoding/json"
	"errors"
	"fmt"

	"github.com/golang-migrate/migrate/v4"
	migratepgx "github.com/golang-migrate/migrate/v4/database/pgx/v5"
	"github.com/golang-migrate/migrate/v4/source/iofs"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"

	_ "github.com/jackc/pgx/v5/stdlib" // register the "pgx" database/sql driver for migrate

	"git.d-ma.be/mathias/tapir/internal/domain"
)

//go:embed migrations/*.sql
var migrationsFS embed.FS

// Store persists summaries to Postgres and answers durable dedup queries.
type Store struct {
	pool *pgxpool.Pool
}

// New connects a pool to dsn, applies all pending migrations, and verifies the
// connection. The caller owns the lifetime: call Close when done.
func New(ctx context.Context, dsn string) (*Store, error) {
	if dsn == "" {
		return nil, errors.New("store: empty DSN")
	}
	if err := Migrate(dsn); err != nil {
		return nil, err
	}
	pool, err := pgxpool.New(ctx, dsn)
	if err != nil {
		return nil, fmt.Errorf("store: connect pool: %w", err)
	}
	if err := pool.Ping(ctx); err != nil {
		pool.Close()
		return nil, fmt.Errorf("store: ping: %w", err)
	}
	return &Store{pool: pool}, nil
}

// Migrate applies all pending up-migrations against dsn. It opens its own
// short-lived connection (golang-migrate uses database/sql) and closes it before
// returning, so it can run before the pool is created or be invoked standalone.
func Migrate(dsn string) error {
	db, err := sql.Open("pgx", dsn)
	if err != nil {
		return fmt.Errorf("store: open migrate db: %w", err)
	}
	defer func() { _ = db.Close() }()

	drv, err := migratepgx.WithInstance(db, &migratepgx.Config{})
	if err != nil {
		return fmt.Errorf("store: migrate driver: %w", err)
	}
	src, err := iofs.New(migrationsFS, "migrations")
	if err != nil {
		return fmt.Errorf("store: migrate source: %w", err)
	}
	m, err := migrate.NewWithInstance("iofs", src, "pgx", drv)
	if err != nil {
		return fmt.Errorf("store: migrator: %w", err)
	}
	if err := m.Up(); err != nil && !errors.Is(err, migrate.ErrNoChange) {
		return fmt.Errorf("store: migrate up: %w", err)
	}
	return nil
}

// Close releases the connection pool.
func (s *Store) Close() {
	s.pool.Close()
}

// Name identifies this sink in delivery records.
func (s *Store) Name() string { return "store" }

// withUser is the single choke point through which EVERY DB access in this
// package flows, so per-user isolation is structural — not a per-query opt-in
// someone can forget. It:
//
//   - BEGINs a transaction,
//   - sets the per-request GUC tapir.current_user_id via
//     set_config('tapir.current_user_id', $1, true). The set_config form is used
//     instead of `SET LOCAL` because it is parameterizable (SET cannot bind a
//     value through the driver); the third arg true = local = transaction-scoped,
//     so it auto-resets on commit/rollback and a pooled connection never leaks one
//     request's user into the next,
//   - runs fn against that transaction,
//   - COMMITs (or ROLLBACKs on error).
//
// The migration-003 RLS policies key off this GUC: a row is visible/writable only
// when its owner = current_setting('tapir.current_user_id'). RLS enforces only
// when the app connects as a non-superuser, non-BYPASSRLS role (in production the
// table owner tapir, made subject via FORCE ROW LEVEL SECURITY). A superuser DSN
// bypasses RLS regardless — see rls_test.go, which connects as a dedicated
// non-superuser role to prove the enforcement is real.
func (s *Store) withUser(ctx context.Context, userID string, fn func(pgx.Tx) error) error {
	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

	if _, err := tx.Exec(ctx,
		`SELECT set_config('tapir.current_user_id', $1, true)`, userID); err != nil {
		return fmt.Errorf("store: scope user: %w", err)
	}
	if err := fn(tx); err != nil {
		return err
	}
	if err := tx.Commit(ctx); err != nil {
		return fmt.Errorf("store: commit: %w", err)
	}
	return nil
}

// Deliver upserts the summary idempotently on (user_id, video_id) and records the
// store delivery. Re-delivering the same summary updates in place — it never
// errors or duplicates. The whole write is one transaction so a summary and its
// delivery row stay consistent.
func (s *Store) Deliver(ctx context.Context, sum domain.Summary) error {
	highlights, err := marshalList(sum.Highlights)
	if err != nil {
		return fmt.Errorf("store: marshal highlights: %w", err)
	}
	takeaways, err := marshalList(sum.Takeaways)
	if err != nil {
		return fmt.Errorf("store: marshal takeaways: %w", err)
	}

	return s.withUser(ctx, sum.UserID, func(tx pgx.Tx) error {
		// Ensure the owning user exists (FK target). The store sink receives only
		// a Summary, so a minimal user row is enough at Stage 0.
		if _, err := tx.Exec(ctx,
			`INSERT INTO users (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`,
			sum.UserID); err != nil {
			return fmt.Errorf("store: upsert user: %w", err)
		}

		var summaryID string
		if err := tx.QueryRow(ctx,
			`INSERT INTO summaries
			     (user_id, video_id, summary, highlights, takeaways, ai_provider, ai_model, fallback_used)
			 VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
			 ON CONFLICT (user_id, video_id) DO UPDATE SET
			     summary       = EXCLUDED.summary,
			     highlights    = EXCLUDED.highlights,
			     takeaways     = EXCLUDED.takeaways,
			     ai_provider   = EXCLUDED.ai_provider,
			     ai_model      = EXCLUDED.ai_model,
			     fallback_used = EXCLUDED.fallback_used
			 RETURNING id`,
			sum.UserID, sum.VideoID, sum.Summary, highlights, takeaways,
			sum.AIProvider, sum.AIModel, sum.FallbackUsed,
		).Scan(&summaryID); err != nil {
			return fmt.Errorf("store: upsert summary: %w", err)
		}

		if _, err := tx.Exec(ctx,
			`INSERT INTO sink_deliveries (summary_id, sink, status)
			 VALUES ($1, 'store', 'delivered')
			 ON CONFLICT (summary_id, sink) DO UPDATE SET
			     status     = 'delivered',
			     detail     = NULL,
			     updated_at = NOW()`,
			summaryID); err != nil {
			return fmt.Errorf("store: record delivery: %w", err)
		}
		return nil
	})
}

// HasSummary reports whether a summary already exists for (userID, videoID).
// This is the per-video durable dedup check.
func (s *Store) HasSummary(ctx context.Context, userID, videoID string) (bool, error) {
	var exists bool
	if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
		return tx.QueryRow(ctx,
			`SELECT EXISTS(SELECT 1 FROM summaries WHERE user_id = $1 AND video_id = $2)`,
			userID, videoID).Scan(&exists)
	}); err != nil {
		return false, fmt.Errorf("store: has summary: %w", err)
	}
	return exists, nil
}

// SeenVideoIDs returns the set of video IDs that already have a summary for the
// user. The watcher uses it to skip re-summarizing across restarts. Scoped by
// user_id, so one user never sees another's videos.
func (s *Store) SeenVideoIDs(ctx context.Context, userID string) (map[string]bool, error) {
	seen := make(map[string]bool)
	if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
		rows, err := tx.Query(ctx,
			`SELECT video_id FROM summaries WHERE user_id = $1`, userID)
		if err != nil {
			return fmt.Errorf("store: seen video ids: %w", err)
		}
		defer rows.Close()

		for rows.Next() {
			var id string
			if err := rows.Scan(&id); err != nil {
				return fmt.Errorf("store: scan video id: %w", err)
			}
			seen[id] = true
		}
		if err := rows.Err(); err != nil {
			return fmt.Errorf("store: iterate video ids: %w", err)
		}
		return nil
	}); err != nil {
		return nil, err
	}
	return seen, nil
}

// marshalList renders a string slice as a JSON array, normalising nil to "[]" so
// the jsonb columns never hold SQL/JSON null.
func marshalList(xs []string) ([]byte, error) {
	if xs == nil {
		xs = []string{}
	}
	return json.Marshal(xs)
}