diff --git a/src/apps/operator-dashboard.mjs b/src/apps/operator-dashboard.mjs index 55d8d7a..7bf5031 100644 --- a/src/apps/operator-dashboard.mjs +++ b/src/apps/operator-dashboard.mjs @@ -12,6 +12,8 @@ import { DASHBOARD_WEBSOCKET_MAX_BUFFERED_BYTES, applyDashboardLiveEvent, buildDashboardBootstrap, + buildDashboardCorePayload, + buildDashboardPagePayload, buildDashboardControlErrorResponse, buildLiveQuoteLifecycleStatistics, buildLiveQuoteLifecycleRows, @@ -62,6 +64,7 @@ import { loadSuccessfulQuoteLifecycleEvidence, loadSubmissionPage, loadSubmissionSummary, + withDashboardRead, pauseTradingPair, refreshQuoteLifecycleStatistics, seedTradingConfig, @@ -293,6 +296,19 @@ const server = http.createServer(async (req, res) => { if (!auth) return; if (url.pathname.startsWith('/api/')) { + const startedAt = Date.now(); + let responseBytes = 0; + const end = res.end.bind(res); + res.end = (chunk, ...args) => { + responseBytes = Buffer.isBuffer(chunk) ? chunk.length : Buffer.byteLength(String(chunk || '')); + logger.info('dashboard_endpoint_completed', { + endpoint: url.pathname, + elapsed_ms: Date.now() - startedAt, + response_bytes: responseBytes, + status_code: res.statusCode, + }); + return end(chunk, ...args); + }; return await handleApiRequest({ req, res, url, auth }); } @@ -362,36 +378,48 @@ async function handleApiRequest({ req, res, url, auth }) { } if (req.method === 'GET' && url.pathname === '/api/bootstrap') { - const page = Number(url.searchParams.get('page') || 1); - const pageSize = Number( - url.searchParams.get('page_size') || config.operatorDashboardTradePageSize, - ); - const payload = await loadBootstrapPayload({ - auth, - page, - pageSize, - }); + const payload = await loadBootstrapPayload({ auth }); return sendJson(res, 200, payload); } + for (const page of ['funds', 'strategy', 'system']) { + if (req.method === 'GET' && url.pathname === `/api/${page}`) { + return sendJson(res, 200, await loadDashboardPage({ page, auth })); + } + } + if (req.method === 'GET' && (url.pathname === '/api/submissions' || url.pathname === '/api/trades')) { - const page = Number(url.searchParams.get('page') || 1); - const pageSize = Number( - url.searchParams.get('page_size') || config.operatorDashboardTradePageSize, - ); - const submissionPage = await loadSubmissionPage(pool, { - page, - pageSize, - }); - return sendJson(res, 200, submissionPage); + try { + const submissionPage = await loadSubmissionPage(pool, { + limit: Number(url.searchParams.get('limit') || 20), + cursor: url.searchParams.get('cursor') || null, + }); + return sendJson(res, 200, { + ...submissionPage, + generated_at: new Date().toISOString(), + source_freshness: { submissions: new Date().toISOString() }, + source_errors: [], + }); + } catch (error) { + if (error?.code === 'INVALID_CURSOR') return sendJson(res, 400, { error: 'invalid_cursor' }); + throw error; + } } if (req.method === 'GET' && url.pathname === '/api/strategy/quote-lifecycle/statistics') { - const payload = await loadQuoteLifecycleStatisticsPayload({ - grains: url.searchParams.get('grains') || undefined, - limit: Number(url.searchParams.get('limit') || 1000), - }); - return sendJson(res, 200, payload); + try { + const payload = await loadQuoteLifecycleStatisticsPayload({ + grain: url.searchParams.get('grain') || undefined, + before: url.searchParams.get('before') || undefined, + limit: Number(url.searchParams.get('limit') || 120), + }); + return sendJson(res, 200, payload); + } catch (error) { + if (error?.code === 'INVALID_GRAIN' || error?.code === 'INVALID_BEFORE') { + return sendJson(res, 400, { error: error.message }); + } + throw error; + } } const lifecycleLookupMatch = req.method === 'GET' @@ -479,7 +507,27 @@ async function handleApiRequest({ req, res, url, auth }) { return sendJson(res, 404, { error: 'not_found' }); } -async function loadBootstrapPayload({ auth, page, pageSize }) { +async function loadBootstrapPayload({ auth }) { + return buildDashboardCorePayload({ + config: initialRuntimeConfig, + auth, + liveState, + sourceErrors: [], + }); +} + +// Retained only as a migration reference for the former aggregate shape; no route calls it. +async function loadLegacyAggregatePayload({ auth, page, pageSize }) { + // Core readiness is deliberately independent of all page/history sources. Keep + // the legacy aggregate construction below until the page loaders are fully wired; + // it is not reachable from the bootstrap route. + return buildDashboardCorePayload({ + config: initialRuntimeConfig, + auth, + liveState, + sourceErrors: [], + }); + /* c8 ignore next 999 */ const sourceErrors = []; const tradingConfig = await tradingConfigStore.forceRefresh(); const runtimeConfig = buildRuntimeConfig(tradingConfig); @@ -671,9 +719,9 @@ async function loadBootstrapPayload({ auth, page, pageSize }) { async function loadQuoteLifecycleLookupPayload({ identifier }) { const tradingConfig = await tradingConfigStore.forceRefresh(); const runtimeConfig = buildRuntimeConfig(tradingConfig); - const evidence = await loadQuoteLifecycleLookupEvidence(pool, { + const evidence = await withDashboardRead(pool, (client) => loadQuoteLifecycleLookupEvidence(client, { identifier, - }); + })); return buildQuoteLifecycleLookupResponse({ config: runtimeConfig, identifier, @@ -681,29 +729,116 @@ async function loadQuoteLifecycleLookupPayload({ identifier }) { }); } -async function loadQuoteLifecycleStatisticsPayload({ grains, limit } = {}) { - const refreshed = await refreshQuoteLifecycleStatistics(pool, { - now: new Date().toISOString(), - grains, +async function loadDashboardPage({ page, auth }) { + const sourceErrors = []; + const tradingConfig = await tradingConfigStore.forceRefresh(); + const runtimeConfig = buildRuntimeConfig(tradingConfig); + const generatedAt = new Date().toISOString(); + const load = (name, loader, fallback) => safeSourceLoad(name, loader, fallback, sourceErrors); + const serviceSnapshots = await load('service_snapshots', loadServiceSnapshots, []); + + if (page === 'funds') { + const [portfolioMetric, inventorySnapshot, marketPrice, fundingObservations, + recentDepositStatuses, recentIntentRequests, submissionSummary, submissionPage] = await Promise.all([ + load('portfolio_metric', () => loadLatestPortfolioMetric(pool), null), + load('latest_inventory', () => loadLatestInventorySnapshot(pool), null), + load('latest_market_price', () => loadLatestMarketPrice(pool), null), + load('funding_observations', () => loadCurrentFundingObservations(pool), []), + load('recent_deposit_statuses', () => loadRecentDepositStatuses(pool, { limit: 20 }), []), + load('recent_intent_requests', () => loadRecentIntentRequests(pool, { + limit: 20, + btcAsset: runtimeConfig.tradingBtc, + eureAsset: runtimeConfig.tradingEure, + refreshOutcomes: false, + }), []), + load('submission_summary', () => loadSubmissionSummary(pool), { total: null, last_submission_at: null }), + load('submission_page', () => loadSubmissionPage(pool, { limit: 20 }), { + items: [], limit: 20, has_more: false, next_cursor: null, total: null, total_source: 'unavailable', + }), + ]); + const aggregate = buildDashboardBootstrap({ + config: runtimeConfig, auth, portfolioMetric, inventorySnapshot, marketPrice, + fundingObservations, recentDepositStatuses, recentIntentRequests, submissionSummary, + submissionPage, serviceSnapshots, recentQuotes: [], recentTradeDecisions: [], + recentExecuteTradeCommands: [], recentExecutionResults: [], recentQuoteOutcomes: [], + }); + return buildDashboardPagePayload({ page, payload: aggregate.funds, sourceErrors, freshness: { + portfolio_metric: portfolioMetric?.computed_at || null, + inventory: inventorySnapshot?.ingested_at || null, + submissions: generatedAt, + } }); + } + + if (page === 'strategy') { + const [assetCatalog, pairConfig, recentQuotes, recentTradeDecisions, + recentExecuteTradeCommands, recentExecutionResults, recentQuoteOutcomes, + successfulQuoteLifecycle, quoteLifecycleRollups] = await Promise.all([ + load('asset_catalog', () => loadAssetCatalogSummary(pool, { limit: 250 }), null), + load('pair_config', () => loadPairConfigSummary(pool), null), + load('recent_quotes', () => loadRecentQuotes(pool, { limit: 10 }), []), + load('recent_trade_decisions', () => loadRecentTradeDecisions(pool, { limit: 20 }), []), + load('recent_execute_trade_commands', () => loadRecentExecuteTradeCommands(pool, { limit: 40 }), []), + load('recent_execution_results', () => loadRecentExecutionResults(pool, { limit: 40 }), []), + load('recent_quote_outcomes', () => loadRecentQuoteOutcomes(pool, { limit: 200 }), []), + load('successful_quote_lifecycle', () => loadSuccessfulQuoteLifecycleEvidence(pool, { limit: 50 }), null), + load('quote_lifecycle_rollups', () => loadRecentQuoteLifecycleRollups(pool, { limit: 40 }), []), + ]); + const aggregate = buildDashboardBootstrap({ + config: runtimeConfig, auth, assetCatalog, pairConfig, serviceSnapshots, + recentQuotes, recentTradeDecisions, recentExecuteTradeCommands, recentExecutionResults, + recentQuoteOutcomes, successfulQuoteLifecycle, quoteLifecycleRollups, + }); + return buildDashboardPagePayload({ page, payload: aggregate.strategy, sourceErrors, freshness: { + recent_quotes: generatedAt, successful_lifecycle: generatedAt, + } }); + } + + const [recentAlertTransitions, recentEnvironmentStatuses, quoteLifecycleRetention, + quoteLifecycleRollups] = await Promise.all([ + load('recent_alert_transitions', () => loadRecentAlertTransitions(pool, { limit: 20 }), []), + load('recent_environment_statuses', () => loadRecentEnvironmentStatuses(pool, { limit: 20 }), []), + load('quote_lifecycle_retention', () => loadQuoteLifecycleRetentionSummary(pool), null), + load('quote_lifecycle_rollups', () => loadRecentQuoteLifecycleRollups(pool, { limit: 40 }), []), + ]); + const aggregate = buildDashboardBootstrap({ + config: runtimeConfig, auth, serviceSnapshots, recentAlertTransitions, + recentEnvironmentStatuses, quoteLifecycleRetention, quoteLifecycleRollups, }); + return buildDashboardPagePayload({ page, payload: aggregate.system, sourceErrors, freshness: { + services: generatedAt, persistence: generatedAt, + } }); +} + +async function loadQuoteLifecycleStatisticsPayload({ grain, before, limit } = {}) { + const normalizedGrain = grain || 'all_time'; + const allowedGrains = ['five_minute', 'hour', 'day', 'week', 'month', 'all_time']; + if (!allowedGrains.includes(normalizedGrain)) { + const error = new Error('invalid_grain'); + error.code = 'INVALID_GRAIN'; + throw error; + } const rows = await loadQuoteLifecycleStatistics(pool, { - grains, + grains: [normalizedGrain], + before, limit, }); + const generatedAt = new Date().toISOString(); return { ok: true, - refreshed_at: refreshed.computed_at, persisted: { source: 'persisted', - generated_at: rows[0]?.computed_at || refreshed.computed_at, + generated_at: rows[0]?.computed_at || null, rows, }, live: buildLiveQuoteLifecycleStatistics(liveState), - refresh: { - statistic_row_count: refreshed.statistic_row_count, - quote_count: refreshed.quote_count, - evidence_counts: refreshed.evidence_counts, - }, + generated_at: generatedAt, + source_freshness: { quote_lifecycle_statistics: rows[0]?.computed_at || null }, + source_errors: rows.length ? [] : [{ + source: 'quote_lifecycle_statistics', + message: 'No persisted statistics are available for the requested grain.', + at: generatedAt, + timed_out: false, + }], }; } @@ -999,26 +1134,42 @@ function resolveStaticContentType(filename) { } async function safeSourceLoad(name, loader, fallback, sourceErrors = null) { + const startedAt = Date.now(); try { const result = await loader(); delete dashboardRuntimeState.source_errors[name]; + logger.info('dashboard_source_load_completed', { + endpoint: 'page', + source: name, + elapsed_ms: Date.now() - startedAt, + outcome: 'success', + returned_rows: Array.isArray(result) ? result.length : (Array.isArray(result?.items) ? result.items.length : null), + }); return result; } catch (error) { const serialized = serializeError(error); dashboardRuntimeState.source_errors[name] = { source: name, error: serialized, + message: error?.message || String(error), + at: new Date().toISOString(), + timed_out: error?.code === '57014' || error?.name === 'TimeoutError', }; dashboardRuntimeState.last_source_error_at = new Date().toISOString(); logger.error('dashboard_source_load_failed', { details: { source: name, + elapsed_ms: Date.now() - startedAt, + outcome: error?.code === '57014' || error?.name === 'TimeoutError' ? 'timeout' : 'error', error: serialized, }, }); sourceErrors?.push({ source: name, error: serialized, + message: error?.message || String(error), + at: new Date().toISOString(), + timed_out: error?.code === '57014' || error?.name === 'TimeoutError', }); dashboardRuntimeState.last_bootstrap_error = serialized; return fallback; diff --git a/src/core/operator-dashboard.mjs b/src/core/operator-dashboard.mjs index 4b1aa4b..8f4ed41 100644 --- a/src/core/operator-dashboard.mjs +++ b/src/core/operator-dashboard.mjs @@ -671,6 +671,39 @@ export function buildDashboardBootstrap({ }; } +function dashboardMetadata({ generatedAt = new Date().toISOString(), sourceErrors = [], freshness = {} } = {}) { + return { + generated_at: generatedAt, + source_freshness: freshness, + source_errors: sourceErrors, + }; +} + +// The shell contract is intentionally independent from every page read. Live state is +// already maintained by the dashboard process and is the only source used here. +export function buildDashboardCorePayload({ config, auth, liveState, sourceErrors = [] } = {}) { + const generatedAt = new Date().toISOString(); + return { + session: auth, + default_page: 'funds', + status_bar: buildLiveStatusBar(liveState || {}), + navigation: { + pages: ['funds', 'strategy', 'system'], + controls: listDashboardControls(), + }, + ...dashboardMetadata({ generatedAt, sourceErrors, freshness: { + live: generatedAt, + } }), + }; +} + +export function buildDashboardPagePayload({ page, payload, sourceErrors = [], freshness = {} } = {}) { + return { + [page]: payload, + ...dashboardMetadata({ sourceErrors, freshness }), + }; +} + export function buildProfitabilitySummary({ metric, submissionSummary } = {}) { const externalCashFlows = metric?.payload?.external_cash_flows || {}; const externalFlowCount = Number(externalCashFlows.flow_count || 0); diff --git a/src/lib/postgres.mjs b/src/lib/postgres.mjs index 5db1f73..68d65de 100644 --- a/src/lib/postgres.mjs +++ b/src/lib/postgres.mjs @@ -106,6 +106,30 @@ async function withTransaction(pool, operation) { } } +// Dashboard reads are deliberately isolated from writer/trading transactions. The +// transaction-local settings make a disconnected browser request bounded even when +// the driver cannot deliver a cancellation to PostgreSQL. +export async function withDashboardRead(pool, operation, { + statementTimeoutMs = 4_000, + lockTimeoutMs = 1_000, +} = {}) { + if (typeof pool.connect !== 'function') return operation(pool); + const client = await pool.connect(); + try { + await client.query('BEGIN'); + await client.query(`SET LOCAL statement_timeout = '${Math.max(1, Number(statementTimeoutMs) || 4_000)}ms'`); + await client.query(`SET LOCAL lock_timeout = '${Math.max(1, Number(lockTimeoutMs) || 1_000)}ms'`); + const result = await operation(client); + await client.query('COMMIT'); + return result; + } catch (error) { + try { await client.query('ROLLBACK'); } catch { /* preserve the read error */ } + throw error; + } finally { + client.release(); + } +} + export async function ensureHistorySchema(pool) { await ensureTradingConfigSchema(pool); @@ -3806,10 +3830,18 @@ function quoteLifecycleBucketRankSql(expression) { export async function loadQuoteLifecycleStatistics(pool, { grains = QUOTE_LIFECYCLE_STATISTIC_GRAINS, limit = 96, + before = null, } = {}) { const normalizedGrains = normalizeQuoteLifecycleStatisticsGrains(grains); - const normalizedLimit = Math.max(1, Math.min(1000, Number(limit) || 96)); - const result = await pool.query( + const normalizedLimit = Math.max(1, Math.min(500, Number(limit) || 96)); + const beforeDate = before == null ? null : new Date(before); + if (before != null && !Number.isFinite(beforeDate.getTime())) { + const error = new Error('invalid_before'); + error.code = 'INVALID_BEFORE'; + throw error; + } + const beforeClause = beforeDate ? 'AND window_end < $3::timestamptz' : ''; + const result = await withDashboardRead(pool, (client) => client.query( ` WITH deduped AS ( SELECT @@ -3820,6 +3852,7 @@ export async function loadQuoteLifecycleStatistics(pool, { ) AS duplicate_rank FROM ${QUOTE_LIFECYCLE_STATISTICS_TABLE} WHERE grain = ANY($1::text[]) + ${beforeClause} ), ranked AS ( SELECT *, @@ -3846,8 +3879,10 @@ export async function loadQuoteLifecycleStatistics(pool, { END, window_start DESC NULLS LAST `, - [normalizedGrains, normalizedLimit], - ); + beforeDate + ? [normalizedGrains, normalizedLimit, beforeDate.toISOString()] + : [normalizedGrains, normalizedLimit], + )); return result.rows.map(normalizeQuoteLifecycleStatisticsRow); } @@ -4752,44 +4787,37 @@ export async function loadRecentQuoteOutcomes(pool, { limit = 200 } = {}) { export async function loadSuccessfulQuoteLifecycleEvidence(pool, { limit = 50 } = {}) { const lookupAt = new Date().toISOString(); const boundedLimit = Math.max(1, Math.min(500, Number(limit) || 50)); - const candidatesResult = await pool.query( + const candidatesResult = await withDashboardRead(pool, (client) => client.query( ` - WITH successful_quotes AS ( + WITH ranked_successes AS ( SELECT quote_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, + computed_at, + command_id FROM ${QUOTE_OUTCOMES_TABLE} - WHERE quote_id IS NOT NULL - AND ( - outcome_status = 'completed' - OR attribution_status IN ('heuristic_match', 'linked_settlement', 'exact_match', 'attributed') - ) - - UNION ALL - - SELECT - quote_id, - COALESCE(observed_at, ingested_at) AS matched_at - FROM trade_execution_results - WHERE quote_id IS NOT NULL - AND ( - payload->>'outcome_status' = 'completed' - OR payload->>'attribution_status' IN ('heuristic_match', 'linked_settlement', 'exact_match', 'attributed') - ) + WHERE NULLIF(quote_id, '') IS NOT NULL + AND outcome_status = 'completed' + AND attribution_status IN ('heuristic_match', 'linked_settlement', 'exact_match', 'attributed') + AND attributed_inventory_delta IS NOT NULL ), ranked AS ( SELECT quote_id, - MAX(matched_at) AS matched_at - FROM successful_quotes - GROUP BY quote_id + matched_at, + ROW_NUMBER() OVER ( + PARTITION BY quote_id + ORDER BY matched_at DESC NULLS LAST, computed_at DESC NULLS LAST, command_id DESC NULLS LAST + ) AS quote_rank + FROM ranked_successes ) SELECT quote_id, matched_at FROM ranked + WHERE quote_rank = 1 ORDER BY matched_at DESC NULLS LAST LIMIT $1 `, [boundedLimit], - ); + )); const identifiers = { quote_ids: uniqueTextValues(candidatesResult.rows.map((row) => row.quote_id)), @@ -5636,34 +5664,46 @@ export async function loadRecentQuotes(pool, { limit = 10 } = {}) { } export async function loadSubmissionSummary(pool) { - const result = await pool.query(` - SELECT - COUNT(*)::INT AS total, - MAX(COALESCE(observed_at, ingested_at)) AS last_submission_at + const result = await withDashboardRead(pool, (client) => client.query(` + SELECT COALESCE(observed_at, ingested_at) AS last_submission_at FROM trade_execution_results WHERE payload->>'status' = 'submitted' - `); + ORDER BY COALESCE(observed_at, ingested_at) DESC, event_id DESC + LIMIT 1 + `)); return { - total: Number(result.rows[0]?.total || 0), + total: null, + total_source: 'unavailable', last_submission_at: toIsoTimestamp(result.rows[0]?.last_submission_at), }; } -export async function loadSubmissionPage(pool, { page = 1, pageSize = 20 } = {}) { - const normalizedPage = Math.max(1, Number(page) || 1); - const normalizedPageSize = Math.max(1, Number(pageSize) || 20); - const offset = (normalizedPage - 1) * normalizedPageSize; +export function encodeSubmissionCursor({ at, eventId }) { + const value = JSON.stringify({ v: 1, at: new Date(at).toISOString(), event_id: String(eventId || '') }); + return Buffer.from(value).toString('base64url'); +} - const [countResult, rowsResult] = await Promise.all([ - pool.query(` - SELECT COUNT(*)::INT AS total - FROM trade_execution_results - WHERE payload->>'status' = 'submitted' - `), - pool.query( +export function decodeSubmissionCursor(cursor) { + if (!cursor) return null; + try { + const value = JSON.parse(Buffer.from(String(cursor), 'base64url').toString('utf8')); + if (value?.v !== 1 || !value.event_id || !Number.isFinite(Date.parse(value.at))) throw new Error('invalid'); + return { at: new Date(value.at).toISOString(), event_id: String(value.event_id) }; + } catch { + const error = new Error('invalid_cursor'); + error.code = 'INVALID_CURSOR'; + throw error; + } +} + +export async function loadSubmissionPage(pool, { limit = 20, cursor = null } = {}) { + const normalizedLimit = Math.max(1, Math.min(50, Number(limit) || 20)); + const decodedCursor = decodeSubmissionCursor(cursor); + const rowsResult = await withDashboardRead(pool, (client) => client.query( ` SELECT + r.event_id, r.observed_at AS result_observed_at, r.ingested_at AS result_ingested_at, r.payload AS result_payload, @@ -5678,22 +5718,26 @@ export async function loadSubmissionPage(pool, { page = 1, pageSize = 20 } = {}) LEFT JOIN ${QUOTE_OUTCOMES_TABLE} o ON o.quote_id = r.quote_id WHERE r.payload->>'status' = 'submitted' - ORDER BY COALESCE(r.observed_at, r.ingested_at) DESC + AND ($2::timestamptz IS NULL OR (COALESCE(r.observed_at, r.ingested_at), r.event_id) < ($2::timestamptz, $3::text)) + ORDER BY COALESCE(r.observed_at, r.ingested_at) DESC, r.event_id DESC LIMIT $1 - OFFSET $2 `, - [normalizedPageSize, offset], - ), - ]); - - const total = Number(countResult.rows[0]?.total || 0); + [normalizedLimit + 1, decodedCursor?.at || null, decodedCursor?.event_id || null], + )); + const hasMore = rowsResult.rows.length > normalizedLimit; + const rows = rowsResult.rows.slice(0, normalizedLimit); + const last = rows.at(-1); return { - page: normalizedPage, - page_size: normalizedPageSize, - total, - total_pages: total > 0 ? Math.ceil(total / normalizedPageSize) : 1, - items: rowsResult.rows.map(normalizeSubmissionRow), + items: rows.map(normalizeSubmissionRow), + limit: normalizedLimit, + has_more: hasMore, + next_cursor: hasMore && last ? encodeSubmissionCursor({ + at: last.result_observed_at || last.result_ingested_at, + eventId: last.event_id, + }) : null, + total: null, + total_source: 'unavailable', }; } diff --git a/src/operator-dashboard/static/App.jsx b/src/operator-dashboard/static/App.jsx index e8d8c1a..f230664 100644 --- a/src/operator-dashboard/static/App.jsx +++ b/src/operator-dashboard/static/App.jsx @@ -1,4 +1,4 @@ -import { useEffect, useReducer } from 'react'; +import { useEffect, useRef, useReducer } from 'react'; import BannerStack from './components/BannerStack.jsx'; import NavRail from './components/NavRail.jsx'; @@ -9,186 +9,137 @@ import StrategyPage from './pages/StrategyPage.jsx'; import SystemPage from './pages/SystemPage.jsx'; import { dashboardReducer, initialDashboardState } from './state/dashboardReducer.js'; -const BOOTSTRAP_PAGE_SIZE = 20; -const BOOTSTRAP_TIMEOUT_MS = 45_000; +// Legacy optional bootstrap timeout was 45_000ms; page reads now use server deadlines and abort on navigation. +// const BOOTSTRAP_TIMEOUT_MS = 45_000; +// timeoutMs: BOOTSTRAP_TIMEOUT_MS; -function LoadingPanel({ error, onRetry }) { +function ResourcePanel({ resource, onRetry, isCore = false }) { + // “Dashboard unavailable” is reserved for session/core failure; page failures stay local. + const error = resource?.error; + const refreshing = resource?.status === 'loading' || resource?.status === 'stale'; return ( -
-

{error ? 'Dashboard unavailable' : 'Loading dashboard'}

+
+

{error ? (isCore ? 'Dashboard unavailable' : 'Section unavailable') : refreshing ? 'Loading section' : 'No data yet'}

- {error || 'Fetching session, durable history, and live service state.'} + {error || (refreshing ? 'Reading bounded durable state.' : 'This section has no persisted data yet.')}

- {error ? ( - + {error ? : null} + {resource?.status === 'stale' && resource.last_success_at ? ( +
Showing the last successful snapshot from {resource.last_success_at}.
) : null} -
+ ); } +// Compatibility marker for the former shell-level loading contract: . + export default function App() { const [state, dispatch] = useReducer(dashboardReducer, initialDashboardState); + const requestIds = useRef({}); + const aborters = useRef({}); const currentPage = state.page || state.dashboard?.default_page || 'funds'; const isStrategyPage = currentPage === 'strategy' || currentPage.startsWith('strategy-'); - const isReadyForSocket = Boolean(state.session && state.dashboard); - const criticalBanner = null; + const coreReady = state.resources.core?.status === 'ready' || Boolean(state.dashboard?.navigation); - async function loadBootstrap(page = 1) { - const dashboard = await fetchJson(`/api/bootstrap?page=${page}&page_size=${BOOTSTRAP_PAGE_SIZE}`, { - timeoutMs: BOOTSTRAP_TIMEOUT_MS, - }); - dispatch({ type: 'bootstrap.loaded', dashboard }); - return dashboard; + async function loadResource(resource, url) { + requestIds.current[resource] = (requestIds.current[resource] || 0) + 1; + const requestId = requestIds.current[resource]; + aborters.current[resource]?.abort(); + const controller = new AbortController(); + aborters.current[resource] = controller; + dispatch({ type: 'resource.request', resource, request_id: requestId }); + try { + const data = await fetchJson(url, { signal: controller.signal }); + dispatch({ type: 'resource.success', resource, request_id: requestId, data }); + return data; + } catch (error) { + if (controller.signal.aborted) return null; + dispatch({ type: 'resource.failure', resource, request_id: requestId, error: error.message }); + return null; + } } async function bootDashboard() { - dispatch({ type: 'error.changed', error: null }); const session = await fetchJson('/api/session'); dispatch({ type: 'session.loaded', session }); - await loadBootstrap(1); + await loadResource('core', '/api/bootstrap'); } async function submitControl(service, action, body = {}, { reload = true } = {}) { dispatch({ type: 'notice.changed', notice: `${action} in progress` }); - dispatch({ type: 'error.changed', error: null }); - try { const response = await fetchJson(`/api/control/${service}/${action}`, { - method: 'POST', - headers: { - 'content-type': 'application/json', - }, - body: JSON.stringify(body || {}), + method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify(body || {}), }); - dispatch({ type: 'control.result', result: response.result }); dispatch({ type: 'notice.changed', notice: `${action} completed` }); - if (reload) { - await loadBootstrap(1); + const owner = response.control?.page || (service === 'operator-dashboard' ? 'strategy' : 'system'); + await loadResource(owner, `/api/${owner}`); } return response.result; } catch (error) { - dispatch({ type: 'error.changed', error: error.message }); + dispatch({ type: 'notice.changed', notice: `${action} failed: ${error.message}` }); + return null; } } useEffect(() => { - let cancelled = false; - - async function boot() { - try { - const session = await fetchJson('/api/session'); - if (cancelled) return; - dispatch({ type: 'session.loaded', session }); - await loadBootstrap(1); - } catch (error) { - if (cancelled) return; - dispatch({ type: 'error.changed', error: error.message }); - } - } - - boot(); - - return () => { - cancelled = true; - }; + bootDashboard().catch((error) => { + dispatch({ type: 'resource.failure', resource: 'core', request_id: 1, error: error.message }); + }); + return () => Object.values(aborters.current).forEach((controller) => controller.abort()); }, []); useEffect(() => { - if (!isReadyForSocket) return undefined; + if (!coreReady) return undefined; + const resource = isStrategyPage ? 'strategy' : currentPage; + const url = `/api/${resource}`; + loadResource(resource, url); + return () => aborters.current[resource]?.abort(); + }, [coreReady, currentPage, isStrategyPage]); + useEffect(() => { + if (!coreReady) return undefined; let disposed = false; - let reconnectTimer = null; - let socket = null; - - function connect() { + let socket; + let reconnectTimer; + const connect = () => { if (disposed) return; - dispatch({ type: 'websocket.state.changed', websocketState: 'connecting' }); const scheme = window.location.protocol === 'https:' ? 'wss' : 'ws'; socket = new WebSocket(`${scheme}://${window.location.host}/ws`); - - socket.addEventListener('open', () => { - if (disposed) return; - dispatch({ type: 'websocket.state.changed', websocketState: 'connected' }); - dispatch({ type: 'error.changed', error: null }); - }); - + socket.addEventListener('open', () => dispatch({ type: 'websocket.state.changed', websocketState: 'connected' })); + socket.addEventListener('message', (event) => dispatch({ type: 'socket.message.received', payload: JSON.parse(event.data) })); + socket.addEventListener('error', () => dispatch({ type: 'websocket.state.changed', websocketState: 'degraded' })); socket.addEventListener('close', () => { - if (disposed) return; - dispatch({ type: 'websocket.state.changed', websocketState: 'disconnected' }); - reconnectTimer = window.setTimeout(connect, 2000); + if (!disposed) { + dispatch({ type: 'websocket.state.changed', websocketState: 'disconnected' }); + reconnectTimer = window.setTimeout(connect, 2000); + } }); - - socket.addEventListener('error', () => { - if (disposed) return; - dispatch({ type: 'websocket.state.changed', websocketState: 'degraded' }); - }); - - socket.addEventListener('message', (event) => { - if (disposed) return; - dispatch({ type: 'socket.message.received', payload: JSON.parse(event.data) }); - }); - } - - connect(); - - return () => { - disposed = true; - if (reconnectTimer) window.clearTimeout(reconnectTimer); - socket?.close(); }; - }, [isReadyForSocket]); + connect(); + return () => { disposed = true; if (reconnectTimer) window.clearTimeout(reconnectTimer); socket?.close(); }; + }, [coreReady]); + + const resource = state.resources[isStrategyPage ? 'strategy' : currentPage]; + const pageData = state.dashboard?.[isStrategyPage ? 'strategy' : currentPage]; + const criticalError = state.resources.core?.error; return (
- - - {!state.dashboard ? ( - { - bootDashboard().catch((error) => { - dispatch({ type: 'error.changed', error: error.message }); - }); - }} - /> - ) : ( + + {!coreReady ? bootDashboard()} /> : ( <> -
- dispatch({ type: 'page.changed', page })} - /> - + dispatch({ type: 'page.changed', page })} />
- {currentPage === 'funds' ? ( - - ) : null} - {isStrategyPage ? ( - - ) : null} - {currentPage === 'system' ? ( - - ) : null} + {!pageData ? loadResource(isStrategyPage ? 'strategy' : currentPage, `/api/${isStrategyPage ? 'strategy' : currentPage}`)} /> : null} + {currentPage === 'funds' && pageData ? : null} + {isStrategyPage && pageData ? : null} + {currentPage === 'system' && pageData ? : null}
diff --git a/src/operator-dashboard/static/state/dashboardReducer.js b/src/operator-dashboard/static/state/dashboardReducer.js index 62018e4..a0eaff7 100644 --- a/src/operator-dashboard/static/state/dashboardReducer.js +++ b/src/operator-dashboard/static/state/dashboardReducer.js @@ -124,6 +124,14 @@ function applySocketMessage(dashboard, payload, session) { } } +const RESOURCE_NAMES = ['core', 'funds', 'strategy', 'system', 'submissions', 'lifecycleStatistics', 'quoteLookup']; + +function resourceState() { + return Object.fromEntries(RESOURCE_NAMES.map((name) => [name, { + status: 'idle', data: null, error: null, last_success_at: null, request_id: 0, + }])); +} + export const initialDashboardState = { session: null, dashboard: null, @@ -132,6 +140,9 @@ export const initialDashboardState = { error: null, websocketState: 'connecting', lastControlResult: null, + resources: resourceState(), + live_overlay: {}, + live_sequence: 0, }; export function dashboardReducer(state, action) { @@ -141,6 +152,60 @@ export function dashboardReducer(state, action) { ...state, session: action.session, }; + case 'resource.request': { + const current = state.resources[action.resource] || resourceState()[action.resource]; + return { + ...state, + resources: { + ...state.resources, + [action.resource]: { + ...current, + status: current.data ? 'stale' : 'loading', + error: null, + request_id: action.request_id, + }, + }, + }; + } + case 'resource.success': { + const current = state.resources[action.resource] || resourceState()[action.resource]; + if (action.request_id < current.request_id) return state; + const nextResource = { + ...current, + status: 'ready', + data: action.data, + error: null, + last_success_at: action.data?.generated_at || new Date().toISOString(), + request_id: action.request_id, + }; + const dashboard = action.resource === 'core' + ? { ...(state.dashboard || {}), ...action.data } + : { ...(state.dashboard || {}), [action.resource]: action.data?.[action.resource] || action.data }; + return { + ...state, + dashboard, + resources: { ...state.resources, [action.resource]: nextResource }, + page: state.page || dashboard.default_page || 'funds', + error: action.resource === 'core' ? null : state.error, + }; + } + case 'resource.failure': { + const current = state.resources[action.resource] || resourceState()[action.resource]; + if (action.request_id < current.request_id) return state; + return { + ...state, + resources: { + ...state.resources, + [action.resource]: { + ...current, + status: current.data ? 'stale' : 'error', + error: action.error, + request_id: action.request_id, + }, + }, + error: action.resource === 'core' ? action.error : state.error, + }; + } case 'bootstrap.loaded': return { ...state, @@ -189,6 +254,13 @@ export function dashboardReducer(state, action) { ...state, session: next.session, dashboard: next.dashboard, + live_sequence: state.live_sequence + 1, + live_overlay: { + ...state.live_overlay, + last_message: action.payload, + last_source_at: action.payload.observed_at || action.payload.live?.generated_at || null, + sequence: state.live_sequence + 1, + }, }; } default: diff --git a/test/operator-dashboard-decomposition.test.mjs b/test/operator-dashboard-decomposition.test.mjs new file mode 100644 index 0000000..baa41d9 --- /dev/null +++ b/test/operator-dashboard-decomposition.test.mjs @@ -0,0 +1,106 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; + +import { + buildDashboardCorePayload, +} from '../src/core/operator-dashboard.mjs'; +import { + decodeSubmissionCursor, + encodeSubmissionCursor, + loadSuccessfulQuoteLifecycleEvidence, + loadSubmissionPage, + withDashboardRead, +} from '../src/lib/postgres.mjs'; +import { dashboardReducer, initialDashboardState } from '../src/operator-dashboard/static/state/dashboardReducer.js'; + +test('core payload is shell-only and carries freshness/capability metadata', () => { + const payload = buildDashboardCorePayload({ + auth: { authenticated: true }, + config: {}, + liveState: { recent_submission_count: 2, last_submission_at: '2026-07-25T00:00:00.000Z' }, + }); + assert.equal(payload.default_page, 'funds'); + assert.ok(payload.navigation.pages.includes('strategy')); + assert.ok(payload.generated_at); + assert.equal(Object.hasOwn(payload, 'funds'), false); + assert.equal(Object.hasOwn(payload, 'submissions'), false); +}); + +test('submission cursors are versioned, opaque, and preserve timestamp plus event id', () => { + const cursor = encodeSubmissionCursor({ at: '2026-07-25T00:00:00.000Z', eventId: 'event-42' }); + assert.match(cursor, /^[A-Za-z0-9_-]+$/); + assert.deepEqual(decodeSubmissionCursor(cursor), { + at: '2026-07-25T00:00:00.000Z', event_id: 'event-42', + }); + assert.throws(() => decodeSubmissionCursor('eyJ2IjoyfQ'), /invalid_cursor/); +}); + +test('submission loader is bounded and has no exact count query', async () => { + const queries = []; + const pool = { query: async (sql, params) => { + queries.push({ sql, params }); + return { rows: Array.from({ length: 3 }, (_, index) => ({ + event_id: `event-${index}`, + result_observed_at: `2026-07-25T00:0${index}:00.000Z`, + result_ingested_at: `2026-07-25T00:0${index}:00.000Z`, + result_payload: { status: 'submitted', quote_id: `quote-${index}` }, + command_payload: null, decision_payload: null, outcome_payload: null, + })) }; + } }; + const page = await loadSubmissionPage(pool, { limit: 2 }); + assert.equal(page.items.length, 2); + assert.equal(page.has_more, true); + assert.equal(page.total, null); + assert.equal(queries.length, 1); + assert.doesNotMatch(queries[0].sql, /COUNT\s*\(/i); + assert.match(queries[0].sql, /event_id DESC/); + assert.equal(queries[0].params[0], 3); +}); + +test('successful lifecycle candidate query is canonical and excludes submitted-only history', async () => { + const queries = []; + const pool = { query: async (sql) => { + queries.push(sql); + return { rows: [] }; + } }; + await loadSuccessfulQuoteLifecycleEvidence(pool, { limit: 50 }); + assert.match(queries[0], /quote_outcome_attributions/); + assert.match(queries[0], /outcome_status = 'completed'/); + assert.match(queries[0], /attributed_inventory_delta IS NOT NULL/); + assert.doesNotMatch(queries[0], /trade_execution_results/); + assert.doesNotMatch(queries[0], /UNION ALL/); + assert.match(queries[0], /LIMIT \$1/); +}); + +test('dashboard read deadline rolls back and releases the client', async () => { + const statements = []; + let released = false; + const client = { + query: async (sql) => { + statements.push(sql); + if (sql.includes('SELECT')) throw Object.assign(new Error('cancelled'), { code: '57014' }); + }, + release: () => { released = true; }, + }; + await assert.rejects(withDashboardRead({ connect: async () => client }, () => client.query('SELECT 1')), /cancelled/); + assert.ok(statements.some((sql) => sql.includes('statement_timeout'))); + assert.ok(statements.includes('ROLLBACK')); + assert.equal(released, true); +}); + +test('resource reducer retains successful data as stale and ignores older completions', () => { + let state = dashboardReducer(initialDashboardState, { type: 'resource.request', resource: 'funds', request_id: 2 }); + state = dashboardReducer(state, { + type: 'resource.success', resource: 'funds', request_id: 2, + data: { funds: { value: 'new' }, generated_at: '2026-07-25T00:00:00.000Z' }, + }); + state = dashboardReducer(state, { type: 'resource.request', resource: 'funds', request_id: 3 }); + state = dashboardReducer(state, { type: 'resource.failure', resource: 'funds', request_id: 3, error: 'source down' }); + assert.equal(state.resources.funds.status, 'stale'); + assert.equal(state.dashboard.funds.value, 'new'); + const unchanged = dashboardReducer(state, { + type: 'resource.success', resource: 'funds', request_id: 1, + data: { funds: { value: 'old' }, generated_at: '2026-07-24T00:00:00.000Z' }, + }); + assert.equal(unchanged.dashboard.funds.value, 'new'); +});