diff --git a/indexer/index.mjs b/indexer/index.mjs index 998fc11..0ad3be4 100644 --- a/indexer/index.mjs +++ b/indexer/index.mjs @@ -1,4 +1,4 @@ -import { appendFileSync, existsSync, readFileSync, writeFileSync } from "node:fs"; +import { appendFileSync, existsSync, readFileSync, renameSync, writeFileSync } from "node:fs"; import { rpc, scValToNative } from "@stellar/stellar-sdk"; import { withRateLimitBackoff } from "./rpc-backoff.mjs"; @@ -31,20 +31,32 @@ if (!config.ok) { const { RPC_URL, CONTRACT_ID } = config.value; const OUT = process.env.OUT ?? "events.ndjson"; -const STATE = process.env.STATE ?? "state.json"; const POLL_MS = Number(process.env.POLL_MS ?? 10_000); const BACKOFF_INITIAL_MS = Number(process.env.BACKOFF_INITIAL_MS ?? 1_000); const BACKOFF_MAX_MS = Number(process.env.BACKOFF_MAX_MS ?? 60_000); const server = new rpc.Server(RPC_URL); -function loadCursor() { - if (!existsSync(STATE)) return null; - return JSON.parse(readFileSync(STATE, "utf8")).cursor ?? null; +function stateFile() { + return process.env.STATE ?? "state.json"; } -function saveCursor(cursor) { - writeFileSync(STATE, JSON.stringify({ cursor })); +export function loadCursor() { + const file = stateFile(); + if (!existsSync(file)) return null; + try { + return JSON.parse(readFileSync(file, "utf8")).cursor ?? null; + } catch { + console.warn(`[indexer] state file "${file}" is corrupt or empty; starting from default window`); + return null; + } +} + +export function saveCursor(cursor) { + const file = stateFile(); + const tmp = `${file}.tmp`; + writeFileSync(tmp, JSON.stringify({ cursor })); + renameSync(tmp, file); } function decode(ev) { @@ -173,6 +185,8 @@ async function poll() { } } -console.log(`indexing ${CONTRACT_ID} from ${RPC_URL} every ${POLL_MS}ms`); -await poll(); -intervalId = setInterval(() => poll().catch((e) => console.error(e.message ?? e)), POLL_MS); +if (import.meta.filename === process.argv[1]) { + console.log(`indexing ${CONTRACT_ID} from ${RPC_URL} every ${POLL_MS}ms`); + await poll(); + intervalId = setInterval(() => poll().catch((e) => console.error(e.message ?? e)), POLL_MS); +} diff --git a/indexer/index.test.mjs b/indexer/index.test.mjs index 9a0bfc9..958537c 100644 --- a/indexer/index.test.mjs +++ b/indexer/index.test.mjs @@ -1,6 +1,14 @@ import test from 'node:test'; import assert from 'node:assert/strict'; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; import { validateConfig } from './index.mjs'; +import { loadCursor, saveCursor } from './index.mjs'; + +// --------------------------------------------------------------------------- +// validateConfig +// --------------------------------------------------------------------------- test('validateConfig rejects missing required env values', () => { const result = validateConfig({ @@ -23,3 +31,92 @@ test('validateConfig accepts populated env values', () => { assert.equal(result.value.CONTRACT_ID, 'CC123'); assert.equal(result.value.RPC_URL, 'https://example.com'); }); + +// --------------------------------------------------------------------------- +// loadCursor / saveCursor +// --------------------------------------------------------------------------- + +/** + * Create a fresh temporary directory for each test and override the STATE + * environment variable so the functions use our isolated file. + */ +function withTempState(fn) { + return async () => { + const dir = mkdtempSync(join(tmpdir(), 'tributary-test-')); + const stateFile = join(dir, 'state.json'); + const original = process.env.STATE; + process.env.STATE = stateFile; + try { + await fn({ dir, stateFile }); + } finally { + process.env.STATE = original ?? undefined; + rmSync(dir, { recursive: true, force: true }); + } + }; +} + +test('loadCursor returns null when state file does not exist', withTempState(async ({ stateFile }) => { + assert.equal(existsSync(stateFile), false, 'precondition: file must not exist'); + // loadCursor/saveCursor read STATE from process.env at call time via the + // module-level binding, so we call them after setting the env var. + const cursor = loadCursor(); + assert.equal(cursor, null); +})); + +test('loadCursor returns null and warns on empty state file', withTempState(async ({ stateFile }) => { + writeFileSync(stateFile, ''); + + const warnings = []; + const origWarn = console.warn; + console.warn = (...args) => warnings.push(args.join(' ')); + + const cursor = loadCursor(); + + console.warn = origWarn; + + assert.equal(cursor, null); + assert.ok(warnings.length > 0, 'should have logged a warning'); + assert.ok(warnings[0].includes('corrupt or empty'), `unexpected warning: ${warnings[0]}`); +})); + +test('loadCursor returns null and warns on corrupt state file', withTempState(async ({ stateFile }) => { + writeFileSync(stateFile, '{bad json{{'); + + const warnings = []; + const origWarn = console.warn; + console.warn = (...args) => warnings.push(args.join(' ')); + + const cursor = loadCursor(); + + console.warn = origWarn; + + assert.equal(cursor, null); + assert.ok(warnings.length > 0, 'should have logged a warning'); +})); + +test('saveCursor writes atomically and loadCursor reads back the cursor', withTempState(async ({ dir, stateFile }) => { + const testCursor = '1234567890-1'; + + saveCursor(testCursor); + + // The .tmp file must not be left behind after a successful save. + assert.equal(existsSync(`${stateFile}.tmp`), false, '.tmp file should be cleaned up'); + + // The state file must exist and contain the correct cursor. + assert.ok(existsSync(stateFile), 'state.json must exist after saveCursor'); + + const raw = JSON.parse(readFileSync(stateFile, 'utf8')); + assert.equal(raw.cursor, testCursor); + + const loaded = loadCursor(); + assert.equal(loaded, testCursor); +})); + +test('saveCursor overwrites previous cursor atomically', withTempState(async ({ stateFile }) => { + saveCursor('cursor-v1'); + saveCursor('cursor-v2'); + + const loaded = loadCursor(); + assert.equal(loaded, 'cursor-v2'); + assert.equal(existsSync(`${stateFile}.tmp`), false); +}));