Money: ledgers and exactly-once

// claim takes the idempotency key. If the key exists, it returns the first entry, or
// ErrKeyReused when the first request was different.
func claim(ctx context.Context, tx pgx.Tx, e Entry) (Result, error) {
  var sum int64
  for _, leg := range e.Legs {
    sum += leg.Amount
  }
  if sum != 0 || len(e.Legs) < 2 {
    return Result{}, fmt.Errorf("entry %q: %d legs sum to %d, want 2 or more legs that sum to 0", e.Key, len(e.Legs), sum)
  }
  var id int64
  err := tx.QueryRow(ctx, q["claim_key"], e.Key, e.fingerprint(), e.Kind, nullable(e.ExternalRef)).Scan(&id)
  if errors.Is(err, pgx.ErrNoRows) { // the key exists: this request is a retry or a copy
    var first string
    if err := tx.QueryRow(ctx, q["find_entry"], e.Key).Scan(&id, &first); err != nil {
      return Result{}, fmt.Errorf("find entry for key %q: %w", e.Key, err)
    }
    if first != e.fingerprint() {
      return Result{}, fmt.Errorf("%w: %q", ErrKeyReused, e.Key)
    }
    return Result{EntryID: id, Replayed: true}, nil
  }
  if err != nil {
    return Result{}, fmt.Errorf("claim key %q: %w", e.Key, err)
  }
  return Result{EntryID: id}, nil
}

// apply changes each cached balance and writes each posting. The CHECK refuses an overdraft,
// and the deferred trigger refuses an entry that does not sum to zero.
func apply(ctx context.Context, tx pgx.Tx, entryID int64, e Entry) error {
  for i, leg := range e.Legs {
    if _, err := tx.Exec(ctx, q["apply"], leg.Account, leg.Amount); err != nil {
      if isCheck(err, "no_overdraft") {
        return fmt.Errorf("%w: account %d", ErrInsufficientFunds, leg.Account)
      }
      return fmt.Errorf("apply leg to account %d: %w", leg.Account, err)
    }
    if _, err := tx.Exec(ctx, q["post"], entryID, leg.Account, e.Currency, leg.Amount); err != nil {
      return fmt.Errorf("post leg to account %d: %w", leg.Account, err)
    }
    if i == 0 && e.Fault == CrashAfterDebit {
      return ErrCrashed
    }
  }
  return nil
}

-- Claim the idempotency key first. A copy of the same request waits here, then finds the key.
INSERT INTO journal_entries (idempotency_key, fingerprint, kind, external_ref)
VALUES ($1, $2, $3, $4)
ON CONFLICT (idempotency_key) DO NOTHING
RETURNING id;

-- Lock every account of the entry in id order. Two transfers can never wait for each other.
SELECT id, currency FROM accounts WHERE id = ANY($1) ORDER BY id FOR UPDATE;

-- Change the cached balance. The no_overdraft CHECK refuses a balance below the holds.
UPDATE accounts SET balance = balance + $2 WHERE id = $1;

INSERT INTO postings (entry_id, account_id, currency, amount) VALUES ($1, $2, $3, $4);

// Capture closes a hold and moves amount, at most the held amount, to another account. The rest
// of the hold returns to the available balance.
func (l *Ledger) Capture(ctx context.Context, holdID, to, amount int64, key, currency string) (Result, error) {
  return l.run(ctx, NoFault, func(tx pgx.Tx) (Result, error) {
    var from, held int64
    if err := tx.QueryRow(ctx, q["read_hold"], holdID).Scan(&from, &held); err != nil {
      return Result{}, fmt.Errorf("read hold %d: %w", holdID, err)
    }
    if amount <= 0 || amount > held {
      return Result{}, fmt.Errorf("%w: %d of %d", ErrOverCapture, amount, held)
    }
    e := Entry{Key: key, Kind: "capture", Currency: currency, Legs: []Leg{{from, -amount}, {to, amount}}}
    res, err := claim(ctx, tx, e)
    if err != nil || res.Replayed {
      return res, err
    }
    if err := l.lock(ctx, tx, e); err != nil {
      return Result{}, err
    }
    if err := tx.QueryRow(ctx, q["close_hold"], holdID, "captured").Scan(&from, &held); err != nil {
      if errors.Is(err, pgx.ErrNoRows) {
        return Result{}, fmt.Errorf("%w: hold %d", ErrHoldClosed, holdID)
      }
      return Result{}, fmt.Errorf("close hold %d: %w", holdID, err)
    }
    if _, err := tx.Exec(ctx, q["set_held"], from, -held); err != nil {
      return Result{}, fmt.Errorf("lift hold on account %d: %w", from, err)
    }
    return res, apply(ctx, tx, res.EntryID, e)
  })
}

INSERT INTO holds (account_id, amount, idempotency_key) VALUES ($1, $2, $3)
ON CONFLICT (idempotency_key) DO NOTHING
RETURNING id;

-- Holds count against the balance: the same CHECK refuses a hold the account cannot cover.
UPDATE accounts SET held = held + $2 WHERE id = $1;

-- Capture or release a hold once. 0 rows: it was already closed.
UPDATE holds SET status = $2 WHERE id = $1 AND status = 'held'
RETURNING account_id, amount;

CREATE TEMP TABLE processor_report (
  external_ref text   PRIMARY KEY,
  amount       bigint NOT NULL
) ON COMMIT DROP;

-- Match the ledger's processor postings against the processor's report, both ways.
SELECT coalesce(l.external_ref, r.external_ref) AS ref,
       l.amount AS ledger, r.amount AS processor,
       CASE WHEN l.external_ref IS NULL THEN 'missing in the ledger'
            WHEN r.external_ref IS NULL THEN 'missing at the processor'
            WHEN l.amount <> r.amount  THEN 'amounts differ'
            ELSE 'matched' END AS result
FROM (SELECT j.external_ref, -p.amount AS amount
      FROM journal_entries j JOIN postings p ON p.entry_id = j.id
      WHERE p.account_id = $1 AND j.external_ref IS NOT NULL) l
FULL OUTER JOIN processor_report r ON r.external_ref = l.external_ref
ORDER BY 1;