diff --git a/CONTRIBUTOR_SETUP.md b/CONTRIBUTOR_SETUP.md new file mode 100644 index 00000000..89523078 --- /dev/null +++ b/CONTRIBUTOR_SETUP.md @@ -0,0 +1,627 @@ +# Contributor Environment Setup Guide + +This guide walks you through setting up a local development environment for NotifyChain. By the end, you will have the listener service, the dashboard, and the smart contracts building and running on your machine. + +--- + +## Table of Contents + +1. [Required Dependencies](#1-required-dependencies) +2. [Clone the Repository](#2-clone-the-repository) +3. [Listener Service Setup](#3-listener-service-setup) +4. [Dashboard Setup](#4-dashboard-setup) +5. [Smart Contracts Setup](#5-smart-contracts-setup) +6. [Environment Variables Reference](#6-environment-variables-reference) +7. [Running Tests](#7-running-tests) +8. [VS Code Setup (Recommended)](#8-vs-code-setup-recommended) +9. [Troubleshooting & FAQ](#9-troubleshooting--faq) + +--- + +## 1. Required Dependencies + +### Essential Toolchain + +| Dependency | Minimum Version | Install Method | Used By | +|----------------|-----------------|-----------------------------------------|--------------------| +| Rust | stable | [rustup.rs](https://rustup.rs) | Smart contracts | +| `wasm32-unknown-unknown` | — | `rustup target add wasm32-unknown-unknown` | Soroban contracts | +| Stellar CLI | latest | `cargo install stellar-cli` | Contract build/deploy | +| Node.js | **18** (dashboard), **20** (listener) | [nodejs.org](https://nodejs.org) or `nvm` | Listener, Dashboard | +| npm | comes with Node | — | Package management | +| Git | — | Your package manager or [git-scm.com](https://git-scm.com) | Version control | + +### Platform Notes + +- **macOS**: Install Xcode Command Line Tools (`xcode-select --install`) before Rust. +- **Linux**: Install `build-essential` (`sudo apt install build-essential`) before Rust. +- **Windows**: Use [Visual Studio Build Tools](https://visualstudio.microsoft.com/downloads/#build-tools-for-visual-studio-2022) with "C++ build tools" workload for native `sqlite3` bindings. + +### Node.js Version Management + +The project uses **two different Node.js versions** across its components: + +| Workspace | Node Version | CI Reference | +|-------------|-------------|--------------| +| `listener/` | **20** | `.github/workflows/ci.yml` — `node-version: 20` | +| `dashboard/` | **18** | `.github/workflows/ci.yml` — `node-version: 18` | + +**Recommendation**: Use [nvm](https://github.com/nvm-sh/nvm) (Node Version Manager) to switch between versions: + +```bash +# Install nvm (macOS/Linux) +curl -o- https://raw.githubusercontent.com/nvm-sh/nvm/v0.40.3/install.sh | bash + +# Restart your terminal, then install both versions +nvm install 18 +nvm install 20 + +# Use Node 20 for the listener (the more demanding component) +nvm use 20 +``` + +For most local development, **Node 20 is sufficient** for both components. If you encounter CI-related issues, test against the specific version used in CI. + +### Quick Install Commands + +```bash +# ── Rust + WebAssembly + Stellar CLI ── +curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh +source "$HOME/.cargo/env" +rustup target add wasm32-unknown-unknown +cargo install --locked stellar-cli --features opt + +# Verify +rustc --version && cargo --version && stellar --version +``` + +--- + +## 2. Clone the Repository + +```bash +git clone https://github.com/Core-Foundry/Notify-Chain.git +cd Notify-Chain +``` + +If you plan to contribute, fork the repository first, then clone your fork: + +```bash +git clone https://github.com/YOUR-USERNAME/Notify-Chain.git +cd Notify-Chain +git remote add upstream https://github.com/Core-Foundry/Notify-Chain.git +``` + +--- + +## 3. Listener Service Setup + +The listener is the core off-chain service that polls the Stellar network, processes contract events, and delivers notifications. + +### 3.1 Install Dependencies + +```bash +cd listener +npm install +``` + +> **Note**: The `sqlite3` package includes native bindings. If the install fails, see the [Troubleshooting](#9-troubleshooting--faq) section. + +### 3.2 Configure Environment + +```bash +cp .env.example .env +``` + +Open `listener/.env` in your editor. At minimum, set: + +```bash +STELLAR_RPC_URL=https://soroban-testnet.stellar.org:443 +CONTRACT_ADDRESSES=[{"address":"YOUR_CONTRACT_ID","events":["*"]}] +``` + +For a full reference of every variable, see [Environment Variables Reference](#6-environment-variables-reference). + +### 3.3 Initialize the Database + +```bash +npm run migrate +``` + +This creates the SQLite database file (default location: `listener/data/notifications.db`) and runs all schema migrations. + +### 3.4 Run the Listener + +```bash +npm run dev +``` + +The listener starts and immediately begins polling the configured contracts. You should see log output showing poll cycles. + +**Verify it's running:** + +```bash +curl http://localhost:8787/health # Health check +curl http://localhost:8787/api/events # Event feed (may be empty initially) +``` + +### 3.5 Useful Listener Commands + +| Command | Purpose | +|-------------------------|--------------------------------------------------| +| `npm run dev` | Start in development mode (ts-node, hot reload) | +| `npm run build` | Compile TypeScript to JavaScript | +| `npm start` | Run the compiled production build | +| `npm test` | Run all tests | +| `npm run typecheck` | TypeScript type checking (no emit) | +| `npm run migrate` | Initialize or update the SQLite database schema | +| `npm run lint` | Alias for `typecheck` | + +--- + +## 4. Dashboard Setup + +The dashboard is a React + Vite application that visualizes events from the listener's API. + +### 4.1 Install Dependencies + +```bash +cd dashboard +npm install +``` + +### 4.2 Configure Environment + +```bash +cp .env.example .env +``` + +The default `.env.example` is pre-configured for local development: + +```bash +VITE_EVENTS_API_URL=http://localhost:8787/api/events +VITE_STELLAR_NETWORK=TESTNET +``` + +Change `VITE_EVENTS_API_URL` if your listener runs on a different port. + +### 4.3 Run the Dashboard + +```bash +npm run dev +``` + +Open [http://localhost:5173](http://localhost:5173) in your browser. The dashboard fetches events from the listener API and displays them in real-time. + +### 4.4 Useful Dashboard Commands + +| Command | Purpose | +|------------------|--------------------------------------------------| +| `npm run dev` | Start Vite dev server with hot module replacement | +| `npm run build` | TypeScript check + Vite production build | +| `npm test` | Run all tests | +| `npm run lint` | ESLint with zero-tolerance for warnings | +| `npm run preview` | Preview the production build locally | +| `npm run benchmark` | Run performance rendering benchmarks | + +--- + +## 5. Smart Contracts Setup + +The project contains two Soroban smart contracts. Building them is optional — you only need to do this if you are modifying contracts or deploying your own instances. + +### 5.1 Build Contracts + +**AutoShare Contract** (`contract/`): + +```bash +cd contract +stellar contract build +``` + +The compiled `.wasm` file is written to `contract/target/wasm32-unknown-unknown/release/`. + +**TaskBounty Contract** (`Documents/Task Bounty/`): + +```bash +cd Documents/Task\ Bounty +stellar contract build +``` + +### 5.2 Run Contract Tests + +```bash +# AutoShare +cd contract/contracts/hello-world +cargo test + +# TaskBounty +cd Documents/Task\ Bounty +cargo test +``` + +### 5.3 Deploy to Testnet (Optional) + +This requires a funded Stellar testnet identity: + +```bash +# Generate and fund a test identity +stellar keys generate my-identity --network testnet +stellar keys fund my-identity --network testnet + +# Deploy the AutoShare contract +cd contract +stellar contract deploy \ + --wasm target/wasm32-unknown-unknown/release/hello_world.wasm \ + --source my-identity \ + --network testnet + +# The contract ID is printed after successful deployment +``` + +### 5.4 Useful Contract Commands + +| Command | Purpose | +|--------------------------------------------------|-----------------------------| +| `stellar contract build` | Build all contracts | +| `cargo test` | Run contract tests | +| `stellar contract deploy --wasm --source --network testnet` | Deploy to testnet | +| `stellar contract invoke --id --source --network testnet -- [ARGS]` | Call a contract function | +| `stellar contract optimize --wasm ` | Optimize Wasm for production | +| `stellar contract inspect --wasm ` | Inspect contract interface | +| `cargo fmt --all` | Format Rust code | + +--- + +## 6. Environment Variables Reference + +### 6.1 Listener (`listener/.env`) + +| Variable | Required | Default | Description | +|----------|----------|---------|-------------| +| **Stellar Network** | | | | +| `STELLAR_NETWORK` | No | `testnet` | Network passphrase (`testnet`, `pubnet`, or custom) | +| `STELLAR_RPC_URL` | **Yes** | `https://soroban-testnet.stellar.org:443` | Soroban RPC endpoint to poll for events | +| **Contracts** | | | | +| `CONTRACT_ADDRESSES` | **Yes** | `[]` | JSON array of contract configs. Format: `[{"address":"C...","events":["*"]}]`. Each entry has `address` (string, the Stellar contract ID) and `events` (string array of event names to filter on, or `["*"]` for all). | +| **Polling** | | | | +| `POLL_INTERVAL_MS` | No | `30000` | Time between RPC polls (milliseconds) | +| `MAX_RECONNECT_ATTEMPTS` | No | `5` | Max consecutive RPC failures before stopping | +| `RECONNECT_DELAY_MS` | No | `5000` | Base delay between reconnection attempts (exponential backoff) | +| **API Server** | | | | +| `EVENTS_API_PORT` | No | `8787` | HTTP server port for `/health` and `/api/events` | +| `EVENTS_API_CORS_ORIGIN` | No | `http://localhost:5173` | Allowed CORS origin (dashboard URL) | +| `WEBHOOK_SECRETS` | No | `[]` | JSON array of webhook secrets for HMAC verification. Format: `[{"id":"default","secret":"whsec_..."}]` | +| **Discord Notifications** | | | | +| `DISCORD_WEBHOOK_URL` | No | — | Discord webhook URL for sending notifications | +| `DISCORD_WEBHOOK_ID` | No | — | Discord webhook ID (required if webhook URL is set) | +| **Retry Queue** | | | | +| `RETRY_BASE_DELAY_MS` | No | `5000` | Base delay for notification retry exponential backoff | +| `RETRY_MAX_RETRIES` | No | `5` | Max retry attempts for failed Discord notifications | +| **Event Processing Queue** | | | | +| `EVENT_QUEUE_MAX_CONCURRENCY` | No | `1` | Max events to process concurrently (1 = ordered) | +| `EVENT_QUEUE_MAX_RETRIES` | No | `3` | Max retry attempts per event before permanent failure | +| `EVENT_QUEUE_BASE_DELAY_MS` | No | `2000` | Base delay for event retry exponential backoff | +| `EVENT_QUEUE_POLL_INTERVAL_MS` | No | `1000` | How often the queue checks for due events (ms) | +| **Database** | | | | +| `DATABASE_PATH` | No | `./data/notifications.db` | SQLite database file path | +| **Scheduler** | | | | +| `SCHEDULER_ENABLED` | No | `true` | Enable the notification scheduler | +| `SCHEDULER_POLL_INTERVAL_MS` | No | `10000` | How often the scheduler polls for due notifications | +| `SCHEDULER_LOCK_TIMEOUT_MS` | No | `60000` | Distributed lock timeout for scheduler | +| `SCHEDULER_PROCESSOR_ID` | No | auto-generated | Unique ID for this scheduler instance (multi-instance setups) | +| `SCHEDULER_BATCH_SIZE` | No | `10` | Max notifications to process per poll cycle | +| `SCHEDULER_TIMING_BUFFER_MS` | No | `60000` | Buffer to prevent premature notification delivery | +| **Rate Limiting** | | | | +| `RATE_LIMIT_ENABLED` | No | `true` | Enable HTTP API rate limiting | +| `RATE_LIMIT_WINDOW_MS` | No | `60000` | Rate limit window (milliseconds) | +| `RATE_LIMIT_MAX_REQUESTS` | No | `60` | Max requests per window per client | +| `RATE_LIMIT_CLIENT_OVERRIDES` | No | `{}` | JSON object of per-client rate limit overrides | +| **Cleanup** | | | | +| `CLEANUP_INTERVAL_MS` | No | `3600000` | How often to run cleanup jobs (1 hour) | +| `NOTIFICATION_RETENTION_MS` | No | `604800000` | Retain completed notifications (7 days) | +| `RATE_LIMIT_EVENT_RETENTION_MS` | No | `86400000` | Retain rate limit audit events (1 day) | +| `EVENT_RETENTION_MS` | No | `86400000` | Retain in-memory events (1 day) | +| **Logging** | | | | +| `LOG_LEVEL` | No | `info` | Winston log level (`error`, `warn`, `info`, `debug`) | +| `NODE_ENV` | No | — | Set to `production` for newline-delimited JSON log output | + +### 6.2 Dashboard (`dashboard/.env`) + +| Variable | Required | Default | Description | +|----------|----------|---------|-------------| +| `VITE_EVENTS_API_URL` | No | `http://localhost:8787/api/events` | Listener API endpoint for fetching events | +| `VITE_STELLAR_NETWORK` | No | `TESTNET` | Stellar network for wallet integration (`TESTNET` or `MAINNET`) | + +### 6.3 Minimum Viable Configuration + +To run the listener with no optional features: + +```bash +STELLAR_RPC_URL=https://soroban-testnet.stellar.org:443 +CONTRACT_ADDRESSES=[{"address":"C...","events":["*"]}] +EVENTS_API_PORT=8787 +EVENTS_API_CORS_ORIGIN=http://localhost:5173 +``` + +To add Discord notifications, also set: + +```bash +DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/YOUR_WEBHOOK_ID/YOUR_TOKEN +``` + +--- + +## 7. Running Tests + +Always run tests before submitting a pull request. + +### Contracts + +```bash +# AutoShare +cd contract/contracts/hello-world && cargo test + +# TaskBounty +cd Documents/Task\ Bounty && cargo test +``` + +### Listener + +```bash +cd listener +npm test +``` + +### Dashboard + +```bash +cd dashboard +npm test +``` + +### CI Validation (What GitHub Actions Runs) + +The CI pipeline executes these checks on every pull request: + +```bash +# Listener +cd listener +npm run typecheck # TypeScript strict type checking +npm test # Jest test suite + +# Dashboard +cd dashboard +npm run lint # ESLint (zero warnings) +npm run build # TypeScript check + Vite build +npm test # Jest test suite + +# Contracts +cd contract +cargo fmt --all -- --check # Rust formatting check +cargo test --workspace --all-features --verbose +``` + +--- + +## 8. VS Code Setup (Recommended) + +### Extensions + +Install these VS Code extensions: + +1. [rust-analyzer](https://marketplace.visualstudio.com/items?itemName=rust-lang.rust-analyzer) — Rust language support +2. [CodeLLDB](https://marketplace.visualstudio.com/items?itemName=vadimcn.vscode-lldb) — Debugger for Rust +3. [Better TOML](https://marketplace.visualstudio.com/items?itemName=bungcip.better-toml) — TOML file support +4. [ESLint](https://marketplace.visualstudio.com/items?itemName=dbaeumer.vscode-eslint) — TypeScript linting +5. [Prettier](https://marketplace.visualstudio.com/items?itemName=esbenp.prettier-vscode) — Code formatter + +### Settings + +Add this to `.vscode/settings.json` (already present in the repository): + +```json +{ + "rust-analyzer.cargo.target": "wasm32-unknown-unknown", + "rust-analyzer.checkOnSave.allTargets": false +} +``` + +This configures `rust-analyzer` to use the `wasm32` target and avoids false-positive type errors from non-Wasm platform checks. + +--- + +## 9. Troubleshooting & FAQ + +> A more extensive troubleshooting guide is available at [`TROUBLESHOOTING.md`](TROUBLESHOOTING.md). This section covers the most common setup issues. + +### Node.js & npm + +**Q: `npm install` fails with `node-gyp` or `sqlite3` errors.** + +Native `sqlite3` bindings must be compiled for your platform and Node.js version. + +```bash +# Rebuild native bindings +npm rebuild sqlite3 + +# If that fails, reinstall +npm uninstall sqlite3 && npm install +``` + +On Windows, ensure you have Visual Studio Build Tools installed with the "C++ build tools" workload. + +--- + +**Q: Which Node.js version should I use?** + +The listener CI runs on Node 20, the dashboard CI runs on Node 18. Use **Node 20** for local development (it is forward-compatible with the dashboard). Use `nvm` to switch if needed. + +--- + +### Stellar RPC & Network + +**Q: The listener starts but no events appear.** + +1. Verify the contract ID in `CONTRACT_ADDRESSES` is correct and deployed on the same network as `STELLAR_RPC_URL`. +2. Confirm the RPC endpoint is reachable: + ```bash + curl -X POST https://soroban-testnet.stellar.org:443 -H "Content-Type: application/json" -d '{"jsonrpc":"2.0","id":1,"method":"getHealth"}' + ``` +3. Check listener logs for poll cycle output: + ```bash + cd listener && npm run dev + ``` + Look for lines containing `"Received events"` or `"Processing event"`. +4. Verify the API is responding: + ```bash + curl http://localhost:8787/health + curl http://localhost:8787/api/events + ``` + +--- + +**Q: `stellar: command not found`** + +```bash +# Ensure cargo's bin directory is in your PATH +source "$HOME/.cargo/env" + +# Or reinstall +cargo install --locked stellar-cli --features opt +``` + +--- + +### Database + +**Q: `Error: Database not initialized` or `SQLITE_ERROR: no such table`** + +```bash +cd listener +npm run migrate +``` + +If the data directory is missing: + +```bash +mkdir -p listener/data +npm run migrate +``` + +--- + +### Port Conflicts + +**Q: `EADDRINUSE: address already in use :::8787`** + +```bash +# Find the process using the port (macOS/Linux) +lsof -i :8787 +kill -9 + +# Or change the port in listener/.env +EVENTS_API_PORT=8788 +``` + +--- + +### Dashboard + +**Q: Dashboard shows a blank page or "Failed to fetch"** + +1. Is the listener running? (`npm run dev` in `listener/`) +2. Does `VITE_EVENTS_API_URL` in `dashboard/.env` match the listener's port? +3. Restart the Vite dev server after editing `.env`. + +--- + +### Contracts + +**Q: `error: toolchain 'stable' does not support target 'wasm32-unknown-unknown'`** + +```bash +rustup target add wasm32-unknown-unknown +``` + +--- + +**Q: `error[E0463]: can't find crate for 'std'`** + +Build with the correct target: + +```bash +cargo build --target wasm32-unknown-unknown --release +# Or use Stellar CLI (handles the target automatically): +stellar contract build +``` + +--- + +### After a `git pull` + +If things stop working after pulling latest changes: + +```bash +# Contracts +cd contract && stellar contract build + +# Listener +cd listener && npm install && npm run migrate + +# Dashboard +cd dashboard && npm install +``` + +--- + +### Still Stuck? + +1. Search [open issues](https://github.com/Core-Foundry/Notify-Chain/issues) — your problem may already be reported. +2. Read the detailed [Troubleshooting Guide](TROUBLESHOOTING.md). +3. Open a new issue with: + - Your OS and version + - Output of `rustc --version`, `node --version`, `stellar --version` + - The full error message and stack trace + - Steps you have already tried + +--- + +## Project Map + +``` +Notify-Chain/ +├── contract/ # Soroban smart contract workspace +│ ├── contracts/hello-world/ # AutoShare contract (active) +│ └── Cargo.toml # Workspace configuration +│ +├── listener/ # Off-chain listener service +│ ├── src/ +│ │ ├── api/ # HTTP API (events, health, templates) +│ │ ├── services/ # Core: subscriber, dedup, notifier, scheduler +│ │ ├── store/ # In-memory event registry + preferences +│ │ ├── database/ # SQLite schema and client +│ │ ├── types/ # TypeScript type definitions +│ │ ├── utils/ # Logging, formatting, helpers +│ │ └── index.ts # Entry point +│ └── src/__tests__/ # Integration / E2E tests +│ +├── dashboard/ # React + Vite event dashboard +│ └── src/ +│ ├── components/ # UI components (EventCard, filters, etc.) +│ ├── services/ # API client +│ ├── store/ # Zustand state management +│ └── pages/ # Page components +│ +├── Documents/Task Bounty/ # TaskBounty contract (Soroban) +│ +├── frontend/ # Legacy Next.js frontend (not actively maintained) +│ +├── scripts/ # Helper scripts (health check, etc.) +│ +├── CONTRIBUTOR_SETUP.md # This file +├── CONTRIBUTING.md # Contribution guidelines & PR workflow +├── TROUBLESHOOTING.md # Detailed troubleshooting reference +├── ARCHITECTURE_OVERVIEW.md # High-level architecture walkthrough +└── README.md # Project overview and quick start +``` diff --git a/listener/.env.example b/listener/.env.example index f671c874..4656519b 100644 --- a/listener/.env.example +++ b/listener/.env.example @@ -37,6 +37,12 @@ SCHEDULER_PROCESSOR_ID= SCHEDULER_BATCH_SIZE=10 SCHEDULER_TIMING_BUFFER_MS=60000 +# Event Processing Queue Configuration +# EVENT_QUEUE_MAX_CONCURRENCY=1 # Max events to process concurrently +# EVENT_QUEUE_MAX_RETRIES=3 # Max retries per event before permanent failure +# EVENT_QUEUE_BASE_DELAY_MS=2000 # Base delay for exponential backoff (ms) +# EVENT_QUEUE_POLL_INTERVAL_MS=1000 # How often to check for due events (ms) + # Rate Limiting Configuration RATE_LIMIT_ENABLED=true RATE_LIMIT_WINDOW_MS=60000 diff --git a/listener/package-lock.json b/listener/package-lock.json index 03890399..3b1d7031 100644 --- a/listener/package-lock.json +++ b/listener/package-lock.json @@ -63,6 +63,7 @@ "integrity": "sha512-RgHBCvtjbOK2gXSNBNIkNoEc9qoVEtau3hj8gEqKQuL3HZAibKarWFEI3Lfm6EYKkLalOh8eSrj9b+ch9H/VBA==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@babel/code-frame": "^7.29.7", "@babel/generator": "^7.29.7", @@ -1408,6 +1409,7 @@ "dev": true, "hasInstallScript": true, "license": "Apache-2.0", + "peer": true, "dependencies": { "@swc/counter": "^0.1.3", "@swc/types": "^0.1.26" @@ -1825,6 +1827,7 @@ "integrity": "sha512-603BddQMv3pUcr4U2dhujk83N2tTDVr/34wII2B6bJy6g+8WD6yUb11jszNs0gdi4PesVWl7ABt8nYMVpnLUcg==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "undici-types": ">=7.24.0 <7.24.7" } @@ -1938,6 +1941,7 @@ "integrity": "sha512-tbsV1jPne5CkFQCgPBcDOt30ItF7aJoZL997JSF7MhGQqOeT3svWRYxiqlfA5RUdlHN6Fi+EI9bxqbdyAUZjYQ==", "dev": true, "license": "BSD-2-Clause", + "peer": true, "dependencies": { "@typescript-eslint/scope-manager": "6.21.0", "@typescript-eslint/types": "6.21.0", @@ -2166,6 +2170,7 @@ "integrity": "sha512-xRQbDb9BnwDafYNn6Vwl839DYVjqXYb1XVGtWAZ1kcDc6iwAL4hg3B1dZlRiuENFeO2H53gFG3in621AdERVAg==", "dev": true, "license": "MIT", + "peer": true, "bin": { "acorn": "bin/acorn" }, @@ -2678,6 +2683,7 @@ } ], "license": "MIT", + "peer": true, "dependencies": { "baseline-browser-mapping": "^2.10.12", "caniuse-lite": "^1.0.30001782", @@ -3486,6 +3492,7 @@ "deprecated": "This version is no longer supported. Please see https://eslint.org/version-support for other options.", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@eslint-community/eslint-utils": "^4.2.0", "@eslint-community/regexpp": "^4.6.1", @@ -4833,6 +4840,7 @@ "integrity": "sha512-NIy3oAFp9shda19hy4HK0HRTWKtPJmGdnvywu01nOqNC2vZg+Z+fvJDxpMQA88eb2I9EcafcdjYgsDthnYTvGw==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@jest/core": "^29.7.0", "@jest/types": "^29.6.3", @@ -7393,6 +7401,7 @@ "integrity": "sha512-f0FFpIdcHgn8zcPSbf1dRevwt047YMnaiJM3u2w2RewrB+fob/zePZcrOyQoLMMO7aBIddLcQIEK5dYjkLnGrQ==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@cspotcode/source-map-support": "^0.8.0", "@tsconfig/node10": "^1.0.7", @@ -7499,6 +7508,7 @@ "integrity": "sha512-y2TvuxSZPDyQakkFRPZHKFm+KKVqIisdg9/CZwm9ftvKXLP8NRWj38/ODjNbr43SsoXqNuAisEf1GdCxqWcdBw==", "dev": true, "license": "Apache-2.0", + "peer": true, "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" diff --git a/listener/src/config.ts b/listener/src/config.ts index 6c361e4b..42fb83a2 100644 --- a/listener/src/config.ts +++ b/listener/src/config.ts @@ -1,4 +1,4 @@ -import { Config, ContractConfig, DiscordConfig, WebhookSecret, AppCleanupConfig } from './types'; +import { Config, ContractConfig, DiscordConfig, WebhookSecret, AppCleanupConfig, EventQueueConfig } from './types'; export class ConfigError extends Error { constructor(message: string) { @@ -146,6 +146,12 @@ export function loadConfig(): Config { baseDelayMs: parseIntegerEnv('RETRY_BASE_DELAY_MS', '5000'), maxRetries: parseIntegerEnv('RETRY_MAX_RETRIES', '5'), }, + eventQueue: { + maxConcurrency: parseIntegerEnv('EVENT_QUEUE_MAX_CONCURRENCY', '1'), + maxRetries: parseIntegerEnv('EVENT_QUEUE_MAX_RETRIES', '3'), + baseDelayMs: parseIntegerEnv('EVENT_QUEUE_BASE_DELAY_MS', '2000'), + pollIntervalMs: parseIntegerEnv('EVENT_QUEUE_POLL_INTERVAL_MS', '1000'), + }, webhookSecrets: validateWebhookSecrets(rawWebhookSecrets), scheduler: { enabled: trimEnv('SCHEDULER_ENABLED') !== 'false', diff --git a/listener/src/services/event-processing-queue.test.ts b/listener/src/services/event-processing-queue.test.ts new file mode 100644 index 00000000..811f55de --- /dev/null +++ b/listener/src/services/event-processing-queue.test.ts @@ -0,0 +1,579 @@ +import { xdr } from '@stellar/stellar-sdk'; +import * as StellarSDK from '@stellar/stellar-sdk'; +import { + EventProcessingQueue, + EventProcessor, +} from './event-processing-queue'; + +jest.mock('../utils/logger', () => ({ + __esModule: true, + default: { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + }, +})); + +function createMockEvent( + overrides: Partial = {} +): StellarSDK.rpc.Api.EventResponse { + return { + id: 'event-123', + type: 'contract', + ledger: 1000, + ledgerClosedAt: '2026-01-01T00:00:00Z', + transactionIndex: 1, + operationIndex: 0, + inSuccessfulContractCall: true, + txHash: 'abc123', + topic: [xdr.ScVal.scvSymbol('test_event')], + value: xdr.ScVal.scvString('test value'), + ...overrides, + }; +} + +const mockContractConfig = { address: 'CA123', events: ['test_event'] }; + +describe('EventProcessingQueue', () => { + beforeEach(() => { + jest.useFakeTimers(); + jest.clearAllMocks(); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + describe('enqueue', () => { + it('adds an event to the queue', () => { + const processor: EventProcessor = jest.fn(); + const queue = new EventProcessingQueue(processor, { baseDelayMs: 1000 }); + + queue.enqueue(createMockEvent(), mockContractConfig); + + expect(queue.size()).toBe(1); + expect(queue.pendingCount()).toBe(1); + }); + + it('returns true when an event is queued', () => { + const processor: EventProcessor = jest.fn(); + const queue = new EventProcessingQueue(processor, { baseDelayMs: 1000 }); + + const result = queue.enqueue(createMockEvent(), mockContractConfig); + + expect(result).toBe(true); + }); + + it('logs when an event is queued', () => { + const logger = jest.requireMock('../utils/logger').default; + const processor: EventProcessor = jest.fn(); + const queue = new EventProcessingQueue(processor, { baseDelayMs: 1000 }); + + queue.enqueue(createMockEvent({ id: 'evt-q' }), mockContractConfig, 'req-1'); + + expect(logger.info).toHaveBeenCalledWith( + 'Event queued for processing', + expect.objectContaining({ eventId: 'evt-q', requestId: 'req-1' }) + ); + }); + + it('skips duplicate events with the same event id and contract address', () => { + const logger = jest.requireMock('../utils/logger').default; + const processor: EventProcessor = jest.fn(); + const queue = new EventProcessingQueue(processor, { baseDelayMs: 1000 }); + const event = createMockEvent({ id: 'evt-dup' }); + + const firstResult = queue.enqueue(event, mockContractConfig, 'req-1'); + const secondResult = queue.enqueue(event, mockContractConfig, 'req-2'); + + expect(queue.size()).toBe(1); + expect(firstResult).toBe(true); + expect(secondResult).toBe(false); + expect(logger.info).toHaveBeenCalledWith( + 'Skipping duplicate event queue entry', + expect.objectContaining({ + eventId: 'evt-dup', + contractAddress: mockContractConfig.address, + }) + ); + }); + + it('allows the same event id from different contract addresses', () => { + const processor: EventProcessor = jest.fn(); + const queue = new EventProcessingQueue(processor, { baseDelayMs: 1000 }); + const event = createMockEvent({ id: 'evt-same' }); + const otherConfig = { address: 'CB456', events: ['test_event'] }; + + queue.enqueue(event, mockContractConfig); + queue.enqueue(event, otherConfig); + + expect(queue.size()).toBe(2); + }); + + it('allows different event ids from the same contract address', () => { + const processor: EventProcessor = jest.fn(); + const queue = new EventProcessingQueue(processor, { baseDelayMs: 1000 }); + + queue.enqueue(createMockEvent({ id: 'evt-1' }), mockContractConfig); + queue.enqueue(createMockEvent({ id: 'evt-2' }), mockContractConfig); + + expect(queue.size()).toBe(2); + }); + + it('returns false for a duplicate and does not add to the queue', () => { + const processor: EventProcessor = jest.fn(); + const queue = new EventProcessingQueue(processor, { baseDelayMs: 1000 }); + + queue.enqueue(createMockEvent({ id: 'evt-only' }), mockContractConfig); + const result = queue.enqueue( + createMockEvent({ id: 'evt-only' }), + mockContractConfig + ); + + expect(result).toBe(false); + expect(queue.size()).toBe(1); + }); + }); + + describe('processing', () => { + it('processes a queued event after the base delay', async () => { + const processor: EventProcessor = jest.fn().mockResolvedValue(true); + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 1000, + pollIntervalMs: 100, + }); + queue.start(); + + queue.enqueue(createMockEvent(), mockContractConfig); + + // Before delay expires — should not have processed yet + jest.advanceTimersByTime(500); + await Promise.resolve(); + expect(processor).not.toHaveBeenCalled(); + + // After delay expires — should process + jest.advanceTimersByTime(600); + await Promise.resolve(); + await Promise.resolve(); + expect(processor).toHaveBeenCalledTimes(1); + + queue.stop(); + }); + + it('calls the processor with the correct arguments', async () => { + const processor: EventProcessor = jest.fn().mockResolvedValue(true); + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 100, + pollIntervalMs: 50, + }); + queue.start(); + + const event = createMockEvent({ id: 'evt-args' }); + queue.enqueue(event, mockContractConfig, 'req-args'); + + jest.advanceTimersByTime(200); + await Promise.resolve(); + await Promise.resolve(); + + expect(processor).toHaveBeenCalledWith( + expect.objectContaining({ id: 'evt-args' }), + mockContractConfig, + 'req-args' + ); + + queue.stop(); + }); + + it('removes the event from the queue on success', async () => { + const processor: EventProcessor = jest.fn().mockResolvedValue(true); + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 100, + pollIntervalMs: 50, + }); + queue.start(); + + queue.enqueue(createMockEvent(), mockContractConfig); + + jest.advanceTimersByTime(200); + await Promise.resolve(); + await Promise.resolve(); + + expect(queue.size()).toBe(0); + + queue.stop(); + }); + + it('logs success on a successful processing', async () => { + const logger = jest.requireMock('../utils/logger').default; + const processor: EventProcessor = jest.fn().mockResolvedValue(true); + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 100, + pollIntervalMs: 50, + }); + queue.start(); + + queue.enqueue(createMockEvent({ id: 'evt-ok' }), mockContractConfig, 'req-ok'); + + jest.advanceTimersByTime(200); + await Promise.resolve(); + await Promise.resolve(); + + expect(logger.info).toHaveBeenCalledWith( + 'Event processing succeeded', + expect.objectContaining({ eventId: 'evt-ok' }) + ); + + queue.stop(); + }); + + it('processes events in enqueue order (FIFO)', async () => { + const callOrder: string[] = []; + const processor: EventProcessor = jest.fn().mockImplementation(async (event) => { + callOrder.push(event.id); + return true; + }); + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 50, + pollIntervalMs: 50, + }); + queue.start(); + + queue.enqueue(createMockEvent({ id: 'first' }), mockContractConfig); + queue.enqueue(createMockEvent({ id: 'second' }), mockContractConfig); + + // First cycle: process 'first' (available=1, only one at a time) + jest.advanceTimersByTime(100); + await Promise.resolve(); + await Promise.resolve(); + + // Second cycle: process 'second' + jest.advanceTimersByTime(100); + await Promise.resolve(); + await Promise.resolve(); + + expect(callOrder).toEqual(['first', 'second']); + + queue.stop(); + }); + }); + + describe('exponential backoff', () => { + it('doubles the delay on each successive failure', async () => { + const processor: EventProcessor = jest.fn().mockResolvedValue(false); + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 1000, + maxRetries: 5, + pollIntervalMs: 100, + }); + queue.start(); + + queue.enqueue(createMockEvent(), mockContractConfig); + + // Trigger attempt 1 (after 1000 ms base delay) + jest.advanceTimersByTime(1100); + await Promise.resolve(); + await Promise.resolve(); + expect(processor).toHaveBeenCalledTimes(1); + + // Trigger attempt 2 (after 2000 ms from attempt 1) + jest.advanceTimersByTime(2100); + await Promise.resolve(); + await Promise.resolve(); + expect(processor).toHaveBeenCalledTimes(2); + + queue.stop(); + }); + + it('logs a warning with the next retry delay on failure', async () => { + const logger = jest.requireMock('../utils/logger').default; + const processor: EventProcessor = jest.fn().mockResolvedValue(false); + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 1000, + maxRetries: 3, + pollIntervalMs: 100, + }); + queue.start(); + + queue.enqueue(createMockEvent({ id: 'evt-backoff' }), mockContractConfig); + + jest.advanceTimersByTime(1100); + await Promise.resolve(); + await Promise.resolve(); + + expect(logger.warn).toHaveBeenCalledWith( + 'Event processing failed, scheduling retry', + expect.objectContaining({ eventId: 'evt-backoff', attempt: 1, delayMs: 2000 }) + ); + + queue.stop(); + }); + }); + + describe('max retries', () => { + it('stops retrying after maxRetries attempts', async () => { + const processor: EventProcessor = jest.fn().mockResolvedValue(false); + const maxRetries = 3; + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 100, + maxRetries, + pollIntervalMs: 50, + }); + queue.start(); + queue.enqueue(createMockEvent(), mockContractConfig); + + const flush = async () => { + for (let i = 0; i < 5; i++) await Promise.resolve(); + }; + + // attempt 1 fires at t=100ms (base delay) + jest.advanceTimersByTime(100); + await flush(); + expect(processor).toHaveBeenCalledTimes(1); + + // attempt 2 fires at t=300ms (100 + 100*2^1 = 300) + jest.advanceTimersByTime(200); + await flush(); + expect(processor).toHaveBeenCalledTimes(2); + + // attempt 3 fires at t=700ms (300 + 100*2^2 = 700) + jest.advanceTimersByTime(400); + await flush(); + expect(processor).toHaveBeenCalledTimes(maxRetries); + expect(queue.size()).toBe(0); + + queue.stop(); + }); + + it('logs an error when the event permanently fails', async () => { + const logger = jest.requireMock('../utils/logger').default; + const processor: EventProcessor = jest.fn().mockResolvedValue(false); + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 100, + maxRetries: 1, + pollIntervalMs: 50, + }); + queue.start(); + + queue.enqueue(createMockEvent({ id: 'evt-dead' }), mockContractConfig, 'req-dead'); + + jest.advanceTimersByTime(200); + await Promise.resolve(); + await Promise.resolve(); + + expect(logger.error).toHaveBeenCalledWith( + 'Event processing permanently failed after max retries', + expect.objectContaining({ eventId: 'evt-dead', totalAttempts: 1 }) + ); + + queue.stop(); + }); + }); + + describe('error handling', () => { + it('retries when the processor throws an error', async () => { + const processor: EventProcessor = jest + .fn() + .mockRejectedValueOnce(new Error('Unexpected error')) + .mockResolvedValueOnce(true); + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 100, + maxRetries: 3, + pollIntervalMs: 50, + }); + queue.start(); + + queue.enqueue(createMockEvent({ id: 'evt-err' }), mockContractConfig); + + jest.advanceTimersByTime(200); + await Promise.resolve(); + await Promise.resolve(); + expect(processor).toHaveBeenCalledTimes(1); + expect(queue.size()).toBe(1); // still in queue for retry + + jest.advanceTimersByTime(400); + await Promise.resolve(); + await Promise.resolve(); + expect(processor).toHaveBeenCalledTimes(2); + expect(queue.size()).toBe(0); // succeeded on retry + + queue.stop(); + }); + + it('logs error on processor crash and schedules retry', async () => { + const logger = jest.requireMock('../utils/logger').default; + const processor: EventProcessor = jest + .fn() + .mockRejectedValue(new Error('Crash')); + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 100, + maxRetries: 2, + pollIntervalMs: 50, + }); + queue.start(); + + queue.enqueue(createMockEvent({ id: 'evt-crash' }), mockContractConfig); + + jest.advanceTimersByTime(200); + await Promise.resolve(); + await Promise.resolve(); + + expect(logger.error).toHaveBeenCalledWith( + 'Event processing crashed, scheduling retry', + expect.objectContaining({ + eventId: 'evt-crash', + attempt: 1, + delayMs: 200, + }) + ); + + queue.stop(); + }); + + it('clears fingerprint on permanent failure after processor crash', async () => { + const processor: EventProcessor = jest + .fn() + .mockRejectedValue(new Error('Permanent crash')); + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 100, + maxRetries: 1, + pollIntervalMs: 50, + }); + queue.start(); + + queue.enqueue(createMockEvent({ id: 'evt-perm' }), mockContractConfig); + + jest.advanceTimersByTime(200); + await Promise.resolve(); + await Promise.resolve(); + + // After permanent failure, re-enqueueing the same event should work + const reenqueueResult = queue.enqueue( + createMockEvent({ id: 'evt-perm' }), + mockContractConfig + ); + expect(reenqueueResult).toBe(true); + + queue.stop(); + }); + }); + + describe('concurrency', () => { + it('respects maxConcurrency = 1 (default)', async () => { + let concurrent = 0; + let maxObserved = 0; + const processor: EventProcessor = jest.fn().mockImplementation(async () => { + concurrent++; + maxObserved = Math.max(maxObserved, concurrent); + await new Promise((resolve) => setTimeout(resolve, 500)); + concurrent--; + return true; + }); + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 50, + pollIntervalMs: 50, + }); + queue.start(); + + queue.enqueue(createMockEvent({ id: 'evt-con-1' }), mockContractConfig); + queue.enqueue(createMockEvent({ id: 'evt-con-2' }), mockContractConfig); + + // Let first item start processing + jest.advanceTimersByTime(200); + await Promise.resolve(); + await Promise.resolve(); + + // At this point, only one item should be in flight + expect(processor).toHaveBeenCalledTimes(1); + expect(maxObserved).toBe(1); + + queue.stop(); + }); + + it('processes up to maxConcurrency items simultaneously', async () => { + let concurrent = 0; + let maxObserved = 0; + const processor: EventProcessor = jest.fn().mockImplementation(async () => { + concurrent++; + maxObserved = Math.max(maxObserved, concurrent); + await new Promise((resolve) => setTimeout(resolve, 500)); + concurrent--; + return true; + }); + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 50, + maxConcurrency: 3, + pollIntervalMs: 50, + }); + queue.start(); + + queue.enqueue(createMockEvent({ id: 'evt-1' }), mockContractConfig); + queue.enqueue(createMockEvent({ id: 'evt-2' }), mockContractConfig); + queue.enqueue(createMockEvent({ id: 'evt-3' }), mockContractConfig); + + // Let the cycle start processing + jest.advanceTimersByTime(200); + await Promise.resolve(); + await Promise.resolve(); + + expect(maxObserved).toBe(3); + + queue.stop(); + }); + }); + + describe('start / stop', () => { + it('does not process items when stopped', async () => { + const processor: EventProcessor = jest.fn().mockResolvedValue(true); + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 100, + pollIntervalMs: 50, + }); + + queue.enqueue(createMockEvent(), mockContractConfig); + // Never call queue.start() + + jest.advanceTimersByTime(1000); + await Promise.resolve(); + + expect(processor).not.toHaveBeenCalled(); + }); + + it('calling start twice does not double-process items', async () => { + const processor: EventProcessor = jest.fn().mockResolvedValue(true); + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 100, + pollIntervalMs: 50, + }); + queue.start(); + queue.start(); // second call should be a no-op + + queue.enqueue(createMockEvent(), mockContractConfig); + + jest.advanceTimersByTime(200); + await Promise.resolve(); + await Promise.resolve(); + + expect(processor).toHaveBeenCalledTimes(1); + + queue.stop(); + }); + + it('stops processing items after stop is called', async () => { + const processor: EventProcessor = jest.fn().mockResolvedValue(true); + const queue = new EventProcessingQueue(processor, { + baseDelayMs: 100, + pollIntervalMs: 50, + }); + queue.start(); + + queue.enqueue(createMockEvent({ id: 'evt-stop' }), mockContractConfig); + + queue.stop(); + + jest.advanceTimersByTime(1000); + await Promise.resolve(); + + expect(processor).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/listener/src/services/event-processing-queue.ts b/listener/src/services/event-processing-queue.ts new file mode 100644 index 00000000..b5936082 --- /dev/null +++ b/listener/src/services/event-processing-queue.ts @@ -0,0 +1,239 @@ +import * as StellarSDK from '@stellar/stellar-sdk'; +import { ContractConfig } from '../types'; +import logger from '../utils/logger'; + +export interface EventProcessingQueueOptions { + maxConcurrency?: number; + pollIntervalMs?: number; + maxRetries?: number; + baseDelayMs?: number; +} + +export type EventProcessor = ( + event: StellarSDK.rpc.Api.EventResponse, + contractConfig: ContractConfig, + requestId?: string +) => Promise; + +interface QueuedEvent { + event: StellarSDK.rpc.Api.EventResponse; + contractConfig: ContractConfig; + requestId: string; + retryCount: number; + nextRetryAt: number; + fingerprint: string; +} + +const DEFAULTS = { + maxConcurrency: 1, + pollIntervalMs: 1_000, + maxRetries: 3, + baseDelayMs: 2_000, +}; + +export class EventProcessingQueue { + private queue: QueuedEvent[] = []; + private readonly queuedFingerprints: Set = new Set(); + private readonly activeFingerprints: Set = new Set(); + private readonly maxConcurrency: number; + private readonly pollIntervalMs: number; + private readonly maxRetries: number; + private readonly baseDelayMs: number; + private readonly processor: EventProcessor; + private timer: ReturnType | null = null; + + constructor(processor: EventProcessor, options?: EventProcessingQueueOptions) { + this.processor = processor; + this.maxConcurrency = Math.max(1, options?.maxConcurrency ?? DEFAULTS.maxConcurrency); + this.pollIntervalMs = options?.pollIntervalMs ?? DEFAULTS.pollIntervalMs; + this.maxRetries = options?.maxRetries ?? DEFAULTS.maxRetries; + this.baseDelayMs = options?.baseDelayMs ?? DEFAULTS.baseDelayMs; + } + + enqueue( + event: StellarSDK.rpc.Api.EventResponse, + contractConfig: ContractConfig, + requestId?: string + ): boolean { + const fingerprint = buildEventFingerprint(event, contractConfig.address); + + if (this.queuedFingerprints.has(fingerprint)) { + logger.info('Skipping duplicate event queue entry', { + requestId, + eventId: event.id, + contractAddress: contractConfig.address, + fingerprint, + }); + return false; + } + + const delayMs = this.calculateDelay(0); + const nextRetryAt = Date.now() + delayMs; + + logger.info('Event queued for processing', { + requestId, + eventId: event.id, + contractAddress: contractConfig.address, + delayMs, + nextRetryAt: new Date(nextRetryAt).toISOString(), + maxRetries: this.maxRetries, + }); + + this.queuedFingerprints.add(fingerprint); + this.queue.push({ + event, + contractConfig, + requestId: requestId ?? '', + retryCount: 0, + nextRetryAt, + fingerprint, + }); + + return true; + } + + start(): void { + if (this.timer !== null) return; + this.timer = setInterval(() => { + this.processNext().catch((err) => + logger.error('Unexpected error in event processing queue', { error: err }) + ); + }, this.pollIntervalMs); + } + + stop(): void { + if (this.timer !== null) { + clearInterval(this.timer); + this.timer = null; + } + } + + size(): number { + return this.queue.length; + } + + pendingCount(): number { + return this.queue.length; + } + + private async processNext(): Promise { + const available = this.maxConcurrency - this.activeFingerprints.size; + if (available <= 0) return; + + const now = Date.now(); + + const due = this.queue + .filter((item) => item.nextRetryAt <= now && !this.activeFingerprints.has(item.fingerprint)) + .slice(0, available); + + if (due.length === 0) return; + + const selectedFingerprints = new Set(due.map((item) => item.fingerprint)); + + this.queue = this.queue.filter( + (item) => + item.nextRetryAt > now || + this.activeFingerprints.has(item.fingerprint) || + !selectedFingerprints.has(item.fingerprint) + ); + + const results = await Promise.allSettled(due.map((item) => this.processItem(item))); + + for (const result of results) { + if (result.status === 'rejected') { + logger.error('Unexpected rejection in event processing queue', { + error: result.reason, + }); + } + } + } + + private async processItem(item: QueuedEvent): Promise { + this.activeFingerprints.add(item.fingerprint); + + try { + const success = await this.processor(item.event, item.contractConfig, item.requestId); + + if (success) { + this.queuedFingerprints.delete(item.fingerprint); + this.activeFingerprints.delete(item.fingerprint); + logger.info('Event processing succeeded', { + requestId: item.requestId, + eventId: item.event.id, + contractAddress: item.contractConfig.address, + }); + return; + } + + const attempt = item.retryCount + 1; + + if (attempt >= this.maxRetries) { + this.queuedFingerprints.delete(item.fingerprint); + this.activeFingerprints.delete(item.fingerprint); + logger.error('Event processing permanently failed after max retries', { + requestId: item.requestId, + eventId: item.event.id, + contractAddress: item.contractConfig.address, + totalAttempts: attempt, + }); + return; + } + + const delayMs = this.calculateDelay(attempt); + const nextRetryAt = Date.now() + delayMs; + + logger.warn('Event processing failed, scheduling retry', { + requestId: item.requestId, + eventId: item.event.id, + contractAddress: item.contractConfig.address, + attempt, + delayMs, + nextRetryAt: new Date(nextRetryAt).toISOString(), + }); + + this.activeFingerprints.delete(item.fingerprint); + this.queue.push({ ...item, retryCount: attempt, nextRetryAt }); + } catch (error) { + this.activeFingerprints.delete(item.fingerprint); + + const attempt = item.retryCount + 1; + + if (attempt >= this.maxRetries) { + this.queuedFingerprints.delete(item.fingerprint); + logger.error('Event processing crashed after max retries', { + requestId: item.requestId, + eventId: item.event.id, + contractAddress: item.contractConfig.address, + totalAttempts: attempt, + error, + }); + return; + } + + const delayMs = this.calculateDelay(attempt); + const nextRetryAt = Date.now() + delayMs; + + logger.error('Event processing crashed, scheduling retry', { + requestId: item.requestId, + eventId: item.event.id, + contractAddress: item.contractConfig.address, + attempt, + delayMs, + error, + }); + + this.queue.push({ ...item, retryCount: attempt, nextRetryAt }); + } + } + + private calculateDelay(retryCount: number): number { + return this.baseDelayMs * Math.pow(2, retryCount); + } +} + +function buildEventFingerprint( + event: StellarSDK.rpc.Api.EventResponse, + contractAddress: string +): string { + return `${contractAddress}:${event.id}`; +} diff --git a/listener/src/services/event-subscriber.ts b/listener/src/services/event-subscriber.ts index 40b6acc5..9569ace6 100644 --- a/listener/src/services/event-subscriber.ts +++ b/listener/src/services/event-subscriber.ts @@ -12,6 +12,7 @@ import { import { DiscordNotificationService } from './discord-notification'; import { NotificationRetryQueue } from './notification-retry-queue'; import { EventDeduplicationService } from './event-deduplication-service'; +import { EventProcessingQueue } from './event-processing-queue'; export class EventSubscriber { private config: Config; @@ -22,6 +23,7 @@ export class EventSubscriber { private discordService: DiscordNotificationService | null = null; private retryQueue: NotificationRetryQueue | null = null; private deduplicationService: EventDeduplicationService | null = null; + private eventQueue: EventProcessingQueue | null = null; constructor(config: Config, deduplicationService?: EventDeduplicationService) { this.config = config; @@ -35,6 +37,13 @@ export class EventSubscriber { config.retryQueue ); } + if (config.eventQueue) { + this.eventQueue = new EventProcessingQueue( + (event, contractConfig, requestId) => + this.processEvent(event, contractConfig, requestId), + config.eventQueue + ); + } } async start(): Promise { @@ -45,12 +54,14 @@ export class EventSubscriber { this.isRunning = true; logger.info('Starting event subscriber service'); + this.eventQueue?.start(); this.retryQueue?.start(); this.poll(); } async stop(): Promise { this.isRunning = false; + this.eventQueue?.stop(); this.retryQueue?.stop(); logger.info('Stopping event subscriber service'); } @@ -122,7 +133,11 @@ export class EventSubscriber { } for (const event of processableEvents) { - await this.processEvent(event, contractConfig, requestId); + if (this.eventQueue) { + this.eventQueue.enqueue(event, contractConfig, requestId); + } else { + await this.processEvent(event, contractConfig, requestId); + } } if (response.cursor) { @@ -212,7 +227,7 @@ export class EventSubscriber { event: StellarSDK.rpc.Api.EventResponse, contractConfig: ContractConfig, requestId: string = '' - ): Promise { + ): Promise { const eventStart = Date.now(); const eventName = getEventName(event.topic); @@ -238,7 +253,7 @@ export class EventSubscriber { 'SKIPPED' ); - return; + return true; } } @@ -322,6 +337,11 @@ export class EventSubscriber { notificationSent, durationMs: Date.now() - eventStart, }); + + if (!this.discordService) return true; + if (notificationSent) return true; + if (processingError && this.retryQueue) return true; + return false; } private async handleReconnection(requestId?: string): Promise { diff --git a/listener/src/types/index.ts b/listener/src/types/index.ts index d9d29b6e..24f7dda1 100644 --- a/listener/src/types/index.ts +++ b/listener/src/types/index.ts @@ -41,6 +41,7 @@ export interface Config { eventsApiCorsOrigin: string; discord?: DiscordConfig; retryQueue?: RetryQueueConfig; + eventQueue?: EventQueueConfig; webhookSecrets?: WebhookSecret[]; scheduler?: SchedulerConfig; databasePath?: string; @@ -57,6 +58,17 @@ export interface SchedulerConfig { timingBufferMs: number; } +export interface EventQueueConfig { + /** Maximum number of events to process concurrently (default: 1, must be >= 1). */ + maxConcurrency?: number; + /** Maximum retry attempts per event (default: 3). */ + maxRetries?: number; + /** Base delay in ms for exponential backoff (default: 2000). */ + baseDelayMs?: number; + /** How often to poll the queue for due events in ms (default: 1000). */ + pollIntervalMs?: number; +} + export interface AppCleanupConfig { /** How often to run cleanup jobs (ms). */ intervalMs: number;