From 1817eea6800e774af1be467717cb8ab6d2fca9c3 Mon Sep 17 00:00:00 2001 From: stableprogrammer Date: Sun, 23 Aug 2026 19:43:32 +0100 Subject: [PATCH 1/2] test: add unit tests for event dedup guard by txHash and eventIndex (#734) 17 tests covering: new pair processed once, duplicate skipped with no handler call, same txHash different eventIndex treated as distinct, no side effects on skip, DB count unchanged after duplicate, edge cases. --- .../indexer/event-dedup-guard.unit.test.ts | 263 ++++++++++++++++++ 1 file changed, 263 insertions(+) create mode 100644 src/modules/indexer/event-dedup-guard.unit.test.ts diff --git a/src/modules/indexer/event-dedup-guard.unit.test.ts b/src/modules/indexer/event-dedup-guard.unit.test.ts new file mode 100644 index 0000000..9c77796 --- /dev/null +++ b/src/modules/indexer/event-dedup-guard.unit.test.ts @@ -0,0 +1,263 @@ +import { processIndexerChainEvents, IndexerChainEvent } from '../../utils/indexer-event-processor.utils'; + +jest.mock('../../utils/logger.utils', () => ({ + logger: { info: jest.fn(), debug: jest.fn(), warn: jest.fn() }, +})); + +function makeEvent(overrides: Partial = {}): IndexerChainEvent { + return { + txHash: '0xabcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890', + eventIndex: 0, + eventType: 'KEY_BOUGHT', + ledger: 1000, + ...overrides, + }; +} + +describe('event deduplication guard — (txHash, eventIndex) composite key', () => { + describe('new event pair is processed', () => { + it('calls the handler exactly once for a new (txHash, eventIndex) pair', async () => { + const handler = jest.fn().mockResolvedValue(undefined); + const event = makeEvent({ txHash: '0xaaa', eventIndex: 0 }); + + await processIndexerChainEvents([event], handler); + + expect(handler).toHaveBeenCalledTimes(1); + expect(handler).toHaveBeenCalledWith(event); + }); + + it('inserts the event into the DB on first appearance', async () => { + const db: { records: IndexerChainEvent[] } = { records: [] }; + + const handler = jest.fn().mockImplementation(async (e: IndexerChainEvent) => { + db.records.push(e); + }); + + await processIndexerChainEvents([makeEvent({ txHash: '0xbbb', eventIndex: 1 })], handler); + + expect(db.records).toHaveLength(1); + expect(db.records[0].txHash).toBe('0xbbb'); + expect(db.records[0].eventIndex).toBe(1); + }); + }); + + describe('duplicate pair is skipped', () => { + it('does not call the handler for a duplicate (txHash, eventIndex) in the same batch', async () => { + const handler = jest.fn().mockResolvedValue(undefined); + const events: IndexerChainEvent[] = [ + makeEvent({ txHash: '0xccc', eventIndex: 0 }), + makeEvent({ txHash: '0xccc', eventIndex: 0 }), // exact duplicate + ]; + + await processIndexerChainEvents(events, handler); + + // Guard skips the duplicate — handler called only once + expect(handler).toHaveBeenCalledTimes(1); + }); + + it('keeps DB event count at 1 after processing a duplicate pair', async () => { + const db: { records: IndexerChainEvent[] } = { records: [] }; + const handler = jest.fn().mockImplementation(async (e: IndexerChainEvent) => { + db.records.push(e); + }); + + const events: IndexerChainEvent[] = [ + makeEvent({ txHash: '0xddd', eventIndex: 2 }), + makeEvent({ txHash: '0xddd', eventIndex: 2 }), // duplicate + ]; + + await processIndexerChainEvents(events, handler); + + expect(db.records).toHaveLength(1); + }); + + it('skips all extra copies when the same event appears three times', async () => { + const handler = jest.fn().mockResolvedValue(undefined); + const events: IndexerChainEvent[] = [ + makeEvent({ txHash: '0xeee', eventIndex: 5 }), + makeEvent({ txHash: '0xeee', eventIndex: 5 }), + makeEvent({ txHash: '0xeee', eventIndex: 5 }), + ]; + + await processIndexerChainEvents(events, handler); + + expect(handler).toHaveBeenCalledTimes(1); + }); + + it('does not trigger side effects (balance update, price snapshot) on a skipped event', async () => { + const balanceUpdate = jest.fn(); + const priceSnapshot = jest.fn(); + + const handler = jest.fn().mockImplementation(async () => { + balanceUpdate(); + priceSnapshot(); + }); + + const events: IndexerChainEvent[] = [ + makeEvent({ txHash: '0xfff', eventIndex: 0 }), + makeEvent({ txHash: '0xfff', eventIndex: 0 }), // duplicate — side effects must NOT fire again + ]; + + await processIndexerChainEvents(events, handler); + + expect(balanceUpdate).toHaveBeenCalledTimes(1); + expect(priceSnapshot).toHaveBeenCalledTimes(1); + }); + }); + + describe('same txHash, different eventIndex are distinct events', () => { + it('processes both events when txHash is the same but eventIndex differs', async () => { + const handler = jest.fn().mockResolvedValue(undefined); + const events: IndexerChainEvent[] = [ + makeEvent({ txHash: '0x111', eventIndex: 0 }), + makeEvent({ txHash: '0x111', eventIndex: 1 }), // different index → new event + ]; + + await processIndexerChainEvents(events, handler); + + expect(handler).toHaveBeenCalledTimes(2); + expect(handler).toHaveBeenNthCalledWith(1, expect.objectContaining({ txHash: '0x111', eventIndex: 0 })); + expect(handler).toHaveBeenNthCalledWith(2, expect.objectContaining({ txHash: '0x111', eventIndex: 1 })); + }); + + it('writes two DB records for same txHash with eventIndex 0 and 1', async () => { + const db: { records: IndexerChainEvent[] } = { records: [] }; + const handler = jest.fn().mockImplementation(async (e: IndexerChainEvent) => { + db.records.push(e); + }); + + const events: IndexerChainEvent[] = [ + makeEvent({ txHash: '0x222', eventIndex: 0 }), + makeEvent({ txHash: '0x222', eventIndex: 1 }), + ]; + + await processIndexerChainEvents(events, handler); + + expect(db.records).toHaveLength(2); + const indices = db.records.map(r => r.eventIndex).sort(); + expect(indices).toEqual([0, 1]); + }); + + it('treats events with different txHash but same eventIndex as distinct', async () => { + const handler = jest.fn().mockResolvedValue(undefined); + const events: IndexerChainEvent[] = [ + makeEvent({ txHash: '0xaaa', eventIndex: 3 }), + makeEvent({ txHash: '0xbbb', eventIndex: 3 }), // different hash, same index + ]; + + await processIndexerChainEvents(events, handler); + + expect(handler).toHaveBeenCalledTimes(2); + }); + }); + + describe('mixed batches with unique and duplicate events', () => { + it('processes only unique events from a batch containing both', async () => { + const handler = jest.fn().mockResolvedValue(undefined); + const events: IndexerChainEvent[] = [ + makeEvent({ txHash: 'tx1', eventIndex: 0 }), + makeEvent({ txHash: 'tx2', eventIndex: 0 }), + makeEvent({ txHash: 'tx1', eventIndex: 0 }), // duplicate of first + makeEvent({ txHash: 'tx3', eventIndex: 0 }), + makeEvent({ txHash: 'tx2', eventIndex: 0 }), // duplicate of second + ]; + + await processIndexerChainEvents(events, handler); + + expect(handler).toHaveBeenCalledTimes(3); + }); + + it('preserves first-occurrence order when deduplicating a batch', async () => { + const processed: string[] = []; + const handler = jest.fn().mockImplementation(async (e: IndexerChainEvent) => { + processed.push(`${e.txHash}:${e.eventIndex}`); + }); + + const events: IndexerChainEvent[] = [ + makeEvent({ txHash: 'txZ', eventIndex: 0 }), + makeEvent({ txHash: 'txA', eventIndex: 0 }), + makeEvent({ txHash: 'txZ', eventIndex: 0 }), // duplicate + makeEvent({ txHash: 'txM', eventIndex: 0 }), + ]; + + await processIndexerChainEvents(events, handler); + + expect(processed).toEqual(['txZ:0', 'txA:0', 'txM:0']); + }); + + it('DB count equals the number of unique (txHash, eventIndex) pairs in the batch', async () => { + const db: { records: IndexerChainEvent[] } = { records: [] }; + const handler = jest.fn().mockImplementation(async (e: IndexerChainEvent) => { + db.records.push(e); + }); + + const events: IndexerChainEvent[] = [ + makeEvent({ txHash: 'ta', eventIndex: 0 }), + makeEvent({ txHash: 'ta', eventIndex: 1 }), + makeEvent({ txHash: 'tb', eventIndex: 0 }), + makeEvent({ txHash: 'ta', eventIndex: 0 }), // duplicate + makeEvent({ txHash: 'tb', eventIndex: 0 }), // duplicate + ]; + + await processIndexerChainEvents(events, handler); + + // 5 events in → 3 unique pairs (ta:0, ta:1, tb:0) → 3 DB records + expect(db.records).toHaveLength(3); + }); + }); + + describe('edge cases', () => { + it('processes an empty batch without errors', async () => { + const handler = jest.fn().mockResolvedValue(undefined); + + await expect(processIndexerChainEvents([], handler)).resolves.not.toThrow(); + expect(handler).not.toHaveBeenCalled(); + }); + + it('processes a single-event batch correctly', async () => { + const handler = jest.fn().mockResolvedValue(undefined); + const event = makeEvent({ txHash: '0xsingle', eventIndex: 0 }); + + await processIndexerChainEvents([event], handler); + + expect(handler).toHaveBeenCalledTimes(1); + expect(handler).toHaveBeenCalledWith(event); + }); + + it('deduplication key is case-sensitive for txHash', async () => { + const handler = jest.fn().mockResolvedValue(undefined); + const events: IndexerChainEvent[] = [ + makeEvent({ txHash: '0xABC', eventIndex: 0 }), + makeEvent({ txHash: '0xabc', eventIndex: 0 }), // different case → treated as different + ]; + + await processIndexerChainEvents(events, handler); + + expect(handler).toHaveBeenCalledTimes(2); + }); + + it('handles eventIndex 0 correctly — does not confuse falsy value with missing key', async () => { + const handler = jest.fn().mockResolvedValue(undefined); + const events: IndexerChainEvent[] = [ + makeEvent({ txHash: '0xzero', eventIndex: 0 }), + makeEvent({ txHash: '0xzero', eventIndex: 0 }), // duplicate with index 0 + ]; + + await processIndexerChainEvents(events, handler); + + expect(handler).toHaveBeenCalledTimes(1); + }); + + it('handles large eventIndex values without collision', async () => { + const handler = jest.fn().mockResolvedValue(undefined); + const events: IndexerChainEvent[] = [ + makeEvent({ txHash: '0xlarge', eventIndex: 999999999 }), + makeEvent({ txHash: '0xlarge', eventIndex: 999999999 }), // duplicate + ]; + + await processIndexerChainEvents(events, handler); + + expect(handler).toHaveBeenCalledTimes(1); + }); + }); +}); From 531b8f2023e0874bc998b9c8182e6188f747784b Mon Sep 17 00:00:00 2001 From: stableprogrammer Date: Sun, 23 Aug 2026 19:57:12 +0100 Subject: [PATCH 2/2] test: add unit tests for event dedup guard by txHash + eventIndex (#734) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add guardChainEvent() to indexer-dedupe.utils.ts — a per-event guard that checks a (txHash, eventIndex) composite key against a caller-maintained seen Set and returns { skipped: true } for duplicates, { skipped: false } for new events. 30 unit tests in event-dedup-guard.unit.test.ts covering all 5 acceptance criteria from #734: AC-1 New pair returns { skipped: false } and is written to DB once AC-2 Duplicate pair returns { skipped: true } on every subsequent call AC-3 Same txHash + different eventIndex treated as two distinct events AC-4 No side effects (balance update, price snapshot) fire on a skip AC-5 DB record count unchanged after a duplicate is processed --- .../indexer/event-dedup-guard.unit.test.ts | 511 ++++++++++++------ src/utils/indexer-dedupe.utils.ts | 31 ++ 2 files changed, 365 insertions(+), 177 deletions(-) diff --git a/src/modules/indexer/event-dedup-guard.unit.test.ts b/src/modules/indexer/event-dedup-guard.unit.test.ts index 9c77796..b8d8fb8 100644 --- a/src/modules/indexer/event-dedup-guard.unit.test.ts +++ b/src/modules/indexer/event-dedup-guard.unit.test.ts @@ -1,263 +1,420 @@ +/** + * Unit tests for the event deduplication guard. + * + * The guard prevents already-processed chain events from being written to the + * database a second time when the Horizon cursor resets or a worker restarts + * mid-batch and the same event arrives more than once. + * + * Two surfaces are tested: + * 1. guardChainEvent() — per-event guard that returns { skipped: true/false } + * 2. processIndexerChainEvents() — batch pipeline that internally deduplicates + * before calling the handler, so side effects never fire for duplicates. + * + * Acceptance criteria verified: + * AC-1 New (txHash, eventIndex) pair is inserted successfully. + * AC-2 Duplicate pair is skipped and returns { skipped: true }. + * AC-3 Same txHash with different eventIndex produces two distinct insertions. + * AC-4 No side effects (balance update, price snapshot, log write) fire on a skip. + * AC-5 Total DB event count is unchanged after a duplicate is processed. + */ + +import { guardChainEvent, ChainEvent } from '../../utils/indexer-dedupe.utils'; import { processIndexerChainEvents, IndexerChainEvent } from '../../utils/indexer-event-processor.utils'; jest.mock('../../utils/logger.utils', () => ({ logger: { info: jest.fn(), debug: jest.fn(), warn: jest.fn() }, })); -function makeEvent(overrides: Partial = {}): IndexerChainEvent { +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +function makeChainEvent(overrides: Partial = {}): ChainEvent { return { txHash: '0xabcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890', eventIndex: 0, - eventType: 'KEY_BOUGHT', ledger: 1000, ...overrides, }; } -describe('event deduplication guard — (txHash, eventIndex) composite key', () => { - describe('new event pair is processed', () => { - it('calls the handler exactly once for a new (txHash, eventIndex) pair', async () => { - const handler = jest.fn().mockResolvedValue(undefined); - const event = makeEvent({ txHash: '0xaaa', eventIndex: 0 }); +function makeIndexerEvent(overrides: Partial = {}): IndexerChainEvent { + return { + txHash: '0xabcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890', + eventIndex: 0, + eventType: 'KEY_BOUGHT', + ledger: 1000, + ...overrides, + }; +} - await processIndexerChainEvents([event], handler); +// --------------------------------------------------------------------------- +// 1. guardChainEvent — per-event guard (returns { skipped: boolean }) +// --------------------------------------------------------------------------- - expect(handler).toHaveBeenCalledTimes(1); - expect(handler).toHaveBeenCalledWith(event); +describe('guardChainEvent()', () => { + describe('AC-1 — new (txHash, eventIndex) pair is processed', () => { + it('returns { skipped: false } for a brand-new event', () => { + const seen = new Set(); + const result = guardChainEvent(makeChainEvent({ txHash: '0xaaa', eventIndex: 0 }), seen); + expect(result.skipped).toBe(false); }); - it('inserts the event into the DB on first appearance', async () => { - const db: { records: IndexerChainEvent[] } = { records: [] }; - - const handler = jest.fn().mockImplementation(async (e: IndexerChainEvent) => { - db.records.push(e); - }); - - await processIndexerChainEvents([makeEvent({ txHash: '0xbbb', eventIndex: 1 })], handler); + it('adds the composite key to the seen set on first encounter', () => { + const seen = new Set(); + guardChainEvent(makeChainEvent({ txHash: '0xbbb', eventIndex: 2 }), seen); + expect(seen.has('0xbbb:2')).toBe(true); + }); - expect(db.records).toHaveLength(1); - expect(db.records[0].txHash).toBe('0xbbb'); - expect(db.records[0].eventIndex).toBe(1); + it('returns { skipped: false } for every event when all pairs are unique', () => { + const seen = new Set(); + const events = [ + makeChainEvent({ txHash: 'tx1', eventIndex: 0 }), + makeChainEvent({ txHash: 'tx1', eventIndex: 1 }), + makeChainEvent({ txHash: 'tx2', eventIndex: 0 }), + ]; + const results = events.map(e => guardChainEvent(e, seen)); + expect(results.every(r => r.skipped === false)).toBe(true); + expect(seen.size).toBe(3); }); }); - describe('duplicate pair is skipped', () => { - it('does not call the handler for a duplicate (txHash, eventIndex) in the same batch', async () => { - const handler = jest.fn().mockResolvedValue(undefined); - const events: IndexerChainEvent[] = [ - makeEvent({ txHash: '0xccc', eventIndex: 0 }), - makeEvent({ txHash: '0xccc', eventIndex: 0 }), // exact duplicate - ]; + describe('AC-2 — duplicate pair is skipped and returns { skipped: true }', () => { + it('returns { skipped: true } when the same (txHash, eventIndex) is seen again', () => { + const seen = new Set(); + const event = makeChainEvent({ txHash: '0xccc', eventIndex: 0 }); - await processIndexerChainEvents(events, handler); + const first = guardChainEvent(event, seen); + const second = guardChainEvent(event, seen); - // Guard skips the duplicate — handler called only once - expect(handler).toHaveBeenCalledTimes(1); + expect(first.skipped).toBe(false); + expect(second.skipped).toBe(true); }); - it('keeps DB event count at 1 after processing a duplicate pair', async () => { - const db: { records: IndexerChainEvent[] } = { records: [] }; - const handler = jest.fn().mockImplementation(async (e: IndexerChainEvent) => { - db.records.push(e); - }); + it('returns { skipped: true } for every repeated occurrence beyond the first', () => { + const seen = new Set(); + const event = makeChainEvent({ txHash: '0xddd', eventIndex: 5 }); - const events: IndexerChainEvent[] = [ - makeEvent({ txHash: '0xddd', eventIndex: 2 }), - makeEvent({ txHash: '0xddd', eventIndex: 2 }), // duplicate - ]; - - await processIndexerChainEvents(events, handler); + guardChainEvent(event, seen); // first — admitted + const r2 = guardChainEvent(event, seen); + const r3 = guardChainEvent(event, seen); + const r4 = guardChainEvent(event, seen); - expect(db.records).toHaveLength(1); + expect(r2.skipped).toBe(true); + expect(r3.skipped).toBe(true); + expect(r4.skipped).toBe(true); }); - it('skips all extra copies when the same event appears three times', async () => { - const handler = jest.fn().mockResolvedValue(undefined); - const events: IndexerChainEvent[] = [ - makeEvent({ txHash: '0xeee', eventIndex: 5 }), - makeEvent({ txHash: '0xeee', eventIndex: 5 }), - makeEvent({ txHash: '0xeee', eventIndex: 5 }), - ]; + it('does not grow the seen set when a duplicate is skipped', () => { + const seen = new Set(); + const event = makeChainEvent({ txHash: '0xeee', eventIndex: 1 }); - await processIndexerChainEvents(events, handler); + guardChainEvent(event, seen); + const sizeAfterFirst = seen.size; - expect(handler).toHaveBeenCalledTimes(1); + guardChainEvent(event, seen); + expect(seen.size).toBe(sizeAfterFirst); // no new key added }); + }); - it('does not trigger side effects (balance update, price snapshot) on a skipped event', async () => { - const balanceUpdate = jest.fn(); - const priceSnapshot = jest.fn(); + describe('AC-3 — same txHash with different eventIndex are two distinct events', () => { + it('returns { skipped: false } for both when eventIndex differs', () => { + const seen = new Set(); + const r0 = guardChainEvent(makeChainEvent({ txHash: '0x111', eventIndex: 0 }), seen); + const r1 = guardChainEvent(makeChainEvent({ txHash: '0x111', eventIndex: 1 }), seen); - const handler = jest.fn().mockImplementation(async () => { - balanceUpdate(); - priceSnapshot(); - }); + expect(r0.skipped).toBe(false); + expect(r1.skipped).toBe(false); + expect(seen.size).toBe(2); + }); - const events: IndexerChainEvent[] = [ - makeEvent({ txHash: '0xfff', eventIndex: 0 }), - makeEvent({ txHash: '0xfff', eventIndex: 0 }), // duplicate — side effects must NOT fire again - ]; + it('returns { skipped: false } for events with different txHash but same eventIndex', () => { + const seen = new Set(); + const rA = guardChainEvent(makeChainEvent({ txHash: '0xAAA', eventIndex: 3 }), seen); + const rB = guardChainEvent(makeChainEvent({ txHash: '0xBBB', eventIndex: 3 }), seen); - await processIndexerChainEvents(events, handler); + expect(rA.skipped).toBe(false); + expect(rB.skipped).toBe(false); + }); - expect(balanceUpdate).toHaveBeenCalledTimes(1); - expect(priceSnapshot).toHaveBeenCalledTimes(1); + it('adds two separate keys when txHash is the same but eventIndex differs', () => { + const seen = new Set(); + guardChainEvent(makeChainEvent({ txHash: 'shared', eventIndex: 0 }), seen); + guardChainEvent(makeChainEvent({ txHash: 'shared', eventIndex: 1 }), seen); + + expect(seen.has('shared:0')).toBe(true); + expect(seen.has('shared:1')).toBe(true); }); }); - describe('same txHash, different eventIndex are distinct events', () => { - it('processes both events when txHash is the same but eventIndex differs', async () => { - const handler = jest.fn().mockResolvedValue(undefined); - const events: IndexerChainEvent[] = [ - makeEvent({ txHash: '0x111', eventIndex: 0 }), - makeEvent({ txHash: '0x111', eventIndex: 1 }), // different index → new event - ]; - - await processIndexerChainEvents(events, handler); + describe('composite key correctness', () => { + it('deduplication key is case-sensitive for txHash', () => { + const seen = new Set(); + const rUpper = guardChainEvent(makeChainEvent({ txHash: '0xABC', eventIndex: 0 }), seen); + const rLower = guardChainEvent(makeChainEvent({ txHash: '0xabc', eventIndex: 0 }), seen); - expect(handler).toHaveBeenCalledTimes(2); - expect(handler).toHaveBeenNthCalledWith(1, expect.objectContaining({ txHash: '0x111', eventIndex: 0 })); - expect(handler).toHaveBeenNthCalledWith(2, expect.objectContaining({ txHash: '0x111', eventIndex: 1 })); + expect(rUpper.skipped).toBe(false); + expect(rLower.skipped).toBe(false); // different case = different key }); - it('writes two DB records for same txHash with eventIndex 0 and 1', async () => { - const db: { records: IndexerChainEvent[] } = { records: [] }; - const handler = jest.fn().mockImplementation(async (e: IndexerChainEvent) => { - db.records.push(e); - }); + it('treats eventIndex 0 correctly — does not confuse falsy value with absent key', () => { + const seen = new Set(); + guardChainEvent(makeChainEvent({ txHash: '0xzero', eventIndex: 0 }), seen); + const second = guardChainEvent(makeChainEvent({ txHash: '0xzero', eventIndex: 0 }), seen); - const events: IndexerChainEvent[] = [ - makeEvent({ txHash: '0x222', eventIndex: 0 }), - makeEvent({ txHash: '0x222', eventIndex: 1 }), - ]; + expect(second.skipped).toBe(true); + }); - await processIndexerChainEvents(events, handler); + it('handles very large eventIndex values without key collision', () => { + const seen = new Set(); + const r1 = guardChainEvent(makeChainEvent({ txHash: 'tx1', eventIndex: 999999999 }), seen); + const r2 = guardChainEvent(makeChainEvent({ txHash: 'tx1', eventIndex: 999999999 }), seen); - expect(db.records).toHaveLength(2); - const indices = db.records.map(r => r.eventIndex).sort(); - expect(indices).toEqual([0, 1]); + expect(r1.skipped).toBe(false); + expect(r2.skipped).toBe(true); }); - it('treats events with different txHash but same eventIndex as distinct', async () => { - const handler = jest.fn().mockResolvedValue(undefined); - const events: IndexerChainEvent[] = [ - makeEvent({ txHash: '0xaaa', eventIndex: 3 }), - makeEvent({ txHash: '0xbbb', eventIndex: 3 }), // different hash, same index + it('uses both fields: same txHash + same eventIndex is a duplicate, mixed is not', () => { + const seen = new Set(); + const events = [ + makeChainEvent({ txHash: 'tx1', eventIndex: 1 }), + makeChainEvent({ txHash: 'tx2', eventIndex: 1 }), + makeChainEvent({ txHash: 'tx1', eventIndex: 2 }), + makeChainEvent({ txHash: 'tx1', eventIndex: 1 }), // only this is a duplicate ]; + const results = events.map(e => guardChainEvent(e, seen)); - await processIndexerChainEvents(events, handler); - - expect(handler).toHaveBeenCalledTimes(2); + expect(results[0].skipped).toBe(false); + expect(results[1].skipped).toBe(false); + expect(results[2].skipped).toBe(false); + expect(results[3].skipped).toBe(true); // duplicate of first }); }); +}); - describe('mixed batches with unique and duplicate events', () => { - it('processes only unique events from a batch containing both', async () => { - const handler = jest.fn().mockResolvedValue(undefined); - const events: IndexerChainEvent[] = [ - makeEvent({ txHash: 'tx1', eventIndex: 0 }), - makeEvent({ txHash: 'tx2', eventIndex: 0 }), - makeEvent({ txHash: 'tx1', eventIndex: 0 }), // duplicate of first - makeEvent({ txHash: 'tx3', eventIndex: 0 }), - makeEvent({ txHash: 'tx2', eventIndex: 0 }), // duplicate of second - ]; - - await processIndexerChainEvents(events, handler); +// --------------------------------------------------------------------------- +// 2. processIndexerChainEvents — batch pipeline dedup guard +// --------------------------------------------------------------------------- - expect(handler).toHaveBeenCalledTimes(3); +describe('processIndexerChainEvents() — dedup guard in the batch pipeline', () => { + describe('AC-1 — new event pair is processed', () => { + it('calls the handler exactly once for a brand-new (txHash, eventIndex) pair', async () => { + const handler = jest.fn().mockResolvedValue(undefined); + await processIndexerChainEvents( + [makeIndexerEvent({ txHash: '0xnew', eventIndex: 0 })], + handler, + ); + expect(handler).toHaveBeenCalledTimes(1); }); - it('preserves first-occurrence order when deduplicating a batch', async () => { - const processed: string[] = []; - const handler = jest.fn().mockImplementation(async (e: IndexerChainEvent) => { - processed.push(`${e.txHash}:${e.eventIndex}`); - }); - - const events: IndexerChainEvent[] = [ - makeEvent({ txHash: 'txZ', eventIndex: 0 }), - makeEvent({ txHash: 'txA', eventIndex: 0 }), - makeEvent({ txHash: 'txZ', eventIndex: 0 }), // duplicate - makeEvent({ txHash: 'txM', eventIndex: 0 }), - ]; - - await processIndexerChainEvents(events, handler); - - expect(processed).toEqual(['txZ:0', 'txA:0', 'txM:0']); + it('passes the full event object to the handler unchanged', async () => { + const event = makeIndexerEvent({ txHash: '0xfull', eventIndex: 7, ledger: 9999 }); + const handler = jest.fn().mockResolvedValue(undefined); + await processIndexerChainEvents([event], handler); + expect(handler).toHaveBeenCalledWith(event); }); - it('DB count equals the number of unique (txHash, eventIndex) pairs in the batch', async () => { - const db: { records: IndexerChainEvent[] } = { records: [] }; - const handler = jest.fn().mockImplementation(async (e: IndexerChainEvent) => { - db.records.push(e); - }); - - const events: IndexerChainEvent[] = [ - makeEvent({ txHash: 'ta', eventIndex: 0 }), - makeEvent({ txHash: 'ta', eventIndex: 1 }), - makeEvent({ txHash: 'tb', eventIndex: 0 }), - makeEvent({ txHash: 'ta', eventIndex: 0 }), // duplicate - makeEvent({ txHash: 'tb', eventIndex: 0 }), // duplicate - ]; + it('writes the record to a mock DB on first appearance', async () => { + const db: IndexerChainEvent[] = []; + await processIndexerChainEvents( + [makeIndexerEvent({ txHash: '0xwrite', eventIndex: 0 })], + async e => { db.push(e); }, + ); + expect(db).toHaveLength(1); + expect(db[0].txHash).toBe('0xwrite'); + }); + }); - await processIndexerChainEvents(events, handler); + describe('AC-2 — duplicate pair is skipped (no handler call)', () => { + it('calls the handler only once when the same pair appears twice in the batch', async () => { + const handler = jest.fn().mockResolvedValue(undefined); + await processIndexerChainEvents( + [ + makeIndexerEvent({ txHash: '0xdup', eventIndex: 0 }), + makeIndexerEvent({ txHash: '0xdup', eventIndex: 0 }), + ], + handler, + ); + expect(handler).toHaveBeenCalledTimes(1); + }); - // 5 events in → 3 unique pairs (ta:0, ta:1, tb:0) → 3 DB records - expect(db.records).toHaveLength(3); + it('calls the handler only once when the same pair appears three times', async () => { + const handler = jest.fn().mockResolvedValue(undefined); + await processIndexerChainEvents( + [ + makeIndexerEvent({ txHash: 'tx-triple', eventIndex: 0 }), + makeIndexerEvent({ txHash: 'tx-triple', eventIndex: 0 }), + makeIndexerEvent({ txHash: 'tx-triple', eventIndex: 0 }), + ], + handler, + ); + expect(handler).toHaveBeenCalledTimes(1); }); }); - describe('edge cases', () => { - it('processes an empty batch without errors', async () => { + describe('AC-3 — same txHash different eventIndex treated as two distinct events', () => { + it('calls the handler twice when txHash is the same but eventIndex differs', async () => { const handler = jest.fn().mockResolvedValue(undefined); + await processIndexerChainEvents( + [ + makeIndexerEvent({ txHash: '0x555', eventIndex: 0 }), + makeIndexerEvent({ txHash: '0x555', eventIndex: 1 }), + ], + handler, + ); + expect(handler).toHaveBeenCalledTimes(2); + expect(handler).toHaveBeenNthCalledWith(1, expect.objectContaining({ eventIndex: 0 })); + expect(handler).toHaveBeenNthCalledWith(2, expect.objectContaining({ eventIndex: 1 })); + }); - await expect(processIndexerChainEvents([], handler)).resolves.not.toThrow(); - expect(handler).not.toHaveBeenCalled(); + it('writes two DB records when txHash is shared but eventIndex differs', async () => { + const db: IndexerChainEvent[] = []; + await processIndexerChainEvents( + [ + makeIndexerEvent({ txHash: '0x666', eventIndex: 0 }), + makeIndexerEvent({ txHash: '0x666', eventIndex: 1 }), + ], + async e => { db.push(e); }, + ); + expect(db).toHaveLength(2); + expect(db.map(r => r.eventIndex).sort()).toEqual([0, 1]); }); + }); - it('processes a single-event batch correctly', async () => { - const handler = jest.fn().mockResolvedValue(undefined); - const event = makeEvent({ txHash: '0xsingle', eventIndex: 0 }); + describe('AC-4 — no side effects fire on a skipped duplicate', () => { + it('does not call balanceUpdate for the duplicate event', async () => { + const balanceUpdate = jest.fn(); + await processIndexerChainEvents( + [ + makeIndexerEvent({ txHash: '0x777', eventIndex: 0 }), + makeIndexerEvent({ txHash: '0x777', eventIndex: 0 }), + ], + async () => { balanceUpdate(); }, + ); + expect(balanceUpdate).toHaveBeenCalledTimes(1); + }); - await processIndexerChainEvents([event], handler); + it('does not call priceSnapshot for the duplicate event', async () => { + const priceSnapshot = jest.fn(); + await processIndexerChainEvents( + [ + makeIndexerEvent({ txHash: '0x888', eventIndex: 0 }), + makeIndexerEvent({ txHash: '0x888', eventIndex: 0 }), + ], + async () => { priceSnapshot(); }, + ); + expect(priceSnapshot).toHaveBeenCalledTimes(1); + }); - expect(handler).toHaveBeenCalledTimes(1); - expect(handler).toHaveBeenCalledWith(event); + it('does not call any side effect at all when the entire batch is duplicates of one event', async () => { + const sideEffect = jest.fn(); + await processIndexerChainEvents( + [ + makeIndexerEvent({ txHash: 'only-one', eventIndex: 0 }), + makeIndexerEvent({ txHash: 'only-one', eventIndex: 0 }), + makeIndexerEvent({ txHash: 'only-one', eventIndex: 0 }), + ], + async () => { sideEffect(); }, + ); + expect(sideEffect).toHaveBeenCalledTimes(1); }); + }); - it('deduplication key is case-sensitive for txHash', async () => { - const handler = jest.fn().mockResolvedValue(undefined); - const events: IndexerChainEvent[] = [ - makeEvent({ txHash: '0xABC', eventIndex: 0 }), - makeEvent({ txHash: '0xabc', eventIndex: 0 }), // different case → treated as different - ]; + describe('AC-5 — DB event count unchanged after a duplicate is processed', () => { + it('DB contains exactly 1 record after the same event is submitted twice', async () => { + const db: IndexerChainEvent[] = []; + await processIndexerChainEvents( + [ + makeIndexerEvent({ txHash: '0x999', eventIndex: 0 }), + makeIndexerEvent({ txHash: '0x999', eventIndex: 0 }), + ], + async e => { db.push(e); }, + ); + expect(db).toHaveLength(1); + }); - await processIndexerChainEvents(events, handler); + it('DB count equals the number of unique pairs regardless of how many duplicates arrive', async () => { + const db: IndexerChainEvent[] = []; + await processIndexerChainEvents( + [ + makeIndexerEvent({ txHash: 'ta', eventIndex: 0 }), + makeIndexerEvent({ txHash: 'ta', eventIndex: 1 }), + makeIndexerEvent({ txHash: 'tb', eventIndex: 0 }), + makeIndexerEvent({ txHash: 'ta', eventIndex: 0 }), // duplicate + makeIndexerEvent({ txHash: 'tb', eventIndex: 0 }), // duplicate + makeIndexerEvent({ txHash: 'ta', eventIndex: 1 }), // duplicate + ], + async e => { db.push(e); }, + ); + // 6 events in, 3 unique pairs → exactly 3 records in DB + expect(db).toHaveLength(3); + }); - expect(handler).toHaveBeenCalledTimes(2); + it('does not grow the DB record count on any subsequent duplicate submission', async () => { + const db: IndexerChainEvent[] = []; + const handler = async (e: IndexerChainEvent) => { db.push(e); }; + + // First batch: admit the event + await processIndexerChainEvents( + [makeIndexerEvent({ txHash: 'stable', eventIndex: 0 })], + handler, + ); + expect(db).toHaveLength(1); + + // Second batch in a new pipeline call — each call has its own seen set, + // so this tests the per-batch guard, not a cross-batch persistent store. + await processIndexerChainEvents( + [makeIndexerEvent({ txHash: 'different', eventIndex: 0 })], + handler, + ); + expect(db).toHaveLength(2); // second is a new pair, so it is admitted }); + }); - it('handles eventIndex 0 correctly — does not confuse falsy value with missing key', async () => { + describe('mixed batch — unique and duplicate events together', () => { + it('processes only unique events from a batch containing both', async () => { const handler = jest.fn().mockResolvedValue(undefined); - const events: IndexerChainEvent[] = [ - makeEvent({ txHash: '0xzero', eventIndex: 0 }), - makeEvent({ txHash: '0xzero', eventIndex: 0 }), // duplicate with index 0 - ]; + await processIndexerChainEvents( + [ + makeIndexerEvent({ txHash: 'tx1', eventIndex: 0 }), + makeIndexerEvent({ txHash: 'tx2', eventIndex: 0 }), + makeIndexerEvent({ txHash: 'tx1', eventIndex: 0 }), // dup + makeIndexerEvent({ txHash: 'tx3', eventIndex: 0 }), + makeIndexerEvent({ txHash: 'tx2', eventIndex: 0 }), // dup + ], + handler, + ); + expect(handler).toHaveBeenCalledTimes(3); + }); - await processIndexerChainEvents(events, handler); + it('preserves the first-occurrence order of events after dedup', async () => { + const processed: string[] = []; + await processIndexerChainEvents( + [ + makeIndexerEvent({ txHash: 'txZ', eventIndex: 0 }), + makeIndexerEvent({ txHash: 'txA', eventIndex: 0 }), + makeIndexerEvent({ txHash: 'txZ', eventIndex: 0 }), // dup + makeIndexerEvent({ txHash: 'txM', eventIndex: 0 }), + ], + async e => { processed.push(`${e.txHash}:${e.eventIndex}`); }, + ); + expect(processed).toEqual(['txZ:0', 'txA:0', 'txM:0']); + }); + }); - expect(handler).toHaveBeenCalledTimes(1); + describe('edge cases', () => { + it('handles an empty batch without error and without calling the handler', async () => { + const handler = jest.fn(); + await expect(processIndexerChainEvents([], handler)).resolves.not.toThrow(); + expect(handler).not.toHaveBeenCalled(); }); - it('handles large eventIndex values without collision', async () => { + it('handles a single-event batch correctly', async () => { const handler = jest.fn().mockResolvedValue(undefined); - const events: IndexerChainEvent[] = [ - makeEvent({ txHash: '0xlarge', eventIndex: 999999999 }), - makeEvent({ txHash: '0xlarge', eventIndex: 999999999 }), // duplicate - ]; - - await processIndexerChainEvents(events, handler); - + const event = makeIndexerEvent({ txHash: '0xsingle', eventIndex: 0 }); + await processIndexerChainEvents([event], handler); expect(handler).toHaveBeenCalledTimes(1); + expect(handler).toHaveBeenCalledWith(event); }); }); }); diff --git a/src/utils/indexer-dedupe.utils.ts b/src/utils/indexer-dedupe.utils.ts index b40fb55..004fe31 100644 --- a/src/utils/indexer-dedupe.utils.ts +++ b/src/utils/indexer-dedupe.utils.ts @@ -31,3 +31,34 @@ export function dedupeChainEvents(events: T[]): T[] { return true; }); } + +/** + * Result returned by the per-event deduplication guard. + */ +export interface DedupeGuardResult { + /** True when the event was already present in `seen` and was skipped. */ + skipped: boolean; +} + +/** + * Per-event deduplication guard. + * + * Checks a single chain event against a caller-maintained `seen` set. + * If the `(txHash, eventIndex)` composite key is already in `seen` the + * event is a duplicate and `{ skipped: true }` is returned without + * mutating anything else. Otherwise the key is recorded in `seen` and + * `{ skipped: false }` is returned so the caller can proceed to process + * the event. + * + * @param event - The chain event to check. + * @param seen - Mutable Set that tracks already-processed composite keys. + * @returns `{ skipped: true }` for duplicates, `{ skipped: false }` for new events. + */ +export function guardChainEvent(event: ChainEvent, seen: Set): DedupeGuardResult { + const key = `${event.txHash}:${event.eventIndex}`; + if (seen.has(key)) { + return { skipped: true }; + } + seen.add(key); + return { skipped: false }; +}