feat(logging): emit structured ingestion failures

docs/logging.md

@@ -0,0 +1,14 @@
+# Structured ingestion logs
+
+Ingestion failures are emitted as one JSON object per line. Stable fields such
+as `route`, `venue`, `pair`, `network`, and `status` can be indexed by the log
+collector while `message` stays readable. Error details are kept in `error`.
+
+With Loki, count failures for one pair over the last hour with:
+
+```logql
+sum(count_over_time({app="lens"} | json | level="error" | pair="XLM/USDC" [1h]))
+```
+
+The application does not currently assign request IDs to background ingestion
+cycles, so those events omit `requestId` instead of inventing one.

src/ingest/venues/amm.ts

@@ -6,6 +6,7 @@ import { dispatchPriceUpdate } from '../../webhookDispatcher'
 import { publishPriceUpdate } from '../../events'
 import type { WatchedPair } from '../../types'
 import type { VenueAdapter, VenuePricePoint } from './types'
+import { logEvent } from '../../logging'
 import {
   getLastPrice,
   resetLastPrice,
@@ -42,7 +43,14 @@ export async function fetchPools(pair: WatchedPair, network: NetworkName = activ
     if (!data._embedded?.records) return []
     return data._embedded.records
   } catch (err) {
-    console.error(`[amm] Failed to fetch pools for ${pair.pairKey}:`, (err as Error).message)
+    logEvent('error', 'Failed to fetch AMM pools', {
+      route: '/liquidity_pools',
+      venue: 'AMM',
+      pair: pair.pairKey,
+      network,
+      status: 'error',
+      error: (err as Error).message,
+    })
     return []
   }
 }

src/ingest/venues/runner.ts

@@ -5,6 +5,7 @@ import { dispatchPriceUpdate } from '../../webhookDispatcher'
 import { publishPriceUpdate } from '../../events'
 import type { WatchedPair } from '../../types'
 import type { VenueAdapter, VenuePricePoint } from './types'
+import { logEvent } from '../../logging'
 
 // ── Centralized Last Price Cache ──────────────────────────────────────────────
 // Keyed by `${venue}:${network}:${pairKey}`
@@ -99,19 +100,29 @@ export async function startVenueAdapter(
 
   let consecutiveErrors = 0
   while (true) {
+    const cycleStartedAt = Date.now()
+    let currentPair: string | undefined
     try {
       const pairs = getActivePairs()
       for (const pair of pairs) {
+        currentPair = pair.pairKey
         await adapter.ingestPair(pair, network)
       }
       consecutiveErrors = 0
     } catch (err) {
       consecutiveErrors++
       const backoffMs = Math.min(10_000 * Math.pow(2, consecutiveErrors - 1), 60_000)
-      console.error(
-        `[${adapter.name}] Cycle failed on ${network} (attempt ${consecutiveErrors}), retrying in ${backoffMs}ms:`,
-        (err as Error).message
-      )
+      logEvent('error', 'Venue ingestion cycle failed', {
+        route: `ingest/${adapter.name}`,
+        venue: adapter.name,
+        network,
+        pair: currentPair,
+        status: 'error',
+        attempt: consecutiveErrors,
+        retryMs: backoffMs,
+        durationMs: Date.now() - cycleStartedAt,
+        error: (err as Error).message,
+      })
       await sleep(backoffMs)
       continue
     }

src/logging.ts

@@ -0,0 +1,13 @@
+type LogFields = Record<string, string | number | boolean | null | undefined>
+
+/** Emit one JSON log event so operators can filter and aggregate field values. */
+export function logEvent(
+  level: 'info' | 'warn' | 'error',
+  message: string,
+  fields: LogFields = {},
+): void {
+  const record = JSON.stringify({ level, message, ...fields })
+  if (level === 'error') console.error(record)
+  else if (level === 'warn') console.warn(record)
+  else console.log(record)
+}