From 5542a6f00bc6569dc9f38bf2b9a213d3c190e596 Mon Sep 17 00:00:00 2001 From: philipp Date: Fri, 12 Jun 2026 15:11:25 +0200 Subject: [PATCH] Fix late settlement ntfy notifications Proof: Targeted notification and outcome refresh tests pass; full npm test passes with 284 tests; operator dashboard bundle builds. Assumptions: A completed settlement can be computed after its inventory movement timestamp, and notification_deliveries remains the canonical once-only de-dup guard for ntfy delivery. Still fake: Venue-native terminal fill ids and fee-complete realized PnL remain unavailable; ntfy is still an external utility service outside the unrip app deployment ownership. --- src/apps/history-writer.mjs | 6 +++--- src/core/notification-layer.mjs | 5 +++++ src/lib/postgres.mjs | 16 +++++++++++---- test/notification-layer.test.mjs | 20 +++++++++++++++++++ test/postgres-intent-requests.test.mjs | 1 + test/postgres-quote-outcomes-refresh.test.mjs | 1 + 6 files changed, 42 insertions(+), 7 deletions(-) diff --git a/src/apps/history-writer.mjs b/src/apps/history-writer.mjs index f59f24a..1d3729c 100644 --- a/src/apps/history-writer.mjs +++ b/src/apps/history-writer.mjs @@ -10,7 +10,7 @@ import { buildLiquidityActionNotification, buildQuoteOutcomeNotification, createNotificationDispatcher, - observedAtOrAfter, + outcomeNotificationObservedAtOrAfter, } from '../core/notification-layer.mjs'; import { createNtfyNotificationClient } from '../core/notification-client.mjs'; import { buildPortfolioMetricId, computePortfolioMetric } from '../core/portfolio-metrics.mjs'; @@ -725,14 +725,14 @@ async function publishLiquidityNotification({ topic, event }) { async function publishQuoteOutcomeNotifications(records, { minObservedAt = null } = {}) { for (const record of records || []) { - if (!observedAtOrAfter(record.outcome_observed_at, minObservedAt)) continue; + if (!outcomeNotificationObservedAtOrAfter(record, minObservedAt)) continue; await publishNotification(buildQuoteOutcomeNotification({ record, config })); } } async function publishIntentRequestOutcomeNotifications(records, { minObservedAt = null } = {}) { for (const record of records || []) { - if (!observedAtOrAfter(record.outcome_observed_at, minObservedAt)) continue; + if (!outcomeNotificationObservedAtOrAfter(record, minObservedAt)) continue; await publishNotification(buildIntentRequestOutcomeNotification({ record, config })); } } diff --git a/src/core/notification-layer.mjs b/src/core/notification-layer.mjs index 1aefcc7..e20e8fe 100644 --- a/src/core/notification-layer.mjs +++ b/src/core/notification-layer.mjs @@ -149,6 +149,11 @@ export function observedAtOrAfter(value, minObservedAt) { return Number.isFinite(valueTs) && Number.isFinite(minTs) && valueTs >= minTs; } +export function outcomeNotificationObservedAtOrAfter(record = {}, minObservedAt) { + return observedAtOrAfter(record.outcome_observed_at || record.payload?.outcome_observed_at, minObservedAt) + || observedAtOrAfter(record.computed_at || record.payload?.computed_at, minObservedAt); +} + export function hasProvenSettlement(record = {}) { const payload = record.payload || {}; const status = record.outcome_status || payload.outcome_status; diff --git a/src/lib/postgres.mjs b/src/lib/postgres.mjs index da638f3..fc87d37 100644 --- a/src/lib/postgres.mjs +++ b/src/lib/postgres.mjs @@ -3745,14 +3745,18 @@ export async function refreshQuoteOutcomes(pool, { const computedAt = new Date( typeof now === 'number' ? now : Date.parse(now), ).toISOString(); - for (const record of records) { + const computedRecords = records.map((record) => ({ + ...record, + computed_at: computedAt, + })); + for (const record of computedRecords) { await upsertQuoteOutcome(pool, { ...record, computedAt, }); } - return records; + return computedRecords; } export async function upsertQuoteOutcome(pool, { @@ -4000,13 +4004,17 @@ export async function refreshIntentRequestOutcomes(pool, { const computedAt = new Date( typeof now === 'number' ? now : Date.parse(now), ).toISOString(); - for (const record of records) { + const computedRecords = records.map((record) => ({ + ...record, + computed_at: computedAt, + })); + for (const record of computedRecords) { await upsertIntentRequestOutcome(pool, { ...record, computedAt, }); } - return records; + return computedRecords; } async function loadIntentRequestPreflightRefreshCandidates(pool, { limit }) { diff --git a/test/notification-layer.test.mjs b/test/notification-layer.test.mjs index 8efa70e..570c87c 100644 --- a/test/notification-layer.test.mjs +++ b/test/notification-layer.test.mjs @@ -7,6 +7,7 @@ import { buildQuoteOutcomeNotification, createNotificationDispatcher, hasProvenSettlement, + outcomeNotificationObservedAtOrAfter, observedAtOrAfter, } from '../src/core/notification-layer.mjs'; @@ -203,3 +204,22 @@ test('outcome recency filter keeps startup refresh from sending historical notif assert.equal(observedAtOrAfter('2026-04-16T10:00:00.000Z', '2026-04-16T09:59:59.000Z'), true); assert.equal(observedAtOrAfter('2026-04-16T09:00:00.000Z', '2026-04-16T09:59:59.000Z'), false); }); + +test('outcome notification recency accepts settlements computed after their observed movement', () => { + assert.equal(outcomeNotificationObservedAtOrAfter({ + outcome_observed_at: '2026-06-12T13:02:32.053Z', + computed_at: '2026-06-12T13:08:42.800Z', + }, '2026-06-12T13:08:10.000Z'), true); + + assert.equal(outcomeNotificationObservedAtOrAfter({ + payload: { + outcome_observed_at: '2026-06-12T13:02:32.053Z', + computed_at: '2026-06-12T13:08:42.800Z', + }, + }, '2026-06-12T13:08:10.000Z'), true); + + assert.equal(outcomeNotificationObservedAtOrAfter({ + outcome_observed_at: '2026-06-12T13:02:32.053Z', + computed_at: '2026-06-12T13:07:42.800Z', + }, '2026-06-12T13:08:10.000Z'), false); +}); diff --git a/test/postgres-intent-requests.test.mjs b/test/postgres-intent-requests.test.mjs index 6f8fdb8..0b12112 100644 --- a/test/postgres-intent-requests.test.mjs +++ b/test/postgres-intent-requests.test.mjs @@ -259,6 +259,7 @@ test('intent request outcome refresh bounds source queries to refresh candidates assert.equal(records.length, 1); assert.equal(records[0].request_id, 'request-1'); assert.equal(records[0].outcome_status, 'completed'); + assert.equal(records[0].computed_at, '2026-05-13T10:00:20.000Z'); assert.equal(queries.filter((entry) => entry.sql.includes('INSERT INTO intent_request_outcomes')).length, 1); }); diff --git a/test/postgres-quote-outcomes-refresh.test.mjs b/test/postgres-quote-outcomes-refresh.test.mjs index 9103860..9003634 100644 --- a/test/postgres-quote-outcomes-refresh.test.mjs +++ b/test/postgres-quote-outcomes-refresh.test.mjs @@ -125,6 +125,7 @@ test('quote outcome refresh bounds source queries and joins by recent quote ids' assert.equal(records.length, 1); assert.equal(records[0].quote_id, 'quote-1'); + assert.equal(records[0].computed_at, '2026-05-13T10:00:20.000Z'); assert.equal(records[0].payload.edge_bps, '10'); assert.equal(queries.filter((entry) => entry.sql.includes('INSERT INTO quote_outcome_attributions')).length, 1); });