// Package ledger implements append-only double-entry accounting. // // Invariants, enforced here and again by database constraints and triggers: // - every transaction's postings sum to exactly zero // - no account balance may go negative // - rows are never updated or deleted; corrections are compensating entries // // Every balance change is explained by a posting that records what happened, // when, which round it belonged to, and the balance either side of it. package ledger import ( "context" "errors" "fmt" "math/big" "sort" "strings" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) var ( ErrUnbalanced = errors.New("ledger: postings do not sum to zero") ErrInsufficientFunds = errors.New("ledger: insufficient funds") ErrEmptyTransaction = errors.New("ledger: transaction has no postings") ErrNonPositiveAmount = errors.New("ledger: amount must be positive") ) // Posting is a single leg of a transaction. Positive credits, negative debits. type Posting struct { AccountID int64 AmountMsat int64 } // Entry is a posting as seen from one account's history. type Entry struct { TransactionID int64 Kind string RoundID *int64 AmountMsat int64 BalanceBefore int64 BalanceAfter int64 CreatedAt time.Time } type Ledger struct{ pool *pgxpool.Pool } func New(pool *pgxpool.Pool) *Ledger { return &Ledger{pool: pool} } // Post writes one balanced transaction atomically. // // Accounts are locked in ascending id order so that concurrent transactions // touching the same accounts cannot deadlock, and so a balance read cannot be // stale by the time the posting is written. func (l *Ledger) Post(ctx context.Context, kind string, roundID *int64, postings []Posting) (int64, error) { if len(postings) == 0 { return 0, ErrEmptyTransaction } // The zero-sum check must be overflow-safe. Accumulating into an int64 // lets a crafted posting set wrap to zero — two legs of MaxInt64 and one // of 2 sum to 0 in wrapping arithmetic — which would mint money out of // nothing. big.Int has no such boundary. sum := new(big.Int) for _, p := range postings { sum.Add(sum, big.NewInt(p.AmountMsat)) } if sum.Sign() != 0 { return 0, fmt.Errorf("%w: sum is %s", ErrUnbalanced, sum.String()) } tx, err := l.pool.Begin(ctx) if err != nil { return 0, err } defer tx.Rollback(ctx) var txID int64 if err := tx.QueryRow(ctx, `INSERT INTO transactions (kind, round_id) VALUES ($1, $2) RETURNING id`, kind, roundID).Scan(&txID); err != nil { return 0, err } // Merge duplicate accounts before locking. A transaction that touched the // same account twice would otherwise read a stale balance for the second // leg and write a posting that contradicts the first. merged := make(map[int64]int64, len(postings)) order := make([]int64, 0, len(postings)) for _, p := range postings { if _, seen := merged[p.AccountID]; !seen { order = append(order, p.AccountID) } merged[p.AccountID] += p.AmountMsat } sort.Slice(order, func(i, j int) bool { return order[i] < order[j] }) ids := make([]int64, 0, len(order)) amounts := make([]int64, 0, len(order)) for _, id := range order { if merged[id] == 0 { continue // legs cancelled out; nothing to record } ids = append(ids, id) amounts = append(amounts, merged[id]) } if len(ids) == 0 { // Every leg cancelled. The transaction row stands as a record that // something was attempted, but there is no balance change to write. if err := tx.Commit(ctx); err != nil { return 0, err } return txID, nil } // Lock every account first, in ascending id order so concurrent // transactions cannot deadlock against each other. // // This must be its own statement. A single statement — even one whose CTE // does FOR UPDATE — evaluates against one snapshot taken before the locks // are held, so the balance read would see pre-lock data and concurrent // transactions would silently overwrite each other. Splitting it means the // second statement takes a fresh snapshot, by which point we hold the // locks and no other writer can commit against these accounts. if _, err := tx.Exec(ctx, `SELECT id FROM accounts WHERE id = ANY($1) ORDER BY id FOR UPDATE`, ids); err != nil { return 0, fmt.Errorf("locking accounts: %w", err) } // Now write every posting in one statement, however many legs there are. // Doing this per-posting cost three round trips each, which made // settlement scale in network latency rather than in real work. rows, err := tx.Query(ctx, ` WITH input AS ( SELECT unnest($2::bigint[]) AS account_id, unnest($3::bigint[]) AS amount ), current AS ( SELECT i.account_id, i.amount, COALESCE((SELECT p.balance_after FROM postings p WHERE p.account_id = i.account_id ORDER BY p.id DESC LIMIT 1), 0) AS balance_before FROM input i ) INSERT INTO postings (transaction_id, account_id, amount_msat, balance_before, balance_after) SELECT $1, account_id, amount, balance_before, balance_before + amount FROM current RETURNING account_id, balance_after`, txID, ids, amounts) if err != nil { // The balance floor is enforced by a database trigger, so an overdraft // surfaces here. Translate it into the domain error callers expect. if isBalanceFloorViolation(err) { return 0, fmt.Errorf("%w: %v", ErrInsufficientFunds, err) } return 0, err } written := 0 for rows.Next() { var acct, after int64 if err := rows.Scan(&acct, &after); err != nil { rows.Close() return 0, err } written++ } rows.Close() if err := rows.Err(); err != nil { if isBalanceFloorViolation(err) { return 0, fmt.Errorf("%w: %v", ErrInsufficientFunds, err) } return 0, err } if written != len(ids) { return 0, fmt.Errorf("ledger: wrote %d postings for %d accounts; "+ "an account id does not exist", written, len(ids)) } if err := tx.Commit(ctx); err != nil { if isBalanceFloorViolation(err) { return 0, fmt.Errorf("%w: %v", ErrInsufficientFunds, err) } return 0, err } return txID, nil } // isBalanceFloorViolation reports whether an error is the database refusing to // let an account go negative. func isBalanceFloorViolation(err error) bool { if err == nil { return false } msg := err.Error() return strings.Contains(msg, "may not go negative") || strings.Contains(msg, "balance_nonnegative") } // Transfer moves funds between two accounts. This is the peer-to-peer path. func (l *Ledger) Transfer(ctx context.Context, from, to int64, amountMsat int64) (int64, error) { if amountMsat <= 0 { return 0, ErrNonPositiveAmount } return l.Post(ctx, "transfer", nil, []Posting{ {AccountID: from, AmountMsat: -amountMsat}, {AccountID: to, AmountMsat: amountMsat}, }) } // Deposit credits a player from the Lightning bridge account. func (l *Ledger) Deposit(ctx context.Context, player int64, amountMsat int64) (int64, error) { if amountMsat <= 0 { return 0, ErrNonPositiveAmount } bridge, err := l.AccountByName(ctx, "lightning_bridge") if err != nil { return 0, err } return l.Post(ctx, "deposit", nil, []Posting{ {AccountID: bridge, AmountMsat: -amountMsat}, {AccountID: player, AmountMsat: amountMsat}, }) } // Withdraw debits a player back to the Lightning bridge account. func (l *Ledger) Withdraw(ctx context.Context, player int64, amountMsat int64) (int64, error) { if amountMsat <= 0 { return 0, ErrNonPositiveAmount } bridge, err := l.AccountByName(ctx, "lightning_bridge") if err != nil { return 0, err } return l.Post(ctx, "withdraw", nil, []Posting{ {AccountID: player, AmountMsat: -amountMsat}, {AccountID: bridge, AmountMsat: amountMsat}, }) } // Balance returns the account's current balance in millisatoshis. func (l *Ledger) Balance(ctx context.Context, accountID int64) (int64, error) { var bal int64 err := l.pool.QueryRow(ctx, `SELECT COALESCE( (SELECT balance_after FROM postings WHERE account_id = $1 ORDER BY id DESC LIMIT 1), 0)`, accountID).Scan(&bal) return bal, err } // History returns an account's postings, newest first. func (l *Ledger) History(ctx context.Context, accountID int64, limit int) ([]Entry, error) { rows, err := l.pool.Query(ctx, `SELECT p.transaction_id, t.kind, t.round_id, p.amount_msat, p.balance_before, p.balance_after, p.created_at FROM postings p JOIN transactions t ON t.id = p.transaction_id WHERE p.account_id = $1 ORDER BY p.id DESC LIMIT $2`, accountID, limit) if err != nil { return nil, err } defer rows.Close() var out []Entry for rows.Next() { var e Entry if err := rows.Scan(&e.TransactionID, &e.Kind, &e.RoundID, &e.AmountMsat, &e.BalanceBefore, &e.BalanceAfter, &e.CreatedAt); err != nil { return nil, err } out = append(out, e) } return out, rows.Err() } // EnsurePlayer returns the account id for a public key, creating it if needed. func (l *Ledger) EnsurePlayer(ctx context.Context, pubkey []byte) (int64, error) { var id int64 err := l.pool.QueryRow(ctx, `INSERT INTO accounts (kind, pubkey) VALUES ('player', $1) ON CONFLICT (pubkey) DO UPDATE SET pubkey = EXCLUDED.pubkey RETURNING id`, pubkey).Scan(&id) return id, err } // AccountByName resolves a system account such as "house_pot". func (l *Ledger) AccountByName(ctx context.Context, name string) (int64, error) { var id int64 err := l.pool.QueryRow(ctx, `SELECT id FROM accounts WHERE name = $1`, name).Scan(&id) if errors.Is(err, pgx.ErrNoRows) { return 0, fmt.Errorf("ledger: no account named %q", name) } return id, err } // TotalIssued is the value held inside the system by players and the house — // every account except the external Lightning bridge. It changes only when // funds genuinely enter or leave, never through internal play. func (l *Ledger) TotalIssued(ctx context.Context) (int64, error) { return l.sumBalances(ctx, `WHERE NOT a.allow_negative`) } // ConservationCheck sums every account including the bridge. Because each // transaction sums to zero, this must always be exactly zero. A non-zero // result means the books are corrupt, and is the top-level audit alarm. func (l *Ledger) ConservationCheck(ctx context.Context) (int64, error) { return l.sumBalances(ctx, ``) } // sumBalances totals account balances. // // Postgres SUM() over bigint returns numeric, which can exceed int64 even // though no single balance can. Scanning it as text and parsing through // big.Int means a corrupt ledger reports a clear error instead of a scan // failure — the alarm must survive the very condition it exists to detect. func (l *Ledger) sumBalances(ctx context.Context, where string) (int64, error) { var text string err := l.pool.QueryRow(ctx, `SELECT COALESCE(SUM(b.balance_msat), 0)::text FROM account_balances b JOIN accounts a ON a.id = b.account_id `+where).Scan(&text) if err != nil { return 0, err } total, ok := new(big.Int).SetString(text, 10) if !ok { return 0, fmt.Errorf("ledger: unparseable balance total %q", text) } if !total.IsInt64() { return 0, fmt.Errorf("ledger: balance total %s exceeds int64; the books are corrupt", text) } return total.Int64(), nil }