diff --git a/docs/indexer/PIPELINE_REFERENCE.md b/docs/indexer/PIPELINE_REFERENCE.md new file mode 100644 index 0000000..acd489f --- /dev/null +++ b/docs/indexer/PIPELINE_REFERENCE.md @@ -0,0 +1,353 @@ +# Indexer Event Processing Pipeline Reference + +This document is a comprehensive guide to the indexer's event processing pipeline. It explains how events flow from the polling worker, through deduplication and processing, how the system recovers from crashes, and where to find detailed documentation for each concern. + +**Quick links:** + +- [Event Processing](./EVENT_PROCESSING.md) — deduplication and idempotency detail +- [Retry and Backoff](./RETRY_BACKOFF.md) — retry strategy and exhaustion behaviour +- [Architecture](./ARCHITECTURE.md) — end-to-end write path and data models +- [Local Setup Guide](../README.md) — how to set up the indexer for development + +--- + +## Table of Contents + +1. [Polling Mechanism](#polling-mechanism) +2. [Event Ordering](#event-ordering) +3. [Retry Strategy](#retry-strategy) +4. [Crash Recovery](#crash-recovery) +5. [Processing Guarantees](#processing-guarantees) +6. [Troubleshooting](#troubleshooting) + +--- + +## Polling Mechanism + +### Overview + +The indexer uses an external polling worker to fetch events from the Stellar blockchain. This server (access-layer-server) does not perform the polling itself — it receives events as `ChainEvent[]` payloads and processes them idempotently. + +### Ledger Range Fetching + +The external worker: + +1. **Reads the resume point** from this server's `IndexedLedger` table (row ID = `1`) + - Field `ledger`: the highest ledger already fully processed + - Field `cursor`: an opaque pagination token for the Soroban RPC + - Field `updatedAt`: ISO timestamp of the last update + +2. **Fetches a batch** from the Stellar blockchain + - Uses `STELLAR_SOROBAN_RPC_URL` and `STELLAR_HORIZON_URL` (configured in the external worker, not this server) + - Retrieves all events in ledgers from `ledger + 1` to `ledger + BATCH_SIZE` + - Typical batch size: 100–500 ledgers (configurable in the external worker) + +3. **Polls continuously** at an interval + - Typical polling interval: every 5–30 seconds (configurable in the external worker) + - If no new events are found, the ledger range is advanced with the same retry logic + +4. **Posts events to this server** + - Calls the indexer event sink endpoint (typically `POST /api/v1/indexer/events`) + - Sends the batch as `ChainEvent[]` JSON + - This server returns synchronously after processing or moving to the DLQ + +### Progress Tracking + +After successfully processing a batch, the external worker calls `updateIndexedLedger(ledger, cursor)` to persist: + +```typescript +// Located in src/modules/indexer/ledger-gap-detection.service.ts +export async function updateIndexedLedger( + ledger: number, + cursor: string +): Promise { + await prisma.indexedLedger.update({ + where: { id: 1 }, + data: { ledger, cursor, updatedAt: new Date() }, + }); +} +``` + +If this call succeeds, the next fetch will start from `ledger + 1`. If it fails, the batch must be retried (see [Retry Strategy](#retry-strategy)). + +### Gap Detection + +The function `detectLedgerGap()` in `src/modules/indexer/ledger-gap-detection.service.ts` compares the server's `IndexedLedger.ledger` against the Stellar network head. If the gap exceeds **10 ledgers** (~50 seconds at 5 s/ledger): + +```typescript +if (networkHeadLedger - indexedLedger.ledger > 10) { + logger.warn( + { gap, indexedLedger: indexedLedger.ledger }, + 'Large ledger gap detected' + ); +} +``` + +**Note:** In development, the network head is mocked to `12_400`. A `// TODO` comment marks where the real RPC call will go. Do not rely on this in production until resolved. + +### Cursor Staleness + +If `IndexedLedger.updatedAt` is older than `INDEXER_CURSOR_STALE_AGE_WARNING_MS` (default: 5 minutes), the helper `checkIndexerCursorStalenessFromStore()` emits a warn log: + +``` +wallet_address: N/A +message: "Indexer cursor is stale (X minutes without progress)" +``` + +This warns that the external worker has not made progress recently. Enable/disable with the feature flag `ENABLE_INDEXER_CURSOR_STALENESS_WARNING` (see [Feature Flags](./FEATURE_FLAGS.md)). + +--- + +## Event Ordering + +### Order Within a Ledger + +Events arrive as `ChainEvent[]` from the external worker, already sorted by the Stellar blockchain: + +1. **Ledger sequence** — events are grouped by ledger (ascending) +2. **Transaction order** — within a ledger, events are sorted by transaction index +3. **Event index** — within a transaction, events are sorted by `eventIndex` (ascending) + +This ordering is preserved by the Stellar RPC APIs (`STELLAR_SOROBAN_RPC_URL`), so events always flow downstream in a deterministic order. + +### Processing Order + +`processIndexerChainEvents(events, handler)` in `src/utils/indexer-event-processor.utils.ts`: + +1. **Deduplicates** by `txHash:eventIndex` (removes duplicates from overlapping batches) +2. **Maintains order** — deduped events retain their original sequence +3. **Processes sequentially** — one event per iteration, using a `for` loop +4. **Emits one log per event** — after each event completes successfully + +```typescript +export async function processIndexerChainEvents( + events: T[], + handler: (event: T) => Promise +): Promise { + const uniqueEvents = dedupeChainEvents(events); + + for (const event of uniqueEvents) { + await processIndexerChainEvent(event, handler); + } +} +``` + +### Guarantees + +- **No out-of-order processing**: events are handled in the order they appear on-chain +- **No skipped events**: every unique event is processed exactly once per batch (dedup removes only true duplicates) +- **No parallelization**: sequential processing ensures balance changes and price updates are idempotent + +See [Event Processing](./EVENT_PROCESSING.md) for deduplication details. + +--- + +## Retry Strategy + +### What Triggers a Retry + +A retry is triggered when: + +1. **Database write fails** (e.g. constraint violation, connection timeout) +2. **Webhook dispatch fails** (event delivered to callback URL) +3. **Ledger cursor update fails** (the `updateIndexedLedger` call) +4. **External API timeout** (if any RPC calls occur during processing) + +When any of these fails, the entire batch is retried (or the job is moved to the DLQ if retries are exhausted). + +### How Many Retries + +The retry count is determined by `INDEXER_MAX_RETRIES` or a sensible default (typically 5–10 attempts). Each retry uses an exponential backoff strategy: + +``` +delay = applyJitter( min(maxDelayMs, baseDelayMs * 2^attempt), jitterFactor ) +``` + +**Default backoff schedule** (before jitter): + +| Attempt | Delay | +| ------- | ------- | +| 0 | 1s | +| 1 | 2s | +| 2 | 4s | +| 3 | 8s | +| 4 | 16s | +| 5+ | 30s cap | + +After applying jitter (default `INDEXER_JITTER_FACTOR = 0.1`), delays vary by ±10%. + +See [Retry and Backoff](./RETRY_BACKOFF.md) for full details, including configuration env vars. + +### Exhaustion Behavior + +When a job exhausts all retries: + +1. **Move to DLQ** — the batch is moved to the `indexerDLQ` table via `moveToDLQ()` +2. **Record captured** — stores `jobType`, `payload`, `retryCount`, `failureReason`, and `errorDetails` +3. **Job is paused** — no further retries are attempted automatically +4. **Observable** — `getDLQDepth()` exposes the DLQ backlog to monitoring + +**If DLQ is disabled** (`ENABLE_INDEXER_DLQ=false`), retry-exhausted jobs are logged and dropped (silent job loss). + +### Idempotency and Retries + +Because retries may re-process the same event: + +- **All writes must be idempotent** (upsert, not insert) +- The `activity` table uses a unique constraint on `(txHash, eventIndex)` (if present) +- `updateOwnership()` uses `upsert` with `{ increment: balanceChange }` +- `upsertPriceSnapshot()` checks `tradeAt ≤ lastTradeAt` to skip stale events + +--- + +## Crash Recovery + +### How Crash Recovery Works + +If the indexer or this server crashes mid-batch: + +1. **Last update persists** — `IndexedLedger` was last written after the previous batch + - If batch N crashes, `IndexedLedger.ledger` still points to batch N−1 +2. **Restart reads the resume point** — on restart, the external worker reads `IndexedLedger` again +3. **Re-fetches batch N** — the worker starts from `IndexedLedger.ledger + 1` (same as before the crash) +4. **Dedup absorbs overlaps** — events from batch N are deduped; any already-written events are skipped by idempotent operations + +### Resume Point Logic + +The resume point is determined by: + +```typescript +const resumeLedger = indexedLedger.ledger; // Highest fully-processed ledger +const nextBatch = resumeLedger + 1; // Start here +``` + +This means: + +- **After batch 100 succeeds**: `IndexedLedger.ledger = 100`, next fetch starts at ledger 101 +- **If batch 101 crashes**: `IndexedLedger.ledger` is still 100, next restart fetches 101 again +- **On restart, batch 101 is retried**: events are deduped and idempotent ops (upsert) ensure no duplicates + +### No Events Are Lost + +Because: + +1. **Ledger cursor is durable** — persisted to `IndexedLedger` table +2. **Batch requests are idempotent** — re-fetching the same ledger range yields the same events +3. **Event processing is idempotent** — re-processing the same event via upsert/check has no side-effect + +--- + +## Processing Guarantees + +### Exactly-Once Delivery (Per Event) + +Events are processed exactly once per batch because: + +1. **Deduplication** removes event duplicates (same `txHash:eventIndex`) +2. **Idempotent writes** ensure re-processing has no side-effect +3. **Cursor update is atomic** — either the batch succeeds fully, or it is retried in toto + +### Strong Ordering + +Events are processed in the order they appear on-chain: + +1. **Ledger order preserved** — ledgers are processed sequentially +2. **Event order preserved** — within a ledger, events are in transaction order +3. **Sequential processing** — no parallelization within a batch + +This ensures balance changes and price updates happen in the correct order. + +### Atomic Cursor Advance + +The batch is considered complete only after: + +1. All events are processed (inserted to `activity`, upserted to `keyOwnership`, etc.) +2. All webhooks are dispatched (or queued for retry) +3. The cursor is updated via `updateIndexedLedger()` + +If step 3 fails, the batch is retried. If steps 1–2 partially succeed but step 3 fails, re-processing is safe due to idempotency. + +--- + +## Troubleshooting + +### Indexer Falls Behind (Large Ledger Gap) + +**Symptom**: `detectLedgerGap()` logs show gap > 10 ledgers + +**Possible causes**: + +1. External worker is not running or is paused +2. Network connectivity issue between worker and this server +3. Database is slow or unresponsive +4. Batch size is too small (fetch overhead dominates) + +**Steps**: + +1. Check if the external worker is running: `ps aux | grep indexer` +2. Check `IndexedLedger.updatedAt` — is it recent? +3. Check the server logs for errors during event processing +4. Increase the batch size in the external worker config if appropriate + +### Cursor is Stale + +**Symptom**: `checkIndexerCursorStalenessFromStore()` logs "cursor is stale" + +**Possible causes**: + +1. External worker crashed or is stuck +2. Event processing is very slow (bottleneck in database, webhooks, etc.) +3. Network latency or RPC timeout + +**Steps**: + +1. Check the external worker logs for errors +2. Check if this server is responsive: `curl http://localhost:3000/health` +3. Check database connection pool stats (if exposed) +4. Restart the external worker + +### Dead-Letter Queue Growing + +**Symptom**: `getDLQDepth()` increases; `indexerDLQ` table has many rows + +**Possible causes**: + +1. Systematic failure in event processing (e.g. bad contract data from RPC) +2. Feature flag misconfiguration (e.g. invalid rate limit values) +3. Database constraint violation (unique constraint on an already-inserted record) + +**Steps**: + +1. Inspect the DLQ records: `SELECT * FROM indexer_dlq ORDER BY createdAt DESC LIMIT 10;` +2. Check the `failureReason` and `errorDetails` for clues +3. Check the server logs during the failure period +4. Fix the root cause and replay the DLQ (see [Dead-Letter Queue Workflow](./DLQ_WORKFLOW.md)) + +### Duplicate Records Despite Deduplication + +**Symptom**: Same event appears twice in the `activity` table + +**Possible causes**: + +1. Deduplication is disabled (`ENABLE_INDEXER_DEDUPE=false`) +2. Events have the same `txHash` but different `eventIndex` (not duplicates) +3. Multiple external workers are running without coordination (each posts a separate batch) + +**Steps**: + +1. Check the `eventIndex` — if different, they are separate events +2. Verify `ENABLE_INDEXER_DEDUPE=true` at startup +3. Ensure only one external worker is polling (or use a distributed lock) +4. Check the activity records' `createdAt` — did they come from different batches? + +--- + +## See Also + +- [Event Processing](./EVENT_PROCESSING.md) — in-depth deduplication and idempotency +- [Retry and Backoff](./RETRY_BACKOFF.md) — retry configuration and DLQ behavior +- [Architecture](./ARCHITECTURE.md) — write path, data models, and webhook dispatch +- [Dead-Letter Queue Workflow](./DLQ_WORKFLOW.md) — DLQ investigation and replay +- [Feature Flags](./FEATURE_FLAGS.md) — startup validation and flag reference +- [Recovery](./RECOVERY.md) — outage response and data consistency +- [Local Setup](../README.md) — how to run and test the indexer locally diff --git a/src/utils/__tests__/indexer-dedupe.utils.jest.test.ts b/src/utils/__tests__/indexer-dedupe.utils.jest.test.ts new file mode 100644 index 0000000..73887d0 --- /dev/null +++ b/src/utils/__tests__/indexer-dedupe.utils.jest.test.ts @@ -0,0 +1,241 @@ +/** + * Unit tests for indexer deduplication logic. + * + * Tests confirm that duplicate events (identified by txHash + eventIndex composite key) + * are detected and discarded without creating duplicate database records or throwing errors. + * + * Scope: + * - First event with a given hash+index is inserted successfully + * - Second event with the same hash+index is discarded (no duplicate record, no error thrown) + * - Event with the same hash but different index is treated as a new record + * - Event with different hash but same index is treated as a new record + */ + +import { dedupeChainEvents, ChainEvent } from '../indexer-dedupe.utils'; + +describe('indexer-dedupe.utils — dedupeChainEvents()', () => { + describe('composite key deduplication (txHash + eventIndex)', () => { + it('first event with a given hash+index is included', () => { + const events: ChainEvent[] = [ + { txHash: 'abc123', eventIndex: 0, ledger: 100 }, + ]; + + const deduped = dedupeChainEvents(events); + + expect(deduped).toHaveLength(1); + expect(deduped[0].txHash).toBe('abc123'); + expect(deduped[0].eventIndex).toBe(0); + }); + + it('second event with the same hash+index is discarded', () => { + const events: ChainEvent[] = [ + { txHash: 'abc123', eventIndex: 0, ledger: 100 }, + { txHash: 'abc123', eventIndex: 0, ledger: 100 }, // Exact duplicate + ]; + + const deduped = dedupeChainEvents(events); + + expect(deduped).toHaveLength(1); + expect(deduped[0].txHash).toBe('abc123'); + expect(deduped[0].eventIndex).toBe(0); + }); + + it('multiple duplicates of the same event are all discarded', () => { + const events: ChainEvent[] = [ + { txHash: 'abc123', eventIndex: 0 }, + { txHash: 'abc123', eventIndex: 0 }, + { txHash: 'abc123', eventIndex: 0 }, + { txHash: 'abc123', eventIndex: 0 }, + ]; + + const deduped = dedupeChainEvents(events); + + expect(deduped).toHaveLength(1); + }); + + it('event with the same hash but different index is treated as new', () => { + const events: ChainEvent[] = [ + { txHash: 'abc123', eventIndex: 0 }, + { txHash: 'abc123', eventIndex: 1 }, // Same hash, different index + ]; + + const deduped = dedupeChainEvents(events); + + expect(deduped).toHaveLength(2); + expect(deduped[0].eventIndex).toBe(0); + expect(deduped[1].eventIndex).toBe(1); + }); + + it('event with different hash but same index is treated as new', () => { + const events: ChainEvent[] = [ + { txHash: 'abc123', eventIndex: 5 }, + { txHash: 'xyz789', eventIndex: 5 }, // Different hash, same index + ]; + + const deduped = dedupeChainEvents(events); + + expect(deduped).toHaveLength(2); + expect(deduped[0].txHash).toBe('abc123'); + expect(deduped[1].txHash).toBe('xyz789'); + }); + }); + + describe('batch deduplication', () => { + it('dedupes a batch with mixed unique and duplicate events', () => { + const events: ChainEvent[] = [ + { txHash: 'tx1', eventIndex: 0 }, + { txHash: 'tx2', eventIndex: 0 }, + { txHash: 'tx1', eventIndex: 0 }, // Duplicate + { txHash: 'tx3', eventIndex: 0 }, + { txHash: 'tx2', eventIndex: 0 }, // Duplicate + { txHash: 'tx4', eventIndex: 0 }, + ]; + + const deduped = dedupeChainEvents(events); + + expect(deduped).toHaveLength(4); + expect(deduped.map(e => e.txHash)).toEqual(['tx1', 'tx2', 'tx3', 'tx4']); + }); + + it('preserves order of first occurrences', () => { + const events: ChainEvent[] = [ + { txHash: 'z', eventIndex: 0 }, + { txHash: 'a', eventIndex: 0 }, + { txHash: 'z', eventIndex: 0 }, // Duplicate + { txHash: 'm', eventIndex: 0 }, + { txHash: 'a', eventIndex: 0 }, // Duplicate + ]; + + const deduped = dedupeChainEvents(events); + + expect(deduped.map(e => e.txHash)).toEqual(['z', 'a', 'm']); + }); + }); + + describe('edge cases', () => { + it('handles empty event list', () => { + const events: ChainEvent[] = []; + + const deduped = dedupeChainEvents(events); + + expect(deduped).toHaveLength(0); + }); + + it('preserves all fields from events (not just txHash and eventIndex)', () => { + const events: ChainEvent[] = [ + { + txHash: 'tx1', + eventIndex: 0, + ledger: 1000, + someOtherField: 'data', + }, + ]; + + const deduped = dedupeChainEvents(events); + + expect(deduped[0]).toEqual({ + txHash: 'tx1', + eventIndex: 0, + ledger: 1000, + someOtherField: 'data', + }); + }); + + it('keeps first occurrence when same event appears twice with different extra fields', () => { + const events: ChainEvent[] = [ + { txHash: 'tx1', eventIndex: 0, metadata: 'first' }, + { txHash: 'tx1', eventIndex: 0, metadata: 'second' }, + ]; + + const deduped = dedupeChainEvents(events); + + expect(deduped).toHaveLength(1); + expect(deduped[0].metadata).toBe('first'); + }); + + it('handles events with eventIndex 0', () => { + const events: ChainEvent[] = [ + { txHash: 'tx1', eventIndex: 0 }, + { txHash: 'tx1', eventIndex: 0 }, + ]; + + const deduped = dedupeChainEvents(events); + + expect(deduped).toHaveLength(1); + }); + + it('handles large eventIndex values', () => { + const events: ChainEvent[] = [ + { txHash: 'tx1', eventIndex: 999999999 }, + { txHash: 'tx1', eventIndex: 999999999 }, + ]; + + const deduped = dedupeChainEvents(events); + + expect(deduped).toHaveLength(1); + }); + + it('composite key is case-sensitive for txHash', () => { + const events: ChainEvent[] = [ + { txHash: 'ABC123', eventIndex: 0 }, + { txHash: 'abc123', eventIndex: 0 }, // Different case + ]; + + const deduped = dedupeChainEvents(events); + + // Different case means different txHash, so both are kept + expect(deduped).toHaveLength(2); + }); + + it('does not throw error on duplicate event', () => { + const events: ChainEvent[] = [ + { txHash: 'tx1', eventIndex: 0 }, + { txHash: 'tx1', eventIndex: 0 }, + ]; + + expect(() => { + dedupeChainEvents(events); + }).not.toThrow(); + }); + }); + + describe('correctness of composite key', () => { + it('the composite key is specifically txHash:eventIndex (not just one field)', () => { + // This test verifies that deduplication uses BOTH fields + const events: ChainEvent[] = [ + { txHash: 'tx1', eventIndex: 1 }, + { txHash: 'tx2', eventIndex: 1 }, + { txHash: 'tx1', eventIndex: 2 }, + { txHash: 'tx1', eventIndex: 1 }, // Only this one is a duplicate + ]; + + const deduped = dedupeChainEvents(events); + + expect(deduped).toHaveLength(3); + const ids = deduped.map(e => `${e.txHash}:${e.eventIndex}`); + expect(ids).toEqual(['tx1:1', 'tx2:1', 'tx1:2']); + }); + }); + + describe('returns new array (immutability)', () => { + it('returns a new array instance', () => { + const events: ChainEvent[] = [{ txHash: 'tx1', eventIndex: 0 }]; + + const deduped = dedupeChainEvents(events); + + expect(deduped).not.toBe(events); + }); + + it('does not mutate the input array', () => { + const events: ChainEvent[] = [ + { txHash: 'tx1', eventIndex: 0 }, + { txHash: 'tx1', eventIndex: 0 }, + ]; + const original = [...events]; + + dedupeChainEvents(events); + + expect(events).toEqual(original); + }); + }); +}); diff --git a/src/utils/__tests__/rate-limit-tracking.utils.test.ts b/src/utils/__tests__/rate-limit-tracking.utils.test.ts new file mode 100644 index 0000000..89f541d --- /dev/null +++ b/src/utils/__tests__/rate-limit-tracking.utils.test.ts @@ -0,0 +1,330 @@ +/** + * Unit tests for rate limit tracking and threshold logging. + * + * Tests confirm that: + * - 80% threshold log emitted once per window when crossed + * - 100% threshold log emitted once per window when crossed + * - Wallet address truncated in log + * - Both logs contain all five fields (wallet_address, request_count, limit, window_reset_at, threshold) + */ + +import { + truncateWalletAddress, + checkRateLimitThresholds, + cleanupRateLimitState, + resetRateLimitState, +} from '../rate-limit-tracking.utils'; +import { logger } from '../logger.utils'; + +jest.mock('../logger.utils', () => ({ + logger: { + warn: jest.fn(), + }, +})); + +const mockLogger = logger as unknown as { warn: jest.Mock }; + +describe('rate-limit-tracking.utils', () => { + beforeEach(() => { + jest.clearAllMocks(); + resetRateLimitState(); + }); + + describe('truncateWalletAddress()', () => { + it('truncates long addresses to first 4 + last 4 characters', () => { + const address = + 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; + + const truncated = truncateWalletAddress(address); + + expect(truncated).toBe('GAAA…AAAA'); + }); + + it('returns the address unchanged if 8 characters or less', () => { + const short = 'ABCD1234'; + + expect(truncateWalletAddress(short)).toBe('ABCD1234'); + }); + + it('handles addresses with exactly 9 characters', () => { + const address = 'ABCDEFGHI'; + + const truncated = truncateWalletAddress(address); + + expect(truncated).toBe('ABCD…EFHI'); + }); + + it('uses ellipsis character in the middle', () => { + const address = + 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; + + const truncated = truncateWalletAddress(address); + + expect(truncated).toContain('…'); + expect(truncated.split('…')).toHaveLength(2); + }); + }); + + describe('checkRateLimitThresholds() — 80% threshold', () => { + it('emits warn log when crossing 80% threshold', () => { + const wallet = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; + const limit = 100; + const windowReset = new Date('2026-01-15T13:00:00Z'); + + // Trigger at 80% + checkRateLimitThresholds(wallet, 80, limit, windowReset); + + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + const call = mockLogger.warn.mock.calls[0]; + expect(call[0]).toMatchObject({ + wallet_address: 'GAAA…AAAA', + request_count: 80, + limit: 100, + window_reset_at: '2026-01-15T13:00:00.000Z', + threshold: '80%', + }); + }); + + it('emits log only once per window at 80% threshold', () => { + const wallet = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; + const limit = 100; + const windowReset = new Date('2026-01-15T13:00:00Z'); + + // Trigger at 80% + checkRateLimitThresholds(wallet, 80, limit, windowReset); + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + + // Trigger again at 81% (should not log) + checkRateLimitThresholds(wallet, 81, limit, windowReset); + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + + // Trigger at 99% (should not log) + checkRateLimitThresholds(wallet, 99, limit, windowReset); + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + }); + + it('emits log again in a new window', () => { + const wallet = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; + const limit = 100; + const windowReset1 = new Date('2026-01-15T13:00:00Z'); + const windowReset2 = new Date('2026-01-15T13:15:00Z'); + + // First window + checkRateLimitThresholds(wallet, 80, limit, windowReset1); + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + + // Second window with new reset time + checkRateLimitThresholds(wallet, 85, limit, windowReset2); + expect(mockLogger.warn).toHaveBeenCalledTimes(2); + }); + }); + + describe('checkRateLimitThresholds() — 100% threshold', () => { + it('emits warn log when hitting 100% of limit', () => { + const wallet = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; + const limit = 100; + const windowReset = new Date('2026-01-15T13:00:00Z'); + + checkRateLimitThresholds(wallet, 100, limit, windowReset); + + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + const call = mockLogger.warn.mock.calls[0]; + expect(call[0]).toMatchObject({ + wallet_address: 'GAAA…AAAA', + request_count: 100, + limit: 100, + window_reset_at: '2026-01-15T13:00:00.000Z', + threshold: '100%', + }); + }); + + it('emits log only once per window at 100% threshold', () => { + const wallet = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; + const limit = 100; + const windowReset = new Date('2026-01-15T13:00:00Z'); + + // Trigger at 100% + checkRateLimitThresholds(wallet, 100, limit, windowReset); + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + + // Trigger again at 100% (should not log) + checkRateLimitThresholds(wallet, 100, limit, windowReset); + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + + // Trigger at 101% (should not log again) + checkRateLimitThresholds(wallet, 101, limit, windowReset); + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + }); + }); + + describe('checkRateLimitThresholds() — both thresholds', () => { + it('emits both 80% and 100% logs when crossing 100%', () => { + const wallet = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; + const limit = 100; + const windowReset = new Date('2026-01-15T13:00:00Z'); + + // Jump directly to 100%, should log both 80% and 100% + checkRateLimitThresholds(wallet, 100, limit, windowReset); + + // Both thresholds should be logged + expect(mockLogger.warn).toHaveBeenCalledTimes(2); + const calls = mockLogger.warn.mock.calls; + expect(calls[0][0].threshold).toBe('80%'); + expect(calls[1][0].threshold).toBe('100%'); + }); + + it('logs only 80% if request count goes from 80 to 99', () => { + const wallet = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; + const limit = 100; + const windowReset = new Date('2026-01-15T13:00:00Z'); + + checkRateLimitThresholds(wallet, 80, limit, windowReset); + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + + checkRateLimitThresholds(wallet, 99, limit, windowReset); + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + }); + }); + + describe('checkRateLimitThresholds() — log fields', () => { + it('includes all five required fields in the warn log', () => { + const wallet = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; + const limit = 100; + const windowReset = new Date('2026-01-15T13:00:00Z'); + + checkRateLimitThresholds(wallet, 80, limit, windowReset); + + const logData = mockLogger.warn.mock.calls[0][0]; + expect(logData).toHaveProperty('wallet_address'); + expect(logData).toHaveProperty('request_count'); + expect(logData).toHaveProperty('limit'); + expect(logData).toHaveProperty('window_reset_at'); + expect(logData).toHaveProperty('threshold'); + }); + + it('wallet_address is truncated in the log', () => { + const wallet = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; + const limit = 100; + const windowReset = new Date('2026-01-15T13:00:00Z'); + + checkRateLimitThresholds(wallet, 80, limit, windowReset); + + const logData = mockLogger.warn.mock.calls[0][0]; + expect(logData.wallet_address).toBe('GAAA…AAAA'); + expect(logData.wallet_address.length).toBeLessThan(wallet.length); + }); + + it('window_reset_at is in ISO format', () => { + const wallet = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; + const limit = 100; + const windowReset = new Date('2026-01-15T13:00:00Z'); + + checkRateLimitThresholds(wallet, 80, limit, windowReset); + + const logData = mockLogger.warn.mock.calls[0][0]; + const resetTime = new Date(logData.window_reset_at); + expect(resetTime.getTime()).toBe(windowReset.getTime()); + }); + }); + + describe('checkRateLimitThresholds() — edge cases', () => { + it('ignores requests with empty wallet address', () => { + checkRateLimitThresholds('', 80, 100, new Date()); + expect(mockLogger.warn).not.toHaveBeenCalled(); + }); + + it('ignores requests with 0 request count', () => { + checkRateLimitThresholds( + 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA', + 0, + 100, + new Date() + ); + expect(mockLogger.warn).not.toHaveBeenCalled(); + }); + + it('ignores requests with 0 limit', () => { + checkRateLimitThresholds( + 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA', + 50, + 0, + new Date() + ); + expect(mockLogger.warn).not.toHaveBeenCalled(); + }); + + it('handles fractional percentages (e.g., 80.5%)', () => { + const wallet = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; + const limit = 1000; + const windowReset = new Date('2026-01-15T13:00:00Z'); + + checkRateLimitThresholds(wallet, 805, limit, windowReset); + + expect(mockLogger.warn).toHaveBeenCalled(); + const logData = mockLogger.warn.mock.calls[0][0]; + expect(logData.threshold).toBe('80%'); + }); + }); + + describe('cleanupRateLimitState()', () => { + it('removes old entries from tracking state', () => { + const wallet = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; + const limit = 100; + const oldWindow = new Date(Date.now() - 60 * 60 * 1000); // 1 hour ago + const recentWindow = new Date(); + + // Add old and recent entries + checkRateLimitThresholds(wallet, 80, limit, oldWindow); + checkRateLimitThresholds(wallet, 80, limit, recentWindow); + + // Both should have logged + expect(mockLogger.warn).toHaveBeenCalledTimes(2); + + // Cleanup old entries (older than 30 minutes ago) + const thirtyMinutesAgo = Date.now() - 30 * 60 * 1000; + cleanupRateLimitState(thirtyMinutesAgo); + + // Trying to log for the old window should log again (state was cleaned) + mockLogger.warn.mockClear(); + checkRateLimitThresholds(wallet, 80, limit, oldWindow); + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + }); + }); + + describe('resetRateLimitState()', () => { + it('clears all tracking state', () => { + const wallet = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; + const limit = 100; + const windowReset = new Date(); + + checkRateLimitThresholds(wallet, 80, limit, windowReset); + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + + resetRateLimitState(); + mockLogger.warn.mockClear(); + + // After reset, same threshold should log again + checkRateLimitThresholds(wallet, 80, limit, windowReset); + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + }); + }); + + describe('different wallets tracked independently', () => { + it('tracks different wallets separately', () => { + const wallet1 = 'GAAA1111111111111111111111111111111111111111111111111111'; + const wallet2 = 'GBBB2222222222222222222222222222222222222222222222222222'; + const limit = 100; + const windowReset = new Date(); + + checkRateLimitThresholds(wallet1, 80, limit, windowReset); + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + + checkRateLimitThresholds(wallet2, 80, limit, windowReset); + expect(mockLogger.warn).toHaveBeenCalledTimes(2); + + // Same threshold for wallet1 should not log again + checkRateLimitThresholds(wallet1, 81, limit, windowReset); + expect(mockLogger.warn).toHaveBeenCalledTimes(2); + }); + }); +}); diff --git a/src/utils/__tests__/trading-volume.utils.test.ts b/src/utils/__tests__/trading-volume.utils.test.ts new file mode 100644 index 0000000..3dd1331 --- /dev/null +++ b/src/utils/__tests__/trading-volume.utils.test.ts @@ -0,0 +1,231 @@ +import { compute24hVolume } from '../trading-volume.utils'; +import { prisma } from '../../utils/prisma.utils'; + +jest.mock('../../utils/prisma.utils', () => ({ + prisma: { + activity: { + findMany: jest.fn(), + }, + }, +})); + +jest.mock('../../utils/logger.utils', () => ({ + logger: { + error: jest.fn(), + }, +})); + +const mockPrisma = prisma as unknown as { + activity: { findMany: jest.Mock }; +}; + +describe('compute24hVolume()', () => { + const CREATOR_ID = 'test-creator-123'; + const NOW = new Date('2026-01-15T12:00:00Z'); + + beforeEach(() => { + jest.clearAllMocks(); + jest.useFakeTimers(); + jest.setSystemTime(NOW); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('sums all trades within the rolling 24h window', async () => { + const twentyFourHoursAgo = new Date(NOW.getTime() - 24 * 60 * 60 * 1000); + + const trades = [ + { + payload: { price: '1000' }, + }, + { + payload: { price: '2000' }, + }, + { + payload: { price: '3000' }, + }, + ]; + + mockPrisma.activity.findMany.mockResolvedValue(trades); + + const volume = await compute24hVolume(CREATOR_ID); + + expect(volume).toBe(6000n); + expect(mockPrisma.activity.findMany).toHaveBeenCalledWith({ + where: { + creatorId: CREATOR_ID, + type: { in: ['KEY_BOUGHT', 'KEY_SOLD'] }, + createdAt: { + gte: twentyFourHoursAgo, + lte: NOW, + }, + }, + select: { + payload: true, + }, + }); + }); + + it('excludes trades outside the 24h window', async () => { + const trades = [ + { + payload: { price: '1000' }, // Inside window + }, + { + payload: { price: '2000' }, // Inside window + }, + ]; + + mockPrisma.activity.findMany.mockResolvedValue(trades); + + const volume = await compute24hVolume(CREATOR_ID); + + // The mock only returns trades inside the window since that's what + // the query filter would return + expect(volume).toBe(3000n); + }); + + it('returns 0 when no trades exist within the window', async () => { + mockPrisma.activity.findMany.mockResolvedValue([]); + + const volume = await compute24hVolume(CREATOR_ID); + + expect(volume).toBe(0n); + }); + + it('handles large trade volumes without overflow', async () => { + const trades = [ + { + payload: { price: '9223372036854775800' }, // Near max BigInt + }, + { + payload: { price: '100' }, + }, + ]; + + mockPrisma.activity.findMany.mockResolvedValue(trades); + + const volume = await compute24hVolume(CREATOR_ID); + + expect(volume).toBe(9223372036854775900n); + }); + + it('ignores trades with missing price in payload', async () => { + const trades = [ + { + payload: { price: '1000' }, + }, + { + payload: { amount: '5' }, // Missing price + }, + { + payload: { price: '2000' }, + }, + ]; + + mockPrisma.activity.findMany.mockResolvedValue(trades); + + const volume = await compute24hVolume(CREATOR_ID); + + // Should only sum the two trades with valid prices + expect(volume).toBe(3000n); + }); + + it('handles both KEY_BOUGHT and KEY_SOLD activity types', async () => { + // The query filter should include both types + const trades = [ + { + payload: { price: '1000' }, + }, + { + payload: { price: '2000' }, + }, + ]; + + mockPrisma.activity.findMany.mockResolvedValue(trades); + + await compute24hVolume(CREATOR_ID); + + // Verify that the filter includes both types + const callArgs = mockPrisma.activity.findMany.mock.calls[0][0]; + expect(callArgs.where.type).toEqual({ in: ['KEY_BOUGHT', 'KEY_SOLD'] }); + }); + + it('queries with rolling 24h window from current time', async () => { + mockPrisma.activity.findMany.mockResolvedValue([]); + + await compute24hVolume(CREATOR_ID); + + const callArgs = mockPrisma.activity.findMany.mock.calls[0][0]; + const { gte, lte } = callArgs.where.createdAt; + + // gte should be 24 hours before NOW + expect(gte.getTime()).toBe(NOW.getTime() - 24 * 60 * 60 * 1000); + // lte should be NOW + expect(lte.getTime()).toBe(NOW.getTime()); + }); + + it('handles string prices by converting to BigInt', async () => { + const trades = [ + { + payload: { price: '12345' }, + }, + { + payload: { price: '67890' }, + }, + ]; + + mockPrisma.activity.findMany.mockResolvedValue(trades); + + const volume = await compute24hVolume(CREATOR_ID); + + expect(volume).toBe(80235n); + }); + + it('throws error when database query fails', async () => { + const error = new Error('Database connection failed'); + mockPrisma.activity.findMany.mockRejectedValue(error); + + await expect(compute24hVolume(CREATOR_ID)).rejects.toThrow( + 'Database connection failed' + ); + }); + + it('handles null payload gracefully', async () => { + const trades = [ + { + payload: null, + }, + { + payload: { price: '1000' }, + }, + ]; + + mockPrisma.activity.findMany.mockResolvedValue(trades); + + const volume = await compute24hVolume(CREATOR_ID); + + // Should only sum the valid trade + expect(volume).toBe(1000n); + }); + + it('handles empty payload object gracefully', async () => { + const trades = [ + { + payload: {}, + }, + { + payload: { price: '1000' }, + }, + ]; + + mockPrisma.activity.findMany.mockResolvedValue(trades); + + const volume = await compute24hVolume(CREATOR_ID); + + // Should only sum the valid trade + expect(volume).toBe(1000n); + }); +}); diff --git a/src/utils/rate-limit-tracking.utils.ts b/src/utils/rate-limit-tracking.utils.ts new file mode 100644 index 0000000..4573727 --- /dev/null +++ b/src/utils/rate-limit-tracking.utils.ts @@ -0,0 +1,114 @@ +import { logger } from './logger.utils'; + +/** + * Shared state for tracking rate limit thresholds per wallet per window. + * Key format: `${walletAddress}:${windowStart}` + * Value: { requestCount, thresholdsLogged: Set<'80%' | '100%'> } + */ +const rateLimitState = new Map< + string, + { requestCount: number; thresholdsLogged: Set<'80%' | '100%'> } +>(); + +/** + * Truncate a wallet address to first 4 and last 4 characters. + * + * @param address - The full wallet address + * @returns Truncated address in format: ABCD…WXYZ + */ +export function truncateWalletAddress(address: string): string { + if (address.length <= 8) { + return address; + } + return `${address.slice(0, 4)}…${address.slice(-4)}`; +} + +/** + * Emits warn logs when a wallet's request count crosses 80% or 100% of the rate limit. + * + * - Logs are emitted once per threshold per window + * - Wallet address is truncated in the log + * - Includes all required fields: wallet_address, request_count, limit, window_reset_at, threshold + * + * @param walletAddress - The wallet address making the request + * @param requestCount - Current request count within the window + * @param limit - The rate limit (max requests per window) + * @param windowResetAt - ISO timestamp when the current window resets + */ +export function checkRateLimitThresholds( + walletAddress: string, + requestCount: number, + limit: number, + windowResetAt: Date +): void { + if (!walletAddress || requestCount <= 0 || limit <= 0) { + return; + } + + // Create a key that includes the window reset time to allow fresh thresholds per window + const windowKey = `${walletAddress}:${windowResetAt.getTime()}`; + + // Get or initialize tracking state for this wallet+window + let state = rateLimitState.get(windowKey); + if (!state) { + state = { requestCount, thresholdsLogged: new Set() }; + rateLimitState.set(windowKey, state); + } + state.requestCount = requestCount; + + const percentage = (requestCount / limit) * 100; + const truncatedAddress = truncateWalletAddress(walletAddress); + + // Check 80% threshold + if (percentage >= 80 && !state.thresholdsLogged.has('80%')) { + state.thresholdsLogged.add('80%'); + logger.warn( + { + wallet_address: truncatedAddress, + request_count: requestCount, + limit, + window_reset_at: windowResetAt.toISOString(), + threshold: '80%', + }, + 'Wallet approaching rate limit (80% of limit)' + ); + } + + // Check 100% threshold + if (percentage >= 100 && !state.thresholdsLogged.has('100%')) { + state.thresholdsLogged.add('100%'); + logger.warn( + { + wallet_address: truncatedAddress, + request_count: requestCount, + limit, + window_reset_at: windowResetAt.toISOString(), + threshold: '100%', + }, + 'Wallet hit rate limit (100% of limit)' + ); + } +} + +/** + * Cleans up old tracking state to prevent memory leaks. + * Should be called periodically or when windows reset. + * + * @param olderThanTime - Remove entries older than this timestamp (ms) + */ +export function cleanupRateLimitState(olderThanTime: number): void { + for (const [key] of rateLimitState.entries()) { + // Key format: `${walletAddress}:${windowStart}` + const windowStart = Number(key.split(':').pop()); + if (windowStart < olderThanTime) { + rateLimitState.delete(key); + } + } +} + +/** + * Resets all tracking state (useful for testing). + */ +export function resetRateLimitState(): void { + rateLimitState.clear(); +} diff --git a/src/utils/trading-volume.utils.ts b/src/utils/trading-volume.utils.ts new file mode 100644 index 0000000..2737ca5 --- /dev/null +++ b/src/utils/trading-volume.utils.ts @@ -0,0 +1,56 @@ +import { prisma } from './prisma.utils'; +import { logger } from './logger.utils'; + +/** + * Computes the 24h trading volume for a creator by summing all trades + * (KEY_BOUGHT and KEY_SOLD) within the last 24 hours. + * + * The time window is rolling — based on current time minus 24 hours, not calendar day boundaries. + * + * @param creatorId - The ID of the creator + * @returns Promise resolving to the sum of prices (in stroops) for all trades within the window + */ +export async function compute24hVolume(creatorId: string): Promise { + try { + // Calculate the 24-hour cutoff timestamp + const now = new Date(); + const twentyFourHoursAgo = new Date(now.getTime() - 24 * 60 * 60 * 1000); + + // Query all trades (KEY_BOUGHT and KEY_SOLD) for the creator within the window + const trades = await prisma.activity.findMany({ + where: { + creatorId, + type: { in: ['KEY_BOUGHT', 'KEY_SOLD'] }, + createdAt: { + gte: twentyFourHoursAgo, + lte: now, + }, + }, + select: { + payload: true, + }, + }); + + // Sum the price field from each trade's payload + let totalVolume = 0n; + + for (const trade of trades) { + const payload = trade.payload as Record; + if (payload && typeof payload.price !== 'undefined') { + const price = BigInt(payload.price); + totalVolume += price; + } + } + + return totalVolume; + } catch (error) { + logger.error( + { + error: error instanceof Error ? error.message : String(error), + creatorId, + }, + 'Failed to compute 24h trading volume' + ); + throw error; + } +}