Fix/sr observability aml notification gaps

backend/docs/SR-108-job-correlation-id.md

@@ -0,0 +1,46 @@
+# Job correlation ID propagation (SR-108 follow-up)
+
+## Problem
+
+`tracedJob()` in `src/tracing/job-tracer.ts` generated a per-run
+`correlationId` and attached it only as an OpenTelemetry span attribute
+(`correlation_id`). It never called `correlationStorage.run(correlationId,
+fn)` from `src/correlation-id.ts`, so every job function passed to
+`tracedJob` — `revalidate-stale-assets`, `poll-kyc-statuses`,
+`poll-sep24-transactions`, `extend-contract-storage-ttl`,
+`notify-kyc-expiries`, `check-anchor-health`, `retire-webhook-secrets` —
+logged via `createLogger(...).info/warn/error`, which reads the
+correlation ID from `correlationStorage`. Since that storage was never
+populated for the job's execution, every structured log line emitted
+during a scheduled job showed `correlationId: undefined`, while the span
+for that same run carried a different, disconnected `correlation_id`
+attribute. The two could never be cross-referenced.
+
+Additionally, three AML jobs (`aml-periodic-rescreening`,
+`aml-travel-rule-transmit`, `aml-data-retention`) were not wrapped in
+`tracedJob` at all, so they had neither a span nor a correlation ID.
+
+## What changed
+
+- `tracedJob()` now runs the job body inside
+  `correlationStorage.run(correlationId, () => fn())`, so every log line
+  emitted anywhere inside the job (directly, or by any function it calls)
+  picks up the same correlation ID that is on the span.
+- `src/scheduler.ts`: the three AML jobs are now wrapped in `tracedJob`,
+  matching every other scheduled job in the file.
+- Added `src/__tests__/job-tracer-correlation-id.test.ts`, which mocks
+  `@opentelemetry/api` to capture the span attributes passed to
+  `startActiveSpan` and asserts:
+  1. `getCorrelationId()` called from inside a `tracedJob`-wrapped
+     function returns the same ID recorded on the span.
+  2. A structured log line (`createLogger(...).info(...)`) emitted inside
+     the job body has a `correlationId` field equal to the span's
+     `correlation_id` attribute.
+
+## Why this matters operationally
+
+Before this change, an on-call engineer looking at a failed job's trace in
+the OTel backend had no way to pull the matching application logs — the
+`correlation_id` on the span didn't appear anywhere in the log stream.
+After this change, filtering logs by the span's `correlation_id` attribute
+returns exactly the log lines produced by that job run.

backend/migrations/sar_reference_counters.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS sar_reference_counters;

backend/migrations/sar_reference_counters.sql

@@ -0,0 +1,31 @@
+-- SR-112 follow-up: atomic SAR reference numbering.
+--
+-- SarWorkflowService.nextReference() previously derived the next sequence
+-- number from `SELECT COUNT(*) FROM sar_reports WHERE reference LIKE ...`
+-- with no locking. Two concurrent createFromAlerts() calls within the same
+-- calendar year could read the same count before either INSERT committed,
+-- producing a unique-constraint violation on sar_reports.reference for one
+-- of the two officers.
+--
+-- This table gives each calendar year a single counter row. Reference
+-- assignment becomes one atomic `INSERT ... ON CONFLICT DO UPDATE RETURNING`
+-- statement — Postgres serializes concurrent upserts against the same row,
+-- so no two callers can ever be handed the same sequence number.
+
+CREATE TABLE IF NOT EXISTS sar_reference_counters (
+  year          INTEGER PRIMARY KEY,
+  last_sequence INTEGER NOT NULL DEFAULT 0,
+  updated_at    TIMESTAMP NOT NULL DEFAULT NOW()
+);
+
+-- Seed the counter for years that already have SAR reports, so numbering
+-- continues from the highest existing sequence rather than restarting at 1.
+INSERT INTO sar_reference_counters (year, last_sequence)
+SELECT
+  split_part(reference, '-', 2)::int AS year,
+  MAX(split_part(reference, '-', 3)::int) AS last_sequence
+FROM sar_reports
+WHERE reference ~ '^SAR-\d{4}-\d{4}$'
+GROUP BY split_part(reference, '-', 2)::int
+ON CONFLICT (year) DO UPDATE
+  SET last_sequence = GREATEST(sar_reference_counters.last_sequence, EXCLUDED.last_sequence);

backend/src/__tests__/job-tracer-correlation-id.test.ts

@@ -0,0 +1,63 @@
+import { describe, it, expect, vi } from 'vitest';
+
+// Capture the attributes passed to startActiveSpan so we can compare the
+// span's correlation_id attribute against the correlation ID observed by a
+// logger running inside the traced job body.
+const spanAttributes: Record<string, unknown>[] = [];
+
+vi.mock('@opentelemetry/api', () => {
+  const fakeSpan = {
+    setStatus: vi.fn(),
+    recordException: vi.fn(),
+    end: vi.fn(),
+  };
+  return {
+    trace: {
+      getTracer: () => ({
+        startActiveSpan: (_name: string, opts: { attributes: Record<string, unknown> }, fn: (span: unknown) => unknown) => {
+          spanAttributes.push(opts.attributes);
+          return fn(fakeSpan);
+        },
+      }),
+    },
+    context: {},
+    SpanStatusCode: { OK: 1, ERROR: 2 },
+  };
+});
+
+describe('tracedJob correlation ID propagation', () => {
+  it('makes the span correlation_id available to getCorrelationId() inside the job body', async () => {
+    const { tracedJob } = await import('../tracing/job-tracer');
+    const { getCorrelationId } = await import('../correlation-id');
+
+    let observedDuringRun: string | undefined;
+
+    await tracedJob('test-job', async () => {
+      observedDuringRun = getCorrelationId();
+    });
+
+    const recordedCorrelationId = spanAttributes[spanAttributes.length - 1]['correlation_id'] as string;
+
+    expect(observedDuringRun).toBeDefined();
+    expect(observedDuringRun).toBe(recordedCorrelationId);
+  });
+
+  it('produces a structured log line whose correlationId matches the span attribute', async () => {
+    const { tracedJob } = await import('../tracing/job-tracer');
+    const { createLogger } = await import('../correlation-id');
+
+    const logSpy = vi.spyOn(console, 'log').mockImplementation(() => {});
+
+    await tracedJob('test-job-with-log', async () => {
+      const logger = createLogger('test');
+      logger.info('doing work');
+    });
+
+    const logged = JSON.parse(logSpy.mock.calls[logSpy.mock.calls.length - 1][0] as string);
+    const recordedCorrelationId = spanAttributes[spanAttributes.length - 1]['correlation_id'] as string;
+
+    expect(logged.correlationId).toBe(recordedCorrelationId);
+
+    logSpy.mockRestore();
+  });
+});

backend/src/__tests__/remittance-aml-wiring.test.ts

@@ -0,0 +1,168 @@
+/**
+ * SR-112 wiring test — POST /api/remittance must run transaction monitoring
+ * and travel-rule assessment inline, without requiring a compliance officer
+ * to separately call /api/aml/monitoring/evaluate or /api/aml/travel-rule.
+ */
+
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+import request from 'supertest';
+
+const { mockPool, insertedAlerts, insertedTransfers } = vi.hoisted(() => {
+  const insertedAlerts: any[] = [];
+  const insertedTransfers: any[] = [];
+
+  const mockPool = {
+    query: vi.fn(async (sql: string, params: any[] = []) => {
+      const s = sql.replace(/\s+/g, ' ').toUpperCase();
+
+      if (s.includes('INSERT INTO TRANSACTIONS')) {
+        return {
+          rows: [{ transaction_id: params[0], anchor_id: params[1], amount_in: params[2] }],
+          rowCount: 1,
+        };
+      }
+
+      // Transaction-monitoring rule set: a single VELOCITY_COUNT rule with
+      // max_count 0 so any transfer trips it deterministically.
+      if (s.includes('FROM AML_MONITORING_RULES')) {
+        return {
+          rows: [
+            {
+              code: 'VELOCITY_COUNT',
+              name: 'Velocity by count',
+              severity: 'high',
+              enabled: true,
+              params: { lookback_hours: 24, max_count: 0 },
+            },
+          ],
+          rowCount: 1,
+        };
+      }
+
+      // No prior transfer history / known corridors / reporting threshold.
+      if (s.includes('SENDER_ADDRESS = $1') && s.includes('FROM TRANSACTIONS')) {
+        return { rows: [], rowCount: 0 };
+      }
+      if (s.includes('DISTINCT CORRIDOR')) {
+        return { rows: [], rowCount: 0 };
+      }
+      if (s.includes('FROM COMPLIANCE_THRESHOLDS')) {
+        return { rows: [], rowCount: 0 };
+      }
+
+      // Alert insert (raiseAlert) — used by both monitoring and travel rule.
+      if (s.includes('INSERT INTO AML_ALERTS')) {
+        const row = { id: insertedAlerts.length + 1, rule_code: params[0], dedupe_key: params[7] };
+        insertedAlerts.push(row);
+        return { rows: [row], rowCount: 1 };
+      }
+
+      // Travel rule: no configured threshold rows -> fail-safe "required".
+      if (s.includes('FROM TRAVEL_RULE_THRESHOLDS')) {
+        return { rows: [], rowCount: 0 };
+      }
+      if (s.includes('INSERT INTO TRAVEL_RULE_TRANSFERS')) {
+        const row = { id: insertedTransfers.length + 1, transaction_id: params[0] };
+        insertedTransfers.push(row);
+        return { rows: [row], rowCount: 1 };
+      }
+
+      return { rows: [], rowCount: 0 };
+    }),
+  };
+
+  return { mockPool, insertedAlerts, insertedTransfers };
+});
+
+vi.mock('../database', () => ({
+  getPool: () => mockPool,
+  getAssetVerification: vi.fn(),
+  saveAssetVerification: vi.fn(),
+  reportSuspiciousAsset: vi.fn(),
+  getVerifiedAssets: vi.fn(),
+  saveFxRate: vi.fn(),
+  getFxRate: vi.fn(),
+  saveAnchorKycConfig: vi.fn(),
+  getUserKycStatus: vi.fn(),
+  saveUserKycStatus: vi.fn(),
+  saveAssetReport: vi.fn(),
+  getActiveWebhookSubscribers: vi.fn().mockResolvedValue([]),
+  getPendingWebhookDeliveries: vi.fn().mockResolvedValue([]),
+  saveContractEvent: vi.fn(),
+  queryContractEvents: vi.fn().mockResolvedValue({ events: [], total: 0 }),
+}));
+
+vi.mock('../stellar', () => ({
+  storeVerificationOnChain: vi.fn(),
+  simulateSettlement: vi.fn(),
+}));
+
+vi.mock('../metrics', () => ({
+  getMetricsService: () => ({ getMetrics: vi.fn().mockResolvedValue('') }),
+}));
+
+vi.mock('../fx-rate-cache', () => ({
+  getFxRateCache: () => ({ getCurrentRate: vi.fn() }),
+}));
+
+vi.mock('../kyc-upsert-service', () => ({
+  KycUpsertService: vi.fn().mockImplementation(() => ({
+    getStatusForUser: vi.fn(),
+  })),
+}));
+
+vi.mock('../transfer-guard', () => ({
+  createTransferGuard: () => (_req: any, _res: any, next: any) => next(),
+}));
+
+vi.mock('../sep24-service', () => ({
+  Sep24Service: vi.fn().mockImplementation(() => ({
+    initialize: vi.fn(),
+    initiateFlow: vi.fn(),
+    getTransactionStatus: vi.fn(),
+  })),
+  Sep24ConfigError: class Sep24ConfigError extends Error {},
+  Sep24AnchorError: class Sep24AnchorError extends Error {},
+}));
+
+import app from '../api';
+
+const AUTH_HEADER = { 'x-user-id': 'user-test-1' };
+
+beforeEach(() => {
+  insertedAlerts.length = 0;
+  insertedTransfers.length = 0;
+  mockPool.query.mockClear();
+});
+
+describe('POST /api/remittance — inline AML monitoring (SR-112)', () => {
+  it('raises an aml_alerts row for a rule-tripping transfer with no manual /api/aml call', async () => {
+    const res = await request(app)
+      .post('/api/remittance')
+      .set(AUTH_HEADER)
+      .send({
+        sender: 'GSENDERADDRESS000000000000000000000000000000000000000000',
+        agent: 'anchor-test',
+        amount: '500.00',
+      });
+
+    expect(res.status).toBe(201);
+
+    // VELOCITY_COUNT alert raised purely as a side effect of creating the
+    // remittance — no call was made to /api/aml/monitoring/evaluate.
+    expect(insertedAlerts.some((a) => a.rule_code === 'VELOCITY_COUNT')).toBe(true);
+  });
+
+  it('records a travel-rule transfer row for the same remittance', async () => {
+    await request(app)
+      .post('/api/remittance')
+      .set(AUTH_HEADER)
+      .send({
+        sender: 'GSENDERADDRESS000000000000000000000000000000000000000000',
+        agent: 'anchor-test',
+        amount: '500.00',
+      });
+
+    expect(insertedTransfers.length).toBeGreaterThan(0);
+  });
+});