From 43a6c3642b2d87dafcf4ac83109d1b8b0ad2b70e Mon Sep 17 00:00:00 2001 From: Gorka Date: Thu, 18 Jun 2026 20:46:12 -0300 Subject: [PATCH] refactor(event-watcher): drop Deno.KV state store; boot syncs from oldest available MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Removes the second state store so a Postgres wipe fully resets the instance. - EventWatcher holds no durable cursor. A fresh watcher syncs ALL available history from the oldest ledger the RPC still retains (getHealth.oldestLedger); BOOT_SYNC_START_LEDGER_BLOCK overrides the start. Never defaults to "latest". - ChannelRegistry is rebuilt in-memory each boot (converge-by-query + this PP's own DB are the authoritative sources) — no KV persistence. - Sessions move to a plain in-memory Map; delete persistence/kv/config.ts disk KV and its entity. No Deno.openKv / .data / memory-kvdb anywhere in the path. - Inject rpc + startLedgerBlock into EventWatcher (decouples process.ts from env so the boot-ledger behavior is deterministically testable). Version bump 0.8.0 -> 0.9.0. --- deno.json | 4 +- src/config/env.ts | 17 ++++ .../sessions/in-memory-session-manager.ts | 37 +++++--- .../service/event-watcher/channel-registry.ts | 82 ++--------------- .../event-watcher.process.test.ts | 87 +++++++++++++++++++ .../event-watcher/event-watcher.process.ts | 73 +++++----------- src/core/service/event-watcher/index.ts | 13 ++- .../event-watcher/start-ledger.test.ts | 43 +++++++++ .../service/event-watcher/start-ledger.ts | 26 ++++++ src/models/auth/session/session.model.ts | 11 --- src/persistence/kv/config.ts | 16 ---- src/persistence/kv/entity/session.entity.ts | 25 ------ 12 files changed, 243 insertions(+), 191 deletions(-) create mode 100644 src/core/service/event-watcher/event-watcher.process.test.ts create mode 100644 src/core/service/event-watcher/start-ledger.test.ts create mode 100644 src/core/service/event-watcher/start-ledger.ts delete mode 100644 src/persistence/kv/config.ts delete mode 100644 src/persistence/kv/entity/session.entity.ts diff --git a/deno.json b/deno.json index 6ed16df..d7de48d 100644 --- a/deno.json +++ b/deno.json @@ -1,10 +1,10 @@ { "name": "@moonlight-protocol/provider-platform", - "version": "0.8.0", + "version": "0.9.0", "license": "MIT", "exports": "./src/main.ts", "tasks": { - "serve": "deno run --allow-all --unstable-kv src/main.ts", + "serve": "deno run --allow-all src/main.ts", "test": "deno test --allow-all --no-check --ignore=src/http/v1/pay/tests,src/http/v1/entities/tests src/ && deno test --allow-all --no-check --config src/http/v1/pay/tests/deno.json src/http/v1/pay/tests/ && deno test --allow-all --no-check --config src/http/v1/entities/tests/deno.json src/http/v1/entities/tests/", "test:integration": "deno test --allow-all --no-check --config tests/deno.json tests/", "test:unit": "deno test --allow-all --no-check --ignore=src/http/v1/pay/tests,src/http/v1/entities/tests src/", diff --git a/src/config/env.ts b/src/config/env.ts index 6ad3362..0855a35 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -121,3 +121,20 @@ export const TRANSACTION_EXPIRATION_OFFSET = _rawTxExpirationOffset; export const EVENT_WATCHER_INTERVAL_MS = Number( Deno.env.get("EVENT_WATCHER_INTERVAL_MS") ?? "30000", ); + +// Where a fresh watcher (no stored position) begins polling. Unset → SYNC ALL +// AVAILABLE: start from the oldest ledger the RPC still retains, so no in-window +// event is skipped on a cold boot. Set → start at exactly that ledger (e.g. to +// skip ancient history on a known-fresh deploy). Never defaults to "latest". +const _rawBootSyncStart = loadOptionalEnv("BOOT_SYNC_START_LEDGER_BLOCK"); +let _bootSyncStart: number | null = null; +if (_rawBootSyncStart !== undefined && _rawBootSyncStart !== "") { + const n = Number(_rawBootSyncStart); + if (!Number.isInteger(n) || n < 0) { + throw new Error( + `BOOT_SYNC_START_LEDGER_BLOCK must be a non-negative integer, got: "${_rawBootSyncStart}"`, + ); + } + _bootSyncStart = n; +} +export const BOOT_SYNC_START_LEDGER_BLOCK = _bootSyncStart; diff --git a/src/core/service/auth/sessions/in-memory-session-manager.ts b/src/core/service/auth/sessions/in-memory-session-manager.ts index 7539694..5863938 100644 --- a/src/core/service/auth/sessions/in-memory-session-manager.ts +++ b/src/core/service/auth/sessions/in-memory-session-manager.ts @@ -1,15 +1,24 @@ import { SESSION_TTL } from "@/config/env.ts"; -import { memDb } from "@/persistence/kv/config.ts"; import type { Session } from "@/models/auth/session/session.model.ts"; import type { Logger } from "@/utils/logger/index.ts"; +/** + * Live store of in-flight handshake sessions, keyed by challenge tx hash. + * + * Truly in-memory (a plain Map) — these are short-TTL auth handshakes, not + * derived state, so there is no benefit to persisting them. A machine + * stop/redeploy clears them and the client simply re-authenticates. Nothing + * here touches disk or Deno.KV. + */ export class InMemorySessionManager { + private sessions: Map = new Map(); private log: Logger; constructor(deps: { log: Logger }) { this.log = deps.log.scope("InMemorySessionManager"); } + // deno-lint-ignore require-await -- async to preserve the manager's contract public async addSession( txHash: string, clientAccount: string, @@ -19,7 +28,7 @@ export class InMemorySessionManager { this.log.info("addSession"); this.log.debug("txHash", txHash); - const cr = await memDb.sessions.add({ + this.sessions.set(txHash, { txHash, clientAccount, requestId, @@ -27,23 +36,22 @@ export class InMemorySessionManager { status: "PENDING", }); - if (cr.ok) { - this.log.event("session added to store"); - } + this.log.event("session added to store"); } + // deno-lint-ignore require-await -- async to preserve the manager's contract public async getSession(txHash: string): Promise { this.log.info("getSession"); this.log.debug("txHash", txHash); this.log.event("looking up session by txHash"); - const session = await memDb.sessions.findByPrimaryIndex("txHash", txHash); - if (session && Date.now() < session.value.expiresAt.getTime()) { + const session = this.sessions.get(txHash); + if (session && Date.now() < session.expiresAt.getTime()) { this.log.event("session found and valid"); - return session.value; + return session; } this.log.event("session missing or expired, deleting"); - await memDb.sessions.delete(txHash); + this.sessions.delete(txHash); return undefined; } @@ -60,17 +68,20 @@ export class InMemorySessionManager { } this.log.event("persisting session update"); - await memDb.sessions.update(session.txHash, session); + this.sessions.set(session.txHash, session); } + // deno-lint-ignore require-await -- async to preserve the manager's contract public async cleanupExpired(): Promise { const now = Date.now(); this.log.info("cleanupExpired"); - await memDb.sessions.deleteMany({ - filter: (doc) => doc.value.expiresAt.getTime() < now, - }); + for (const [txHash, session] of this.sessions) { + if (session.expiresAt.getTime() < now) { + this.sessions.delete(txHash); + } + } this.log.event("expired sessions cleaned"); } diff --git a/src/core/service/event-watcher/channel-registry.ts b/src/core/service/event-watcher/channel-registry.ts index fe85e8e..5ad5b88 100644 --- a/src/core/service/event-watcher/channel-registry.ts +++ b/src/core/service/event-watcher/channel-registry.ts @@ -24,10 +24,6 @@ export interface ChannelRecord { removedAtLedger?: number; } -const REGISTRY_KV_KEY = ["channel-registry", "channels"]; -const KV_DIR = new URL("../../../../.data", import.meta.url).pathname; -const KV_PATH = `${KV_DIR}/memory-kvdb.db`; - /** * In-memory registry of channels this PP is registered in. * @@ -35,14 +31,14 @@ const KV_PATH = `${KV_DIR}/memory-kvdb.db`; * The `configuredChannels` set represents which channels the operator * has chosen to actively serve (from instance config). * - * State is persisted to Deno KV so restarts don't lose channel info. - * On startup, the registry restores from KV first, then seeds any - * configured channels that aren't already tracked. + * State is NOT persisted: every record it holds is authoritatively + * re-queryable (channel membership from this PP's own DB, asset-channel status + * from the council via converge-by-query on boot), so the registry is rebuilt + * from scratch on each start rather than restored from a durable store. */ export class ChannelRegistry { private channels: Map = new Map(); private configuredChannels: Set; - private kv: Deno.Kv | null = null; private log: Logger; constructor(configuredChannelIds: string[], deps: { log: Logger }) { @@ -50,40 +46,6 @@ export class ChannelRegistry { this.log = deps.log.scope("ChannelRegistry"); } - /** - * Initialize the registry: restore from KV, then seed any configured - * channels not already present. Must be called before use. - */ - async initialize(): Promise { - await Deno.mkdir(KV_DIR, { recursive: true }); - this.kv = await Deno.openKv(KV_PATH); - - // Restore persisted state - const stored = await this.kv.get(REGISTRY_KV_KEY); - if (stored.value && stored.value.length > 0) { - for (const record of stored.value) { - this.channels.set(record.contractId, record); - } - this.log.debug("count", stored.value.length); - this.log.event("ChannelRegistry restored from KV"); - } - - // Seed configured channels that aren't already tracked - for (const contractId of this.configuredChannels) { - if (!this.channels.has(contractId)) { - this.channels.set(contractId, { - contractId, - state: "active", - registeredAtLedger: 0, - }); - this.log.debug("contractId", contractId); - this.log.event("seeded configured channel as active"); - } - } - - await this.persist(); - } - /** Dynamically add a channel to track (e.g., when a PP joins a new council). */ addChannel(contractId: string): void { this.log.info("addChannel"); @@ -137,6 +99,7 @@ export class ChannelRegistry { * `enabled=true` → `active` (full service resumes). Idempotent: re-applying * the same state is a no-op. */ + // deno-lint-ignore require-await -- async to preserve the awaited mutation contract async applyChannelState( channelContractId: string, enabled: boolean, @@ -160,7 +123,6 @@ export class ChannelRegistry { this.log.event( enabled ? "asset channel enabled" : "asset channel disabled", ); - await this.persist(); } /** @@ -172,6 +134,7 @@ export class ChannelRegistry { return this.channels.get(channelContractId)?.state === "disabled"; } + // deno-lint-ignore require-await -- async to preserve the awaited mutation contract private async onProviderAdded(event: ChannelAuthEvent): Promise { const isConfigured = this.configuredChannels.has(event.contractId); const state: ChannelState = isConfigured ? "active" : "pending"; @@ -186,10 +149,9 @@ export class ChannelRegistry { this.log.debug("state", state); this.log.debug("ledger", event.ledger); this.log.event("provider registered in channel"); - - await this.persist(); } + // deno-lint-ignore require-await -- async to preserve the awaited mutation contract private async onProviderRemoved(event: ChannelAuthEvent): Promise { const existing = this.channels.get(event.contractId); if (existing) { @@ -207,20 +169,6 @@ export class ChannelRegistry { this.log.debug("contractId", event.contractId); this.log.debug("ledger", event.ledger); this.log.event("provider removed from channel"); - - await this.persist(); - } - - /** - * Persist current registry state to KV. - */ - private async persist(): Promise { - if (!this.kv) return; - try { - await this.kv.set(REGISTRY_KV_KEY, this.getAll()); - } catch (error) { - this.log.error(error, "failed to persist channel registry"); - } } /** @@ -247,6 +195,7 @@ export class ChannelRegistry { /** * Mark a channel as configured (operator activated it). */ + // deno-lint-ignore require-await -- async to preserve the awaited mutation contract async activateChannel(contractId: string): Promise { this.log.info("activateChannel"); this.log.debug("contractId", contractId); @@ -255,13 +204,13 @@ export class ChannelRegistry { if (channel && channel.state === "pending") { channel.state = "active"; this.log.event("channel transitioned to active"); - await this.persist(); } } /** * Mark a channel as no longer configured by the operator. */ + // deno-lint-ignore require-await -- async to preserve the awaited mutation contract async deactivateChannel(contractId: string): Promise { this.log.info("deactivateChannel"); this.log.debug("contractId", contractId); @@ -270,19 +219,6 @@ export class ChannelRegistry { if (channel && channel.state === "active") { channel.state = "pending"; this.log.event("channel transitioned to pending"); - await this.persist(); - } - } - - /** - * Close the KV handle. Call on shutdown. - */ - close(): void { - this.log.info("close"); - if (this.kv) { - this.kv.close(); - this.kv = null; - this.log.event("KV handle closed"); } } } diff --git a/src/core/service/event-watcher/event-watcher.process.test.ts b/src/core/service/event-watcher/event-watcher.process.test.ts new file mode 100644 index 0000000..21b4682 --- /dev/null +++ b/src/core/service/event-watcher/event-watcher.process.test.ts @@ -0,0 +1,87 @@ +import { assertEquals } from "@std/assert"; +import type { Server } from "stellar-sdk/rpc"; +import { EventWatcher } from "./event-watcher.process.ts"; +import { newNoop } from "@/utils/logger/index.ts"; + +const CONTRACT = "CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAFCT4"; + +/** + * Records the `startLedger` of each getEvents call so a test can assert exactly + * where a fresh watcher began polling. `latestLedger` controls how far the + * in-memory cursor advances after a poll. + */ +function mockRpc(opts: { oldestLedger: number; latestLedger: number }) { + const startLedgers: number[] = []; + const rpc = { + // deno-lint-ignore require-await -- mock satisfies async getHealth contract + getHealth: async () => ({ oldestLedger: opts.oldestLedger }), + // deno-lint-ignore require-await -- mock satisfies async getEvents contract + getEvents: async (req: { startLedger: number }) => { + startLedgers.push(req.startLedger); + return { events: [], latestLedger: opts.latestLedger }; + }, + // deno-lint-ignore require-await -- mock satisfies async getLatestLedger contract + getLatestLedger: async () => ({ sequence: opts.latestLedger }), + } as unknown as Server; + return { rpc, startLedgers }; +} + +Deno.test("EventWatcher - override set → first getEvents at exactly that ledger", async () => { + const { rpc, startLedgers } = mockRpc({ + oldestLedger: 5000, + latestLedger: 5100, + }); + const watcher = new EventWatcher( + { contractId: CONTRACT, intervalMs: 60_000 }, + { log: newNoop(), rpc, startLedgerBlock: 12345 }, + ); + + await watcher.start(); + watcher.stop(); + + assertEquals(startLedgers[0], 12345); +}); + +Deno.test("EventWatcher - override unset → first getEvents at oldest available", async () => { + const { rpc, startLedgers } = mockRpc({ + oldestLedger: 5000, + latestLedger: 5100, + }); + const watcher = new EventWatcher( + { contractId: CONTRACT, intervalMs: 60_000 }, + { log: newNoop(), rpc, startLedgerBlock: null }, + ); + + await watcher.start(); + watcher.stop(); + + assertEquals(startLedgers[0], 5000); +}); + +// Let the fire-and-forget first poll (scheduled by start()) run to completion. +const tick = () => new Promise((resolve) => setTimeout(resolve, 0)); + +Deno.test("EventWatcher - holds no durable cursor: a restart re-syncs from oldest", async () => { + // First watcher polls once and advances its in-memory cursor past 5100. + const first = mockRpc({ oldestLedger: 5000, latestLedger: 5100 }); + const w1 = new EventWatcher( + { contractId: CONTRACT, intervalMs: 60_000 }, + { log: newNoop(), rpc: first.rpc, startLedgerBlock: null }, + ); + await w1.start(); + await tick(); // wait for the first poll to advance the in-memory cursor + w1.stop(); + assertEquals(w1.getLastLedger(), 5101); // advanced in memory + + // A fresh watcher (simulating a process restart) must start from oldest + // again — nothing was persisted, so it does not resume at 5101. + const second = mockRpc({ oldestLedger: 5000, latestLedger: 5100 }); + const w2 = new EventWatcher( + { contractId: CONTRACT, intervalMs: 60_000 }, + { log: newNoop(), rpc: second.rpc, startLedgerBlock: null }, + ); + await w2.start(); + w2.stop(); + + assertEquals(second.startLedgers[0], 5000); +}); diff --git a/src/core/service/event-watcher/event-watcher.process.ts b/src/core/service/event-watcher/event-watcher.process.ts index 2cd2289..6716a6e 100644 --- a/src/core/service/event-watcher/event-watcher.process.ts +++ b/src/core/service/event-watcher/event-watcher.process.ts @@ -1,5 +1,5 @@ import type { Logger } from "@/utils/logger/index.ts"; -import { NETWORK_RPC_SERVER } from "@/config/env.ts"; +import type { Server } from "stellar-sdk/rpc"; import { fetchChannelAuthEvents } from "./event-watcher.service.ts"; import type { ChannelAuthEvent, @@ -7,10 +7,7 @@ import type { } from "./event-watcher.types.ts"; import { withSpan } from "@/core/tracing.ts"; import { recoverFromOutOfRetention } from "./retention.ts"; - -function cursorKvKey(contractId: string): Deno.KvKey { - return ["event-watcher", contractId, "lastLedger"]; -} +import { resolveBootStartLedger } from "./start-ledger.ts"; export type EventHandler = (event: ChannelAuthEvent) => void | Promise; export type ResyncHandler = () => void | Promise; @@ -22,8 +19,11 @@ export type ResyncHandler = () => void | Promise; * Uses a self-scheduling pattern (setTimeout after each poll completes) * to prevent concurrent polls when RPC is slow. * - * Persists the last processed ledger to Deno KV so restarts don't - * lose event history. + * The watcher holds NO durable cursor: provider-platform reconstructs all + * derived state by querying the council on boot (converge-by-query), so a fresh + * watcher simply syncs all available history forward from the resolved boot + * ledger (see `resolveBootStartLedger`). Events are a live delta on top of that + * baseline. * * Consumers register handlers via `onEvent()` and the watcher * dispatches parsed events as they arrive. @@ -35,17 +35,20 @@ export class EventWatcher { private config: EventWatcherConfig; private handlers: EventHandler[] = []; private resyncHandlers: ResyncHandler[] = []; - private kv: Deno.Kv | null = null; + private rpc: Server; + private startLedgerBlock: number | null; private log: Logger; constructor( config: { contractId: string; intervalMs?: number }, - deps: { log: Logger }, + deps: { log: Logger; rpc: Server; startLedgerBlock: number | null }, ) { this.config = { contractId: config.contractId, intervalMs: config.intervalMs ?? 30_000, }; + this.rpc = deps.rpc; + this.startLedgerBlock = deps.startLedgerBlock; this.log = deps.log.scope("EventWatcher"); } @@ -75,26 +78,15 @@ export class EventWatcher { this.isRunning = true; - // Open KV for cursor persistence - await Deno.mkdir(".data", { recursive: true }); - this.kv = await Deno.openKv("./.data/memory-kvdb.db"); - - // Restore cursor from KV, or fall back to current network ledger - const stored = await this.kv.get( - cursorKvKey(this.config.contractId), + // No persisted cursor: resolve the boot start ledger (oldest available, or + // the configured override) and sync forward. + this.lastLedger = await resolveBootStartLedger( + this.rpc, + this.startLedgerBlock, ); - if (stored.value !== null) { - this.lastLedger = stored.value; - this.log.debug("contractId", this.config.contractId); - this.log.debug("startLedger", this.lastLedger); - this.log.event("EventWatcher restored cursor from KV"); - } else { - const latestLedger = await NETWORK_RPC_SERVER.getLatestLedger(); - this.lastLedger = latestLedger.sequence; - this.log.debug("contractId", this.config.contractId); - this.log.debug("startLedger", this.lastLedger); - this.log.event("EventWatcher initialized from network (no saved cursor)"); - } + this.log.debug("contractId", this.config.contractId); + this.log.debug("startLedger", this.lastLedger); + this.log.event("EventWatcher initialized boot start ledger"); // Start the self-scheduling loop this.scheduleNext(); @@ -111,10 +103,6 @@ export class EventWatcher { clearTimeout(this.timeoutId); this.timeoutId = null; } - if (this.kv) { - this.kv.close(); - this.kv = null; - } this.log.event("EventWatcher stopped"); } @@ -150,7 +138,7 @@ export class EventWatcher { if (this.lastLedger === null) return; const { events, latestLedger } = await fetchChannelAuthEvents( - NETWORK_RPC_SERVER, + this.rpc, this.config.contractId, this.lastLedger, { log: this.log }, @@ -169,16 +157,9 @@ export class EventWatcher { } } - // Advance cursor past the latest ledger we've seen + // Advance cursor past the latest ledger we've seen (in memory only — + // there is no durable cursor; converge-by-query is the recovery path). this.lastLedger = latestLedger + 1; - - // Persist cursor to KV - if (this.kv) { - await this.kv.set( - cursorKvKey(this.config.contractId), - this.lastLedger, - ); - } } catch (error) { span.addEvent("poll_error", { "error.message": error instanceof Error @@ -189,19 +170,13 @@ export class EventWatcher { try { const recoveredCursor = await recoverFromOutOfRetention( error, - () => NETWORK_RPC_SERVER.getLatestLedger(), + () => this.rpc.getLatestLedger(), () => this.fireResync(), this.log, ); if (recoveredCursor !== null) { span.addEvent("out_of_retention_recovery"); this.lastLedger = recoveredCursor; - if (this.kv) { - await this.kv.set( - cursorKvKey(this.config.contractId), - this.lastLedger, - ); - } return; } } catch (recoveryError) { diff --git a/src/core/service/event-watcher/index.ts b/src/core/service/event-watcher/index.ts index e7ed816..9d92486 100644 --- a/src/core/service/event-watcher/index.ts +++ b/src/core/service/event-watcher/index.ts @@ -1,4 +1,9 @@ -import { CHALLENGE_TTL, EVENT_WATCHER_INTERVAL_MS } from "@/config/env.ts"; +import { + BOOT_SYNC_START_LEDGER_BLOCK, + CHALLENGE_TTL, + EVENT_WATCHER_INTERVAL_MS, + NETWORK_RPC_SERVER, +} from "@/config/env.ts"; import { EventWatcher } from "./event-watcher.process.ts"; import { ChannelRegistry } from "./channel-registry.ts"; import { @@ -63,7 +68,11 @@ async function ensureWatcher(channelAuthId: string): Promise { const watcher = new EventWatcher({ contractId: channelAuthId, intervalMs: EVENT_WATCHER_INTERVAL_MS, - }, { log: watcherLog! }); + }, { + log: watcherLog!, + rpc: NETWORK_RPC_SERVER, + startLedgerBlock: BOOT_SYNC_START_LEDGER_BLOCK, + }); // Re-query the council and reconcile asset-channel statuses when the event // cursor falls out of Stellar RPC retention (events may have been missed; the diff --git a/src/core/service/event-watcher/start-ledger.test.ts b/src/core/service/event-watcher/start-ledger.test.ts new file mode 100644 index 0000000..5b9bfe7 --- /dev/null +++ b/src/core/service/event-watcher/start-ledger.test.ts @@ -0,0 +1,43 @@ +import { assertEquals } from "@std/assert"; +import { type BootSyncRpc, resolveBootStartLedger } from "./start-ledger.ts"; + +function rpcWithOldest(oldestLedger: number): BootSyncRpc & { + healthCalls: number; +} { + return { + healthCalls: 0, + // deno-lint-ignore require-await -- mock satisfies async getHealth contract + async getHealth() { + this.healthCalls++; + return { oldestLedger }; + }, + }; +} + +Deno.test("resolveBootStartLedger - override set → starts at exactly that ledger", async () => { + const rpc = rpcWithOldest(5000); + + const start = await resolveBootStartLedger(rpc, 12345); + + assertEquals(start, 12345); + // The override wins without consulting the RPC at all. + assertEquals(rpc.healthCalls, 0); +}); + +Deno.test("resolveBootStartLedger - override unset → starts at oldest available", async () => { + const rpc = rpcWithOldest(5000); + + const start = await resolveBootStartLedger(rpc, null); + + assertEquals(start, 5000); + assertEquals(rpc.healthCalls, 1); +}); + +Deno.test("resolveBootStartLedger - override of 0 is honored (not treated as unset)", async () => { + const rpc = rpcWithOldest(5000); + + const start = await resolveBootStartLedger(rpc, 0); + + assertEquals(start, 0); + assertEquals(rpc.healthCalls, 0); +}); diff --git a/src/core/service/event-watcher/start-ledger.ts b/src/core/service/event-watcher/start-ledger.ts new file mode 100644 index 0000000..5a8363d --- /dev/null +++ b/src/core/service/event-watcher/start-ledger.ts @@ -0,0 +1,26 @@ +/** + * Minimal RPC surface needed to resolve a boot start ledger. The Stellar RPC + * `getHealth` response carries `oldestLedger` — the oldest ledger still inside + * the server's retention window. + */ +export interface BootSyncRpc { + getHealth(): Promise<{ oldestLedger: number }>; +} + +/** + * Resolve the ledger a fresh watcher (no stored position) should begin polling. + * + * Default is SYNC ALL AVAILABLE: start from the oldest ledger the RPC still + * retains, so no on-chain event inside the retention window is skipped on a cold + * boot. An explicit `BOOT_SYNC_START_LEDGER_BLOCK` override starts at that exact + * ledger instead. This never falls back to "latest" — that would silently skip + * the gap between the last poll and now. + */ +export async function resolveBootStartLedger( + rpc: BootSyncRpc, + startLedgerBlock: number | null, +): Promise { + if (startLedgerBlock !== null) return startLedgerBlock; + const { oldestLedger } = await rpc.getHealth(); + return oldestLedger; +} diff --git a/src/models/auth/session/session.model.ts b/src/models/auth/session/session.model.ts index 0d9b585..7ccd333 100644 --- a/src/models/auth/session/session.model.ts +++ b/src/models/auth/session/session.model.ts @@ -1,4 +1,3 @@ -import { collection } from "@olli/kvdex"; import { z } from "zod"; const sessionStatusModel = z.enum(["PENDING", "ACTIVE"]); @@ -13,13 +12,3 @@ const sessionModel = z.object({ export type SessionStatus = z.infer; export type Session = z.infer; - -export const sessionCollection = collection(sessionModel, { - idGenerator: (session) => session.txHash, - indices: { - txHash: "primary", - clientAccount: "secondary", - expiresAt: "secondary", - status: "secondary", - }, -}); diff --git a/src/persistence/kv/config.ts b/src/persistence/kv/config.ts deleted file mode 100644 index d135560..0000000 --- a/src/persistence/kv/config.ts +++ /dev/null @@ -1,16 +0,0 @@ -import { kvdex } from "@olli/kvdex"; -import { sessionCollection } from "./entity/session.entity.ts"; - -// Memory Database -// Ensure .data directory exists before opening KV -await Deno.mkdir(".data", { recursive: true }); -const memoryKv = await Deno.openKv("./.data/memory-kvdb.db"); - -const memDb = kvdex({ - kv: memoryKv, - schema: { - sessions: sessionCollection, - }, -}); - -export { memDb, memoryKv }; diff --git a/src/persistence/kv/entity/session.entity.ts b/src/persistence/kv/entity/session.entity.ts deleted file mode 100644 index bf75deb..0000000 --- a/src/persistence/kv/entity/session.entity.ts +++ /dev/null @@ -1,25 +0,0 @@ -import { collection } from "@olli/kvdex"; -import { z } from "zod"; - -const sessionStatusEnum = z.enum(["PENDING", "ACTIVE"]); - -const sessionEntity = z.object({ - txHash: z.string(), - clientAccount: z.string(), - expiresAt: z.date(), - status: sessionStatusEnum, - requestId: z.string(), -}); - -export type SessionStatus = typeof sessionStatusEnum; -export type Session = z.infer; - -export const sessionCollection = collection(sessionEntity, { - idGenerator: (session) => session.txHash, - indices: { - txHash: "primary", - clientAccount: "secondary", - expiresAt: "secondary", - status: "secondary", - }, -});