diff --git a/internal/adapters/store/uploaded_transcripts.go b/internal/adapters/store/uploaded_transcripts.go new file mode 100644 index 0000000..7252761 --- /dev/null +++ b/internal/adapters/store/uploaded_transcripts.go @@ -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 +}