package postgres import ( "context" "errors" "fmt" "log/slog" "github.com/jackc/pgx/v5/pgconn" "gl/domain/ledger" ) func authorityError(err error) error { var pgErr *pgconn.PgError if errors.As(err, &pgErr) { switch pgErr.Code { case "PGL01": return ledger.ErrModeRequired case "PGL02": return ledger.ErrInsufficientFunds case "40P01", "40001": return ledger.ErrConcurrentPosting } } return nil } func (r *JournalRepository) OperatingMode(ctx context.Context) (ledger.OperatingState, error) { return scanOperatingState(r.database.QueryRow(ctx, operatingStateSQL)) } func scanOperatingState(row Row) (ledger.OperatingState, error) { var state ledger.OperatingState var mode string err := row.Scan(&mode, &state.Revision, &state.ActorID, &state.Reason, &state.ChangedAt) state.Mode = ledger.OperatingMode(mode) return state, err } const operatingStateSQL = "SELECT mode, revision, actor_id, reason, changed_at FROM ledger_operating_state WHERE singleton" func (r *JournalRepository) SetOperatingMode(ctx context.Context, mode ledger.OperatingMode, revision int64, actor, reason string, reconciled bool) (ledger.OperatingState, error) { tx, err := r.database.Begin(ctx) if err != nil { return ledger.OperatingState{}, err } defer tx.Rollback(ctx) current, err := scanOperatingState(tx.QueryRow(ctx, operatingStateSQL+" FOR UPDATE")) if err != nil { return ledger.OperatingState{}, err } if current.Revision != revision { return ledger.OperatingState{}, ledger.ErrModeConflict } if current.Mode == mode { return current, nil } if mode == ledger.ModeNormal && !reconciled { return ledger.OperatingState{}, fmt.Errorf("%w: reconciliation must be confirmed before returning to NORMAL", ledger.ErrModeConflict) } if mode == ledger.ModeGLOnly { var inFlight bool if err := tx.QueryRow(ctx, "SELECT EXISTS (SELECT 1 FROM kuknos_settlements WHERE locked_by <> '')").Scan(&inFlight); err != nil { return ledger.OperatingState{}, err } if inFlight { return ledger.OperatingState{}, ledger.ErrSettlementInFlight } var negativeBalance bool if err := tx.QueryRow(ctx, `SELECT EXISTS ( SELECT a.id FROM ledger_accounts a JOIN journal_entries e ON e.account_id=a.id JOIN journals j ON j.id=e.journal_id AND j.sealed_at IS NOT NULL WHERE a.class IN ('USER_AVAILABLE','USER_FROZEN') GROUP BY a.id HAVING sum(e.amount)<0 )`).Scan(&negativeBalance); err != nil { return ledger.OperatingState{}, err } if negativeBalance { return ledger.OperatingState{}, ledger.ErrInsufficientFunds } } state, err := scanOperatingState(tx.QueryRow(ctx, `UPDATE ledger_operating_state SET mode=$1, revision=revision+1, actor_id=$2, reason=$3, changed_at=clock_timestamp() WHERE singleton RETURNING mode, revision, actor_id, reason, changed_at`, string(mode), actor, reason)) if err != nil { return ledger.OperatingState{}, err } if _, err := tx.Exec(ctx, `INSERT INTO ledger_mode_changes (revision, previous_mode, mode, actor_id, reason, reconciliation_confirmed, changed_at) VALUES ($1,$2,$3,$4,$5,$6,$7)`, state.Revision, string(current.Mode), string(mode), actor, reason, reconciled, state.ChangedAt); err != nil { return ledger.OperatingState{}, err } if err := tx.Commit(ctx); err != nil { return ledger.OperatingState{}, err } slog.WarnContext(ctx, "ledger operating mode changed", "previous_mode", current.Mode, "mode", mode, "revision", state.Revision, "actor_id", actor, "reason", reason) return state, nil } // One statement provides a consistent snapshot across available and frozen funds. func (r *JournalRepository) UserBalances(ctx context.Context, ownerType, ownerID string, assetID *int64) ([]ledger.UserAssetBalance, error) { rows, err := r.database.Query(ctx, `SELECT a.asset_id, COALESCE(sum(e.amount) FILTER (WHERE a.class='USER_AVAILABLE'),0)::text, COALESCE(sum(e.amount) FILTER (WHERE a.class='USER_FROZEN'),0)::text FROM ledger_accounts a LEFT JOIN journal_entries e ON e.account_id=a.id LEFT JOIN journals j ON j.id=e.journal_id AND j.sealed_at IS NOT NULL WHERE a.owner_type=$1 AND a.owner_id=$2 AND a.class IN ('USER_AVAILABLE','USER_FROZEN') AND ($3::bigint IS NULL OR a.asset_id=$3) AND (e.journal_id IS NULL OR j.id IS NOT NULL) GROUP BY a.asset_id ORDER BY a.asset_id`, ownerType, ownerID, assetID) if err != nil { return nil, err } defer rows.Close() result := []ledger.UserAssetBalance{} for rows.Next() { var balance ledger.UserAssetBalance var available, frozen string if err := rows.Scan(&balance.AssetID, &available, &frozen); err != nil { return nil, err } balance.Available, err = ledger.ParseAmount(available) if err != nil { return nil, err } balance.Frozen, err = ledger.ParseAmount(frozen) if err != nil { return nil, err } result = append(result, balance) } if err := rows.Err(); err != nil { return nil, err } if assetID != nil && len(result) == 0 { result = append(result, ledger.UserAssetBalance{AssetID: *assetID}) } return result, nil }