Phase 12 batch 7: F06 — queue persistence

New queue_entries table + replace_queue_snapshot(jsonb) RPC
(migration 20260422_003, applied prod). In-memory queue stays the hot
path; every mutation fires an async snapshot-replace, and boot-time
loadQueueFromDB() rehydrates the array before the server listener
begins accepting requests.

Mirror sites instrumented:
  - /join push
  - /join/bulk push (loop)
  - settlement observer paymentId stamp (F20)
  - ws.close paid → disconnected flag, unpaid → remove
  - startGame pull + rollback
  - stale-queue sweep filter

Previously a pm2 restart silently lost every in-flight paid entry
with no refund path. The stale-queue sweeper remains the final
arbiter if a rehydrated entry's payment never reconciles.

Tests: 192/192 including a new queue-persist round-trip suite.
Boot-tested locally — rehydrate fires on startup, health endpoint
responsive.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: back2matching

CHANGELOG.md

@@ -30,6 +30,7 @@ Multi-agent audit ran 2026-04-22; produced 43 ranked findings (see internal `.lo
 - **F37** — Jackpot-pool deduction on a win is now atomic via the `deduct_jackpot_pool_for_win` RPC (migration `20260422_002_jackpot_pool_deduct_atomic.sql`). Previously `updateJackpotPoolAfterWin` wrote the pre-computed `remainingPool` as a raw overwrite, so any concurrent `/join` contribution landing between the roll-read and the roll-write was silently clobbered.
 - **F35** — `POST /faucet` gained an IP-based rate limit (5/min prod, 30/min dev; `FAUCET_RATE_LIMIT` override). The per-wallet DB cap only applied AFTER the CDP transfer, so a script cycling fresh `walletAddress` values could exhaust the project-wide CDP faucet quota.
 - **F26** — Jackpot tier selection now uses cumulative thresholds (`roll < ultra` → `roll < ultra + mega` → `roll < ultra + mega + mini`) so the marginal MINI/MEGA/ULTRA probabilities exactly match the advertised 3% / 0.3% / 0.04%. Previously sequential `roll < tier.chance` checks made the effective MEGA probability 0.26% and MINI 2.7% — users under-delivered vs. the marketing copy.
+- **F06** — Queue entries now persist to a new `queue_entries` table (migration `20260422_003_queue_entries_persistence.sql`) via the atomic `replace_queue_snapshot(jsonb)` RPC after every mutation (push, splice, filter, paymentId stamp, disconnect flag, startGame pull). On server boot the in-memory queue is rehydrated from this table before the HTTP listener begins processing requests. Previously the queue was pure in-memory state, so every pm2 restart or crash silently dropped in-flight paid entries with no refund path. Persistence is fire-and-forget from request handlers (memory is still the hot path); the stale-queue sweeper is the final arbiter if a payment never reconciles.
 
 Pre-Phase-12 items from this Unreleased window:
 

scripts/queue-persist.test.js

@@ -0,0 +1,161 @@
+/**
+ * F06 — Queue persistence round-trip tests.
+ *
+ * Pure-function tests for queueEntryToRow/rowToQueueEntry — no DB required.
+ * Verifies that a full queue entry (bulk + single + paid + unpaid) survives
+ * a serialize → deserialize round-trip so rehydrate on boot preserves state.
+ */
+
+const { test, describe } = require('node:test');
+const assert = require('node:assert');
+
+const { queueEntryToRow, rowToQueueEntry } = require('../src/lib/supabase');
+
+describe('F06 — queue entry serialization', () => {
+  test('single paid entry round-trips', () => {
+    const entry = {
+      playerId: 'agent-alice',
+      displayName: 'Alice',
+      walletAddress: '0xAbCdEf0123456789aBcDeF0123456789AbCdEf01',
+      gamePoolContribution: 0.15,
+      jackpotContribution: 0.10,
+      callbackUrl: 'https://alice.example.com/hook',
+      webhookSecret: 'whsec_abcdef0123456789abcdef0123456789abcdef0123456789',
+      idempotencyKey: 'idem-12345',
+      paymentId: 'pay_1234567890_abcd',
+      joinedAt: 1700000000000,
+      disconnected: false
+    };
+
+    const row = queueEntryToRow(entry);
+    const back = rowToQueueEntry({
+      ...row,
+      // simulate what Supabase returns: nulls for '' fields, preserved types
+      wallet_address: row.wallet_address || null,
+      callback_url: row.callback_url || null,
+      webhook_secret: row.webhook_secret || null,
+      idempotency_key: row.idempotency_key || null,
+      payment_id: row.payment_id || null,
+      bulk_parent_id: null,
+      bulk_index: null,
+      bulk_count: null,
+      disconnected_at: null
+    });
+
+    assert.strictEqual(back.playerId, entry.playerId);
+    assert.strictEqual(back.displayName, entry.displayName);
+    assert.strictEqual(back.walletAddress, entry.walletAddress.toLowerCase());
+    assert.strictEqual(back.gamePoolContribution, entry.gamePoolContribution);
+    assert.strictEqual(back.jackpotContribution, entry.jackpotContribution);
+    assert.strictEqual(back.callbackUrl, entry.callbackUrl);
+    assert.strictEqual(back.webhookSecret, entry.webhookSecret);
+    assert.strictEqual(back.idempotencyKey, entry.idempotencyKey);
+    assert.strictEqual(back.paymentId, entry.paymentId);
+    assert.strictEqual(back.joinedAt, entry.joinedAt);
+    assert.strictEqual(back.disconnected, false);
+  });
+
+  test('bulk sub-entry round-trips with index/count/parent', () => {
+    const entry = {
+      playerId: 'agent-bob#1700000000000-7',
+      displayName: 'Bob',
+      walletAddress: '0x1111111111111111111111111111111111111111',
+      gamePoolContribution: 0.15,
+      jackpotContribution: 0.10,
+      callbackUrl: null,
+      webhookSecret: null,
+      idempotencyKey: null,
+      bulkParentId: 'agent-bob',
+      bulkIndex: 7,
+      bulkCount: 50,
+      joinedAt: 1700000000007,
+      disconnected: false
+    };
+
+    const row = queueEntryToRow(entry);
+    assert.strictEqual(row.bulk_index, '7');
+    assert.strictEqual(row.bulk_count, '50');
+    assert.strictEqual(row.bulk_parent_id, 'agent-bob');
+
+    const back = rowToQueueEntry({
+      ...row,
+      wallet_address: row.wallet_address,
+      bulk_index: 7,
+      bulk_count: 50,
+      bulk_parent_id: 'agent-bob',
+      callback_url: null,
+      webhook_secret: null,
+      idempotency_key: null,
+      payment_id: null,
+      disconnected_at: null
+    });
+
+    assert.strictEqual(back.bulkIndex, 7);
+    assert.strictEqual(back.bulkCount, 50);
+    assert.strictEqual(back.bulkParentId, 'agent-bob');
+  });
+
+  test('disconnected entry preserves disconnectedAt', () => {
+    const now = 1700000500000;
+    const entry = {
+      playerId: 'agent-charlie',
+      displayName: 'Charlie',
+      walletAddress: null,
+      gamePoolContribution: 0,
+      jackpotContribution: 0,
+      joinedAt: 1700000000000,
+      disconnected: true,
+      disconnectedAt: now
+    };
+
+    const row = queueEntryToRow(entry);
+    assert.strictEqual(row.disconnected, true);
+    assert.strictEqual(new Date(row.disconnected_at).getTime(), now);
+
+    const back = rowToQueueEntry({
+      ...row,
+      wallet_address: null,
+      callback_url: null,
+      webhook_secret: null,
+      idempotency_key: null,
+      payment_id: null,
+      bulk_parent_id: null,
+      bulk_index: null,
+      bulk_count: null
+    });
+
+    assert.strictEqual(back.disconnected, true);
+    assert.strictEqual(back.disconnectedAt, now);
+  });
+
+  test('unpaid entry with no wallet round-trips', () => {
+    const entry = {
+      playerId: 'guest-1',
+      displayName: 'Guest',
+      walletAddress: null,
+      gamePoolContribution: 0,
+      jackpotContribution: 0,
+      joinedAt: 1700000000000,
+      disconnected: false
+    };
+
+    const row = queueEntryToRow(entry);
+    assert.strictEqual(row.wallet_address, '');
+
+    const back = rowToQueueEntry({
+      ...row,
+      wallet_address: null,
+      callback_url: null,
+      webhook_secret: null,
+      idempotency_key: null,
+      payment_id: null,
+      bulk_parent_id: null,
+      bulk_index: null,
+      bulk_count: null,
+      disconnected_at: null
+    });
+
+    assert.strictEqual(back.walletAddress, null);
+    assert.strictEqual(back.gamePoolContribution, 0);
+  });
+});

src/lib/supabase.js

@@ -936,6 +936,94 @@ async function deductFromJackpotPoolForWin(payoutAmount, tier) {
   }
 }
 
+// =============================================================================
+// F06 — Queue persistence
+// =============================================================================
+// The in-memory `queue` array in src/server.js is the hot path; this module
+// mirrors it to the `queue_entries` table on every mutation. On boot the
+// queue is re-hydrated from the table so restarts/crashes don't lose paid
+// in-flight entries.
+
+function queueEntryToRow(entry) {
+  return {
+    player_id: String(entry.playerId || ''),
+    display_name: String(entry.displayName || ''),
+    wallet_address: entry.walletAddress ? String(entry.walletAddress).toLowerCase() : '',
+    game_pool_contribution: Number(entry.gamePoolContribution || 0),
+    jackpot_contribution: Number(entry.jackpotContribution || 0),
+    callback_url: entry.callbackUrl || '',
+    webhook_secret: entry.webhookSecret || '',
+    idempotency_key: entry.idempotencyKey || '',
+    payment_id: entry.paymentId || '',
+    bulk_parent_id: entry.bulkParentId || '',
+    bulk_index: (entry.bulkIndex !== undefined && entry.bulkIndex !== null) ? String(entry.bulkIndex) : '',
+    bulk_count: (entry.bulkCount !== undefined && entry.bulkCount !== null) ? String(entry.bulkCount) : '',
+    joined_at: new Date(entry.joinedAt || Date.now()).toISOString(),
+    disconnected: !!entry.disconnected,
+    disconnected_at: entry.disconnectedAt ? new Date(entry.disconnectedAt).toISOString() : ''
+  };
+}
+
+function rowToQueueEntry(row) {
+  return {
+    playerId: row.player_id,
+    displayName: row.display_name,
+    walletAddress: row.wallet_address || null,
+    gamePoolContribution: Number(row.game_pool_contribution || 0),
+    jackpotContribution: Number(row.jackpot_contribution || 0),
+    callbackUrl: row.callback_url || null,
+    webhookSecret: row.webhook_secret || null,
+    idempotencyKey: row.idempotency_key || null,
+    paymentId: row.payment_id || null,
+    bulkParentId: row.bulk_parent_id || null,
+    bulkIndex: row.bulk_index !== null ? Number(row.bulk_index) : undefined,
+    bulkCount: row.bulk_count !== null ? Number(row.bulk_count) : undefined,
+    joinedAt: row.joined_at ? Date.parse(row.joined_at) : Date.now(),
+    disconnected: !!row.disconnected,
+    disconnectedAt: row.disconnected_at ? Date.parse(row.disconnected_at) : null
+  };
+}
+
+/**
+ * Atomically replace the `queue_entries` table contents with the supplied
+ * in-memory queue. Fire-and-forget callers should `.catch(log)` — a lost
+ * snapshot write is a degradation (we still have the memory queue), not a
+ * request failure.
+ */
+async function persistQueueSnapshot(queueArray) {
+  if (!supabaseAdmin) return null;
+  const payload = (queueArray || []).map(queueEntryToRow);
+  try {
+    const { data, error } = await supabaseAdmin.rpc('replace_queue_snapshot', {
+      p_entries: payload
+    });
+    if (error) throw error;
+    return data;
+  } catch (err) {
+    log.warn({ err: err.message, size: payload.length }, 'Queue snapshot write failed (non-fatal)');
+    return null;
+  }
+}
+
+/**
+ * Rehydrate the queue from the durable log on server boot.
+ * Returns an array ordered by joined_at (oldest first) to preserve FIFO.
+ */
+async function loadQueueFromDB() {
+  if (!supabaseAdmin) return [];
+  try {
+    const { data, error } = await supabaseAdmin
+      .from('queue_entries')
+      .select('*')
+      .order('joined_at', { ascending: true });
+    if (error) throw error;
+    return (data || []).map(rowToQueueEntry);
+  } catch (err) {
+    log.error({ err: err.message }, 'Queue rehydrate failed — starting with empty queue');
+    return [];
+  }
+}
+
 /**
  * Check database connection
  * @returns {Promise<boolean>}
@@ -973,6 +1061,11 @@ module.exports = {
   updateJackpotPool,
   addToJackpotPool,
   deductFromJackpotPoolForWin,
+  // F06 — Queue persistence
+  persistQueueSnapshot,
+  loadQueueFromDB,
+  queueEntryToRow,
+  rowToQueueEntry,
   // Jackpot Tickets
   addJackpotTickets,
   getAllJackpotTickets,