// Snapshot store adapter: wraps the SnapStore Durable Object (production) or an in-memory object (tests), and runs
// refreshes (cron every 5 min; a lazy refresh is kicked in the background when a read finds the snapshot missing/old).
import { buildSnapshot } from './engine.js';
import { withDeadline } from './deadline.js';

export const LAZY_REFRESH_AFTER_MS = 8 * 60_000;

/** Refresh once under the store lock. Returns a short status object (never throws). */
export async function refresh(stub, { fetchImpl = globalThis.fetch, now = Date.now() } = {}) {
  try {
    if (!(await stub.lock(now))) return { ok: false, skipped: 'locked' };
    try {
      const prev = await stub.cohort();
      const { snapshot, cohortRecord } = await buildSnapshot({ fetchImpl, now, cohortRecord: prev });
      await stub.put(snapshot, cohortRecord !== prev ? cohortRecord : null);
      return { ok: true, covered: snapshot.covered, size: snapshot.cohort_size, coins: Object.keys(snapshot.coins).length, failed: snapshot.failed };
    } catch (e) { try { await stub.unlock(); } catch { /* lock expires */ } return { ok: false, error: String(e?.message || e).slice(0, 160) }; }
  } catch (e) { return { ok: false, error: String(e?.message || e).slice(0, 160) }; }
}

/** Store used by the app: read() with a 20 s per-isolate cache; a stale/missing snapshot kicks a background refresh. */
// `stub` may be a function returning a fresh stub: Durable Object stubs are bound to the request that created them, so the
// Worker passes () => env.SNAP.get(id) and a new stub is made per read (reusing one across requests throws cross-request I/O).
export function makeStore(stubOrFn, { fetchImpl = globalThis.fetch, now = () => Date.now(), readMs = 4000 } = {}) {
  let cached = null; let at = 0; let kicked = 0;
  return {
    async read(waitUntil) {
      const t = now();
      if (cached && t - at < 20_000) return cached;
      const stub = typeof stubOrFn === 'function' ? stubOrFn() : stubOrFn;
      const r = await withDeadline(stub.read(), readMs, 'snapshot store');
      cached = r; at = t;
      const age = r.snapshot ? t - r.snapshot.at : Infinity;
      if (age > LAZY_REFRESH_AFTER_MS && t - kicked > 60_000 && waitUntil) { kicked = t; try { waitUntil(refresh(stub, { fetchImpl, now: t })); } catch { /* not on Workers */ } }
      return r;
    },
    async cohort() { try { return (await this.read()).cohort; } catch { return null; } },
  };
}

/** In-memory stub with the SnapStore interface (tests, local node server). */
export function memoryStub() {
  const s = { snapshot: null, cohortRecord: null, hist: [], lockAt: 0 };
  return {
    state: s,
    async lock(now) { if (now - s.lockAt < 120_000) return false; s.lockAt = now; return true; },
    async unlock() { s.lockAt = 0; return true; },
    async cohort() { return s.cohortRecord; },
    async put(snapshot, cohortRecord) {
      s.snapshot = snapshot; if (cohortRecord) s.cohortRecord = cohortRecord; s.lockAt = 0;
      s.hist.push({ at: snapshot.at, coins: Object.fromEntries(Object.entries(snapshot.coins).map(([c, a]) => [c, { net: a.net, ln: a.ln, sn: a.sn, tl: a.tl, ts: a.ts }])) });
      return true;
    },
    async read() { return { snapshot: s.snapshot, cohort: s.cohortRecord?.cohort || [], hist: s.hist.filter((h) => !s.snapshot || h.at < s.snapshot.at) }; },
  };
}

agentgate-source: code mirror on Cloudflare Pages. All repos