ultimate-media-tracker/server/store.go

package main

import (
	"context"
	"database/sql"
	"errors"
	"time"

	_ "github.com/jackc/pgx/v5/stdlib"
)

// errRevisionConflict is returned by putBlob when the caller's baseRevision no
// longer matches the revision currently stored for that user. The client must
// re-pull and resolve before pushing again (optimistic concurrency).
var errRevisionConflict = errors.New("revision conflict")

// libraryBlob is the encrypted, opaque library snapshot stored for one user.
// The server never sees the plaintext or the passphrase — payload is ciphertext.
type libraryBlob struct {
	Revision  int64
	UpdatedAt time.Time
	Payload   []byte
}

// store wraps the PostgreSQL connection used for the zero-knowledge library
// sync. It is optional: when DATABASE_URL is unset the server runs as a plain
// TMDB proxy and store is nil.
type store struct {
	db *sql.DB
}

// openStore connects to PostgreSQL and ensures the schema exists.
func openStore(dsn string) (*store, error) {
	db, err := sql.Open("pgx", dsn)
	if err != nil {
		return nil, err
	}
	db.SetMaxOpenConns(10)
	db.SetConnMaxIdleTime(5 * time.Minute)

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if err := db.PingContext(ctx); err != nil {
		db.Close()
		return nil, err
	}
	if err := migrate(ctx, db); err != nil {
		db.Close()
		return nil, err
	}
	return &store{db: db}, nil
}

func (s *store) Close() error { return s.db.Close() }

func migrate(ctx context.Context, db *sql.DB) error {
	_, err := db.ExecContext(ctx, `
CREATE TABLE IF NOT EXISTS library_blobs (
    user_subject TEXT PRIMARY KEY,
    user_email   TEXT,
    revision     BIGINT NOT NULL,
    updated_at   TIMESTAMPTZ NOT NULL,
    payload      BYTEA NOT NULL
);`)
	return err
}

// getBlob loads the stored snapshot for a user. It returns (nil, nil) when the
// user has never pushed a library yet.
func (s *store) getBlob(ctx context.Context, subject string) (*libraryBlob, error) {
	row := s.db.QueryRowContext(ctx,
		`SELECT revision, updated_at, payload FROM library_blobs WHERE user_subject = $1`,
		subject)
	var b libraryBlob
	switch err := row.Scan(&b.Revision, &b.UpdatedAt, &b.Payload); err {
	case nil:
		return &b, nil
	case sql.ErrNoRows:
		return nil, nil
	default:
		return nil, err
	}
}

// putBlob writes a new snapshot using optimistic concurrency: the write only
// succeeds when the stored revision equals baseRevision (or the row is absent
// and baseRevision is 0). On success the revision is incremented and returned.
// On mismatch it returns the current revision and errRevisionConflict.
func (s *store) putBlob(ctx context.Context, subject, email string, baseRevision int64, payload []byte) (int64, error) {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return 0, err
	}
	defer tx.Rollback()

	var current int64
	var exists bool
	row := tx.QueryRowContext(ctx,
		`SELECT revision FROM library_blobs WHERE user_subject = $1 FOR UPDATE`, subject)
	switch err := row.Scan(&current); err {
	case nil:
		exists = true
	case sql.ErrNoRows:
		exists = false
	default:
		return 0, err
	}

	if exists {
		if current != baseRevision {
			return current, errRevisionConflict
		}
	} else if baseRevision != 0 {
		// Client thinks it is updating an existing snapshot, but none exists.
		return 0, errRevisionConflict
	}

	next := baseRevision + 1
	now := time.Now().UTC()
	if exists {
		_, err = tx.ExecContext(ctx,
			`UPDATE library_blobs SET user_email = $2, revision = $3, updated_at = $4, payload = $5 WHERE user_subject = $1`,
			subject, email, next, now, payload)
	} else {
		_, err = tx.ExecContext(ctx,
			`INSERT INTO library_blobs (user_subject, user_email, revision, updated_at, payload) VALUES ($1, $2, $3, $4, $5)`,
			subject, email, next, now, payload)
	}
	if err != nil {
		return 0, err
	}
	if err := tx.Commit(); err != nil {
		return 0, err
	}
	return next, nil
}