Rebuilding from the ledger

The rebuild. One UPDATE. The ledger reseeds the cache.

-- Lock the wallets you're about to rewrite, so concurrent posts wait.
BEGIN;
SELECT user_id FROM wallet FOR UPDATE;

-- Recompute every wallet balance from the ledger. Liabilities are
-- credit-positive, so balance = SUM(credit) - SUM(debit) for the linked
-- account code. (Assets flip the sign; do the same scan with the other
-- side for asset accounts.)
UPDATE wallet w
   SET balance = COALESCE((
         SELECT SUM(jl.credit) - SUM(jl.debit)
           FROM journal_line jl
          WHERE jl.account_code = w.account_code
       ), 0),
       updated_at = now();

COMMIT;

Same rebuild, scriptable for ops use.

def rebuild_wallet_balances(conn) -> None:
    with conn.transaction():
        # Lock every wallet row so concurrent posts wait.
        conn.execute("SELECT user_id FROM wallet FOR UPDATE")

        conn.execute(
            """
            UPDATE wallet w
               SET balance = COALESCE((
                     SELECT SUM(jl.credit) - SUM(jl.debit)
                       FROM journal_line jl
                      WHERE jl.account_code = w.account_code
                   ), 0),
                   updated_at = now()
            """
        )

Same rebuild, callable from an ops command.

func RebuildWalletBalances(ctx context.Context, db *sql.DB) error {
    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    if _, err := tx.ExecContext(ctx, `SELECT user_id FROM wallet FOR UPDATE`); err != nil {
        return err
    }

    if _, err := tx.ExecContext(ctx, `
        UPDATE wallet w
           SET balance = COALESCE((
                 SELECT SUM(jl.credit) - SUM(jl.debit)
                   FROM journal_line jl
                  WHERE jl.account_code = w.account_code
               ), 0),
               updated_at = now()
    `); err != nil {
        return err
    }
    return tx.Commit()
}