package pgstore import ( "context" "errors" "fmt" "net" "strings" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" "textmachine/platform/internal/money" ) var ( // ErrInsufficientCredit is a refusal, not a failure: the account has less than the run needs. ErrInsufficientCredit = errors.New("pgstore: insufficient credit") // ErrNoReservation means the hold this settlement refers to is not open. ErrNoReservation = errors.New("pgstore: no open reservation") // ErrDuplicateHold is a second hold on an attempt id that already has one. It is an error, not // a no-op: a hold that debits nothing reserves nothing while reporting that it did. ErrDuplicateHold = errors.New("pgstore: attempt already has a hold") // ErrNotOwner is a book that does not belong to the account being charged for it. ErrNotOwner = errors.New("pgstore: book belongs to another account") // ErrNoAccount separates "this account has nothing" from "this account does not exist" — the // difference between a balance of zero and a typo in an admin command. ErrNoAccount = errors.New("pgstore: no such account") ) // Grant credits an account and reports whether this call is what credited it. The free tier is one // of these and nothing more. // // (source, sourceID) is the idempotency key, scoped to the account by the schema. applied is false // when the key was already spent: the caller must say so rather than print a success it did not // cause. func (s *Store) Grant(ctx context.Context, userID string, amount money.MicroUSD, source, sourceID, note string, now time.Time) (applied bool, err error) { if amount <= 0 { return false, fmt.Errorf("pgstore: grant must be positive, got %d", amount) } if source == "" || sourceID == "" { return false, errors.New("pgstore: grant needs an idempotency key") } err = s.inTx(ctx, func(tx pgx.Tx) error { applied, err = appendLedger(ctx, tx, userID, "grant", amount, source, sourceID, note, now) return err }) return applied, err } // Adjust corrects a balance. A ledger row is never edited: the correction is another row, which is // what keeps the sum reproducible. The note is mandatory, in the DDL as well as here. func (s *Store) Adjust(ctx context.Context, userID string, amount money.MicroUSD, source, sourceID, note string, now time.Time) (applied bool, err error) { if amount == 0 || note == "" { return false, errors.New("pgstore: an adjustment needs a non-zero amount and a reason") } if source == "" || sourceID == "" { return false, errors.New("pgstore: adjustment needs an idempotency key") } err = s.inTx(ctx, func(tx pgx.Tx) error { applied, err = appendLedger(ctx, tx, userID, "adjustment", amount, source, sourceID, note, now) return err }) return applied, err } // Balance is what the account may still spend, read from the cache that every ledger write updates // in its own transaction. func (s *Store) Balance(ctx context.Context, userID string) (money.MicroUSD, error) { var v *int64 err := s.pool.QueryRow(ctx, ` select b.balance_micro_usd from users u left join account_balances b on b.user_id = u.id where u.id = $1`, userID).Scan(&v) if errors.Is(err, pgx.ErrNoRows) { return 0, ErrNoAccount } if err != nil { return 0, fmt.Errorf("pgstore: balance: %w", err) } if v == nil { return 0, nil // an account with no ledger rows has no credit, which is not an error } return money.MicroUSD(*v), nil } // Account is what an operator needs to see about one account's money. type Account struct { Balance money.MicroUSD Reserved money.MicroUSD // LedgerSum is recomputed from the rows. It exists to be COMPARED with Balance, and both are // read in one snapshot: reading them separately reports drift that a concurrent grant caused // between the two queries. LedgerSum money.MicroUSD } // ReadAccount returns the money view in a single consistent snapshot. func (s *Store) ReadAccount(ctx context.Context, userID string) (Account, error) { var a Account const q = ` select coalesce((select balance_micro_usd from account_balances where user_id = $1), 0), coalesce((select sum(amount_micro_usd) from credit_ledger where user_id = $1), 0), coalesce((select sum(amount_micro_usd) from reservations where user_id = $1 and state = 'open'), 0) from users where id = $1` err := s.pool.QueryRow(ctx, q, userID).Scan(&a.Balance, &a.LedgerSum, &a.Reserved) if errors.Is(err, pgx.ErrNoRows) { return Account{}, ErrNoAccount } if err != nil { return Account{}, fmt.Errorf("pgstore: read account: %w", err) } return a, nil } // Reservation is an open hold as an operator sees it. type Reservation struct { EngineRunID string BookID string Amount money.MicroUSD Ceiling money.MicroUSD OpenedAt time.Time } // OpenReservations lists holds that were taken and never closed. Without this they are money that // is gone from the balance and invisible to everything that could give it back. func (s *Store) OpenReservations(ctx context.Context, userID string) ([]Reservation, error) { const q = ` select engine_run_id, book_id, amount_micro_usd, ceiling_micro_usd, opened_at from reservations where user_id = $1 and state = 'open' order by opened_at` rows, err := s.pool.Query(ctx, q, userID) if err != nil { return nil, fmt.Errorf("pgstore: open reservations: %w", err) } defer rows.Close() var out []Reservation for rows.Next() { var r Reservation if err := rows.Scan(&r.EngineRunID, &r.BookID, &r.Amount, &r.Ceiling, &r.OpenedAt); err != nil { return nil, fmt.Errorf("pgstore: scan reservation: %w", err) } out = append(out, r) } return out, rows.Err() } // Hold reserves credit before a run is spawned. Together with the per-book ceiling handed to the // engine it is the enforcement half of the money design: the hold makes the credit unavailable to // the next run, and the engine stops itself at the ceiling, so an overspend is impossible even // while the platform is blind. The event stream is freshness only. // // The ceiling to hand the engine is the amount held; read it back with OpenReservations. func (s *Store) Hold(ctx context.Context, userID, bookID, engineRunID string, amount money.MicroUSD, now time.Time) error { return s.inTx(ctx, func(tx pgx.Tx) error { return holdTx(ctx, tx, userID, bookID, engineRunID, amount, now) }) } // holdTx is Hold inside a caller's transaction. It exists because admitting a run writes the run // row, its attempt, the hold and the queue entry together or not at all (StartRun): a hold in its // own transaction is a hold that can outlive the thing it was taken for. func holdTx(ctx context.Context, tx pgx.Tx, userID, bookID, engineRunID string, amount money.MicroUSD, now time.Time) error { if amount <= 0 { return fmt.Errorf("pgstore: hold must be positive, got %d", amount) } // account_balances is locked FIRST here and in every other operation. A path that locked // the reservation first would invert the order against this one and deadlock — measured, // not hypothetical. balance, err := lockBalance(ctx, tx, userID) if err != nil { return err } if balance < amount { return ErrInsufficientCredit } // The book must belong to the account being charged. The foreign key only proves the book // exists, which is not the same question. tag, err := tx.Exec(ctx, ` insert into reservations (engine_run_id, user_id, book_id, amount_micro_usd, ceiling_micro_usd, state, opened_at) select $1, $2, $3, $4, $4, 'open', $5 from books where id = $3 and owner_id = $2`, engineRunID, userID, bookID, int64(amount), now) if err != nil { // A LIVE reservation on this attempt collides on the primary key, and that is the ordinary // way a duplicate hold happens — the declared ErrDuplicateHold below is reachable only in the // rarer shape where the reservation row was removed and the ledger key was not. Both are the // same fact to a caller, and a raw SQLSTATE is not something a worker can act on (PD-81). if isUnique(err, "reservations_pkey") { return ErrDuplicateHold } return fmt.Errorf("pgstore: open reservation: %w", err) } if tag.RowsAffected() == 0 { return ErrNotOwner } applied, err := appendLedger(ctx, tx, userID, "hold", -amount, "run", engineRunID, "", now) if err != nil { return err } if !applied { // The ledger already holds this attempt id, so nothing was debited. Reporting success // would spawn a run against credit that was never reserved. return ErrDuplicateHold } return nil } // Settle closes a reservation with what the attempt actually cost: the hold comes back and the real // cost is charged, in one transaction. Settling twice is refused — the reservation is no longer // open — which is what makes it safe on a retried path. // // A cost above the hold is CAPPED at the hold and the row says so. Spending more than was reserved // means the engine's ceiling did not hold, and the account is not the place to absorb that. func (s *Store) Settle(ctx context.Context, engineRunID string, spent money.MicroUSD, now time.Time) error { if spent < 0 { return fmt.Errorf("pgstore: spend cannot be negative, got %d", spent) } return s.inTx(ctx, func(tx pgx.Tx) error { userID, held, err := closeReservation(ctx, tx, engineRunID, "settled", now) if err != nil { return err } note := "" if spent > held { note = fmt.Sprintf("capped at the hold; the engine reported %s", spent.USD()) spent = held } if err := releaseHold(ctx, tx, userID, engineRunID, held, now); err != nil { return err } _, err = appendLedger(ctx, tx, userID, "settlement", -spent, "run_settle", engineRunID, note, now) return err }) } // ErrReleaseKeySpent is a reservation whose release key has already been used. It means the money // came back once and the reservation was opened again on the same attempt id — an out-of-band // sequence, and the only one where closing the reservation would return nothing (PD-97). Reported // instead of applied, so the transaction rolls back and the row stays open for a human. var ErrReleaseKeySpent = errors.New("pgstore: the release of this attempt was already posted") // releaseHold puts the reserved amount back and refuses to pretend it did when the key was spent. func releaseHold(ctx context.Context, tx pgx.Tx, userID, engineRunID string, held money.MicroUSD, now time.Time) error { applied, err := appendLedger(ctx, tx, userID, "hold_release", held, "run_release", engineRunID, "", now) if err != nil { return err } if !applied { return fmt.Errorf("%w: %s", ErrReleaseKeySpent, engineRunID) } return nil } // Release gives a reservation back untouched: the run never started, or it cost nothing. func (s *Store) Release(ctx context.Context, engineRunID string, now time.Time) error { return s.inTx(ctx, func(tx pgx.Tx) error { userID, held, err := closeReservation(ctx, tx, engineRunID, "released", now) if err != nil { return err } return releaseHold(ctx, tx, userID, engineRunID, held, now) }) } // ErrAttemptSpawned is an attempt that has a unit, refused where the caller believed it had none. var ErrAttemptSpawned = errors.New("pgstore: the attempt was spawned") // ReleaseUnspawned is Release for the one caller that has to be sure: the reconciler settling an // attempt it believes never started a process, whose hold therefore comes back WHOLE. // // The belief comes from a snapshot the sweep took at its start, and between that read and this write // the queue worker can spawn the attempt, the engine can spend and the unit can exit — which is not // a corner case but the ordinary shape of a fast run in a sweep with other runs ahead of it in the // list. Measured: an attempt that spent $0.50 was charged $0.000000 and the whole hold went back, an // underpayment nothing later looks for. So the belief is re-checked HERE, in the same transaction as // the money, against the row rather than against the snapshot. func (s *Store) ReleaseUnspawned(ctx context.Context, engineRunID string, attemptID int64, now time.Time) error { return s.inTx(ctx, func(tx pgx.Tx) error { var unit *string err := tx.QueryRow(ctx, `select unit_name from run_attempts where id = $1 for update`, attemptID).Scan(&unit) if errors.Is(err, pgx.ErrNoRows) { return ErrNoRun } if err != nil { return fmt.Errorf("pgstore: read attempt: %w", err) } if unit != nil { return fmt.Errorf("%w: %s", ErrAttemptSpawned, *unit) } userID, held, err := closeReservation(ctx, tx, engineRunID, "released", now) if err != nil { return err } return releaseHold(ctx, tx, userID, engineRunID, held, now) }) } // lockBalance takes the account's row lock and returns the balance under it. Every money operation // starts here, so they all take their locks in the same order. func lockBalance(ctx context.Context, tx pgx.Tx, userID string) (money.MicroUSD, error) { var v int64 err := tx.QueryRow(ctx, `select balance_micro_usd from account_balances where user_id = $1 for update`, userID).Scan(&v) if errors.Is(err, pgx.ErrNoRows) { // No balance row is TWO different facts, and answering both with "insufficient credit" told // an operator who mistyped an account id that the account was broke (PD-82). The extra query // runs only on this path, and only the nullable-side lock it replaces would have been free — // Postgres refuses FOR UPDATE on the nullable side of an outer join, so there is no one-query // form of this question. var exists bool switch err := tx.QueryRow(ctx, `select true from users where id = $1`, userID).Scan(&exists); { case errors.Is(err, pgx.ErrNoRows): return 0, ErrNoAccount case err != nil: return 0, fmt.Errorf("pgstore: read account: %w", err) } return 0, ErrInsufficientCredit // the account exists and has no ledger rows: no credit } if err != nil { return 0, fmt.Errorf("pgstore: read balance: %w", err) } return money.MicroUSD(v), nil } func closeReservation(ctx context.Context, tx pgx.Tx, engineRunID, state string, now time.Time) (string, money.MicroUSD, error) { // Read the owner unlocked, lock the balance, then close under the state guard. The guard is // what makes the unlocked read safe: a reservation closed by someone else in between makes the // update match nothing. var userID string err := tx.QueryRow(ctx, `select user_id from reservations where engine_run_id = $1`, engineRunID).Scan(&userID) if errors.Is(err, pgx.ErrNoRows) { return "", 0, ErrNoReservation } if err != nil { return "", 0, fmt.Errorf("pgstore: find reservation: %w", err) } if _, err := lockBalance(ctx, tx, userID); err != nil && !errors.Is(err, ErrInsufficientCredit) { return "", 0, err } var amount int64 err = tx.QueryRow(ctx, ` update reservations set state = $2, closed_at = $3 where engine_run_id = $1 and state = 'open' returning amount_micro_usd`, engineRunID, state, now).Scan(&amount) if errors.Is(err, pgx.ErrNoRows) { return "", 0, ErrNoReservation } if err != nil { return "", 0, fmt.Errorf("pgstore: close reservation: %w", err) } return userID, money.MicroUSD(amount), nil } // appendLedger writes one row and moves the cached balance with it, in the caller's transaction. // The two are never written apart: a cache that can lag its source is a second answer about money. // applied is false when the idempotency key was already spent. func appendLedger(ctx context.Context, tx pgx.Tx, userID, kind string, amount money.MicroUSD, source, sourceID, note string, now time.Time) (bool, error) { tag, err := tx.Exec(ctx, ` insert into credit_ledger (user_id, kind, amount_micro_usd, source, source_id, note, created_at) values ($1, $2, $3, $4, $5, $6, $7) on conflict (user_id, source, source_id) do nothing`, userID, kind, int64(amount), source, sourceID, note, now) if err != nil { // A typo in an account id is the commonest way an operator gets here, and Balance already // answers it with ErrNoAccount. Reporting the same fact as a raw constraint name reads as a // broken database (PD-56). var pg *pgconn.PgError if errors.As(err, &pg) && pg.ConstraintName == "credit_ledger_user_id_fkey" { return false, ErrNoAccount } return false, fmt.Errorf("pgstore: append ledger: %w", err) } if tag.RowsAffected() == 0 { return false, nil } if _, err := tx.Exec(ctx, ` insert into account_balances (user_id, balance_micro_usd, updated_at) values ($1, $2, $3) on conflict (user_id) do update set balance_micro_usd = account_balances.balance_micro_usd + excluded.balance_micro_usd, updated_at = excluded.updated_at`, userID, int64(amount), now); err != nil { return false, fmt.Errorf("pgstore: update balance: %w", err) } return true, nil } // lockBook takes the book's row lock, which is the FIRST lock of every transaction in this package // that touches more than one of the tables below. // // The order is: books → runs → run_attempts → account_balances → reservations. It is written down // here because it is not derivable from any one function — a transaction that takes two of these in // the other order deadlocks with a transaction that takes them in this one, and Postgres resolves // that by aborting a side, which costs a sweep or an API call rather than corrupting anything. That // is exactly what happened while the materializer locked the attempt first and the reconciler locked // the book first: 258 of 300 concurrent pairs aborted. // // A missing book is not an error here: the caller's own statements find nothing and decide what that // means. Locking is not the place to invent a not-found policy. func lockBook(ctx context.Context, tx pgx.Tx, bookID string) error { var id string err := tx.QueryRow(ctx, `select id from books where id = $1 for update`, bookID).Scan(&id) if err != nil && !errors.Is(err, pgx.ErrNoRows) { return fmt.Errorf("pgstore: lock book: %w", err) } return nil } // IsTransient reports whether an error means "run the same transaction again" rather than "this // write must never be attempted again". // // It exists because the two used to be conflated: any error out of the materializer quarantined the // attempt, so one aborted deadlock blinded the projection of a live, paying run permanently. The // first version of this function knew only about deadlocks, and a re-check found the commoner half — // a Postgres RESTART, which is a planned event on any managed database, arrives as 57P01 or as a // broken connection, and left the same permanent blindness behind. // // Retrying is safe for every caller of this: the materializer is idempotent by its high-water mark, // so a transaction that may or may not have committed is re-read and re-checked rather than // re-applied. // // - class 08 — connection exception, including 08006 "connection failure"; // - 57P01/57P02/57P03 — the server is shutting down, crashed a sibling, or is not accepting // connections yet, which is precisely what a rolling restart looks like from here; // - 40001/40P01 — serialization failure and a deadlock Postgres broke; // - anything pgx itself marks safe to retry, and any net.Error: a reset connection is not a // journal this platform cannot read. File errors are *fs.PathError and do NOT satisfy net.Error, // so a malformed journal still stops the projection as it must. func IsTransient(err error) bool { if err == nil { return false } var pg *pgconn.PgError if errors.As(err, &pg) { switch { case pg.Code == "40001", pg.Code == "40P01": return true case strings.HasPrefix(pg.Code, "08"), strings.HasPrefix(pg.Code, "57P"): return true } return false } if pgconn.SafeToRetry(err) { return true } var ne net.Error return errors.As(err, &ne) } func (s *Store) inTx(ctx context.Context, fn func(pgx.Tx) error) error { tx, err := s.pool.Begin(ctx) if err != nil { return fmt.Errorf("pgstore: begin: %w", err) } defer func() { _ = tx.Rollback(ctx) }() if err := fn(tx); err != nil { return err } if err := tx.Commit(ctx); err != nil { return fmt.Errorf("pgstore: commit: %w", err) } return nil } // DeleteBook removes a book and the CLOSED reservations that referenced it. An OPEN one blocks the // delete (the foreign key is RESTRICT), which is the point: removing a book with money reserved // against it would leave the hold in the ledger with nothing left to release it. // // Closed reservations carry no financial fact the ledger does not already hold — they are // operational state — so removing them with the book loses nothing. func (s *Store) DeleteBook(ctx context.Context, bookID string) error { return s.inTx(ctx, func(tx pgx.Tx) error { // Book first, like everything else that touches two of these tables (lockBook). This one took // the reservations first, which is the inverted order — no cycle exists for it today, and that // is exactly the kind of reasoning an invariant written in one place is supposed to replace. if err := lockBook(ctx, tx, bookID); err != nil { return err } if _, err := tx.Exec(ctx, `delete from reservations where book_id = $1 and state <> 'open'`, bookID); err != nil { return fmt.Errorf("pgstore: clear reservations: %w", err) } if _, err := tx.Exec(ctx, `delete from books where id = $1`, bookID); err != nil { return fmt.Errorf("pgstore: delete book: %w", err) } return nil }) }