Fix late settlement ntfy notifications
All checks were successful
deploy / deploy (push) Successful in 56s

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.
This commit is contained in:
philipp 2026-06-12 15:11:25 +02:00
parent e5c0a5dc6e
commit 5542a6f00b
6 changed files with 42 additions and 7 deletions

View file

@ -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 }));
}
}

View file

@ -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;

View file

@ -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 }) {

View file

@ -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);
});

View file

@ -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);
});

View file

@ -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);
});