From 2e909026a7a1444d45fbe27fbdf7435e8cfc13e4 Mon Sep 17 00:00:00 2001
From: mpabi <mpabi@mpabi.pl>
Date: Sun, 12 Apr 2026 18:30:30 +0200
Subject: [PATCH] feat(sol): align canary ingestor and api auth

---
 .../workflows/deploy-trade-r001-canary.yaml   | 88 ++++++++++++++++++-
 environments/sol/trade-r001-canary/README.md  |  5 +-
 .../trade-r001-canary/api-token-seed-job.yaml | 76 ++++++++++++++++
 .../assets/ingestor/dlob-ingestor.mjs         | 38 ++++++--
 .../sol/trade-r001-canary/kustomization.yaml  |  1 +
 5 files changed, 199 insertions(+), 9 deletions(-)
 create mode 100644 environments/sol/trade-r001-canary/api-token-seed-job.yaml

diff --git a/.gitea/workflows/deploy-trade-r001-canary.yaml b/.gitea/workflows/deploy-trade-r001-canary.yaml
index 63e30e2..897c5e4 100644
--- a/.gitea/workflows/deploy-trade-r001-canary.yaml
+++ b/.gitea/workflows/deploy-trade-r001-canary.yaml
@@ -41,7 +41,7 @@ jobs:
         env:
           KUBECONFIG: /tmp/kubeconfig
         run: |
-          kubectl -n trade-r001-canary delete job postgres-migrate hasura-bootstrap --ignore-not-found=true
+          kubectl -n trade-r001-canary delete job postgres-migrate hasura-bootstrap api-token-seed --ignore-not-found=true
 
       - name: Apply shared host access infrastructure
         env:
@@ -79,6 +79,7 @@ jobs:
         run: |
           kubectl -n trade-r001-canary wait --for=condition=complete job/postgres-migrate --timeout=300s
           kubectl -n trade-r001-canary wait --for=condition=complete job/hasura-bootstrap --timeout=300s
+          kubectl -n trade-r001-canary wait --for=condition=complete job/api-token-seed --timeout=300s
 
       - name: Wait for application rollouts
         env:
@@ -104,6 +105,60 @@ jobs:
           restart_count="$(kubectl -n trade-r001-canary get pod "$pod_name" -o jsonpath='{.status.containerStatuses[0].restartCount}')"
           test "${restart_count}" = "0"
           kubectl -n trade-r001-canary logs "$pod_name" --tail=20
+          kubectl -n trade-r001-canary exec -i "$pod_name" -- node - <<'JS'
+          const endpoint = 'http://hasura:8080/v1/graphql';
+          const adminSecret = process.env.HASURA_ADMIN_SECRET;
+          const query = `
+            query {
+              drift_ticks(limit: 5, order_by: { ts: desc }) {
+                symbol
+                source
+                raw
+                ts
+              }
+            }
+          `;
+          const allowed = new Set(['dlob_hot_derived_latest', 'dlob_all_derived_latest']);
+
+          async function check() {
+            const response = await fetch(endpoint, {
+              method: 'POST',
+              headers: {
+                'content-type': 'application/json',
+                'x-hasura-admin-secret': adminSecret,
+              },
+              body: JSON.stringify({ query }),
+              signal: AbortSignal.timeout(10000),
+            });
+            const payload = await response.json();
+            const rows = payload?.data?.drift_ticks || [];
+            if (!rows.length) {
+              throw new Error('No rows in drift_ticks after trade-ingestor rollout');
+            }
+            const from = rows[0]?.raw?.from || null;
+            if (!allowed.has(from)) {
+              throw new Error(`Unexpected drift_ticks raw.from: ${from}`);
+            }
+            console.log(JSON.stringify(rows[0], null, 2));
+          }
+
+          (async () => {
+            for (let attempt = 0; attempt < 12; attempt += 1) {
+              try {
+                await check();
+                return;
+              } catch (error) {
+                if (attempt === 11) {
+                  throw error;
+                }
+                await new Promise((resolve) => setTimeout(resolve, 5000));
+              }
+            }
+          })().catch((error) => {
+            console.error(String(error && error.message ? error.message : error));
+            process.exit(1);
+          });
+          JS
 
       - name: Verify canary namespace connectivity
         env:
@@ -213,6 +268,37 @@ jobs:
             process.exit(1);
           });
           JS
+          token="$(kubectl -n trade-r001-canary get secret trade-frontend-tokens -o jsonpath='{.data.read\.json}' | base64 -d | jq -r .token)"
+          kubectl -n trade-r001-canary exec -i "$pod_name" -- env API_TOKEN="$token" node - <<'JS'
+          const headers = { Authorization: `Bearer ${process.env.API_TOKEN}` };
+
+          async function getJson(url) {
+            const response = await fetch(url, {
+              headers,
+              signal: AbortSignal.timeout(10000),
+            });
+            const payload = await response.json();
+            if (!response.ok || !payload?.ok) {
+              throw new Error(`${url} failed: ${response.status} ${JSON.stringify(payload)}`);
+            }
+            return payload;
+          }
+
+          (async () => {
+            const ticks = await getJson('http://trade-api:8787/v1/ticks?symbol=SOL-PERP&limit=5');
+            if (!Array.isArray(ticks.ticks) || !ticks.ticks.length) {
+              throw new Error('No SOL-PERP ticks from trade-api');
+            }
+            const chart = await getJson('http://trade-api:8787/v1/chart?symbol=SOL-PERP&tf=1m&limit=20');
+            if (!Array.isArray(chart.candles) || !chart.candles.length) {
+              throw new Error('No SOL-PERP candles from trade-api');
+            }
+            console.log(JSON.stringify({ ticks: ticks.ticks.at(-1), chart: chart.candles.at(-1) }, null, 2));
+          })().catch((err) => {
+            console.error(String(err && err.message ? err.message : err));
+            process.exit(1);
+          });
+          JS
           kubectl -n trade-r001-canary exec -i "$pod_name" -- node - <<'JS'
           const targets = [
             'http://hasura:8080/healthz',
diff --git a/environments/sol/trade-r001-canary/README.md b/environments/sol/trade-r001-canary/README.md
index f8ff627..224c0c3 100644
--- a/environments/sol/trade-r001-canary/README.md
+++ b/environments/sol/trade-r001-canary/README.md
@@ -50,9 +50,10 @@ Minimal canary namespace for migration baseline `R001` on `sol`.
 - The canary workflow re-runs:
   - `postgres-migrate`
   - `hasura-bootstrap`
+  - `api-token-seed`
   before it waits for `Hasura`, `trade-api`, `trade-frontend`, `trade-ingestor`, and the DLOB hot/all-path deployments to become healthy.
-- The current canary `trade-ingestor` is intentionally pinned to the schema already reconstructed on `sol` and reads from `dlob_stats_latest`.
-- The exact live `R001` ingestor path that reads `dlob_*_derived_latest` remains a follow-up substep after the DLOB writer chain is reconstructed.
+- The canary `trade-ingestor` now follows the live `R001` path on `sol`: it reads `dlob_hot_derived_latest` first for hot markets and falls back to `dlob_all_derived_latest`.
+- `api-token-seed` restores the frontend read token in `api_tokens`, so `trade-api` and `trade-frontend` can be validated against the reconstructed derived tables after each deploy.
 
 ## Operator Flow
 
diff --git a/environments/sol/trade-r001-canary/api-token-seed-job.yaml b/environments/sol/trade-r001-canary/api-token-seed-job.yaml
new file mode 100644
index 0000000..7a3eb21
--- /dev/null
+++ b/environments/sol/trade-r001-canary/api-token-seed-job.yaml
@@ -0,0 +1,76 @@
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: api-token-seed
+  namespace: trade-r001-canary
+spec:
+  backoffLimit: 1
+  template:
+    spec:
+      restartPolicy: OnFailure
+      containers:
+        - name: seed
+          image: postgres:16-alpine
+          imagePullPolicy: IfNotPresent
+          command:
+            - sh
+            - -lc
+            - |
+              set -euo pipefail
+
+              read_json="$(tr -d '\n' </tokens/read.json)"
+              token="$(printf '%s' "$read_json" | sed -E 's/.*"token"[[:space:]]*:[[:space:]]*"([^"]+)".*/\1/')"
+              name="$(printf '%s' "$read_json" | sed -E 's/.*"name"[[:space:]]*:[[:space:]]*"([^"]+)".*/\1/')"
+
+              test -n "$token"
+              test -n "$name"
+
+              token_hash="$(printf '%s' "$token" | sha256sum | awk '{print $1}')"
+              export PGPASSWORD="$POSTGRES_PASSWORD"
+
+              psql "host=$PGHOST port=$PGPORT dbname=$POSTGRES_DB user=$POSTGRES_USER" \
+                -v token_name="$name" \
+                -v token_hash="$token_hash" \
+                -v token_meta='{"seed":"trade-frontend-tokens/read.json","namespace":"trade-r001-canary","scopes":["read"]}' <<'SQL'
+              INSERT INTO api_tokens (name, token_hash, scopes, meta, revoked_at)
+              VALUES (
+                :'token_name',
+                :'token_hash',
+                ARRAY['read'],
+                (:'token_meta')::jsonb,
+                NULL
+              )
+              ON CONFLICT (token_hash) DO UPDATE
+              SET name = EXCLUDED.name,
+                  scopes = EXCLUDED.scopes,
+                  meta = EXCLUDED.meta,
+                  revoked_at = NULL;
+              SQL
+          env:
+            - name: PGHOST
+              value: postgres
+            - name: PGPORT
+              value: "5432"
+            - name: POSTGRES_USER
+              valueFrom:
+                secretKeyRef:
+                  name: trade-postgres
+                  key: POSTGRES_USER
+            - name: POSTGRES_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: trade-postgres
+                  key: POSTGRES_PASSWORD
+            - name: POSTGRES_DB
+              valueFrom:
+                secretKeyRef:
+                  name: trade-postgres
+                  key: POSTGRES_DB
+          volumeMounts:
+            - name: frontend-tokens
+              mountPath: /tokens
+              readOnly: true
+      volumes:
+        - name: frontend-tokens
+          secret:
+            secretName: trade-frontend-tokens
diff --git a/environments/sol/trade-r001-canary/assets/ingestor/dlob-ingestor.mjs b/environments/sol/trade-r001-canary/assets/ingestor/dlob-ingestor.mjs
index d49a9c0..dcf6447 100644
--- a/environments/sol/trade-r001-canary/assets/ingestor/dlob-ingestor.mjs
+++ b/environments/sol/trade-r001-canary/assets/ingestor/dlob-ingestor.mjs
@@ -101,10 +101,22 @@ async function graphqlRequest(cfg, query, variables) {
 async function fetchStats(cfg) {
   const query = `
     query DlobStatsLatest($markets: [String!]!) {
-      dlob_stats_latest(where: { market_name: { _in: $markets } }) {
+      dlob_hot_derived_latest(where: { market_name: { _in: $markets }, is_indicative: { _eq: false } }) {
         market_name
         market_index
-        ts
+        ts_ms
+        slot
+        oracle_price
+        mark_price
+        mid_price
+        best_bid_price
+        best_ask_price
+        updated_at
+      }
+      dlob_all_derived_latest(where: { market_name: { _in: $markets }, is_indicative: { _eq: false } }) {
+        market_name
+        market_index
+        ts_ms
         slot
         oracle_price
         mark_price
@@ -117,7 +129,21 @@ async function fetchStats(cfg) {
   `;
 
   const data = await graphqlRequest(cfg, query, { markets: cfg.markets });
-  return Array.isArray(data?.dlob_stats_latest) ? data.dlob_stats_latest : [];
+  const merged = new Map();
+
+  for (const row of Array.isArray(data?.dlob_all_derived_latest) ? data.dlob_all_derived_latest : []) {
+    const marketName = String(row?.market_name || '').trim();
+    if (!marketName) continue;
+    merged.set(marketName, { ...row, __from: 'dlob_all_derived_latest' });
+  }
+
+  for (const row of Array.isArray(data?.dlob_hot_derived_latest) ? data.dlob_hot_derived_latest : []) {
+    const marketName = String(row?.market_name || '').trim();
+    if (!marketName) continue;
+    merged.set(marketName, { ...row, __from: 'dlob_hot_derived_latest' });
+  }
+
+  return Array.from(merged.values());
 }
 
 async function insertTicks(cfg, objects) {
@@ -168,7 +194,7 @@ async function main() {
         if (updatedAt) lastUpdatedAtByMarket.set(marketName, updatedAt);
 
         const marketIndex = toIntOrNull(r?.market_index) ?? 0;
-        const dlobIso = isoFromEpochMs(r?.ts);
+        const dlobIso = isoFromEpochMs(r?.ts_ms);
         const tsIso = dlobIso || nowIso;
 
         const oraclePrice = numStr(r?.oracle_price) || numStr(r?.mark_price) || numStr(r?.mid_price);
@@ -184,11 +210,11 @@ async function main() {
           oracle_slot: r?.slot == null ? null : String(r.slot),
           source: cfg.source,
           raw: {
-            from: 'dlob_stats_latest',
+            from: r?.__from || 'dlob_all_derived_latest',
             market_name: marketName,
             market_index: marketIndex,
             dlob: {
-              ts: r?.ts ?? null,
+              ts_ms: r?.ts_ms ?? null,
               slot: r?.slot ?? null,
               best_bid_price: r?.best_bid_price ?? null,
               best_ask_price: r?.best_ask_price ?? null,
diff --git a/environments/sol/trade-r001-canary/kustomization.yaml b/environments/sol/trade-r001-canary/kustomization.yaml
index 6e2be80..b6c6d72 100644
--- a/environments/sol/trade-r001-canary/kustomization.yaml
+++ b/environments/sol/trade-r001-canary/kustomization.yaml
@@ -16,6 +16,7 @@ resources:
   - hasura-deployment.yaml
   - postgres-migrate-job.yaml
   - hasura-bootstrap-job.yaml
+  - api-token-seed-job.yaml
   - trade-api-service.yaml
   - trade-api-deployment.yaml
   - trade-frontend-service.yaml