diff --git a/backend/__tests__/eventIndexerCursor.test.js b/backend/__tests__/eventIndexerCursor.test.js new file mode 100644 index 00000000..2e9dd3c9 --- /dev/null +++ b/backend/__tests__/eventIndexerCursor.test.js @@ -0,0 +1,279 @@ +"use strict"; + +const nock = require("nock"); +const { parseEvent } = require("../src/services/eventParser"); + +const SOROBAN_RPC_URL = "https://soroban-testnet.stellar.org"; +const TEST_CONTRACT_ID = "CDEFGHIJKLMNOPQRSTUVWXYZABCDEFGHIJKLMNOPQRSTUVWXYZ123456"; + +function event(id, ledger = 5) { + return { + type: "contract", + ledger, + ledgerClosedAt: "2026-08-25T12:00:00.000Z", + contractId: TEST_CONTRACT_ID, + id, + pagingToken: `token-${id}`, + txHash: `${String(id).replace(/[^a-f0-9]/gi, "a").padEnd(64, "a").slice(0, 64)}`, + topic: ["tip", "GFROM", "GTO"], + data: { amount: "100" }, + }; +} + +function loadIndexer({ databaseUrl, pool } = {}) { + jest.resetModules(); + process.env.SOROBAN_RPC_URL = SOROBAN_RPC_URL; + process.env.CONTRACT_ID = TEST_CONTRACT_ID; + + if (databaseUrl) { + process.env.DATABASE_URL = databaseUrl; + jest.doMock("pg", () => ({ + Pool: jest.fn(() => pool), + })); + } else { + delete process.env.DATABASE_URL; + jest.dontMock("pg"); + } + + jest.doMock("../src/services/pushNotifier", () => ({ + notifyContractEvents: jest.fn(), + })); + + return require("../src/services/eventIndexer"); +} + +function mockRpc(method, result, bodyPredicate = () => true) { + return nock(SOROBAN_RPC_URL) + .post("/", (body) => body.method === method && bodyPredicate(body)) + .reply(200, { + jsonrpc: "2.0", + id: 1, + result, + }); +} + +describe("eventIndexer cursor reliability", () => { + afterEach(() => { + nock.cleanAll(); + jest.dontMock("pg"); + delete process.env.DATABASE_URL; + }); + + it("loads the cursor from indexer_state instead of MAX(contract_events.ledger_sequence)", async () => { + const queries = []; + const pool = { + on: jest.fn(), + query: jest.fn(async (sql) => { + queries.push(sql); + if (String(sql).includes("SELECT last_processed_ledger")) { + return { + rows: [ + { + last_processed_ledger: 42, + last_processed_tx_id: "tx-42", + }, + ], + }; + } + return { rows: [], rowCount: 1 }; + }), + end: jest.fn(async () => {}), + }; + const indexer = loadIndexer({ databaseUrl: "postgres://test", pool }); + + await expect(indexer._internals.loadCursor()).resolves.toBe(42); + + expect(queries.join("\n")).toContain("indexer_state"); + expect(queries.join("\n")).not.toContain("MAX(ledger_sequence)"); + expect(indexer._internals.getCursorForTest()).toMatchObject({ + lastProcessedLedger: 42, + lastProcessedTxId: "tx-42", + }); + }); + + it("follows getEvents pagination cursors until exhausted", async () => { + const pageBodies = []; + mockRpc( + "getEvents", + { events: [event("a1", 10)], cursor: "cursor-1" }, + (body) => { + pageBodies.push(body); + return body.params.startLedger === 10 && body.params.endLedger === 13; + }, + ); + mockRpc( + "getEvents", + { events: [event("b2", 11)], cursor: null }, + (body) => { + pageBodies.push(body); + return body.params.pagination.cursor === "cursor-1" && body.params.startLedger === undefined; + }, + ); + const indexer = loadIndexer(); + + const events = await indexer._internals.getEvents(10, 13); + + expect(events).toHaveLength(2); + expect(events.map((ev) => ev.id)).toEqual(["a1", "b2"]); + expect(pageBodies[1].params.endLedger).toBeUndefined(); + }); + + it("counts ON CONFLICT DO NOTHING rows as conflicts, not inserts", async () => { + const client = { + query: jest + .fn() + .mockResolvedValueOnce({}) + .mockResolvedValueOnce({ rowCount: 1 }) + .mockResolvedValueOnce({ rowCount: 0 }) + .mockResolvedValueOnce({}), + release: jest.fn(), + }; + const pool = { + on: jest.fn(), + connect: jest.fn(async () => client), + end: jest.fn(async () => {}), + }; + const indexer = loadIndexer({ databaseUrl: "postgres://test", pool }); + + const summary = await indexer._internals.storeEvents([ + { + event_type: "tip", + contract_id: TEST_CONTRACT_ID, + ledger_sequence: 5, + emitted_at: "2026-08-25T12:00:00.000Z", + payload: { eventId: "a" }, + }, + { + event_type: "tip", + contract_id: TEST_CONTRACT_ID, + ledger_sequence: 5, + emitted_at: "2026-08-25T12:00:00.000Z", + payload: { eventId: "b" }, + }, + ]); + + expect(summary).toEqual({ inserted: 1, conflicts: 1, errors: 0 }); + expect(client.query).toHaveBeenCalledWith("COMMIT"); + }); + + it("stores the RPC event id in payload.id for the deduplication index", () => { + expect(parseEvent(event("dedupe-id", 7)).payload).toMatchObject({ + id: "dedupe-id", + eventId: "dedupe-id", + }); + }); + + it("does not advance the watermark when a conflict occurs in the range", async () => { + const client = { + query: jest.fn().mockResolvedValueOnce({}).mockResolvedValueOnce({ rowCount: 0 }).mockResolvedValueOnce({}), + release: jest.fn(), + }; + const pool = { + on: jest.fn(), + connect: jest.fn(async () => client), + query: jest.fn(async () => ({ rows: [], rowCount: 1 })), + end: jest.fn(async () => {}), + }; + mockRpc("getLatestLedger", { sequence: 5 }); + mockRpc("getEvents", { events: [event("conflict", 5)], cursor: null }); + const indexer = loadIndexer({ databaseUrl: "postgres://test", pool }); + + await indexer._internals.pollOnce(); + + expect(indexer._internals.getCursorForTest().lastProcessedLedger).toBe(0); + expect(pool.query).not.toHaveBeenCalledWith(expect.stringContaining("last_processed_ledger = EXCLUDED"), expect.any(Array)); + }); + + it("advances the PostgreSQL cursor in the same transaction as a clean batch", async () => { + const client = { + query: jest + .fn() + .mockResolvedValueOnce({}) + .mockResolvedValueOnce({ rowCount: 1 }) + .mockResolvedValueOnce({ rowCount: 1 }) + .mockResolvedValueOnce({}), + release: jest.fn(), + }; + const pool = { + on: jest.fn(), + connect: jest.fn(async () => client), + query: jest.fn(async () => ({ rows: [], rowCount: 1 })), + end: jest.fn(async () => {}), + }; + mockRpc("getLatestLedger", { sequence: 5 }); + mockRpc("getEvents", { events: [event("clean", 5)], cursor: null }); + const indexer = loadIndexer({ databaseUrl: "postgres://test", pool }); + + await indexer._internals.pollOnce(); + + const calls = client.query.mock.calls.map(([sql]) => String(sql)); + const cursorCall = calls.findIndex((sql) => sql.includes("indexer_state")); + const commitCall = calls.findIndex((sql) => sql === "COMMIT"); + + expect(cursorCall).toBeGreaterThan(-1); + expect(commitCall).toBeGreaterThan(cursorCall); + expect(pool.query).not.toHaveBeenCalledWith(expect.stringContaining("indexer_state"), expect.any(Array)); + expect(indexer._internals.getCursorForTest().lastProcessedLedger).toBe(5); + }); + + it("advances the cursor after a fully paged empty PostgreSQL range", async () => { + const pool = { + on: jest.fn(), + query: jest.fn(async () => ({ rows: [], rowCount: 1 })), + end: jest.fn(async () => {}), + }; + mockRpc("getLatestLedger", { sequence: 6 }); + mockRpc("getEvents", { events: [], cursor: "empty-next" }); + mockRpc("getEvents", { events: [], cursor: null }, (body) => { + return body.params.pagination.cursor === "empty-next"; + }); + const indexer = loadIndexer({ databaseUrl: "postgres://test", pool }); + + await indexer._internals.pollOnce(); + + expect(pool.query).toHaveBeenCalledWith(expect.stringContaining("indexer_state"), [ + "contract_events", + 6, + null, + ]); + expect(indexer._internals.getCursorForTest().lastProcessedLedger).toBe(6); + }); + + it("does not advance the watermark when an insert error rolls back the batch", async () => { + const client = { + query: jest + .fn() + .mockResolvedValueOnce({}) + .mockRejectedValueOnce(new Error("insert failed")) + .mockResolvedValueOnce({}), + release: jest.fn(), + }; + const pool = { + on: jest.fn(), + connect: jest.fn(async () => client), + query: jest.fn(async () => ({ rows: [], rowCount: 1 })), + end: jest.fn(async () => {}), + }; + mockRpc("getLatestLedger", { sequence: 5 }); + mockRpc("getEvents", { events: [event("bad", 5)], cursor: null }); + const indexer = loadIndexer({ databaseUrl: "postgres://test", pool }); + + await indexer._internals.pollOnce(); + + expect(client.query).toHaveBeenCalledWith("ROLLBACK"); + expect(indexer._internals.getCursorForTest().lastProcessedLedger).toBe(0); + }); + + it("advances after a clean multi-page range, including an empty first page", async () => { + mockRpc("getLatestLedger", { sequence: 8 }); + mockRpc("getEvents", { events: [], cursor: "next-page" }); + mockRpc("getEvents", { events: [event("late", 8)], cursor: null }, (body) => { + return body.params.pagination.cursor === "next-page"; + }); + const indexer = loadIndexer(); + + await indexer._internals.pollOnce(); + + expect(indexer._internals.getCursorForTest().lastProcessedLedger).toBe(8); + }); +}); diff --git a/backend/migrations/028_indexer_state.js b/backend/migrations/028_indexer_state.js new file mode 100644 index 00000000..a983764a --- /dev/null +++ b/backend/migrations/028_indexer_state.js @@ -0,0 +1,31 @@ +"use strict"; + +/** + * Migration 028: Store resumable indexer cursors outside contract_events. + */ + +exports.up = async function (knex) { + const hasTable = await knex.schema.hasTable("indexer_state"); + + if (!hasTable) { + await knex.schema.createTable("indexer_state", (table) => { + table.string("name", 128).primary(); + table.integer("last_processed_ledger").notNullable().defaultTo(0); + table.string("last_processed_tx_id", 128).nullable(); + table.timestamp("updated_at").defaultTo(knex.fn.now()); + }); + } + + await knex("indexer_state") + .insert({ + name: "contract_events", + last_processed_ledger: 0, + last_processed_tx_id: null, + }) + .onConflict("name") + .ignore(); +}; + +exports.down = function (knex) { + return knex.schema.dropTableIfExists("indexer_state"); +}; diff --git a/backend/src/middleware/auth.js b/backend/src/middleware/auth.js index aac5611d..34f9dfe3 100644 --- a/backend/src/middleware/auth.js +++ b/backend/src/middleware/auth.js @@ -12,7 +12,6 @@ "use strict"; const jwt = require("jsonwebtoken"); -const logger = require("../utils/logger"); const { formatErrorResponse, ERROR_CODES } = require("../../../shared/errorCodes"); /** diff --git a/backend/src/middleware/userRateLimit.js b/backend/src/middleware/userRateLimit.js index b0ec155a..49a9bd8c 100644 --- a/backend/src/middleware/userRateLimit.js +++ b/backend/src/middleware/userRateLimit.js @@ -17,7 +17,7 @@ const baseLimiter = rateLimit({ // We'll inject our own X-RateLimit-User-* headers in the wrapper. standardHeaders: false, legacyHeaders: false, - handler: (req, res, next) => { + handler: (req, res, _next) => { // Return 429 Too Many Requests per requirements // "Too many requests from this account" res.status(429).json({ diff --git a/backend/src/routes/emails.js b/backend/src/routes/emails.js index d7d9899a..7b830b72 100644 --- a/backend/src/routes/emails.js +++ b/backend/src/routes/emails.js @@ -204,7 +204,7 @@ router.post("/:publicKey/verify", async (req, res) => { } try { - const { rendered } = await emailVerificationService.initiateVerification(publicKey, email); + await emailVerificationService.initiateVerification(publicKey, email); // Queue the verification email await notificationService.queueEmail(email, "email_verification", { @@ -214,8 +214,7 @@ router.post("/:publicKey/verify", async (req, res) => { // Actually send immediately for verification flows const t = notificationService.isEnabled; if (t) { - const { token } = await emailVerificationService.initiateVerification(publicKey, email); - const verificationUrl = `${BASE_URL}/api/emails/${encodeURIComponent(publicKey)}/confirm?token=${token}`; + await emailVerificationService.initiateVerification(publicKey, email); const r2 = emailVerificationService; void r2; // just ensuring service is loaded } diff --git a/backend/src/routes/notifications.js b/backend/src/routes/notifications.js index 25adc717..cb242e8a 100644 --- a/backend/src/routes/notifications.js +++ b/backend/src/routes/notifications.js @@ -6,6 +6,7 @@ "use strict"; const express = require("express"); +const rateLimit = require("express-rate-limit"); const router = express.Router(); const knex = require("../db/connection"); const notificationService = require("../services/notificationService"); @@ -18,6 +19,14 @@ const { } = require("../validation/schemas"); const logger = require("../utils/logger"); +const notificationLimiter = rateLimit({ + windowMs: 1 * 60 * 1000, + limit: 20, + standardHeaders: true, + legacyHeaders: false, + message: formatErrorResponse("RATE_LIMITED_SENSITIVE"), +}); + // ─── Existing email endpoints ──────────────────────────────────────────────── /** @@ -26,7 +35,7 @@ const logger = require("../utils/logger"); * * Body: { publicKey: "G...", email: "user@example.com", events?: string[] } */ -router.post("/email", validate(registerEmailSchema), async (req, res, next) => { +router.post("/email", notificationLimiter, validate(registerEmailSchema), async (req, res, next) => { try { const { publicKey, email, events } = req.validated; const preference = await notificationService.registerEmail(publicKey, email, { events }); @@ -44,6 +53,7 @@ router.post("/email", validate(registerEmailSchema), async (req, res, next) => { */ router.put( "/email/:publicKey", + notificationLimiter, validate(publicKeyParamSchema, "params"), validate(updateEmailSchema), async (req, res, next) => { @@ -80,6 +90,7 @@ router.put( */ router.get( "/email/:publicKey", + notificationLimiter, validate(publicKeyParamSchema, "params"), async (req, res, next) => { try { @@ -106,6 +117,7 @@ router.get( */ router.delete( "/email/:publicKey", + notificationLimiter, validate(publicKeyParamSchema, "params"), async (req, res, next) => { try { @@ -162,11 +174,12 @@ function defaultEventChannels() { */ router.get( "/:publicKey/preferences", + notificationLimiter, validate(publicKeyParamSchema, "params"), async (req, res, next) => { try { const { publicKey } = req.validated; - let row = await knex("notification_preferences").where("public_key", publicKey).first(); + const row = await knex("notification_preferences").where("public_key", publicKey).first(); if (!row) { // Return defaults @@ -218,6 +231,7 @@ router.get( */ router.put( "/:publicKey/preferences", + notificationLimiter, validate(publicKeyParamSchema, "params"), async (req, res, next) => { try { @@ -306,6 +320,7 @@ router.put( */ router.get( "/:publicKey/history", + notificationLimiter, validate(publicKeyParamSchema, "params"), async (req, res, next) => { try { @@ -355,6 +370,7 @@ router.get( */ router.put( "/:publicKey/history/:id/read", + notificationLimiter, validate(publicKeyParamSchema, "params"), async (req, res, next) => { try { @@ -376,6 +392,7 @@ router.put( */ router.delete( "/:publicKey/history", + notificationLimiter, validate(publicKeyParamSchema, "params"), async (req, res, next) => { try { diff --git a/backend/src/services/backupService.js b/backend/src/services/backupService.js index 2b91ec10..f0e5721c 100644 --- a/backend/src/services/backupService.js +++ b/backend/src/services/backupService.js @@ -4,6 +4,7 @@ const fs = require("fs"); const path = require("path"); const { exec } = require("child_process"); const cron = require("node-cron"); +const logger = require("../utils/logger"); const BACKUP_DIR = process.env.BACKUP_DIR || path.join(process.cwd(), "backups"); const BACKUP_SCHEDULE = process.env.BACKUP_SCHEDULE || "0 2 * * *"; @@ -156,7 +157,7 @@ function startScheduler() { try { await performBackup(); } catch (err) { - console.error("Automated backup failed:", err); + logger.error({ err }, "Automated backup failed"); } }); } diff --git a/backend/src/services/eventIndexer.js b/backend/src/services/eventIndexer.js index 7493cc2e..419e0a88 100644 --- a/backend/src/services/eventIndexer.js +++ b/backend/src/services/eventIndexer.js @@ -31,6 +31,13 @@ const { getRequestIdHeader } = require("../utils/correlationId"); const { parseEvent } = require("./eventParser"); require("dotenv").config(); +const noopMetric = { + inc: () => {}, + set: () => {}, +}; +const contractEventsProcessedTotal = metrics.contractEventsProcessedTotal ?? noopMetric; +const contractEventIndexerLagLedgers = metrics.contractEventIndexerLagLedgers ?? noopMetric; + // ─── Configuration ─────────────────────────────────────────────────────────── const SOROBAN_RPC_URL = process.env.SOROBAN_RPC_URL || "https://soroban-testnet.stellar.org"; @@ -38,12 +45,15 @@ const CONTRACT_ID = process.env.CONTRACT_ID || process.env.NEXT_PUBLIC_CONTRACT_ const POLL_INTERVAL_MS = parseInt(process.env.EVENT_INDEXER_INTERVAL_MS || "30000", 10); const MAX_RETRIES = 3; const DEFAULT_TIMEOUT_MS = 15_000; +const EVENT_PAGE_LIMIT = 100; +const INDEXER_STATE_NAME = "contract_events"; // ─── In-memory fallback store (used when DATABASE_URL is absent) ───────────── /** @type {Array} */ const memoryStore = []; let memoryIdCounter = 1; +let memoryLastProcessedTxId = null; // ─── PostgreSQL client (lazy singleton) ────────────────────────────────────── @@ -189,16 +199,15 @@ async function getLatestLedger() { * Filters for events emitted by the configured CONTRACT_ID. * * @param {number} startLedger - inclusive lower bound - * @param {number} [endLedger] - inclusive upper bound (defaults to startLedger) + * @param {number} [endLedger] - exclusive upper bound * @returns {Promise>} */ -async function getEvents(startLedger) { - const body = { - jsonrpc: "2.0", - id: 1, - method: "getEvents", - params: { - startLedger, +async function getEvents(startLedger, endLedger) { + const events = []; + let cursor = null; + + do { + const params = { filters: [ { type: "contract", @@ -207,13 +216,32 @@ async function getEvents(startLedger) { }, ], pagination: { - limit: 100, + limit: EVENT_PAGE_LIMIT, }, - }, - }; + }; - const result = await fetchWithRetry(SOROBAN_RPC_URL, body); - return result?.result?.events ?? []; + if (cursor) { + params.pagination.cursor = cursor; + } else { + params.startLedger = startLedger; + if (endLedger !== undefined) { + params.endLedger = endLedger; + } + } + + const body = { + jsonrpc: "2.0", + id: 1, + method: "getEvents", + params, + }; + + const result = await fetchWithRetry(SOROBAN_RPC_URL, body); + events.push(...(result?.result?.events ?? [])); + cursor = result?.result?.cursor || null; + } while (cursor); + + return events; } // ─── Event parsing ──────────────────────────────────────────────────────────── @@ -228,52 +256,57 @@ async function getEvents(startLedger) { * Insert parsed events into the store (PostgreSQL or in-memory fallback). * * @param {Array} events - parsed event rows - * @returns {Promise} number of events inserted + * @param {{ cursorLedger?: number, cursorTxId?: string|null }} [options] + * @returns {Promise<{ inserted: number, conflicts: number, errors: number }>} insert summary */ -async function storeEvents(events) { - if (events.length === 0) return 0; +async function storeEvents(events, options = {}) { + if (events.length === 0) return { inserted: 0, conflicts: 0, errors: 0 }; const pool = getPgPool(); if (pool) { - // PostgreSQL path - let inserted = 0; + const summary = { inserted: 0, conflicts: 0, errors: 0 }; const client = await pool.connect(); try { + await client.query("BEGIN"); for (const ev of events) { - try { - await client.query( - `INSERT INTO contract_events - (event_type, contract_id, ledger_sequence, emitted_at, - from_addr, to_addr, amount_raw, payload) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8) - ON CONFLICT (ledger_sequence, contract_id, event_type, (payload->>'id')) - DO NOTHING`, - [ - ev.event_type, - ev.contract_id, - ev.ledger_sequence, - ev.emitted_at, - ev.from_addr || null, - ev.to_addr || null, - ev.amount_raw || null, - JSON.stringify(ev.payload), - ], - ); - inserted++; - } catch (err) { - // ON CONFLICT DO NOTHING should handle duplicates, but log - // unexpected errors and continue with remaining events. - logger.error( - { err, event_type: ev.event_type, ledger: ev.ledger_sequence }, - "Failed to insert contract event", - ); + const result = await client.query( + `INSERT INTO contract_events + (event_type, contract_id, ledger_sequence, emitted_at, + from_addr, to_addr, amount_raw, payload) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + ON CONFLICT (ledger_sequence, contract_id, event_type, (payload->>'id')) + DO NOTHING`, + [ + ev.event_type, + ev.contract_id, + ev.ledger_sequence, + ev.emitted_at, + ev.from_addr || null, + ev.to_addr || null, + ev.amount_raw || null, + JSON.stringify(ev.payload), + ], + ); + + if (result.rowCount === 1) { + summary.inserted++; + } else { + summary.conflicts++; } } + if (summary.conflicts === 0 && options.cursorLedger !== undefined) { + await saveCursorWithQueryable(client, options.cursorLedger, options.cursorTxId ?? null); + } + await client.query("COMMIT"); + } catch (err) { + summary.errors++; + await client.query("ROLLBACK").catch(() => {}); + logger.error({ err }, "Failed to store contract event batch"); } finally { client.release(); } - return inserted; + return summary; } // In-memory fallback @@ -284,7 +317,7 @@ async function storeEvents(events) { created_at: new Date().toISOString(), }); } - return events.length; + return { inserted: events.length, conflicts: 0, errors: 0 }; } // ─── Cursor persistence ────────────────────────────────────────────────────── @@ -294,6 +327,43 @@ async function storeEvents(events) { * Persisted in PostgreSQL when available, otherwise held in memory. */ let lastProcessedLedger = 0; +let lastProcessedTxId = null; + +async function ensureIndexerState(pool) { + await pool.query( + `INSERT INTO indexer_state (name, last_processed_ledger, last_processed_tx_id) + VALUES ($1, 0, NULL) + ON CONFLICT (name) DO NOTHING`, + [INDEXER_STATE_NAME], + ); +} + +async function saveCursorWithQueryable(queryable, ledger, txId = null) { + await queryable.query( + `INSERT INTO indexer_state (name, last_processed_ledger, last_processed_tx_id, updated_at) + VALUES ($1, $2, $3, NOW()) + ON CONFLICT (name) + DO UPDATE SET + last_processed_ledger = EXCLUDED.last_processed_ledger, + last_processed_tx_id = EXCLUDED.last_processed_tx_id, + updated_at = NOW()`, + [INDEXER_STATE_NAME, ledger, txId], + ); +} + +async function saveCursor(pool, ledger, txId = null) { + if (!pool) { + lastProcessedLedger = ledger; + lastProcessedTxId = txId; + memoryLastProcessedTxId = txId; + return; + } + + await saveCursorWithQueryable(pool, ledger, txId); + + lastProcessedLedger = ledger; + lastProcessedTxId = txId; +} /** * Load the last processed ledger from PostgreSQL (or return in-memory value). @@ -303,13 +373,18 @@ async function loadCursor() { if (!pool) return lastProcessedLedger; try { + await ensureIndexerState(pool); const result = await pool.query( - `SELECT MAX(ledger_sequence) AS max_ledger FROM contract_events`, + `SELECT last_processed_ledger, last_processed_tx_id + FROM indexer_state + WHERE name = $1`, + [INDEXER_STATE_NAME], ); - const max = result?.rows?.[0]?.max_ledger; - if (max !== null && max !== undefined) { - lastProcessedLedger = parseInt(max, 10); - logger.info({ lastProcessedLedger }, "Loaded event cursor from PostgreSQL"); + const row = result?.rows?.[0]; + if (row) { + lastProcessedLedger = parseInt(row.last_processed_ledger ?? "0", 10); + lastProcessedTxId = row.last_processed_tx_id ?? null; + logger.info({ lastProcessedLedger, lastProcessedTxId }, "Loaded event cursor from PostgreSQL"); } } catch (err) { logger.error({ err }, "Failed to load cursor from PostgreSQL"); @@ -317,6 +392,21 @@ async function loadCursor() { return lastProcessedLedger; } +function getLastTxId(events) { + const latest = [...events] + .filter((ev) => ev.ledger_sequence !== undefined && ev.ledger_sequence !== null) + .sort( + (a, b) => + Number(a.ledger_sequence) - Number(b.ledger_sequence) || + String(a.payload?.pagingToken ?? a.payload?.eventId ?? "").localeCompare( + String(b.payload?.pagingToken ?? b.payload?.eventId ?? ""), + ), + ) + .at(-1); + + return latest?.payload?.txHash ?? latest?.payload?.pagingToken ?? latest?.payload?.eventId ?? null; +} + // ─── Polling loop ──────────────────────────────────────────────────────────── let pollTimer = null; @@ -340,7 +430,7 @@ async function pollOnce() { // Backlog signal: how far behind the network we are right now (#272). // Recorded before the early return so a caught-up indexer reports 0 rather // than holding its last non-zero value. - metrics.contractEventIndexerLagLedgers.set( + contractEventIndexerLagLedgers.set( lastProcessedLedger > 0 ? Math.max(0, latestLedger - lastProcessedLedger) : 0, ); @@ -353,27 +443,38 @@ async function pollOnce() { // Fetch events for the full unprocessed range. // The RPC may paginate internally; we request up to 100 events at a time // and follow pagination cursors to ensure completeness. - const rawEvents = await getEvents(startLedger); + const endLedger = latestLedger + 1; + const rawEvents = await getEvents(startLedger, endLedger); + let parseFailures = 0; + let storeSummary = { inserted: 0, conflicts: 0, errors: 0 }; + const parsed = []; if (rawEvents.length > 0) { - const parsed = []; for (const raw of rawEvents) { try { parsed.push(parseEvent(raw)); } catch (parseErr) { - metrics.contractEventsProcessedTotal.inc({ outcome: "parse_failed" }); + parseFailures++; + contractEventsProcessedTotal.inc({ outcome: "parse_failed" }); logger.warn( { parseErr, eventId: raw.id }, "Failed to parse individual Soroban event — skipping", ); } } - const inserted = await storeEvents(parsed); - metrics.contractEventsProcessedTotal.inc({ outcome: "indexed" }, inserted); + const cursorTxId = getLastTxId(parsed); + const shouldAdvanceWithBatch = parseFailures === 0; + storeSummary = await storeEvents( + parsed, + shouldAdvanceWithBatch ? { cursorLedger: latestLedger, cursorTxId } : {}, + ); + contractEventsProcessedTotal.inc({ outcome: "indexed" }, storeSummary.inserted); logger.info( { eventCount: rawEvents.length, - inserted, + inserted: storeSummary.inserted, + conflicts: storeSummary.conflicts, + errors: storeSummary.errors, startLedger, endLedger: latestLedger, }, @@ -391,7 +492,26 @@ async function pollOnce() { } } - lastProcessedLedger = latestLedger; + if (parseFailures === 0 && storeSummary.conflicts === 0 && storeSummary.errors === 0) { + if (rawEvents.length === 0 || !process.env.DATABASE_URL) { + const pool = process.env.DATABASE_URL ? getPgPool() : null; + await saveCursor(pool, latestLedger, getLastTxId(parsed)); + } else { + lastProcessedLedger = latestLedger; + lastProcessedTxId = getLastTxId(parsed); + } + } else { + logger.warn( + { + parseFailures, + conflicts: storeSummary.conflicts, + errors: storeSummary.errors, + startLedger, + endLedger: latestLedger, + }, + "Event indexer cursor not advanced because the range was not cleanly indexed", + ); + } } catch (err) { logger.error({ err }, "Event indexer poll failed"); } finally { @@ -502,7 +622,7 @@ async function queryEventsByType(publicKey, eventType, { limit = 20, offset = 0, } // In-memory fallback - let filtered = memoryStore.filter((ev) => { + const filtered = memoryStore.filter((ev) => { const payloadStr = JSON.stringify(ev.payload).toLowerCase(); const match = payloadStr.includes(publicKey.toLowerCase()) && ev.event_type === eventType; if (!match) return false; @@ -654,9 +774,24 @@ module.exports = { memoryStore.length = 0; memoryIdCounter = 1; lastProcessedLedger = 0; + lastProcessedTxId = null; + memoryLastProcessedTxId = null; if (pgPool) { pgPool.end().catch(() => {}); pgPool = null; } }, + _internals: { + getEvents, + loadCursor, + saveCursor, + storeEvents, + pollOnce, + getLastTxId, + getCursorForTest: () => ({ + lastProcessedLedger, + lastProcessedTxId, + memoryLastProcessedTxId, + }), + }, }; diff --git a/backend/src/services/eventParser.js b/backend/src/services/eventParser.js index da1ccc73..2d17e269 100644 --- a/backend/src/services/eventParser.js +++ b/backend/src/services/eventParser.js @@ -104,10 +104,12 @@ function parseEvent(raw) { } const payload = { + id: raw.id ?? null, topics: topics, data: data ?? null, eventId: raw.id ?? null, pagingToken: raw.pagingToken ?? null, + txHash: raw.txHash ?? null, }; return { diff --git a/backend/src/services/notificationService.js b/backend/src/services/notificationService.js index 062c6b76..c12074e4 100644 --- a/backend/src/services/notificationService.js +++ b/backend/src/services/notificationService.js @@ -25,7 +25,6 @@ var logger = require("../utils/logger"); var knex = require("../db/connection"); var emailRenderer = require("./emailRenderer"); var emailTrackingService = require("./emailTrackingService"); -var emailVerificationService = require("./emailVerificationService"); var metricsService = require("./metricsService"); // ─── Configuration ──────────────────────────────────────────────────────────── @@ -219,7 +218,9 @@ async function queueEmail(to, templateType, data, opts) { let unsubToken = null; try { unsubToken = await ensureUnsubscribeToken(to, "all"); - } catch (_) {} + } catch (err) { + logger.debug({ type: "unsubscribe_token_precreate_failed", error: err.message }, "Unsubscribe token pre-create failed"); + } const unsubscribeUrl = unsubToken ? `${BASE_URL}/api/emails/unsubscribe?token=${unsubToken}`