Compare commits
5
Commits
38b3f1cf33
...
836a501576
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
836a501576 | ||
|
|
c29c0a7127 | ||
|
|
4b557209c8 | ||
|
|
e36c4c1e0c | ||
|
|
066ae6966a |
@@ -0,0 +1,3 @@
|
|||||||
|
-- Drops the private uploaded-transcript store and every transcript in it. There
|
||||||
|
-- is no revert of the data: the original audio was never kept (ADR-033).
|
||||||
|
DROP TABLE uploaded_transcripts;
|
||||||
@@ -0,0 +1,48 @@
|
|||||||
|
-- Migration 017: uploaded_transcripts — PRIVATE transcripts of audio a user
|
||||||
|
-- uploaded themselves (ADR-033, tapir#33).
|
||||||
|
--
|
||||||
|
-- This is deliberately NOT the shared `transcripts` table (migration 015,
|
||||||
|
-- ADR-021). That table is non-RLS on the explicit grounds that it holds only
|
||||||
|
-- public caption content and nothing user-identifying. A recorded meeting is the
|
||||||
|
-- opposite: private third-party content owned by exactly one user. Reusing
|
||||||
|
-- `transcripts` — or adding a user_id to it with a partial policy — would put a
|
||||||
|
-- private row in the one table the isolation test asserts is shared, and would
|
||||||
|
-- pass CI green. A separate table keeps ADR-021's exception singular.
|
||||||
|
--
|
||||||
|
-- Row-level security follows migration 003 exactly: ENABLE + FORCE (the app
|
||||||
|
-- connects as the table owner, which bypasses RLS unless forced), and a policy
|
||||||
|
-- keyed off the per-request GUC tapir.current_user_id. With no GUC set the
|
||||||
|
-- predicate is NULL and the table is deny-all.
|
||||||
|
--
|
||||||
|
-- The original audio is never stored (ADR-033): this row is all that is kept.
|
||||||
|
-- segments holds the timestamped cues; re-transcribing 90 minutes costs ~7
|
||||||
|
-- minutes, so discarding the timestamps would be irreversible.
|
||||||
|
CREATE TABLE uploaded_transcripts (
|
||||||
|
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||||
|
user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||||
|
title TEXT NOT NULL,
|
||||||
|
status TEXT NOT NULL DEFAULT 'queued'
|
||||||
|
CHECK (status IN ('queued', 'transcribing', 'done', 'failed')),
|
||||||
|
content TEXT,
|
||||||
|
segments JSONB,
|
||||||
|
endpoint TEXT, -- which backend served it (iguana | berget); a quality fact
|
||||||
|
audio_seconds DOUBLE PRECISION,
|
||||||
|
error_kind TEXT, -- typed, never a provider's error string (ADR-033)
|
||||||
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||||
|
finished_at TIMESTAMPTZ,
|
||||||
|
CONSTRAINT uploaded_transcripts_done_has_content
|
||||||
|
CHECK (status <> 'done' OR content IS NOT NULL),
|
||||||
|
CONSTRAINT uploaded_transcripts_failed_has_error_kind
|
||||||
|
CHECK (status <> 'failed' OR error_kind IS NOT NULL)
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE INDEX idx_uploaded_transcripts_user ON uploaded_transcripts (user_id, created_at DESC);
|
||||||
|
|
||||||
|
ALTER TABLE uploaded_transcripts ENABLE ROW LEVEL SECURITY;
|
||||||
|
ALTER TABLE uploaded_transcripts FORCE ROW LEVEL SECURITY;
|
||||||
|
CREATE POLICY uploaded_transcripts_isolation ON uploaded_transcripts
|
||||||
|
FOR ALL
|
||||||
|
USING (user_id = current_setting('tapir.current_user_id', true)::uuid);
|
||||||
|
|
||||||
|
COMMENT ON TABLE uploaded_transcripts IS
|
||||||
|
'PRIVATE per-user transcripts of user-supplied audio. RLS-scoped (ADR-012). Never merge into the shared transcripts table (ADR-021, ADR-033).';
|
||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -0,0 +1,196 @@
|
|||||||
|
package store_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"sort"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/jackc/pgx/v5/pgxpool"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Isolation proof for ADR-033: an uploaded transcript is PRIVATE content owned by
|
||||||
|
// one user, so it lives in its own RLS-scoped table — never in the shared,
|
||||||
|
// non-RLS `transcripts` (ADR-021). These tests drive the database directly (raw
|
||||||
|
// SQL as the non-superuser `app` role) so they observe row-level security itself,
|
||||||
|
// not any application-layer WHERE clause. The superuser pool is used only to seed
|
||||||
|
// and to read pg_catalog; it bypasses RLS and so must never be the pool whose
|
||||||
|
// answers the assertions rest on.
|
||||||
|
|
||||||
|
// seedUploadedTranscript inserts one finished upload for userID as the superuser
|
||||||
|
// (RLS bypassed so both users' rows land regardless of any GUC) and returns its id.
|
||||||
|
// The text is synthetic — this repo is public, and a real transcript must never
|
||||||
|
// appear in it (#33).
|
||||||
|
func seedUploadedTranscript(t *testing.T, super *pgxpool.Pool, userID, text string) string {
|
||||||
|
t.Helper()
|
||||||
|
ctx := context.Background()
|
||||||
|
_, err := super.Exec(ctx, `INSERT INTO users (id) VALUES ($1) ON CONFLICT DO NOTHING`, userID)
|
||||||
|
require.NoError(t, err)
|
||||||
|
var id string
|
||||||
|
require.NoError(t, super.QueryRow(ctx,
|
||||||
|
`INSERT INTO uploaded_transcripts (user_id, title, status, content)
|
||||||
|
VALUES ($1, 'synthetic fixture', 'done', $2) RETURNING id`,
|
||||||
|
userID, text).Scan(&id))
|
||||||
|
return id
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUploadedTranscriptsRLS_ScopedReadsSeeOnlyOwnRows(t *testing.T) {
|
||||||
|
newStore(t)
|
||||||
|
super := rawPool(t)
|
||||||
|
resetDB(t, super)
|
||||||
|
seedUploadedTranscript(t, super, userA, "text owned by A")
|
||||||
|
seedUploadedTranscript(t, super, userB, "text owned by B")
|
||||||
|
app := appPool(t, super)
|
||||||
|
|
||||||
|
require.Equal(t, 0, scopedCount(t, app, "", "uploaded_transcripts"),
|
||||||
|
"no tapir.current_user_id must be deny-all: this proves RLS is live, not bypassed")
|
||||||
|
require.Equal(t, 1, scopedCount(t, app, userA, "uploaded_transcripts"),
|
||||||
|
"user A must see exactly its own upload")
|
||||||
|
require.Equal(t, 1, scopedCount(t, app, userB, "uploaded_transcripts"),
|
||||||
|
"user B must see exactly its own upload")
|
||||||
|
|
||||||
|
// Count alone cannot distinguish "A's row" from "B's row" — read the content.
|
||||||
|
var got string
|
||||||
|
ctx := context.Background()
|
||||||
|
tx, err := app.Begin(ctx)
|
||||||
|
require.NoError(t, err)
|
||||||
|
defer tx.Rollback(ctx) //nolint:errcheck
|
||||||
|
_, err = tx.Exec(ctx, `SELECT set_config('tapir.current_user_id', $1, true)`, userA)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.NoError(t, tx.QueryRow(ctx, `SELECT content FROM uploaded_transcripts`).Scan(&got))
|
||||||
|
require.Equal(t, "text owned by A", got, "A's scoped read must return A's text, never B's")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUploadedTranscriptsRLS_CrossUserWritesAffectZeroRows(t *testing.T) {
|
||||||
|
newStore(t)
|
||||||
|
super := rawPool(t)
|
||||||
|
resetDB(t, super)
|
||||||
|
seedUploadedTranscript(t, super, userA, "text owned by A")
|
||||||
|
bID := seedUploadedTranscript(t, super, userB, "text owned by B")
|
||||||
|
app := appPool(t, super)
|
||||||
|
|
||||||
|
require.Equal(t, int64(0), scopedRowsAffected(t, app, userA,
|
||||||
|
`UPDATE uploaded_transcripts SET content = 'hacked' WHERE id = $1`, bID),
|
||||||
|
"user A scoped UPDATE must touch zero of B's uploads")
|
||||||
|
require.Equal(t, int64(0), scopedRowsAffected(t, app, userA,
|
||||||
|
`DELETE FROM uploaded_transcripts WHERE id = $1`, bID),
|
||||||
|
"user A scoped DELETE must touch zero of B's uploads")
|
||||||
|
|
||||||
|
var content string
|
||||||
|
require.NoError(t, super.QueryRow(context.Background(),
|
||||||
|
`SELECT content FROM uploaded_transcripts WHERE id = $1`, bID).Scan(&content))
|
||||||
|
require.Equal(t, "text owned by B", content, "B's upload must survive A's writes unchanged")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUploadedTranscriptsRLS_InsertForAnotherUserIsRejected(t *testing.T) {
|
||||||
|
newStore(t)
|
||||||
|
super := rawPool(t)
|
||||||
|
resetDB(t, super)
|
||||||
|
seedUploadedTranscript(t, super, userA, "text owned by A")
|
||||||
|
_, err := super.Exec(context.Background(), `INSERT INTO users (id) VALUES ($1)`, userB)
|
||||||
|
require.NoError(t, err)
|
||||||
|
app := appPool(t, super)
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
tx, err := app.Begin(ctx)
|
||||||
|
require.NoError(t, err)
|
||||||
|
defer tx.Rollback(ctx) //nolint:errcheck
|
||||||
|
_, err = tx.Exec(ctx, `SELECT set_config('tapir.current_user_id', $1, true)`, userA)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
// Scoped as A, plant a row that claims to belong to B. The policy's WITH CHECK
|
||||||
|
// (defaulting to USING under FOR ALL) must refuse it.
|
||||||
|
_, err = tx.Exec(ctx,
|
||||||
|
`INSERT INTO uploaded_transcripts (user_id, title, status, content)
|
||||||
|
VALUES ($1, 'planted', 'done', 'planted')`, userB)
|
||||||
|
require.Error(t, err, "a row scoped as A must not be insertable on behalf of B")
|
||||||
|
require.Contains(t, err.Error(), "row-level security")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUploadedTranscripts_RowSecurityIsEnabledForcedAndKeyedByUser(t *testing.T) {
|
||||||
|
newStore(t)
|
||||||
|
super := rawPool(t)
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
var enabled, forced bool
|
||||||
|
require.NoError(t, super.QueryRow(ctx,
|
||||||
|
`SELECT relrowsecurity, relforcerowsecurity FROM pg_class
|
||||||
|
WHERE oid = 'public.uploaded_transcripts'::regclass`).Scan(&enabled, &forced))
|
||||||
|
require.True(t, enabled, "uploaded_transcripts must have row-level security enabled")
|
||||||
|
require.True(t, forced,
|
||||||
|
"FORCE is what binds the table-owner role tapir; without it the policy is dead in production")
|
||||||
|
|
||||||
|
var userIDNotNull bool
|
||||||
|
require.NoError(t, super.QueryRow(ctx,
|
||||||
|
`SELECT is_nullable = 'NO' FROM information_schema.columns
|
||||||
|
WHERE table_name = 'uploaded_transcripts' AND column_name = 'user_id'`).Scan(&userIDNotNull))
|
||||||
|
require.True(t, userIDNotNull, "an upload with no owner would match no policy and be unreachable")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUploadedTranscripts_DeleteUserRemovesTheirUploads(t *testing.T) {
|
||||||
|
s := newStore(t)
|
||||||
|
super := rawPool(t)
|
||||||
|
resetDB(t, super)
|
||||||
|
seedUploadedTranscript(t, super, userA, "text owned by A")
|
||||||
|
bID := seedUploadedTranscript(t, super, userB, "text owned by B")
|
||||||
|
|
||||||
|
require.NoError(t, s.DeleteUser(context.Background(), userA))
|
||||||
|
|
||||||
|
var aRows, bRows int
|
||||||
|
ctx := context.Background()
|
||||||
|
require.NoError(t, super.QueryRow(ctx,
|
||||||
|
`SELECT count(*) FROM uploaded_transcripts WHERE user_id = $1`, userA).Scan(&aRows))
|
||||||
|
require.NoError(t, super.QueryRow(ctx,
|
||||||
|
`SELECT count(*) FROM uploaded_transcripts WHERE id = $1`, bID).Scan(&bRows))
|
||||||
|
require.Equal(t, 0, aRows, "deleting a user must remove their private transcripts (ON DELETE CASCADE)")
|
||||||
|
require.Equal(t, 1, bRows, "deleting A must not touch B's transcripts")
|
||||||
|
}
|
||||||
|
|
||||||
|
// rowSecurityOffTables are the public-schema tables that deliberately have row
|
||||||
|
// security OFF. The list is exact on purpose: a NEW table added without RLS must
|
||||||
|
// fail this test and force a decision, instead of passing CI by omission — the
|
||||||
|
// failure mode ADR-033 names, where a naive reuse "leaks private content across
|
||||||
|
// tenants and passes CI green".
|
||||||
|
//
|
||||||
|
// - transcripts: shared public caption content, the single exception for CONTENT
|
||||||
|
// (ADR-021). It is the one that must never gain a private row.
|
||||||
|
// - user_identities: the Dex-subject -> user_id map, read at login before any
|
||||||
|
// user is known (migration 004). Holds no user content.
|
||||||
|
// - invitations: exists before the user does, so there is no user to scope by;
|
||||||
|
// the single-use token is the capability (migration 009).
|
||||||
|
// - schema_migrations: golang-migrate bookkeeping.
|
||||||
|
//
|
||||||
|
// #33 describes `transcripts` as "the only non-RLS table". That is true of content
|
||||||
|
// and false of tables; this test asserts the real set.
|
||||||
|
var rowSecurityOffTables = []string{
|
||||||
|
"invitations",
|
||||||
|
"schema_migrations",
|
||||||
|
"transcripts",
|
||||||
|
"user_identities",
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRowSecurityOffTablesAreExactlyTheKnownSet(t *testing.T) {
|
||||||
|
newStore(t)
|
||||||
|
super := rawPool(t)
|
||||||
|
|
||||||
|
rows, err := super.Query(context.Background(),
|
||||||
|
`SELECT c.relname FROM pg_class c
|
||||||
|
JOIN pg_namespace n ON n.oid = c.relnamespace
|
||||||
|
WHERE n.nspname = 'public' AND c.relkind IN ('r', 'p') AND NOT c.relrowsecurity
|
||||||
|
ORDER BY c.relname`)
|
||||||
|
require.NoError(t, err)
|
||||||
|
defer rows.Close()
|
||||||
|
var got []string
|
||||||
|
for rows.Next() {
|
||||||
|
var name string
|
||||||
|
require.NoError(t, rows.Scan(&name))
|
||||||
|
got = append(got, name)
|
||||||
|
}
|
||||||
|
require.NoError(t, rows.Err())
|
||||||
|
|
||||||
|
want := append([]string{}, rowSecurityOffTables...)
|
||||||
|
sort.Strings(want)
|
||||||
|
require.Equal(t, want, got,
|
||||||
|
"every public table must have row security ON unless it is in rowSecurityOffTables; "+
|
||||||
|
"adding a table here is a decision that needs an ADR")
|
||||||
|
}
|
||||||
@@ -0,0 +1,188 @@
|
|||||||
|
package store_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
|
||||||
|
"git.d-ma.be/mathias/tapir/internal/adapters/store"
|
||||||
|
"git.d-ma.be/mathias/tapir/internal/domain"
|
||||||
|
)
|
||||||
|
|
||||||
|
// These exercise the Store API over uploaded_transcripts. The test Store connects
|
||||||
|
// as the superuser, which BYPASSES row-level security — so the cross-user cases
|
||||||
|
// here prove the query's own user_id predicate (defence in depth), while the
|
||||||
|
// database-enforced guarantee is proven separately, as the non-superuser app
|
||||||
|
// role, in uploaded_transcripts_isolation_test.go. Text is synthetic; this repo is
|
||||||
|
// public and no real transcript may appear in it (#33).
|
||||||
|
|
||||||
|
func seedUsers(t *testing.T, ids ...string) {
|
||||||
|
t.Helper()
|
||||||
|
super := rawPool(t)
|
||||||
|
resetDB(t, super)
|
||||||
|
for _, id := range ids {
|
||||||
|
_, err := super.Exec(context.Background(), `INSERT INTO users (id) VALUES ($1)`, id)
|
||||||
|
require.NoError(t, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCreateUpload_StartsQueuedAndIsOwnedByTheCreator(t *testing.T) {
|
||||||
|
s := newStore(t)
|
||||||
|
seedUsers(t, userA)
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
id, err := s.CreateUpload(ctx, userA, "Teams export")
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
got, err := s.GetUpload(ctx, userA, id)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, id, got.ID)
|
||||||
|
assert.Equal(t, userA, got.UserID)
|
||||||
|
assert.Equal(t, "Teams export", got.Title)
|
||||||
|
assert.Equal(t, domain.UploadQueued, got.Status)
|
||||||
|
assert.Empty(t, got.Content)
|
||||||
|
assert.True(t, got.FinishedAt.IsZero(), "an unfinished upload has no finish time")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUploadLifecycle_CompletedKeepsTextSegmentsEndpointAndDuration(t *testing.T) {
|
||||||
|
s := newStore(t)
|
||||||
|
seedUsers(t, userA)
|
||||||
|
ctx := context.Background()
|
||||||
|
id, err := s.CreateUpload(ctx, userA, "m")
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
require.NoError(t, s.StartTranscribing(ctx, userA, id))
|
||||||
|
mid, err := s.GetUpload(ctx, userA, id)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, domain.UploadTranscribing, mid.Status)
|
||||||
|
|
||||||
|
want := domain.Transcription{
|
||||||
|
Text: "God morgon allihopa. Vi går igenom läget.",
|
||||||
|
Segments: []domain.Segment{
|
||||||
|
{Start: 0.0, End: 1.5, Text: "God morgon allihopa."},
|
||||||
|
{Start: 1.5, End: 4.25, Text: "Vi går igenom läget."},
|
||||||
|
},
|
||||||
|
Endpoint: "iguana/kb-whisper",
|
||||||
|
AudioSeconds: 4.25,
|
||||||
|
}
|
||||||
|
require.NoError(t, s.CompleteUpload(ctx, userA, id, want))
|
||||||
|
|
||||||
|
got, err := s.GetUpload(ctx, userA, id)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, domain.UploadDone, got.Status)
|
||||||
|
assert.Equal(t, want.Text, got.Content)
|
||||||
|
assert.Equal(t, want.Segments, got.Segments, "segment times must survive the JSONB round trip")
|
||||||
|
assert.Equal(t, "iguana/kb-whisper", got.Endpoint, "the serving endpoint is a quality fact and is recorded per row")
|
||||||
|
assert.InDelta(t, 4.25, got.AudioSeconds, 1e-9)
|
||||||
|
assert.False(t, got.FinishedAt.IsZero())
|
||||||
|
assert.Empty(t, got.ErrorKind)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUploadLifecycle_FailedRecordsTypedKindAndNoContent(t *testing.T) {
|
||||||
|
s := newStore(t)
|
||||||
|
seedUsers(t, userA)
|
||||||
|
ctx := context.Background()
|
||||||
|
id, err := s.CreateUpload(ctx, userA, "m")
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.NoError(t, s.StartTranscribing(ctx, userA, id))
|
||||||
|
|
||||||
|
require.NoError(t, s.FailUpload(ctx, userA, id, domain.UploadErrTranscriptionFailed))
|
||||||
|
|
||||||
|
got, err := s.GetUpload(ctx, userA, id)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, domain.UploadFailed, got.Status)
|
||||||
|
assert.Equal(t, domain.UploadErrTranscriptionFailed, got.ErrorKind)
|
||||||
|
assert.Empty(t, got.Content)
|
||||||
|
assert.False(t, got.FinishedAt.IsZero())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestGetUpload_AnotherUsersIDIsNotFound(t *testing.T) {
|
||||||
|
s := newStore(t)
|
||||||
|
seedUsers(t, userA, userB)
|
||||||
|
ctx := context.Background()
|
||||||
|
id, err := s.CreateUpload(ctx, userA, "A's meeting")
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
_, err = s.GetUpload(ctx, userB, id)
|
||||||
|
|
||||||
|
assert.ErrorIs(t, err, store.ErrUploadNotFound,
|
||||||
|
"a non-owner must get not-found, not forbidden: existence is not disclosed")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUploadWrites_ByAnotherUserChangeNothing(t *testing.T) {
|
||||||
|
s := newStore(t)
|
||||||
|
seedUsers(t, userA, userB)
|
||||||
|
ctx := context.Background()
|
||||||
|
id, err := s.CreateUpload(ctx, userA, "A's meeting")
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
assert.ErrorIs(t, s.StartTranscribing(ctx, userB, id), store.ErrUploadNotFound)
|
||||||
|
assert.ErrorIs(t, s.CompleteUpload(ctx, userB, id, domain.Transcription{Text: "planted"}), store.ErrUploadNotFound)
|
||||||
|
assert.ErrorIs(t, s.FailUpload(ctx, userB, id, domain.UploadErrTranscriptionFailed), store.ErrUploadNotFound)
|
||||||
|
|
||||||
|
got, err := s.GetUpload(ctx, userA, id)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, domain.UploadQueued, got.Status, "B's writes must leave A's upload exactly as it was")
|
||||||
|
assert.Empty(t, got.Content)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestListUploads_ReturnsOnlyTheCallersNewestFirstWithoutTranscriptText(t *testing.T) {
|
||||||
|
s := newStore(t)
|
||||||
|
seedUsers(t, userA, userB)
|
||||||
|
ctx := context.Background()
|
||||||
|
first, err := s.CreateUpload(ctx, userA, "first")
|
||||||
|
require.NoError(t, err)
|
||||||
|
second, err := s.CreateUpload(ctx, userA, "second")
|
||||||
|
require.NoError(t, err)
|
||||||
|
_, err = s.CreateUpload(ctx, userB, "B's")
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.NoError(t, s.StartTranscribing(ctx, userA, first))
|
||||||
|
require.NoError(t, s.CompleteUpload(ctx, userA, first, domain.Transcription{Text: "long text body", Endpoint: "e"}))
|
||||||
|
|
||||||
|
got, err := s.ListUploads(ctx, userA)
|
||||||
|
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.Len(t, got, 2, "A must see only A's uploads")
|
||||||
|
assert.Equal(t, second, got[0].ID, "newest first")
|
||||||
|
assert.Equal(t, first, got[1].ID)
|
||||||
|
assert.Empty(t, got[1].Content, "a list row must not drag the whole transcript along")
|
||||||
|
assert.Equal(t, domain.UploadDone, got[1].Status)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestFailInterruptedUploads_FailsOnlyUnfinishedRowsForEveryUser(t *testing.T) {
|
||||||
|
s := newStore(t)
|
||||||
|
seedUsers(t, userA, userB)
|
||||||
|
super := rawPool(t)
|
||||||
|
ctx := context.Background()
|
||||||
|
for subject, uid := range map[string]string{"sub-a": userA, "sub-b": userB} {
|
||||||
|
_, err := super.Exec(ctx, `INSERT INTO user_identities (dex_subject, user_id) VALUES ($1, $2)`, subject, uid)
|
||||||
|
require.NoError(t, err)
|
||||||
|
}
|
||||||
|
queued, err := s.CreateUpload(ctx, userA, "queued")
|
||||||
|
require.NoError(t, err)
|
||||||
|
running, err := s.CreateUpload(ctx, userB, "running")
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.NoError(t, s.StartTranscribing(ctx, userB, running))
|
||||||
|
done, err := s.CreateUpload(ctx, userA, "done")
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.NoError(t, s.StartTranscribing(ctx, userA, done))
|
||||||
|
require.NoError(t, s.CompleteUpload(ctx, userA, done, domain.Transcription{Text: "kept", Endpoint: "e"}))
|
||||||
|
|
||||||
|
n, err := s.FailInterruptedUploads(ctx)
|
||||||
|
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, 2, n, "the queued and the transcribing row, across both users")
|
||||||
|
for _, c := range []struct{ user, id string }{{userA, queued}, {userB, running}} {
|
||||||
|
got, err := s.GetUpload(ctx, c.user, c.id)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, domain.UploadFailed, got.Status)
|
||||||
|
assert.Equal(t, domain.UploadErrInterrupted, got.ErrorKind,
|
||||||
|
"audio is not retained, so a restart cannot resume: the honest state is failed/interrupted")
|
||||||
|
}
|
||||||
|
kept, err := s.GetUpload(ctx, userA, done)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, domain.UploadDone, kept.Status, "a finished transcript must survive the sweep")
|
||||||
|
assert.Equal(t, "kept", kept.Content)
|
||||||
|
}
|
||||||
@@ -111,3 +111,62 @@ type Summary struct {
|
|||||||
FallbackUsed bool
|
FallbackUsed bool
|
||||||
CreatedAt time.Time
|
CreatedAt time.Time
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Segment is one timed span of a transcription. Times are seconds from the start
|
||||||
|
// of the audio.
|
||||||
|
type Segment struct {
|
||||||
|
Start float64 `json:"start"`
|
||||||
|
End float64 `json:"end"`
|
||||||
|
Text string `json:"text"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// Transcription is the result of transcribing user-supplied audio (ADR-033):
|
||||||
|
// the full text, its timed segments, which backend served it, and how long the
|
||||||
|
// audio was. Endpoint is recorded because the backends differ in timestamp
|
||||||
|
// granularity — which one served a transcript is a quality fact.
|
||||||
|
type Transcription struct {
|
||||||
|
Text string
|
||||||
|
Segments []Segment
|
||||||
|
Endpoint string
|
||||||
|
AudioSeconds float64
|
||||||
|
}
|
||||||
|
|
||||||
|
// UploadStatus is the state of an uploaded-audio transcription job.
|
||||||
|
type UploadStatus string
|
||||||
|
|
||||||
|
const (
|
||||||
|
UploadQueued UploadStatus = "queued"
|
||||||
|
UploadTranscribing UploadStatus = "transcribing"
|
||||||
|
UploadDone UploadStatus = "done"
|
||||||
|
UploadFailed UploadStatus = "failed"
|
||||||
|
)
|
||||||
|
|
||||||
|
// UploadErrorKind is the typed reason an upload failed. It is deliberately NOT a
|
||||||
|
// provider's error string: the whisper server's "failed to process audio" has had
|
||||||
|
// at least five distinct causes, so nothing may branch on it (ADR-033).
|
||||||
|
type UploadErrorKind string
|
||||||
|
|
||||||
|
const (
|
||||||
|
// UploadErrInterrupted: the process restarted mid-job. The audio is not
|
||||||
|
// retained, so the job cannot resume; the user re-uploads.
|
||||||
|
UploadErrInterrupted UploadErrorKind = "interrupted"
|
||||||
|
// UploadErrTranscriptionFailed: every configured transcription endpoint failed.
|
||||||
|
UploadErrTranscriptionFailed UploadErrorKind = "transcription_failed"
|
||||||
|
)
|
||||||
|
|
||||||
|
// UploadedTranscript is a user's PRIVATE transcript of audio they uploaded. It is
|
||||||
|
// never stored in, or merged with, the shared public-caption Transcript (ADR-021).
|
||||||
|
// Content and Segments are empty until Status is UploadDone.
|
||||||
|
type UploadedTranscript struct {
|
||||||
|
ID string
|
||||||
|
UserID string
|
||||||
|
Title string
|
||||||
|
Status UploadStatus
|
||||||
|
Content string
|
||||||
|
Segments []Segment
|
||||||
|
Endpoint string
|
||||||
|
AudioSeconds float64
|
||||||
|
ErrorKind UploadErrorKind
|
||||||
|
CreatedAt time.Time
|
||||||
|
FinishedAt time.Time // zero until the job reaches done or failed
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user