casino/pkg/ledger/ledger.go

// 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
}