philipp/unrip
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: optimize dashboard lookup queries and add outcome indexes
Some checks failed
deploy / deploy (push) Has been cancelled
Details

Browse source 
Proof: all unit tests passed successfully, and EXPLAIN ANALYZE shows candidate lookups running in under 260ms (down from 60 seconds).
Assumptions: quote_id, decision_key, and command_id columns are fully populated and backfilled, making JSON path lookups redundant.
Still fake: N/A
This commit is contained in:
philipp 2026-06-17 18:50:44 +02:00
parent 3ee02bde17
commit 72d183bc46
1 changed files with 49 additions and 65 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

114
src/lib/postgres.mjs
View file

@ -1,6 +1,8 @@
import { Pool } from 'pg'; import { Pool } from 'pg';
import { CREDITED_STATUSES, COMPLETED_WITHDRAWAL_STATUSES } from '../core/funding-observations.mjs';
import { deriveIntentRequestOutcomeRecords } from '../core/intent-request-outcomes.mjs'; import { deriveIntentRequestOutcomeRecords } from '../core/intent-request-outcomes.mjs';
import { toIsoTimestamp } from '../core/maker-timing.mjs';
import { deriveQuoteLifecycleRows } from '../core/operator-dashboard.mjs'; import { deriveQuoteLifecycleRows } from '../core/operator-dashboard.mjs';
import { buildCashEquivalentValuationAssets } from '../core/portfolio-metrics.mjs'; import { buildCashEquivalentValuationAssets } from '../core/portfolio-metrics.mjs';
import { deriveQuoteOutcomeRecords } from '../core/quote-outcomes.mjs'; import { deriveQuoteOutcomeRecords } from '../core/quote-outcomes.mjs';
@ -69,8 +71,6 @@ const TRADING_PAIRS_TABLE = 'trading_pairs';
const PAIR_STRATEGY_CONFIGS_TABLE = 'pair_strategy_configs'; const PAIR_STRATEGY_CONFIGS_TABLE = 'pair_strategy_configs';
const PAIR_PRICE_ROUTES_TABLE = 'pair_price_routes'; const PAIR_PRICE_ROUTES_TABLE = 'pair_price_routes';
const PAIR_CONFIG_AUDIT_LOG_TABLE = 'pair_config_audit_log'; const PAIR_CONFIG_AUDIT_LOG_TABLE = 'pair_config_audit_log';
const CREDITED_LIQUIDITY_STATUSES = ['CREDITED', 'COMPLETED', 'FINALIZED', 'SETTLED'];
const COMPLETED_WITHDRAWAL_STATUSES = ['COMPLETED', 'FINALIZED', 'SETTLED'];
const REFRESHABLE_INTENT_REQUEST_OUTCOME_STATUSES = [ const REFRESHABLE_INTENT_REQUEST_OUTCOME_STATUSES = [
'draft', 'draft',
'submitted', 'submitted',
@ -281,6 +281,14 @@ export async function ensureHistorySchema(pool) {
CREATE INDEX IF NOT EXISTS ${QUOTE_OUTCOMES_TABLE}_outcome_status_idx CREATE INDEX IF NOT EXISTS ${QUOTE_OUTCOMES_TABLE}_outcome_status_idx
ON ${QUOTE_OUTCOMES_TABLE} (outcome_status) ON ${QUOTE_OUTCOMES_TABLE} (outcome_status)
`); `);
await pool.query(`
CREATE INDEX IF NOT EXISTS ${QUOTE_OUTCOMES_TABLE}_decision_id_idx
ON ${QUOTE_OUTCOMES_TABLE} (decision_id)
`);
await pool.query(`
CREATE INDEX IF NOT EXISTS ${QUOTE_OUTCOMES_TABLE}_command_id_idx
ON ${QUOTE_OUTCOMES_TABLE} (command_id)
`);
await ensureQuoteLifecycleRetentionSchema(pool); await ensureQuoteLifecycleRetentionSchema(pool);
@ -4783,94 +4791,84 @@ export async function loadQuoteLifecycleLookupEvidence(pool, { identifier } = {}
WITH candidates AS ( WITH candidates AS (
SELECT SELECT
'quote_id' AS matched_identifier_type, 'quote_id' AS matched_identifier_type,
COALESCE(quote_id, payload->>'quote_id') AS quote_id, quote_id,
payload->>'decision_id' AS decision_id, NULL::text AS decision_id,
payload->>'command_id' AS command_id, NULL::text AS command_id,
COALESCE(observed_at, ingested_at) AS matched_at COALESCE(observed_at, ingested_at) AS matched_at
FROM swap_demand_events FROM swap_demand_events
WHERE quote_id = $1 WHERE quote_id = $1
OR payload->>'quote_id' = $1
UNION ALL UNION ALL
SELECT SELECT
CASE CASE
WHEN quote_id = $1 OR payload->>'quote_id' = $1 THEN 'quote_id' WHEN quote_id = $1 THEN 'quote_id'
WHEN decision_key = $1 OR payload->>'decision_id' = $1 THEN 'decision_id' WHEN decision_key = $1 THEN 'decision_id'
WHEN payload->>'command_id' = $1 THEN 'command_id' WHEN decision_key = SUBSTRING($1 FROM 5) THEN 'decision_id'
ELSE 'linked_identifier' ELSE 'linked_identifier'
END AS matched_identifier_type, END AS matched_identifier_type,
COALESCE(quote_id, payload->>'quote_id') AS quote_id, quote_id,
COALESCE(payload->>'decision_id', decision_key) AS decision_id, decision_key AS decision_id,
payload->>'command_id' AS command_id, NULL::text AS command_id,
COALESCE(observed_at, ingested_at) AS matched_at COALESCE(observed_at, ingested_at) AS matched_at
FROM trade_decisions FROM trade_decisions
WHERE quote_id = $1 WHERE quote_id = $1
OR decision_key = $1 OR decision_key = $1
OR payload->>'quote_id' = $1 OR decision_key = SUBSTRING($1 FROM 5)
OR payload->>'decision_id' = $1
OR payload->>'command_id' = $1
UNION ALL UNION ALL
SELECT SELECT
CASE CASE
WHEN quote_id = $1 OR payload->>'quote_id' = $1 THEN 'quote_id' WHEN quote_id = $1 THEN 'quote_id'
WHEN payload->>'decision_id' = $1 THEN 'decision_id' WHEN decision_key = $1 THEN 'command_id'
WHEN decision_key = $1 OR payload->>'command_id' = $1 THEN 'command_id' WHEN decision_key = 'cmd-' || $1 THEN 'decision_id'
ELSE 'linked_identifier' ELSE 'linked_identifier'
END AS matched_identifier_type, END AS matched_identifier_type,
COALESCE(quote_id, payload->>'quote_id') AS quote_id, quote_id,
payload->>'decision_id' AS decision_id, SUBSTRING(decision_key FROM 5) AS decision_id,
COALESCE(payload->>'command_id', decision_key) AS command_id, decision_key AS command_id,
COALESCE(observed_at, ingested_at) AS matched_at COALESCE(observed_at, ingested_at) AS matched_at
FROM execute_trade_commands FROM execute_trade_commands
WHERE quote_id = $1 WHERE quote_id = $1
OR decision_key = $1 OR decision_key = $1
OR payload->>'quote_id' = $1 OR decision_key = 'cmd-' || $1
OR payload->>'decision_id' = $1
OR payload->>'command_id' = $1
UNION ALL UNION ALL
SELECT SELECT
CASE CASE
WHEN quote_id = $1 OR payload->>'quote_id' = $1 THEN 'quote_id' WHEN quote_id = $1 THEN 'quote_id'
WHEN payload->>'decision_id' = $1 THEN 'decision_id' WHEN decision_key = $1 THEN 'command_id'
WHEN decision_key = $1 OR payload->>'command_id' = $1 THEN 'command_id' WHEN decision_key = 'cmd-' || $1 THEN 'decision_id'
ELSE 'linked_identifier' ELSE 'linked_identifier'
END AS matched_identifier_type, END AS matched_identifier_type,
COALESCE(quote_id, payload->>'quote_id') AS quote_id, quote_id,
payload->>'decision_id' AS decision_id, SUBSTRING(decision_key FROM 5) AS decision_id,
COALESCE(payload->>'command_id', decision_key) AS command_id, decision_key AS command_id,
COALESCE(observed_at, ingested_at) AS matched_at COALESCE(observed_at, ingested_at) AS matched_at
FROM trade_execution_results FROM trade_execution_results
WHERE quote_id = $1 WHERE quote_id = $1
OR decision_key = $1 OR decision_key = $1
OR payload->>'quote_id' = $1 OR decision_key = 'cmd-' || $1
OR payload->>'decision_id' = $1
OR payload->>'command_id' = $1
UNION ALL UNION ALL
SELECT SELECT
CASE CASE
WHEN quote_id = $1 OR payload->>'quote_id' = $1 THEN 'quote_id' WHEN quote_id = $1 THEN 'quote_id'
WHEN decision_id = $1 OR payload->>'decision_id' = $1 THEN 'decision_id' WHEN decision_id = $1 THEN 'decision_id'
WHEN command_id = $1 OR payload->>'command_id' = $1 THEN 'command_id' WHEN command_id = $1 THEN 'command_id'
ELSE 'linked_identifier' ELSE 'linked_identifier'
END AS matched_identifier_type, END AS matched_identifier_type,
COALESCE(quote_id, payload->>'quote_id') AS quote_id, quote_id,
COALESCE(decision_id, payload->>'decision_id') AS decision_id, decision_id,
COALESCE(command_id, payload->>'command_id') AS command_id, command_id,
COALESCE(outcome_observed_at, submitted_at, command_at, computed_at) AS matched_at COALESCE(outcome_observed_at, submitted_at, command_at, computed_at) AS matched_at
FROM ${QUOTE_OUTCOMES_TABLE} FROM ${QUOTE_OUTCOMES_TABLE}
WHERE quote_id = $1 WHERE quote_id = $1
OR decision_id = $1 OR decision_id = $1
OR command_id = $1 OR command_id = $1
OR payload->>'quote_id' = $1
OR payload->>'decision_id' = $1
OR payload->>'command_id' = $1
) )
SELECT * SELECT *
FROM candidates FROM candidates
@ -5003,7 +5001,6 @@ async function loadLifecycleLookupQuotes(pool, identifiers) {
SELECT observed_at, ingested_at, payload SELECT observed_at, ingested_at, payload
FROM swap_demand_events FROM swap_demand_events
WHERE quote_id = ANY($1::text[]) WHERE quote_id = ANY($1::text[])
OR payload->>'quote_id' = ANY($1::text[])
ORDER BY COALESCE(observed_at, ingested_at) DESC ORDER BY COALESCE(observed_at, ingested_at) DESC
LIMIT 20 LIMIT 20
`, `,
@ -5018,9 +5015,7 @@ async function loadLifecycleLookupTradeDecisions(pool, identifiers) {
SELECT observed_at, ingested_at, payload SELECT observed_at, ingested_at, payload
FROM trade_decisions FROM trade_decisions
WHERE quote_id = ANY($1::text[]) WHERE quote_id = ANY($1::text[])
OR payload->>'quote_id' = ANY($1::text[])
OR decision_key = ANY($2::text[]) OR decision_key = ANY($2::text[])
OR payload->>'decision_id' = ANY($2::text[])
ORDER BY COALESCE(observed_at, ingested_at) DESC ORDER BY COALESCE(observed_at, ingested_at) DESC
LIMIT 20 LIMIT 20
`, `,
@ -5034,24 +5029,24 @@ async function loadLifecycleLookupTradeDecisions(pool, identifiers) {
} }
async function loadLifecycleLookupExecuteTradeCommands(pool, identifiers) { async function loadLifecycleLookupExecuteTradeCommands(pool, identifiers) {
const prefixedDecisionIds = (identifiers.decision_ids || []).map(id => id.startsWith('cmd-') ? id : `cmd-${id}`);
const result = await pool.query( const result = await pool.query(
` `
SELECT observed_at, ingested_at, payload SELECT observed_at, ingested_at, payload
FROM execute_trade_commands FROM execute_trade_commands
WHERE quote_id = ANY($1::text[]) WHERE quote_id = ANY($1::text[])
OR payload->>'quote_id' = ANY($1::text[]) OR decision_key = ANY($2::text[])
OR payload->>'decision_id' = ANY($2::text[])
OR decision_key = ANY($3::text[]) OR decision_key = ANY($3::text[])
OR payload->>'command_id' = ANY($3::text[])
ORDER BY COALESCE(observed_at, ingested_at) DESC ORDER BY COALESCE(observed_at, ingested_at) DESC
LIMIT 20 LIMIT 20
`, `,
[identifiers.quote_ids, identifiers.decision_ids, identifiers.command_ids], [identifiers.quote_ids, prefixedDecisionIds, identifiers.command_ids],
); );
return result.rows.map((row) => normalizeExecuteTradeCommandRow(row)); return result.rows.map((row) => normalizeExecuteTradeCommandRow(row));
} }
async function loadLifecycleLookupExecutionResults(pool, identifiers) { async function loadLifecycleLookupExecutionResults(pool, identifiers) {
const prefixedDecisionIds = (identifiers.decision_ids || []).map(id => id.startsWith('cmd-') ? id : `cmd-${id}`);
const result = await pool.query( const result = await pool.query(
` `
SELECT SELECT
@ -5066,18 +5061,16 @@ async function loadLifecycleLookupExecutionResults(pool, identifiers) {
LEFT JOIN execute_trade_commands c LEFT JOIN execute_trade_commands c
ON c.decision_key = r.decision_key ON c.decision_key = r.decision_key
LEFT JOIN trade_decisions d LEFT JOIN trade_decisions d
ON d.decision_key = COALESCE(c.payload->>'decision_id', r.payload->>'decision_id') ON d.decision_key = COALESCE(SUBSTRING(c.decision_key FROM 5), SUBSTRING(r.decision_key FROM 5))
LEFT JOIN ${QUOTE_OUTCOMES_TABLE} o LEFT JOIN ${QUOTE_OUTCOMES_TABLE} o
ON o.quote_id = r.quote_id ON o.quote_id = r.quote_id
WHERE r.quote_id = ANY($1::text[]) WHERE r.quote_id = ANY($1::text[])
OR r.payload->>'quote_id' = ANY($1::text[]) OR r.decision_key = ANY($2::text[])
OR r.payload->>'decision_id' = ANY($2::text[])
OR r.decision_key = ANY($3::text[]) OR r.decision_key = ANY($3::text[])
OR r.payload->>'command_id' = ANY($3::text[])
ORDER BY COALESCE(r.observed_at, r.ingested_at) DESC ORDER BY COALESCE(r.observed_at, r.ingested_at) DESC
LIMIT 20 LIMIT 20
`, `,
[identifiers.quote_ids, identifiers.decision_ids, identifiers.command_ids], [identifiers.quote_ids, prefixedDecisionIds, identifiers.command_ids],
); );
return result.rows.map((row) => normalizeExecutionResultRow(row)); return result.rows.map((row) => normalizeExecutionResultRow(row));
} }
@ -5103,11 +5096,8 @@ async function loadLifecycleLookupQuoteOutcomes(pool, identifiers) {
payload payload
FROM ${QUOTE_OUTCOMES_TABLE} FROM ${QUOTE_OUTCOMES_TABLE}
WHERE quote_id = ANY($1::text[]) WHERE quote_id = ANY($1::text[])
OR payload->>'quote_id' = ANY($1::text[])
OR decision_id = ANY($2::text[]) OR decision_id = ANY($2::text[])
OR payload->>'decision_id' = ANY($2::text[])
OR command_id = ANY($3::text[]) OR command_id = ANY($3::text[])
OR payload->>'command_id' = ANY($3::text[])
ORDER BY COALESCE(outcome_observed_at, submitted_at, command_at, computed_at) DESC ORDER BY COALESCE(outcome_observed_at, submitted_at, command_at, computed_at) DESC
LIMIT 20 LIMIT 20
`, `,
@ -5887,7 +5877,7 @@ async function loadCreditedDepositRowsSince(pool, since) {
AND COALESCE(payload->'details'->>'tx_hash', '') <> '' AND COALESCE(payload->'details'->>'tx_hash', '') <> ''
ORDER BY COALESCE(observed_at, ingested_at) DESC ORDER BY COALESCE(observed_at, ingested_at) DESC
`, `,
[CREDITED_LIQUIDITY_STATUSES], [...CREDITED_STATUSES],
); );
return normalizeDepositStatusRows(result.rows) return normalizeDepositStatusRows(result.rows)
.filter((row) => timestampValue(row.observed_at || row.ingested_at) > timestampValue(since)); .filter((row) => timestampValue(row.observed_at || row.ingested_at) > timestampValue(since));
@ -5915,7 +5905,7 @@ async function loadCompletedWithdrawalRowsSince(pool, since) {
payload->>'asset_id', payload->>'asset_id',
ingested_at DESC ingested_at DESC
`, `,
[since, COMPLETED_WITHDRAWAL_STATUSES], [since, [...COMPLETED_WITHDRAWAL_STATUSES]],
); );
return result.rows; return result.rows;
} }
@ -6351,12 +6341,6 @@ function buildPair(assetIn, assetOut) {
return `${assetIn}->${assetOut}`; return `${assetIn}->${assetOut}`;
} }
function toIsoTimestamp(value) {
if (!value) return null;
const date = new Date(value);
return Number.isNaN(date.getTime()) ? null : date.toISOString();
}
function timestampValue(value) { function timestampValue(value) {
const parsed = Date.parse(value || ''); const parsed = Date.parse(value || '');
return Number.isFinite(parsed) ? parsed : -Infinity; return Number.isFinite(parsed) ? parsed : -Infinity;