Reference operator wallet

DATABASE_URL=postgres://localhost/operator WALLET_API_KEY=your-api-key WALLET_SIGNING_SECRET=your-api-secret \
  bun examples/operator-wallet/src/main.ts

import {
  createWalletKit,
  PostgresReplayStore,
  type WalletAnswer,
  type WalletHandlers,
  type WalletKit,
  type WalletQuery,
  type WalletTransaction,
  type WalletWriteScope,
} from '@asteriks-aggregator/sdk/wallet'

// The reference operator wallet (S4-04c, D-40 E): an operator's wallet built only on the SDK, the wallet kit over the
// Postgres replay store and a ledger of the operator's own in the same database. The kit checks the D-25 signature,
// the time window, the nonce, the request schema and the first answer per transaction_id; the five handlers only move
// money. Every write runs through `scope.query`, inside the claim's own transaction, so the money and the first answer
// commit together or not at all. Each money handler locks its player's row first, so two transactions of one player
// are serialised, and a bet and its rollback never race. The wallet takes a query and a transaction function of any
// driver; src/main.ts wires postgres.js.

/** A currency of the wallet contract, as a balance answer carries it. */
type Currency = Extract<WalletAnswer, { balance: number }>['currency']

/** The operator's ledger, safe to run again: players, every bet and win, each reversal's answer, the tombstones. */
export const LEDGER_SCHEMA = `
CREATE TABLE IF NOT EXISTS operator_player (
  player_id text PRIMARY KEY,
  currency text NOT NULL,
  balance bigint NOT NULL,
  blocked boolean NOT NULL DEFAULT false
);
CREATE TABLE IF NOT EXISTS operator_entry (
  transaction_id text PRIMARY KEY,
  operation text NOT NULL CONSTRAINT operator_entry_operation CHECK (operation IN ('bet', 'win')),
  player_id text NOT NULL REFERENCES operator_player (player_id),
  currency text NOT NULL,
  amount bigint NOT NULL CONSTRAINT operator_entry_amount CHECK (amount >= 0)
);
CREATE TABLE IF NOT EXISTS operator_reversal (
  refers_to_transaction_id text PRIMARY KEY REFERENCES operator_entry (transaction_id),
  balance bigint NOT NULL,
  currency text NOT NULL
);
CREATE TABLE IF NOT EXISTS operator_tombstone (
  transaction_id text PRIMARY KEY
);
`

/** Thrown when a money handler gets no transaction to write through: the wallet runs on the Postgres store only. */
class LedgerScopeMissing extends Error {
  constructor() {
    super('The reference operator moves money only inside the claim transaction of the Postgres store.')
    this.name = 'LedgerScopeMissing'
  }
}

/** Thrown when the claim no longer holds its transaction_id: nothing is written. */
class LedgerClaimLost extends Error {
  constructor() {
    super('The claim no longer holds this transaction_id; the handler writes nothing.')
    this.name = 'LedgerClaimLost'
  }
}

export type OperatorWalletOptions = {
  /** One autocommit statement of the operator's driver (a balance read, the replay store's own statements). */
  readonly query: WalletQuery
  /** Runs work in one database transaction of the operator's driver. */
  readonly transaction: WalletTransaction
  /** The signing secret of an API key the aggregator signs with, or undefined for an unknown key. */
  readonly secretOf: (apiKey: string) => string | undefined | Promise<string | undefined>
  /** The path the wallet is served under (default `/wallet`). */
  readonly basePath?: string
  /** The clock in milliseconds (default `Date.now`). */
  readonly now?: () => number
}

type Player = { readonly currency: string; readonly balance: string; readonly blocked: boolean }

const refusal = (code: Extract<WalletAnswer, { error: unknown }>['error']['code'], message: string): WalletAnswer => ({
  error: { code, message },
})

const balanceAnswer = (balance: string, currency: string): WalletAnswer => ({
  balance: Number(balance),
  currency: currency as Currency,
})

/**
 * The query a money handler writes through: the claim's own transaction, and only while the claim holds it (the
 * `held()` rule; with the Postgres store the claim's row stays locked while the handler runs).
 */
function ledgerQueryOf(scope: WalletWriteScope): WalletQuery {
  const { query } = scope
  if (query === undefined) throw new LedgerScopeMissing()
  if (!scope.held()) throw new LedgerClaimLost()
  return query
}

/** The player's row, locked for this transaction when `lock` is set, or undefined for an unknown player. */
async function playerOf(query: WalletQuery, playerId: string, lock: boolean): Promise<Player | undefined> {
  const rows = await query(
    `SELECT currency, balance::text AS balance, blocked FROM operator_player WHERE player_id = $1${lock ? ' FOR UPDATE' : ''}`,
    [playerId],
  )
  const row = rows[0]
  if (row === undefined) return undefined
  return { currency: String(row.currency), balance: String(row.balance), blocked: row.blocked === true }
}

/** The player, or the refusal of an unknown player or of a request in another currency. */
function checkedPlayer(player: Player | undefined, currency: string): Player | WalletAnswer {
  if (player === undefined) return refusal('PLAYER_NOT_FOUND', 'The player does not exist.')
  if (player.currency !== currency) return refusal('CURRENCY_MISMATCH', 'The player has another currency.')
  return player
}

const isPlayer = (found: Player | WalletAnswer): found is Player => !('error' in found)

async function hasTombstone(query: WalletQuery, transactionId: string): Promise<boolean> {
  const rows = await query('SELECT 1 FROM operator_tombstone WHERE transaction_id = $1', [transactionId])
  return rows.length > 0
}

/** Adds `delta` (minor units, signed) to the player's balance and returns the new balance. */
async function moveBalance(query: WalletQuery, playerId: string, delta: bigint): Promise<string> {
  const rows = await query(
    'UPDATE operator_player SET balance = balance + $2::bigint WHERE player_id = $1 RETURNING balance::text AS balance',
    [playerId, delta.toString()],
  )
  return String(rows[0]?.balance)
}

const TOMBSTONED = 'A rollback of this transaction arrived before it.'

/** The five handlers of the reference operator over its ledger. */
function ledgerHandlers(query: WalletQuery): WalletHandlers {
  return {
    walletBalance: async (request) => {
      const player = checkedPlayer(await playerOf(query, request.player_id, false), request.currency)
      return isPlayer(player) ? balanceAnswer(player.balance, request.currency) : player
    },
    walletBet: async (request, scope) => {
      const write = ledgerQueryOf(scope)
      const player = checkedPlayer(await playerOf(write, request.player_id, true), request.currency)
      if (!isPlayer(player)) return player
      if (await hasTombstone(write, request.transaction_id)) return refusal('DUPLICATE_TRANSACTION', TOMBSTONED)
      if (player.blocked) return refusal('PLAYER_BLOCKED', 'The player is blocked.')
      const amount = BigInt(request.amount)
      if (amount > BigInt(player.balance)) return refusal('INSUFFICIENT_FUNDS', 'The balance does not cover the bet.')
      const balance = await moveBalance(write, request.player_id, -amount)
      await write(
        "INSERT INTO operator_entry (transaction_id, operation, player_id, currency, amount) VALUES ($1, 'bet', $2, $3, $4::bigint)",
        [request.transaction_id, request.player_id, request.currency, amount.toString()],
      )
      return balanceAnswer(balance, request.currency)
    },
    walletWin: async (request, scope) => {
      const write = ledgerQueryOf(scope)
      const player = checkedPlayer(await playerOf(write, request.player_id, true), request.currency)
      if (!isPlayer(player)) return player
      if (await hasTombstone(write, request.transaction_id)) return refusal('DUPLICATE_TRANSACTION', TOMBSTONED)
      const amount = BigInt(request.amount)
      const balance = await moveBalance(write, request.player_id, amount)
      await write(
        "INSERT INTO operator_entry (transaction_id, operation, player_id, currency, amount) VALUES ($1, 'win', $2, $3, $4::bigint)",
        [request.transaction_id, request.player_id, request.currency, amount.toString()],
      )
      return balanceAnswer(balance, request.currency)
    },
    walletRollback: async (request, scope) => {
      const write = ledgerQueryOf(scope)
      const player = checkedPlayer(await playerOf(write, request.player_id, true), request.currency)
      if (!isPlayer(player)) return player
      const target = request.refers_to_transaction_id
      const entries = await write(
        'SELECT operation, player_id, currency, amount::text AS amount FROM operator_entry WHERE transaction_id = $1',
        [target],
      )
      const entry = entries[0]
      if (entry === undefined) {
        // WALLET-CONTRACT.md section 3: a rollback of an unknown transaction is acknowledged, and the transaction is
        // refused if it arrives later.
        await write('INSERT INTO operator_tombstone (transaction_id) VALUES ($1) ON CONFLICT DO NOTHING', [target])
        return balanceAnswer(player.balance, request.currency)
      }
      if (entry.player_id !== request.player_id || entry.currency !== request.currency)
        return refusal('TRANSACTION_NOT_FOUND', 'This player has no transaction with this id.')
      // One reversal per bet or win: a second rollback of it gets the first one's answer.
      const reversals = await write(
        'SELECT balance::text AS balance, currency FROM operator_reversal WHERE refers_to_transaction_id = $1',
        [target],
      )
      const first = reversals[0]
      if (first !== undefined) return balanceAnswer(String(first.balance), String(first.currency))
      const amount = BigInt(String(entry.amount))
      const balance = await moveBalance(write, request.player_id, entry.operation === 'bet' ? amount : -amount)
      await write(
        'INSERT INTO operator_reversal (refers_to_transaction_id, balance, currency) VALUES ($1, $2::bigint, $3)',
        [target, balance, request.currency],
      )
      return balanceAnswer(balance, request.currency)
    },
    walletRoundFinished: async (request, scope) => {
      const write = ledgerQueryOf(scope)
      const player = checkedPlayer(await playerOf(write, request.player_id, false), request.currency)
      return isPlayer(player) ? balanceAnswer(player.balance, request.currency) : player
    },
  }
}

/** The reference operator wallet: the kit over the Postgres replay store and the ledger handlers. */
export function createOperatorWallet(options: OperatorWalletOptions): WalletKit {
  const now = options.now ?? Date.now
  return createWalletKit({
    handlers: ledgerHandlers(options.query),
    store: new PostgresReplayStore({ query: options.query, transaction: options.transaction, now }),
    secretOf: options.secretOf,
    basePath: options.basePath ?? '/wallet',
    now,
  })
}

import {
  WALLET_REPLAY_SCHEMA,
  type WalletKit,
  type WalletQuery,
  type WalletTransaction,
} from '@asteriks-aggregator/sdk/wallet'
import postgres from 'postgres'
import { createOperatorWallet, LEDGER_SCHEMA } from './wallet.ts'

// The reference operator wallet on postgres.js: one autocommit statement for the reads and the replay store's own
// statements, `sql.begin` for the claim transaction every money write runs in. The replay and ledger tables are
// applied when the wallet opens, so a new database needs no migration step.

/** The aggregator's credential for the operator's wallet, as the admin CLI handed it over. */
export type WalletCredential = { readonly apiKey: string; readonly secret: string }

/** An open wallet and the function that closes its database connections. */
export type OpenWallet = { readonly wallet: WalletKit; readonly close: () => Promise<void> }

/** Opens the reference operator wallet on the Postgres database at `databaseUrl`. */
export async function openOperatorWallet(
  databaseUrl: string,
  credential: WalletCredential,
  now: () => number = Date.now,
): Promise<OpenWallet> {
  const sql = postgres(databaseUrl, { connection: { TimeZone: 'UTC' }, onnotice: () => {} })
  await sql.unsafe(WALLET_REPLAY_SCHEMA + LEDGER_SCHEMA)
  const query: WalletQuery = (text, parameters) => sql.unsafe(text, [...parameters])
  const transaction: WalletTransaction = async <T>(work: (query: WalletQuery) => Promise<T>) =>
    (await sql.begin((tx) => work((text, parameters) => tx.unsafe(text, [...parameters])))) as T
  const wallet = createOperatorWallet({
    query,
    transaction,
    secretOf: (apiKey) => (apiKey === credential.apiKey ? credential.secret : undefined),
    now,
  })
  return { wallet, close: () => sql.end() }
}