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 }