2026-10-03 14:13:37 +00:00
1 changed files with 199 additions and 0 deletions
@@ -0,0 +1,199 @@
+package store
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+
+	"git.d-ma.be/mathias/tapir/internal/domain"
+)
+
+// ErrUploadNotFound is returned when an uploaded transcript does not exist OR is
+// not the caller's. The two are deliberately indistinguishable: a non-owner learns
+// nothing about whether someone else's upload exists (ADR-033).
+var ErrUploadNotFound = errors.New("store: uploaded transcript not found")
+
+// Uploaded transcripts are PRIVATE per-user content, stored in their own
+// RLS-scoped table (migration 017) — never in the shared `transcripts` (ADR-021).
+// Every method goes through withUser, so row-level security confines it to the
+// caller's rows; each statement ALSO names user_id in its predicate, so the
+// isolation does not rest on a single mechanism and stays testable under the
+// superuser connection the unit tests use.
+
+// CreateUpload records a new upload in the queued state and returns its id.
+func (s *Store) CreateUpload(ctx context.Context, userID, title string) (string, error) {
+	var id string
+	if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
+		return tx.QueryRow(ctx,
+			`INSERT INTO uploaded_transcripts (user_id, title) VALUES ($1, $2) RETURNING id`,
+			userID, title).Scan(&id)
+	}); err != nil {
+		return "", fmt.Errorf("store: create upload: %w", err)
+	}
+	return id, nil
+}
+
+// StartTranscribing moves an upload to transcribing.
+func (s *Store) StartTranscribing(ctx context.Context, userID, id string) error {
+	return s.updateUpload(ctx, userID, id, "start transcribing",
+		`UPDATE uploaded_transcripts SET status = 'transcribing'
+		  WHERE id = $1 AND user_id = $2`)
+}
+
+// CompleteUpload stores the finished transcription and moves the upload to done.
+func (s *Store) CompleteUpload(ctx context.Context, userID, id string, t domain.Transcription) error {
+	segs, err := json.Marshal(t.Segments)
+	if err != nil {
+		return fmt.Errorf("store: complete upload: marshal segments: %w", err)
+	}
+	return s.updateUpload(ctx, userID, id, "complete upload",
+		`UPDATE uploaded_transcripts
+		    SET status = 'done', content = $3, segments = $4, endpoint = $5,
+		        audio_seconds = $6, error_kind = NULL, finished_at = NOW()
+		  WHERE id = $1 AND user_id = $2`,
+		t.Text, segs, t.Endpoint, t.AudioSeconds)
+}
+
+// FailUpload moves an upload to failed with a typed reason.
+func (s *Store) FailUpload(ctx context.Context, userID, id string, kind domain.UploadErrorKind) error {
+	return s.updateUpload(ctx, userID, id, "fail upload",
+		`UPDATE uploaded_transcripts
+		    SET status = 'failed', error_kind = $3, finished_at = NOW()
+		  WHERE id = $1 AND user_id = $2`,
+		string(kind))
+}
+
+// updateUpload runs one UPDATE keyed on ($1 = id, $2 = userID) plus extra args,
+// and maps "no row touched" to ErrUploadNotFound — which is what a write to another
+// user's upload looks like under RLS (the row is invisible, not forbidden).
+func (s *Store) updateUpload(ctx context.Context, userID, id, op, sql string, args ...any) error {
+	all := append([]any{id, userID}, args...)
+	return s.withUser(ctx, userID, func(tx pgx.Tx) error {
+		ct, err := tx.Exec(ctx, sql, all...)
+		if err != nil {
+			return fmt.Errorf("store: %s: %w", op, err)
+		}
+		if ct.RowsAffected() == 0 {
+			return ErrUploadNotFound
+		}
+		return nil
+	})
+}
+
+// GetUpload returns one of the caller's uploads, including the transcript text.
+func (s *Store) GetUpload(ctx context.Context, userID, id string) (*domain.UploadedTranscript, error) {
+	var u domain.UploadedTranscript
+	err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
+		row := tx.QueryRow(ctx, uploadSelect+` WHERE id = $1 AND user_id = $2`, id, userID)
+		return scanUpload(row, &u, true)
+	})
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrUploadNotFound
+	}
+	if err != nil {
+		return nil, fmt.Errorf("store: get upload: %w", err)
+	}
+	return &u, nil
+}
+
+// ListUploads returns the caller's uploads, newest first, WITHOUT transcript text
+// or segments: a list view must not drag every transcript along.
+func (s *Store) ListUploads(ctx context.Context, userID string) ([]domain.UploadedTranscript, error) {
+	var out []domain.UploadedTranscript
+	err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
+		rows, err := tx.Query(ctx, uploadSelect+` WHERE user_id = $1 ORDER BY created_at DESC, id`, userID)
+		if err != nil {
+			return err
+		}
+		defer rows.Close()
+		for rows.Next() {
+			var u domain.UploadedTranscript
+			if err := scanUpload(rows, &u, false); err != nil {
+				return err
+			}
+			out = append(out, u)
+		}
+		return rows.Err()
+	})
+	if err != nil {
+		return nil, fmt.Errorf("store: list uploads: %w", err)
+	}
+	return out, nil
+}
+
+// FailInterruptedUploads is the start-up sweep. The audio of an upload is never
+// retained (ADR-033), so a job that was queued or transcribing when the process
+// stopped can never resume; leaving it would be a stuck spinner. Every such row,
+// for every user, becomes failed/interrupted and the user re-uploads. It
+// enumerates users from the un-RLS'd identity map and scopes each pass with
+// withUser, exactly as scheduled discovery does (ADR-018). Returns rows failed.
+func (s *Store) FailInterruptedUploads(ctx context.Context) (int, error) {
+	users, err := s.ListAllUsers(ctx)
+	if err != nil {
+		return 0, fmt.Errorf("store: fail interrupted uploads: %w", err)
+	}
+	total := 0
+	for _, u := range users {
+		err := s.withUser(ctx, u.UserID, func(tx pgx.Tx) error {
+			ct, err := tx.Exec(ctx,
+				`UPDATE uploaded_transcripts
+				    SET status = 'failed', error_kind = $2, finished_at = NOW()
+				  WHERE user_id = $1 AND status IN ('queued', 'transcribing')`,
+				u.UserID, string(domain.UploadErrInterrupted))
+			total += int(ct.RowsAffected())
+			return err
+		})
+		if err != nil {
+			return total, fmt.Errorf("store: fail interrupted uploads: user %s: %w", u.UserID, err)
+		}
+	}
+	return total, nil
+}
+
+const uploadSelect = `SELECT id, user_id, title, status, content, segments, endpoint,
+       audio_seconds, error_kind, created_at, finished_at FROM uploaded_transcripts`
+
+// scanUpload scans one uploadSelect row. withText controls whether the transcript
+// body and segments are kept (GetUpload) or dropped (ListUploads).
+func scanUpload(row pgx.Row, u *domain.UploadedTranscript, withText bool) error {
+	var (
+		status, errKind, endpoint, content *string
+		segments                           []byte
+		seconds                            *float64
+		finished                           *time.Time
+	)
+	if err := row.Scan(&u.ID, &u.UserID, &u.Title, &status, &content, &segments, &endpoint,
+		&seconds, &errKind, &u.CreatedAt, &finished); err != nil {
+		return err
+	}
+	u.Status = domain.UploadStatus(deref(status))
+	u.Endpoint = deref(endpoint)
+	u.ErrorKind = domain.UploadErrorKind(deref(errKind))
+	if seconds != nil {
+		u.AudioSeconds = *seconds
+	}
+	if finished != nil {
+		u.FinishedAt = *finished
+	}
+	if !withText {
+		return nil
+	}
+	u.Content = deref(content)
+	if len(segments) > 0 {
+		if err := json.Unmarshal(segments, &u.Segments); err != nil {
+			return fmt.Errorf("unmarshal segments: %w", err)
+		}
+	}
+	return nil
+}
+
+func deref(s *string) string {
+	if s == nil {
+		return ""
+	}
+	return *s
+}