tapir/internal/adapters/store/uploaded_transcripts.go

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.
// hadVideo records that the file carried a video track (only its audio is used);
// it is kept for the life of the row so the page can say so.
func (s *Store) CreateUpload(ctx context.Context, userID, title string, hadVideo bool) (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, had_video) VALUES ($1, $2, $3) RETURNING id`,
			userID, title, hadVideo).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)
	}
	// A nil Reconciliation must be SQL NULL, not the JSON text "null": the page decides
	// whether to claim a second opinion by whether this column is NULL.
	var rec []byte
	if t.Reconciliation != nil {
		if rec, err = json.Marshal(t.Reconciliation); err != nil {
			return fmt.Errorf("store: complete upload: marshal reconciliation: %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, reconciliation = $7, error_kind = NULL, finished_at = NOW()
		  WHERE id = $1 AND user_id = $2`,
		t.Text, segs, t.Endpoint, t.AudioSeconds, rec)
}

// 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, had_video, reconciliation, 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, reconciliation           []byte
		seconds                            *float64
		finished                           *time.Time
	)
	if err := row.Scan(&u.ID, &u.UserID, &u.Title, &status, &content, &segments, &endpoint,
		&seconds, &errKind, &u.HadVideo, &reconciliation, &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
	}
	if len(reconciliation) > 0 {
		var r domain.Reconciliation
		if err := json.Unmarshal(reconciliation, &r); err != nil {
			return fmt.Errorf("unmarshal reconciliation: %w", err)
		}
		u.Reconciliation = &r
	}
	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
}