From a96a9ae97463417b0a01ec97fe40edfc9f6f052a Mon Sep 17 00:00:00 2001 From: Ekezie Uchechukwu Date: Sat, 7 Mar 2026 00:27:58 +0100 Subject: [PATCH] feat: implement LedgerObserverService for real-time stellar events --- backend/src/index.ts | 6 + .../__tests__/ledgerObserverService.test.ts | 170 ++++++++++++++ backend/src/services/ledgerObserverService.ts | 212 ++++++++++++++++++ 3 files changed, 388 insertions(+) create mode 100644 backend/src/services/__tests__/ledgerObserverService.test.ts create mode 100644 backend/src/services/ledgerObserverService.ts diff --git a/backend/src/index.ts b/backend/src/index.ts index ceb75c8d..077255e9 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -8,6 +8,7 @@ import v1Routes from './routes/v1'; import { initializeSocket, emitTransactionUpdate } from './services/socketService'; import { HealthController } from './controllers/healthController'; import { ThrottlingService } from './services/throttlingService'; +import { LedgerObserverService } from './services/ledgerObserverService'; const app = express(); const httpServer = createServer(app); @@ -97,6 +98,11 @@ const PORT = config.PORT || 3000; httpServer.listen(PORT, () => { console.log(`Server running on port ${PORT}`); console.log(`Environment: ${config.NODE_ENV}`); + + // Start the Ledger Observer Service to listen for Stellar events + LedgerObserverService.start().catch(err => { + console.error('Failed to start LedgerObserverService:', err); + }); }); export default app; diff --git a/backend/src/services/__tests__/ledgerObserverService.test.ts b/backend/src/services/__tests__/ledgerObserverService.test.ts new file mode 100644 index 00000000..d2a83b8c --- /dev/null +++ b/backend/src/services/__tests__/ledgerObserverService.test.ts @@ -0,0 +1,170 @@ +import { LedgerObserverService } from '../ledgerObserverService'; +import { StellarService } from '../stellarService'; +import { pool } from '../../config/database'; +import axios from 'axios'; + +jest.mock('@stellar/stellar-sdk', () => { + return { + ServerApi: {}, + Horizon: { + Server: jest.fn().mockImplementation(() => ({ + payments: jest.fn().mockReturnThis(), + cursor: jest.fn().mockReturnThis(), + stream: jest.fn() + })) + } + }; +}); + +jest.mock('axios'); +jest.mock('../../config/database', () => ({ + pool: { + query: jest.fn(), + }, +})); +jest.mock('../stellarService'); + +describe('LedgerObserverService', () => { + let originalConsoleLog: any; + let originalConsoleError: any; + + beforeAll(() => { + // Suppress console outputs for clean test runs + originalConsoleLog = console.log; + originalConsoleError = console.error; + console.log = jest.fn(); + console.error = jest.fn(); + }); + + afterAll(() => { + console.log = originalConsoleLog; + console.error = originalConsoleError; + }); + + beforeEach(() => { + jest.clearAllMocks(); + // Reset internal state + LedgerObserverService.stop(); + (LedgerObserverService as any).trackedAddresses.clear(); + }); + + describe('Address Tracking', () => { + it('should fetch and cache organization and employee addresses', async () => { + const mockOrgQuery = { rows: [{ id: 1, wallet_address: 'G_ORG_1' }] }; + const mockEmpQuery = { rows: [{ organization_id: 1, wallet_address: 'G_EMP_1' }] }; + + (pool.query as jest.Mock) + .mockResolvedValueOnce(mockOrgQuery) + .mockResolvedValueOnce(mockEmpQuery); + + await (LedgerObserverService as any).refreshTrackedAddresses(); + + const tracked = (LedgerObserverService as any).trackedAddresses; + expect(tracked.size).toBe(2); + expect(tracked.get('G_ORG_1')).toEqual({ address: 'G_ORG_1', organizationId: 1, type: 'organization' }); + expect(tracked.get('G_EMP_1')).toEqual({ address: 'G_EMP_1', organizationId: 1, type: 'employee' }); + }); + }); + + describe('Event Handling', () => { + beforeEach(() => { + // Setup some tracked addresses directly for isolated testing + const tracked = new Map(); + tracked.set('G_KNOWN_ORG', { address: 'G_KNOWN_ORG', organizationId: 99, type: 'organization' }); + (LedgerObserverService as any).trackedAddresses = tracked; + + (pool.query as jest.Mock).mockResolvedValue({ + rows: [{ config_value: JSON.stringify({ webhook_url: 'https://webhook.site/test' }) }] + }); + (axios.post as jest.Mock).mockResolvedValue({ status: 200 }); + }); + + it('should trigger webhook when payment "to" a tracked address is processed', async () => { + const mockPayment = { + id: '1234', + transaction_hash: 'hash1', + type: 'payment', + asset_type: 'native', + amount: '100.0', + from: 'G_UNKNOWN', + to: 'G_KNOWN_ORG' + }; + + await (LedgerObserverService as any).handlePaymentEvent(mockPayment); + + expect(pool.query).toHaveBeenCalledWith( + expect.stringContaining('SELECT config_value FROM tenant_configurations'), + [99] + ); + expect(axios.post).toHaveBeenCalledTimes(1); + expect(axios.post).toHaveBeenCalledWith( + 'https://webhook.site/test', + expect.objectContaining({ + event_type: 'stellar_payment', + address: 'G_KNOWN_ORG', + amount: '100.0' + }), + expect.any(Object) + ); + }); + + it('should not trigger webhook if addresses are not tracked', async () => { + const mockPayment = { + id: '1234', + transaction_hash: 'hash1', + type: 'payment', + from: 'G_UNKNOWN_1', + to: 'G_UNKNOWN_2' + }; + + await (LedgerObserverService as any).handlePaymentEvent(mockPayment); + + expect(axios.post).not.toHaveBeenCalled(); + }); + }); + + describe('Webhook Dispatch Retry', () => { + beforeEach(() => { + jest.useFakeTimers(); + (pool.query as jest.Mock).mockResolvedValue({ + rows: [{ config_value: JSON.stringify({ webhook_url: 'https://webhook.site/fail' }) }] + }); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('should retry failed webhooks up to MAX_RETRIES', async () => { + (axios.post as jest.Mock).mockRejectedValue(new Error('Network Error')); + + // Initial trigger (synchronous error handled, starts timer for retry 1) + const promise = (LedgerObserverService as any).dispatchWebhook(99, { event: 'test' }); + await promise; + expect(axios.post).toHaveBeenCalledTimes(1); + + // Retry 1: Wait 1s + jest.advanceTimersByTime(1000); + await Promise.resolve(); // Allow the promise to execute + await Promise.resolve(); // Allow the catch block to schedule next timer + expect(axios.post).toHaveBeenCalledTimes(2); + + // Retry 2: Wait 2s + jest.advanceTimersByTime(2000); + await Promise.resolve(); + await Promise.resolve(); + expect(axios.post).toHaveBeenCalledTimes(3); + + // Retry 3: Wait 4s + jest.advanceTimersByTime(4000); + await Promise.resolve(); + await Promise.resolve(); + expect(axios.post).toHaveBeenCalledTimes(4); + + // Max retries reached, should not schedule another + jest.advanceTimersByTime(8000); + await Promise.resolve(); + expect(axios.post).toHaveBeenCalledTimes(4); + }); + }); +}); diff --git a/backend/src/services/ledgerObserverService.ts b/backend/src/services/ledgerObserverService.ts new file mode 100644 index 00000000..0d9697ca --- /dev/null +++ b/backend/src/services/ledgerObserverService.ts @@ -0,0 +1,212 @@ +import { StellarService } from './stellarService'; +import { pool } from '../config/database'; +import axios from 'axios'; +import { ServerApi } from '@stellar/stellar-sdk/lib/horizon'; + +interface TrackedAddress { + address: string; + organizationId: number; + type: 'organization' | 'employee'; +} + +export class LedgerObserverService { + private static isRunning = false; + private static closeStream: (() => void) | null = null; + private static trackedAddresses: Map = new Map(); + private static refreshInterval: NodeJS.Timeout | null = null; + private static readonly MAX_WEBHOOK_RETRIES = 3; + + static async start() { + if (this.isRunning) { + console.log('LedgerObserverService is already running.'); + return; + } + + console.log('Starting LedgerObserverService...'); + this.isRunning = true; + + // Fetch initial addresses + await this.refreshTrackedAddresses(); + + // Set up periodic refresh (every 5 minutes) + this.refreshInterval = setInterval(async () => { + await this.refreshTrackedAddresses(); + }, 5 * 60 * 1000); + + // Start listening to the network + this.startStream(); + } + + static stop() { + console.log('Stopping LedgerObserverService...'); + if (this.closeStream) { + this.closeStream(); + this.closeStream = null; + } + if (this.refreshInterval) { + clearInterval(this.refreshInterval); + this.refreshInterval = null; + } + this.isRunning = false; + } + + private static async refreshTrackedAddresses() { + try { + const newTrackedAddresses = new Map(); + + // 1. Fetch Organization Addresses (Tenant Configs or Organizations table if multi-tenant) + // Assuming an organizations table exists with stellar_address/wallet_address + const orgQuery = `SELECT id, wallet_address FROM organizations WHERE wallet_address IS NOT NULL`; + const orgResult = await pool.query(orgQuery); + orgResult.rows.forEach(row => { + newTrackedAddresses.set(row.wallet_address, { + address: row.wallet_address, + organizationId: row.id, + type: 'organization' + }); + }); + + // 2. Fetch Employee Addresses + const empQuery = `SELECT organization_id, wallet_address FROM employees WHERE wallet_address IS NOT NULL AND status = 'active'`; + const empResult = await pool.query(empQuery); + empResult.rows.forEach(row => { + newTrackedAddresses.set(row.wallet_address, { + address: row.wallet_address, + organizationId: row.organization_id, + type: 'employee' + }); + }); + + this.trackedAddresses = newTrackedAddresses; + console.log(`[LedgerObserver] Tracked ${this.trackedAddresses.size} addresses.`); + } catch (error) { + console.error('[LedgerObserver] Failed to refresh tracked addresses:', error); + } + } + + private static startStream() { + const server = StellarService.getServer(); + + try { + this.closeStream = server.payments() + .cursor('now') + .stream({ + onmessage: (record) => { + // Type assertion since stream returns a generic record that we know is an operation + const payment = record as unknown as ServerApi.PaymentOperationRecord; + this.handlePaymentEvent(payment); + }, + onerror: (error) => { + console.error('[LedgerObserver] Stream error:', error); + // Implement backoff or simple restart in production + } + }); + } catch (error) { + console.error('[LedgerObserver] Error starting stream:', error); + this.isRunning = false; + } + } + + private static async handlePaymentEvent(payment: ServerApi.PaymentOperationRecord) { + try { + // Check if 'to' or 'from' is in our tracked addresses + const involvedAddresses = new Set(); + + if (payment.source_account) involvedAddresses.add(payment.source_account); + if ('to' in payment) involvedAddresses.add((payment as any).to); + if ('from' in payment) involvedAddresses.add((payment as any).from); + if ('funder' in payment) involvedAddresses.add((payment as any).funder); + if ('account' in payment) involvedAddresses.add((payment as any).account); + if ('trustor' in payment) involvedAddresses.add((payment as any).trustor); + + for (const addr of involvedAddresses) { + const tracked = this.trackedAddresses.get(addr); + if (tracked) { + console.log(`[LedgerObserver] Relevant event detected for Org ${tracked.organizationId}: ${payment.type} (Tx: ${payment.transaction_hash})`); + + // Construct payload + const payload = { + event_type: 'stellar_payment', + timestamp: new Date().toISOString(), + organization_id: tracked.organizationId, + address_type: tracked.type, + address: addr, + operation_id: payment.id, + transaction_hash: payment.transaction_hash, + type: payment.type, + asset: 'asset_type' in payment ? (payment as any).asset_type : 'native', + amount: 'amount' in payment ? (payment as any).amount : null, + from: 'from' in payment ? (payment as any).from : (payment as any).source_account, + to: 'to' in payment ? (payment as any).to : null + }; + + await this.dispatchWebhook(tracked.organizationId, payload); + + // If one event matches multiple rules, we might want to just notify once per org. + break; + } + } + } catch (error) { + console.error('[LedgerObserver] Error processing event:', error); + } + } + + private static async dispatchWebhook(organizationId: number, payload: unknown, retryCount = 0) { + try { + // Fetch webhook URL directly from DB if tenant configurations exist or fallback to process.env + let webhookUrl = process.env.DEFAULT_WEBHOOK_URL; + + try { + const query = `SELECT config_value FROM tenant_configurations WHERE organization_id = $1 AND config_key = 'notification_settings'`; + const result = await pool.query(query, [organizationId]); + if (result.rows.length > 0 && result.rows[0].config_value) { + const settings = typeof result.rows[0].config_value === 'string' + ? JSON.parse(result.rows[0].config_value) + : result.rows[0].config_value; + if (settings.webhook_url) { + webhookUrl = settings.webhook_url; + } + } + } catch (dbErr) { + // Table might not exist yet, fallback is already set + console.log(`[LedgerObserver] Could not fetch tenant webhook configs, using default if available.`); + } + + if (!webhookUrl) { + // No webhook configured for this organization + return; + } + + console.log(`[LedgerObserver] Dispatching webhook to Org ${organizationId} at ${webhookUrl}`); + + await axios.post(webhookUrl, payload, { + headers: { + 'Content-Type': 'application/json', + 'User-Agent': 'PayD-Ledger-Observer/1.0', + 'X-PayD-Event': (payload as any).event_type + }, + timeout: 5000 // 5 second timeout + }); + + console.log(`[LedgerObserver] Webhook delivered successfully to Org ${organizationId}`); + + } catch (error) { + const errMessage = error instanceof Error ? error.message : String(error); + console.error(`[LedgerObserver] Webhook delivery failed for Org ${organizationId}: ${errMessage}`); + + if (retryCount < this.MAX_WEBHOOK_RETRIES) { + const delay = Math.pow(2, retryCount) * 1000; // Exponential backoff: 1s, 2s, 4s... + console.log(`[LedgerObserver] Retrying webhook in ${delay}ms (Attempt ${retryCount + 1}/${this.MAX_WEBHOOK_RETRIES})`); + + setTimeout(() => { + this.dispatchWebhook(organizationId, payload, retryCount + 1); + }, delay); + } else { + console.error(`[LedgerObserver] Webhook max retries reached for Org ${organizationId}`); + // In a true enterprise system, we might log this to a dead-letter queue or DB table here + } + } + } +} + +export default LedgerObserverService;