From 59eb888f76b8b6786fd3c3c9b84ee9471f249003 Mon Sep 17 00:00:00 2001 From: u1 Date: Sun, 29 Mar 2026 13:14:30 +0200 Subject: [PATCH] chore(snapshot): sync api workspace state --- services/api/server.mjs | 1786 ++++++++++++++++++++++++++++++--------- 1 file changed, 1391 insertions(+), 395 deletions(-) mode change 100644 => 100755 services/api/server.mjs diff --git a/services/api/server.mjs b/services/api/server.mjs old mode 100644 new mode 100755 index 8c63706..3eb8976 --- a/services/api/server.mjs +++ b/services/api/server.mjs @@ -1,12 +1,15 @@ import crypto from 'node:crypto'; import fs from 'node:fs'; import http from 'node:http'; +import { DatabaseSync } from 'node:sqlite'; import path from 'node:path'; import { fileURLToPath } from 'node:url'; const SCRIPT_DIR = path.dirname(fileURLToPath(import.meta.url)); const PROJECT_ROOT = path.resolve(SCRIPT_DIR, '..', '..'); const TOKENS_DIR = path.join(PROJECT_ROOT, 'tokens'); +const diagramDbCache = new Map(); +const DIAGRAM_LAYOUT_REVISION_LIMIT = 120; function readJsonFile(filePath) { try { @@ -17,6 +20,22 @@ function readJsonFile(filePath) { } } +function ensureParentDir(filePath) { + fs.mkdirSync(path.dirname(filePath), { recursive: true }); +} + +function writeJsonFile(filePath, value) { + ensureParentDir(filePath); + const tempPath = `${filePath}.${process.pid}.${Date.now()}.tmp`; + fs.writeFileSync(tempPath, JSON.stringify(value, null, 2)); + fs.renameSync(tempPath, filePath); +} + +function appendLine(filePath, line) { + ensureParentDir(filePath); + fs.appendFileSync(filePath, `${line}\n`, 'utf8'); +} + function getIsoNow() { return new Date().toISOString(); } @@ -55,7 +74,7 @@ function readBearerToken(req) { function withCors(res, corsOrigin) { res.setHeader('access-control-allow-origin', corsOrigin); - res.setHeader('access-control-allow-methods', 'GET,POST,OPTIONS'); + res.setHeader('access-control-allow-methods', 'GET,POST,PATCH,DELETE,OPTIONS'); res.setHeader( 'access-control-allow-headers', 'content-type, authorization, x-api-key, x-admin-secret' @@ -126,6 +145,16 @@ function resolveConfig() { 'get_drift_candles', 'CANDLES_FUNCTION' ); + const dlobPublisherAllHealthUrl = String( + process.env.DLOB_PUBLISHER_ALL_HEALTH_URL || 'http://dlob-publisher-all:8080/health' + ).trim(); + const dlobPublisherAllMetricsUrl = String( + process.env.DLOB_PUBLISHER_ALL_METRICS_URL || 'http://dlob-publisher-all:9464/metrics' + ).trim(); + const diagramStateDir = String(process.env.DIAGRAM_STATE_DIR || path.join(PROJECT_ROOT, '.runtime')).trim(); + const diagramDbFile = String( + process.env.DIAGRAM_DB_FILE || path.join(diagramStateDir, 'diagram.sqlite3') + ).trim(); return { port, @@ -138,6 +167,9 @@ function resolveConfig() { startedAt, ticksTable, candlesFunction, + dlobPublisherAllHealthUrl, + dlobPublisherAllMetricsUrl, + diagramDbFile, }; } @@ -169,6 +201,413 @@ async function hasuraRequest(cfg, { admin }, query, variables) { return json.data; } +async function fetchText(url, { timeoutMs = 2500 } = {}) { + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeoutMs); + try { + const res = await fetch(url, { + method: 'GET', + cache: 'no-store', + signal: controller.signal, + }); + return { ok: res.ok, status: res.status, text: await res.text() }; + } finally { + clearTimeout(timer); + } +} + +function parsePrometheusLabels(raw) { + const labels = {}; + if (!raw) return labels; + const re = /([a-zA-Z_][a-zA-Z0-9_]*)="((?:\\"|[^"])*)"/g; + let match; + while ((match = re.exec(raw))) { + labels[match[1]] = match[2].replace(/\\"/g, '"'); + } + return labels; +} + +function parsePrometheusSamples(text) { + const samples = []; + for (const rawLine of String(text || '').split(/\r?\n/)) { + const line = rawLine.trim(); + if (!line || line.startsWith('#')) continue; + const match = line.match(/^([a-zA-Z_:][a-zA-Z0-9_:]*)(?:\{([^}]*)\})?\s+([^\s]+)(?:\s+(\d+))?$/); + if (!match) continue; + const value = Number(match[3]); + if (!Number.isFinite(value)) continue; + samples.push({ + metric: match[1], + labels: parsePrometheusLabels(match[2] || ''), + value, + timestamp: match[4] ? Number(match[4]) : null, + }); + } + return samples; +} + +async function readPublisherAllRuntime(cfg) { + const runtime = { + id: 'publisher', + title: 'dlob-publisher-all', + status: 'warn', + health: { + httpOk: false, + gauge: null, + }, + metrics: { + transport: 'yellowstone + websocket', + updateIntervalMs: 400, + perpMarkets: 0, + spotMarkets: 0, + totalMarkets: 0, + latestSlot: null, + }, + endpoints: { + health: cfg.dlobPublisherAllHealthUrl, + metrics: cfg.dlobPublisherAllMetricsUrl, + }, + errors: [], + }; + + try { + const health = await fetchText(cfg.dlobPublisherAllHealthUrl); + runtime.health.httpOk = health.ok && /^ok$/i.test(String(health.text || '').trim()); + if (!runtime.health.httpOk) runtime.errors.push(`health_http_${health.status}`); + } catch (err) { + runtime.errors.push(`health_fetch:${String(err?.message || err)}`); + } + + try { + const metricsRes = await fetchText(cfg.dlobPublisherAllMetricsUrl); + if (!metricsRes.ok) { + runtime.errors.push(`metrics_http_${metricsRes.status}`); + } else { + const samples = parsePrometheusSamples(metricsRes.text); + const healthGauge = samples.find((sample) => sample.metric === 'health_status'); + const dlobSlots = samples.filter((sample) => sample.metric === 'dlob_slot'); + const perpMarkets = dlobSlots.filter((sample) => sample.labels.marketType === 'perp'); + const spotMarkets = dlobSlots.filter((sample) => sample.labels.marketType === 'spot'); + const latestSlot = dlobSlots.reduce( + (max, sample) => (sample.value > max ? sample.value : max), + Number.NEGATIVE_INFINITY + ); + + runtime.health.gauge = healthGauge ? healthGauge.value : null; + runtime.metrics.perpMarkets = perpMarkets.length; + runtime.metrics.spotMarkets = spotMarkets.length; + runtime.metrics.totalMarkets = perpMarkets.length + spotMarkets.length; + runtime.metrics.latestSlot = Number.isFinite(latestSlot) ? Math.trunc(latestSlot) : null; + } + } catch (err) { + runtime.errors.push(`metrics_fetch:${String(err?.message || err)}`); + } + + runtime.status = + runtime.health.httpOk && + (runtime.health.gauge == null || runtime.health.gauge === 0) && + runtime.metrics.totalMarkets > 0 + ? 'ok' + : 'warn'; + + return runtime; +} + +async function readDiagramRuntime(cfg) { + return { + ok: true, + version: cfg.appVersion, + fetchedAt: getIsoNow(), + nodes: { + publisher: await readPublisherAllRuntime(cfg), + }, + }; +} + +function normalizeDiagramView(value) { + return String(value || '').trim().toLowerCase() === 'ultra' ? 'ultra' : 'standard'; +} + +function normalizeDiagramLayout(layout) { + if (!layout || typeof layout !== 'object') return {}; + + const next = {}; + for (const [id, rawEntry] of Object.entries(layout)) { + if (!rawEntry || typeof rawEntry !== 'object') continue; + const entry = {}; + + const rawPosition = rawEntry.position; + if (rawPosition && typeof rawPosition === 'object') { + const x = Number(rawPosition.x); + const y = Number(rawPosition.y); + if (Number.isFinite(x) && Number.isFinite(y)) entry.position = { x, y }; + } + + const rawSize = rawEntry.size; + if (rawSize && typeof rawSize === 'object') { + const width = Number(rawSize.width); + const height = Number(rawSize.height); + if (Number.isFinite(width) && Number.isFinite(height)) { + entry.size = { width, height }; + } + } + + if (rawEntry.parentId === null || typeof rawEntry.parentId === 'string') { + entry.parentId = rawEntry.parentId; + } + + if (entry.position || entry.size || Object.prototype.hasOwnProperty.call(entry, 'parentId')) { + next[id] = entry; + } + } + + return next; +} + +function normalizeDiagramLayoutAction(value, fallback = 'save') { + const raw = String(value || fallback) + .trim() + .toLowerCase() + .replace(/[^a-z0-9:_-]+/g, '-') + .slice(0, 32); + return raw || fallback; +} + +function getDiagramDb(cfg) { + const key = cfg.diagramDbFile; + const cached = diagramDbCache.get(key); + if (cached) return cached; + + ensureParentDir(key); + const db = new DatabaseSync(key); + db.exec(` + PRAGMA journal_mode = WAL; + CREATE TABLE IF NOT EXISTS diagram_layout ( + view TEXT PRIMARY KEY, + layout_json TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS diagram_debug_event ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL, + view TEXT NOT NULL, + event TEXT NOT NULL, + node_id TEXT, + viewport_json TEXT, + payload_json TEXT + ); + CREATE TABLE IF NOT EXISTS diagram_layout_revision ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + view TEXT NOT NULL, + updated_at TEXT NOT NULL, + action TEXT NOT NULL, + layout_json TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_diagram_debug_event_ts ON diagram_debug_event(ts); + CREATE INDEX IF NOT EXISTS idx_diagram_layout_revision_view_id + ON diagram_layout_revision(view, id DESC); + `); + diagramDbCache.set(key, db); + return db; +} + +function readDiagramLayoutStore(cfg) { + const db = getDiagramDb(cfg); + const rows = db.prepare('SELECT view, layout_json, updated_at FROM diagram_layout').all(); + const store = { + ok: true, + version: 1, + updatedAt: undefined, + views: { + standard: {}, + ultra: {}, + }, + }; + + for (const row of rows) { + const view = normalizeDiagramView(row.view); + let layout = {}; + try { + layout = normalizeDiagramLayout(JSON.parse(String(row.layout_json || '{}'))); + } catch { + layout = {}; + } + store.views[view] = layout; + if (typeof row.updated_at === 'string' && (!store.updatedAt || row.updated_at > store.updatedAt)) { + store.updatedAt = row.updated_at; + } + } + + return store; +} + +function writeDiagramLayoutStore(cfg, store, views = ['standard', 'ultra']) { + const db = getDiagramDb(cfg); + const updatedAt = store.updatedAt || getIsoNow(); + const upsert = db.prepare(` + INSERT INTO diagram_layout(view, layout_json, updated_at) + VALUES (?, ?, ?) + ON CONFLICT(view) DO UPDATE SET + layout_json = excluded.layout_json, + updated_at = excluded.updated_at + `); + + for (const rawView of views) { + const view = normalizeDiagramView(rawView); + upsert.run( + view, + JSON.stringify(normalizeDiagramLayout(store.views?.[view])), + updatedAt + ); + } +} + +function snapshotDiagramLayoutRevision(cfg, { view, updatedAt, layout, action }) { + const normalizedView = normalizeDiagramView(view); + const normalizedLayout = normalizeDiagramLayout(layout); + const normalizedAction = normalizeDiagramLayoutAction(action); + const ts = String(updatedAt || getIsoNow()); + const json = JSON.stringify(normalizedLayout); + const db = getDiagramDb(cfg); + const last = db + .prepare( + ` + SELECT id, layout_json, action + FROM diagram_layout_revision + WHERE view = ? + ORDER BY id DESC + LIMIT 1 + ` + ) + .get(normalizedView); + + if (last && String(last.layout_json || '') === json && String(last.action || '') === normalizedAction) { + return Number(last.id) || null; + } + + db.prepare( + ` + INSERT INTO diagram_layout_revision(view, updated_at, action, layout_json) + VALUES (?, ?, ?, ?) + ` + ).run(normalizedView, ts, normalizedAction, json); + + const revisionId = Number(db.prepare('SELECT last_insert_rowid() AS id').get()?.id); + + db.prepare( + ` + DELETE FROM diagram_layout_revision + WHERE view = ? + AND id NOT IN ( + SELECT id + FROM diagram_layout_revision + WHERE view = ? + ORDER BY id DESC + LIMIT ? + ) + ` + ).run(normalizedView, normalizedView, DIAGRAM_LAYOUT_REVISION_LIMIT); + + return Number.isFinite(revisionId) ? revisionId : null; +} + +function readCurrentDiagramLayoutRevisionId(cfg, view) { + const row = getDiagramDb(cfg) + .prepare( + ` + SELECT id + FROM diagram_layout_revision + WHERE view = ? + ORDER BY id DESC + LIMIT 1 + ` + ) + .get(normalizeDiagramView(view)); + const id = Number(row?.id); + return Number.isFinite(id) ? id : null; +} + +function readDiagramLayoutHistory(cfg, view, limit = 12) { + const normalizedView = normalizeDiagramView(view); + const clampedLimit = Math.max(1, Math.min(40, Number.parseInt(String(limit || 12), 10) || 12)); + const rows = getDiagramDb(cfg) + .prepare( + ` + SELECT id, updated_at, action, layout_json + FROM diagram_layout_revision + WHERE view = ? + ORDER BY id DESC + LIMIT ? + ` + ) + .all(normalizedView, clampedLimit); + + return rows.map((row) => { + let layout = {}; + try { + layout = normalizeDiagramLayout(JSON.parse(String(row.layout_json || '{}'))); + } catch { + layout = {}; + } + return { + id: Number(row.id), + view: normalizedView, + updatedAt: String(row.updated_at || ''), + action: normalizeDiagramLayoutAction(row.action, 'save'), + entryCount: Object.keys(layout).length, + }; + }); +} + +function readDiagramLayoutRevision(cfg, view, revisionId) { + const normalizedView = normalizeDiagramView(view); + const normalizedRevisionId = Number.parseInt(String(revisionId || ''), 10); + if (!Number.isInteger(normalizedRevisionId) || normalizedRevisionId <= 0) return null; + + const row = getDiagramDb(cfg) + .prepare( + ` + SELECT id, updated_at, action, layout_json + FROM diagram_layout_revision + WHERE view = ? AND id = ? + LIMIT 1 + ` + ) + .get(normalizedView, normalizedRevisionId); + if (!row) return null; + + let layout = {}; + try { + layout = normalizeDiagramLayout(JSON.parse(String(row.layout_json || '{}'))); + } catch { + layout = {}; + } + + return { + id: Number(row.id), + view: normalizedView, + updatedAt: String(row.updated_at || ''), + action: normalizeDiagramLayoutAction(row.action, 'save'), + entryCount: Object.keys(layout).length, + layout, + }; +} + +function appendDiagramDebugEvent(cfg, event) { + const db = getDiagramDb(cfg); + db.prepare(` + INSERT INTO diagram_debug_event(ts, view, event, node_id, viewport_json, payload_json) + VALUES (?, ?, ?, ?, ?, ?) + `).run( + String(event.ts || getIsoNow()), + normalizeDiagramView(event.view), + String(event.event || 'unknown').slice(0, 120), + event.nodeId == null ? null : String(event.nodeId), + event.viewport ? JSON.stringify(event.viewport) : null, + event.payload ? JSON.stringify(event.payload) : null + ); +} + function normalizeScopes(value) { if (!value) return []; if (Array.isArray(value)) return value.map((v) => String(v)).filter(Boolean); @@ -181,7 +620,29 @@ function normalizeScopes(value) { return []; } +function errorMessage(err) { + return String(err?.message || err || ''); +} + +function isMissingHasuraField(err, fieldName) { + const message = errorMessage(err); + return ( + message.includes(`field '${fieldName}' not found in type: 'query_root'`) || + message.includes(`field '${fieldName}' not found in type: 'mutation_root'`) + ); +} + +function isBotControlPlaneSchemaMissing(err) { + return ['bot_config', 'bot_config_by_pk', 'bot_state_by_pk', 'bot_events'].some((fieldName) => + isMissingHasuraField(err, fieldName) + ); +} + async function requireValidToken(cfg, req, requiredScope) { + if (isAdmin(cfg, req)) { + return { ok: true, token: { id: 'admin', name: 'admin' } }; + } + const token = readBearerToken(req); if (!token) return { ok: false, status: 401, error: 'missing_token' }; @@ -347,8 +808,8 @@ function clampInt(value, min, max) { return Math.min(max, Math.max(min, n)); } -function clampNumber(value, min, max, fallback = min) { - const n = typeof value === 'number' ? value : typeof value === 'string' ? Number(value) : NaN; +function clampNumber(value, min, max, fallback) { + const n = typeof value === 'number' ? value : Number(value); if (!Number.isFinite(n)) return fallback; return Math.min(max, Math.max(min, n)); } @@ -357,26 +818,177 @@ function parseNumeric(value) { if (value == null) return null; if (typeof value === 'number') return Number.isFinite(value) ? value : null; if (typeof value === 'string') { - const t = value.trim(); - if (!t) return null; - const n = Number(t); + const s = value.trim(); + if (!s) return null; + const n = Number(s); return Number.isFinite(n) ? n : null; } - if (typeof value?.toString === 'function') { - try { - const n = Number(String(value.toString()).trim()); - return Number.isFinite(n) ? n : null; - } catch { - return null; - } - } return null; } -function clampSeriesLimit(value, fallback) { - const n = Number.parseInt(String(value ?? ''), 10); - if (!Number.isFinite(n) || !Number.isInteger(n)) return fallback; - return Math.min(10_000, Math.max(1, n)); +function tsToUnixSeconds(value) { + if (typeof value === 'number') return Number.isFinite(value) ? value : null; + if (typeof value !== 'string') return null; + const ms = Date.parse(value); + if (!Number.isFinite(ms)) return null; + return Math.floor(ms / 1000); +} + +function flowFromDelta(delta) { + if (delta > 0) return { up: 1, down: 0, flat: 0 }; + if (delta < 0) return { up: 0, down: 1, flat: 0 }; + return { up: 0, down: 0, flat: 1 }; +} + +function computeCandleFlowFromTicks({ candle, bucketSeconds, points, nowSec, isCurrent }) { + const start = candle.time; + const end = start + bucketSeconds; + const progressEnd = isCurrent ? Math.min(end, Math.max(start, nowSec)) : end; + + const totalWindow = progressEnd - start; + if (!(totalWindow > 0)) return flowFromDelta(candle.close - candle.open); + + let up = 0; + let down = 0; + let flat = 0; + + let prevT = start; + let prevP = candle.open; + + for (const pt of points || []) { + const t = clampNumber(pt.t, prevT, progressEnd, prevT); + const dt = t - prevT; + if (dt > 0) { + const delta = pt.p - prevP; + if (delta > 0) up += dt; + else if (delta < 0) down += dt; + else flat += dt; + prevT = t; + } + prevP = pt.p; + if (prevT >= progressEnd) break; + } + + const tail = progressEnd - prevT; + if (tail > 0) flat += tail; + + const sum = up + down + flat; + if (!(sum > 0)) return flowFromDelta(candle.close - candle.open); + return { up: up / sum, down: down / sum, flat: flat / sum }; +} + +function computeCandleFlowRowsFromTicks({ candle, bucketSeconds, points, rows, nowSec, isCurrent }) { + const start = candle.time; + const end = start + bucketSeconds; + const progressEnd = isCurrent ? Math.min(end, Math.max(start, nowSec)) : end; + const totalWindow = progressEnd - start; + if (!(totalWindow > 0)) { + const overall = candle.close - candle.open; + const dir = overall > 0 ? 1 : overall < 0 ? -1 : 0; + return new Array(rows).fill(dir); + } + + const rowDirs = new Array(rows).fill(0); + + const pts = [{ t: start, p: candle.open }, ...(points || []), { t: progressEnd, p: candle.close }]; + for (let i = 1; i < pts.length; i += 1) { + const a = pts[i - 1]; + const b = pts[i]; + const t1 = clampNumber(a.t, start, progressEnd, start); + const t2 = clampNumber(b.t, start, progressEnd, start); + if (!(t2 > t1)) continue; + + const delta = b.p - a.p; + const dir = delta > 0 ? 1 : delta < 0 ? -1 : 0; + + const from = Math.max(0, Math.min(rows - 1, Math.floor(((t1 - start) / bucketSeconds) * rows))); + const to = Math.max(0, Math.min(rows - 1, Math.floor(((t2 - start) / bucketSeconds) * rows))); + for (let r = from; r <= to; r += 1) rowDirs[r] = dir; + } + + return rowDirs; +} + +function computeCandleFlowSlicesFromTicks({ candle, bucketSeconds, points, rows, nowSec, isCurrent }) { + const start = candle.time; + const end = start + bucketSeconds; + const progressEnd = isCurrent ? Math.min(end, Math.max(start, nowSec)) : end; + const dirs = new Array(rows).fill(0); + const moves = new Array(rows).fill(0); + + const totalWindow = progressEnd - start; + if (!(totalWindow > 0)) { + const overall = candle.close - candle.open; + const dir = overall > 0 ? 1 : overall < 0 ? -1 : 0; + dirs.fill(dir); + return { dirs, moves }; + } + + const pts = [...(points || []), { t: progressEnd, p: candle.close }]; + let idx = 0; + let lastP = candle.open; + + for (let r = 0; r < rows; r += 1) { + const sliceStart = start + (r / rows) * bucketSeconds; + if (!(sliceStart < progressEnd)) break; + const sliceEnd = Math.min(progressEnd, start + ((r + 1) / rows) * bucketSeconds); + + while (idx < pts.length && pts[idx].t <= sliceStart) { + lastP = pts[idx].p; + idx += 1; + } + const pStart = lastP; + + while (idx < pts.length && pts[idx].t <= sliceEnd) { + lastP = pts[idx].p; + idx += 1; + } + const pEnd = lastP; + + const delta = pEnd - pStart; + const dir = delta > 0 ? 1 : delta < 0 ? -1 : 0; + const move = Math.abs(delta); + dirs[r] = dir; + moves[r] = Number.isFinite(move) ? move : 0; + } + + return { dirs, moves }; +} + +function computeCandleFlowMovesFromTicks({ candle, bucketSeconds, points, rows, nowSec, isCurrent }) { + const start = candle.time; + const end = start + bucketSeconds; + const progressEnd = isCurrent ? Math.min(end, Math.max(start, nowSec)) : end; + const totalWindow = progressEnd - start; + const out = new Array(rows).fill(0); + if (!(totalWindow > 0)) return out; + + const pts = [...(points || []), { t: progressEnd, p: candle.close }]; + let idx = 0; + let lastP = candle.open; + + for (let r = 0; r < rows; r += 1) { + const sliceStart = start + (r / rows) * bucketSeconds; + if (!(sliceStart < progressEnd)) break; + const sliceEnd = Math.min(progressEnd, start + ((r + 1) / rows) * bucketSeconds); + + while (idx < pts.length && pts[idx].t <= sliceStart) { + lastP = pts[idx].p; + idx += 1; + } + const pStart = lastP; + + while (idx < pts.length && pts[idx].t <= sliceEnd) { + lastP = pts[idx].p; + idx += 1; + } + const pEnd = lastP; + + const move = Math.abs(pEnd - pStart); + out[r] = Number.isFinite(move) ? move : 0; + } + + return out; } function parseTimeframeToSeconds(tf) { @@ -391,43 +1003,6 @@ function parseTimeframeToSeconds(tf) { return num * mult; } -function isTruthy(value) { - if (typeof value === 'boolean') return value; - const v = String(value ?? '').trim().toLowerCase(); - return v === '1' || v === 'true' || v === 'yes' || v === 'on'; -} - -const solPriceCache = { ts: 0, value: null }; - -async function readSolPriceUsd(cfg) { - const now = Date.now(); - if (now - solPriceCache.ts < 5000) return solPriceCache.value; - - const table = cfg.ticksTable; - const query = ` - query SolPrice($where: ${table}_bool_exp!) { - ${table}(where: $where, order_by: {ts: desc}, limit: 1) { - oracle_price - mark_price - ts - } - } - `; - - try { - const data = await hasuraRequest(cfg, { admin: true }, query, { where: { symbol: { _eq: 'SOL-PERP' } } }); - const row = (data?.[table] || [])[0]; - const price = parseNumeric(row?.oracle_price) ?? parseNumeric(row?.mark_price) ?? null; - solPriceCache.ts = now; - solPriceCache.value = price; - return price; - } catch { - solPriceCache.ts = now; - solPriceCache.value = null; - return null; - } -} - function mean(values) { if (!values.length) return 0; return values.reduce((a, b) => a + b, 0) / values.length; @@ -565,6 +1140,20 @@ function isAdmin(cfg, req) { return typeof provided === 'string' && provided === cfg.apiAdminSecret; } +function isUuidLike(value) { + const s = String(value || '').trim(); + return /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(s); +} + +function normalizeBotMode(value) { + const v = String(value || '') + .trim() + .toLowerCase(); + if (!v) return undefined; + if (!['off', 'observe', 'trade', 'panic'].includes(v)) throw new Error(`invalid_mode: ${value}`); + return v; +} + async function handler(cfg, req, res) { if (req.method === 'OPTIONS') { withCors(res, cfg.corsOrigin); @@ -575,6 +1164,7 @@ async function handler(cfg, req, res) { const url = new URL(req.url || '/', `http://${req.headers.host || 'localhost'}`); const pathname = url.pathname; + const segments = pathname.split('/').filter(Boolean); if (req.method === 'GET' && pathname === '/healthz') { sendJson( @@ -593,6 +1183,222 @@ async function handler(cfg, req, res) { return; } + if (req.method === 'GET' && pathname === '/v1/runtime/diagram') { + const auth = await requireValidToken(cfg, req, 'read'); + if (!auth.ok) { + sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); + return; + } + + try { + const runtime = await readDiagramRuntime(cfg); + sendJson(res, 200, runtime, cfg.corsOrigin); + } catch (err) { + sendJson(res, 500, { ok: false, error: String(err?.message || err) }, cfg.corsOrigin); + } + return; + } + + if (req.method === 'GET' && pathname === '/v1/diagram/layout') { + const auth = await requireValidToken(cfg, req, 'read'); + if (!auth.ok) { + sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); + return; + } + + try { + const view = normalizeDiagramView(url.searchParams.get('view')); + const store = readDiagramLayoutStore(cfg); + sendJson( + res, + 200, + { + ok: true, + view, + updatedAt: store.updatedAt, + layout: store.views[view] || {}, + currentRevisionId: readCurrentDiagramLayoutRevisionId(cfg, view), + }, + cfg.corsOrigin + ); + } catch (err) { + sendJson(res, 500, { ok: false, error: String(err?.message || err) }, cfg.corsOrigin); + } + return; + } + + if (req.method === 'PATCH' && pathname === '/v1/diagram/layout') { + const auth = await requireValidToken(cfg, req, 'read'); + if (!auth.ok) { + sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); + return; + } + + try { + const body = await readBodyJson(req, { maxBytes: 512 * 1024 }); + const view = normalizeDiagramView(body.view || url.searchParams.get('view')); + const layout = normalizeDiagramLayout(body.layout); + const action = normalizeDiagramLayoutAction(body.action, 'save'); + const updatedAt = getIsoNow(); + const store = readDiagramLayoutStore(cfg); + store.updatedAt = updatedAt; + store.views[view] = layout; + writeDiagramLayoutStore(cfg, store, [view]); + const revisionId = snapshotDiagramLayoutRevision(cfg, { view, updatedAt, layout, action }); + sendJson(res, 200, { ok: true, view, updatedAt, layout, revisionId }, cfg.corsOrigin); + } catch (err) { + const msg = String(err?.message || err); + const status = msg === 'payload_too_large' ? 413 : msg === 'invalid_json' ? 400 : 500; + sendJson(res, status, { ok: false, error: msg }, cfg.corsOrigin); + } + return; + } + + if (req.method === 'DELETE' && pathname === '/v1/diagram/layout') { + const auth = await requireValidToken(cfg, req, 'read'); + if (!auth.ok) { + sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); + return; + } + + try { + const view = normalizeDiagramView(url.searchParams.get('view')); + const updatedAt = getIsoNow(); + const store = readDiagramLayoutStore(cfg); + store.updatedAt = updatedAt; + store.views[view] = {}; + writeDiagramLayoutStore(cfg, store, [view]); + const revisionId = snapshotDiagramLayoutRevision(cfg, { + view, + updatedAt, + layout: {}, + action: 'reset', + }); + sendJson(res, 200, { ok: true, view, updatedAt, revisionId }, cfg.corsOrigin); + } catch (err) { + sendJson(res, 500, { ok: false, error: String(err?.message || err) }, cfg.corsOrigin); + } + return; + } + + if (req.method === 'GET' && pathname === '/v1/diagram/layout/history') { + const auth = await requireValidToken(cfg, req, 'read'); + if (!auth.ok) { + sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); + return; + } + + try { + const view = normalizeDiagramView(url.searchParams.get('view')); + const revisions = readDiagramLayoutHistory(cfg, view, url.searchParams.get('limit')); + sendJson( + res, + 200, + { + ok: true, + view, + currentRevisionId: readCurrentDiagramLayoutRevisionId(cfg, view), + revisions, + }, + cfg.corsOrigin + ); + } catch (err) { + sendJson(res, 500, { ok: false, error: String(err?.message || err) }, cfg.corsOrigin); + } + return; + } + + if (req.method === 'POST' && pathname === '/v1/diagram/layout/restore') { + const auth = await requireValidToken(cfg, req, 'read'); + if (!auth.ok) { + sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); + return; + } + + try { + const body = await readBodyJson(req, { maxBytes: 64 * 1024 }); + const view = normalizeDiagramView(body.view || url.searchParams.get('view')); + const revisionId = Number.parseInt(String(body.revisionId || ''), 10); + if (!Number.isInteger(revisionId) || revisionId <= 0) { + sendJson(res, 400, { ok: false, error: 'invalid_revision_id' }, cfg.corsOrigin); + return; + } + + const revision = readDiagramLayoutRevision(cfg, view, revisionId); + if (!revision) { + sendJson(res, 404, { ok: false, error: 'revision_not_found' }, cfg.corsOrigin); + return; + } + + const updatedAt = getIsoNow(); + const store = readDiagramLayoutStore(cfg); + store.updatedAt = updatedAt; + store.views[view] = revision.layout; + writeDiagramLayoutStore(cfg, store, [view]); + const restoredRevisionId = snapshotDiagramLayoutRevision(cfg, { + view, + updatedAt, + layout: revision.layout, + action: 'restore', + }); + + sendJson( + res, + 200, + { + ok: true, + view, + updatedAt, + layout: revision.layout, + revisionId: restoredRevisionId, + restoredFromRevisionId: revisionId, + }, + cfg.corsOrigin + ); + } catch (err) { + const msg = String(err?.message || err); + const status = msg === 'payload_too_large' ? 413 : msg === 'invalid_json' ? 400 : 500; + sendJson(res, status, { ok: false, error: msg }, cfg.corsOrigin); + } + return; + } + + if (req.method === 'POST' && pathname === '/v1/diagram/debug-event') { + const auth = await requireValidToken(cfg, req, 'read'); + if (!auth.ok) { + sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); + return; + } + + try { + const body = await readBodyJson(req, { maxBytes: 64 * 1024 }); + const event = { + ts: getIsoNow(), + view: normalizeDiagramView(body.view), + event: String(body.event || 'unknown').slice(0, 120), + nodeId: body.nodeId == null ? undefined : String(body.nodeId), + viewport: + body.viewport && typeof body.viewport === 'object' + ? { + x: Number.isFinite(Number(body.viewport.x)) ? Number(body.viewport.x) : undefined, + y: Number.isFinite(Number(body.viewport.y)) ? Number(body.viewport.y) : undefined, + zoom: Number.isFinite(Number(body.viewport.zoom)) ? Number(body.viewport.zoom) : undefined, + } + : undefined, + payload: body.payload && typeof body.payload === 'object' ? body.payload : undefined, + }; + + console.log('[diagram-debug]', JSON.stringify(event)); + appendDiagramDebugEvent(cfg, event); + sendJson(res, 200, { ok: true, loggedAt: event.ts }, cfg.corsOrigin); + } catch (err) { + const msg = String(err?.message || err); + const status = msg === 'payload_too_large' ? 413 : msg === 'invalid_json' ? 400 : 500; + sendJson(res, status, { ok: false, error: msg }, cfg.corsOrigin); + } + return; + } + if (req.method === 'GET' && pathname === '/v1/chart') { const auth = await requireValidToken(cfg, req, 'read'); if (!auth.ok) { @@ -642,24 +1448,114 @@ async function handler(cfg, req, res) { }); const rows = data?.[fn] || []; - const candles = rows - .slice() - .reverse() - .map((r) => { - const time = Math.floor(Date.parse(r.bucket) / 1000); - const open = Number(r.open); - const high = Number(r.high); - const low = Number(r.low); - const close = Number(r.close); - const oracle = r.oracle_close == null ? null : Number(r.oracle_close); - const volume = Number(r.ticks || 0); - return { time, open, high, low, close, volume, oracle }; - }) - .filter((c) => Number.isFinite(c.time) && Number.isFinite(c.open) && Number.isFinite(c.close)); + const candles = rows + .slice() + .reverse() + .map((r) => { + const time = Math.floor(Date.parse(r.bucket) / 1000); + const open = Number(r.open); + const high = Number(r.high); + const low = Number(r.low); + const close = Number(r.close); + const oracle = r.oracle_close == null ? null : Number(r.oracle_close); + const volume = Number(r.ticks || 0); + return { time, open, high, low, close, volume, oracle }; + }) + .filter((c) => Number.isFinite(c.time) && Number.isFinite(c.open) && Number.isFinite(c.close)); - const times = candles.map((c) => c.time); - const closes = candles.map((c) => c.close); - const oracleSeries = toSeries(times, candles.map((c) => (c.oracle == null ? null : c.oracle))); + // Flow = share of time spent moving up/down/flat inside each bucket. + // Used by the UI to render stacked volume bars describing microstructure. + const nowSec = Math.floor(Date.now() / 1000); + const windowSeconds = bucketSeconds * candles.length; + const canComputeFlow = candles.length > 0 && windowSeconds > 0 && windowSeconds <= 86_400; // cap at 24h + + if (canComputeFlow) { + const firstStart = candles[0].time; + const lastStart = candles[candles.length - 1].time; + const fromIso = new Date(firstStart * 1000).toISOString(); + const toIso = new Date((lastStart + bucketSeconds) * 1000).toISOString(); + + const table = cfg.ticksTable; + const visibleStarts = new Set(candles.map((c) => c.time)); + const pointsByCandle = new Map(); + + const maxTicks = Math.min(200_000, Math.max(5_000, windowSeconds * 20)); + const wantSource = Boolean(source && source.trim()); + const vars = wantSource + ? { symbol, from: fromIso, to: toIso, limit: maxTicks, source: source.trim() } + : { symbol, from: fromIso, to: toIso, limit: maxTicks }; + + const query = ` + query FlowTicks($symbol: String!, $from: timestamptz!, $to: timestamptz!, $limit: Int!${wantSource ? ', $source: String!' : ''}) { + ${table}( + where: {symbol: {_eq: $symbol}, ts: {_gte: $from, _lt: $to}${wantSource ? ', source: {_eq: $source}' : ''}} + order_by: {ts: asc} + limit: $limit + ) { + ts + oracle_price + mark_price + } + } + `; + + try { + const tickData = await hasuraRequest(cfg, { admin: true }, query, vars); + const ticks = tickData?.[table] || []; + for (const r of ticks) { + const t = tsToUnixSeconds(r.ts); + if (t == null) continue; + const p = parseNumeric(r.mark_price) ?? parseNumeric(r.oracle_price); + if (p == null) continue; + const idx = Math.floor((t - firstStart) / bucketSeconds); + const start = firstStart + idx * bucketSeconds; + if (!visibleStarts.has(start)) continue; + const list = pointsByCandle.get(start) || []; + list.push({ t, p }); + pointsByCandle.set(start, list); + } + + const lastCandleTime = candles[candles.length - 1]?.time ?? null; + for (const c of candles) { + const pts = pointsByCandle.get(c.time) || []; + const isCurrent = lastCandleTime != null && c.time === lastCandleTime; + c.flow = computeCandleFlowFromTicks({ candle: c, bucketSeconds, points: pts, nowSec, isCurrent }); + const rows = Math.min(60, Math.max(12, Math.floor(bucketSeconds))); + const slices = computeCandleFlowSlicesFromTicks({ + candle: c, + bucketSeconds, + points: pts, + rows, + nowSec, + isCurrent, + }); + c.flowRows = slices.dirs; + c.flowMoves = slices.moves; + } + } catch { + for (const c of candles) { + const fallback = flowFromDelta(c.close - c.open); + c.flow = fallback; + const rows = Math.min(60, Math.max(12, Math.floor(bucketSeconds))); + const dir = fallback.up ? 1 : fallback.down ? -1 : 0; + c.flowRows = new Array(rows).fill(dir); + c.flowMoves = new Array(rows).fill(0); + } + } + } else { + for (const c of candles) { + const fallback = flowFromDelta(c.close - c.open); + c.flow = fallback; + const rows = Math.min(60, Math.max(12, Math.floor(bucketSeconds))); + const dir = fallback.up ? 1 : fallback.down ? -1 : 0; + c.flowRows = new Array(rows).fill(dir); + c.flowMoves = new Array(rows).fill(0); + } + } + + const times = candles.map((c) => c.time); + const closes = candles.map((c) => c.close); + const oracleSeries = toSeries(times, candles.map((c) => (c.oracle == null ? null : c.oracle))); const sma20 = toSeries(times, sma(closes, 20)); const ema20 = toSeries(times, ema(closes, 20)); @@ -703,329 +1599,6 @@ async function handler(cfg, req, res) { return; } - // Estimate costs for a new contract (read scope). - // Uses DLOB slippage table and simple fee/tx estimates (inputs). - if (req.method === 'POST' && pathname === '/v1/contracts/costs/estimate') { - const auth = await requireValidToken(cfg, req, 'read'); - if (!auth.ok) { - sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); - return; - } - - let body; - try { - body = await readBodyJson(req, { maxBytes: 256 * 1024 }); - } catch (err) { - const msg = String(err?.message || err); - if (msg === 'payload_too_large') { - sendJson(res, 413, { ok: false, error: 'payload_too_large' }, cfg.corsOrigin); - return; - } - sendJson(res, 400, { ok: false, error: 'invalid_json' }, cfg.corsOrigin); - return; - } - - const market = String(body?.market_name || body?.market || '').trim(); - if (!market) { - sendJson(res, 400, { ok: false, error: 'missing_market' }, cfg.corsOrigin); - return; - } - - const notionalUsd = parseNumeric(body?.notional_usd ?? body?.notionalUsd ?? body?.size_usd ?? body?.sizeUsd); - if (notionalUsd == null || !(notionalUsd > 0)) { - sendJson(res, 400, { ok: false, error: 'invalid_notional_usd' }, cfg.corsOrigin); - return; - } - - const sideRaw = String(body?.side || 'long').trim().toLowerCase(); - const entrySide = sideRaw === 'short' || sideRaw === 'sell' ? 'short' : 'long'; - - const defaultTakerBps = parseNumeric(process.env.FEE_TAKER_BPS_DEFAULT) ?? 5; - const defaultMakerBps = parseNumeric(process.env.FEE_MAKER_BPS_DEFAULT) ?? 0; - const takerBps = clampNumber(parseNumeric(body?.fee_taker_bps) ?? defaultTakerBps, 0, 1000, defaultTakerBps); - const makerBps = clampNumber(parseNumeric(body?.fee_maker_bps) ?? defaultMakerBps, -1000, 1000, defaultMakerBps); - - const orderType = String(body?.order_type || body?.orderType || 'market').trim().toLowerCase(); - const isMarket = orderType === 'market' || orderType === 'taker'; - const feeBps = isMarket ? takerBps : makerBps; - - let txFeeUsdEst = parseNumeric(body?.tx_fee_usd_est); - if (txFeeUsdEst == null) { - const baseLamports = parseNumeric(process.env.TX_BASE_FEE_LAMPORTS_EST) ?? 5000; - const sigs = parseNumeric(process.env.TX_SIGNATURES_EST) ?? 1; - const priorityLamports = parseNumeric(process.env.TX_PRIORITY_FEE_LAMPORTS_EST) ?? 0; - const lamports = Math.max(0, baseLamports) * Math.max(1, sigs) + Math.max(0, priorityLamports); - const sol = lamports / 1_000_000_000; - const solUsd = await readSolPriceUsd(cfg); - txFeeUsdEst = solUsd != null ? sol * solUsd : 0; - } - txFeeUsdEst = clampNumber(txFeeUsdEst, 0, 100, 0); - - const defaultReprices = parseNumeric(process.env.EXPECTED_REPRICES_PER_ENTRY_DEFAULT) ?? 0; - const expectedReprices = clampInt(body?.expected_reprices_per_entry ?? body?.expectedReprices ?? String(defaultReprices), 0, 500); - const modifyTxCount = clampInt(body?.modify_tx_count ?? body?.modifyTxCount ?? '2', 0, 10); - - const wantedSide = entrySide === 'long' ? 'buy' : 'sell'; - - const pickNearest = (rows) => { - const candidates = (rows || []).filter((r) => String(r.side || '').toLowerCase() === wantedSide); - let best = null; - let bestD = Number.POSITIVE_INFINITY; - for (const r of candidates) { - const s = parseNumeric(r.size_usd); - if (s == null) continue; - const d = Math.abs(s - notionalUsd); - if (d < bestD) { - bestD = d; - best = r; - } - } - return best; - }; - - const fetchSlippageRows = async (tableName) => { - const query = ` - query Slippage($market: String!, $limit: Int!) { - ${tableName}(where: {market_name: {_eq: $market}}, order_by: {size_usd: asc}, limit: $limit) { - market_name - side - size_usd - mid_price - vwap_price - worst_price - impact_bps - fill_pct - updated_at - } - } - `; - const data = await hasuraRequest(cfg, { admin: true }, query, { market, limit: 500 }); - return data?.[tableName] || []; - }; - - try { - let rows = []; - let usedTable = null; - try { - rows = await fetchSlippageRows('dlob_slippage_latest_v2'); - usedTable = 'dlob_slippage_latest_v2'; - } catch (e) { - rows = await fetchSlippageRows('dlob_slippage_latest'); - usedTable = 'dlob_slippage_latest'; - } - - const best = pickNearest(rows); - const impactBps = parseNumeric(best?.impact_bps) ?? 0; - const slippageUsd = (notionalUsd * impactBps) / 10_000; - const tradeFeeUsd = (notionalUsd * feeBps) / 10_000; - const modifyCostUsd = expectedReprices * modifyTxCount * txFeeUsdEst; - const totalUsd = tradeFeeUsd + slippageUsd + txFeeUsdEst + modifyCostUsd; - const totalBps = (totalUsd / notionalUsd) * 10_000; - - sendJson( - res, - 200, - { - ok: true, - input: { - market_name: market, - notional_usd: notionalUsd, - side: entrySide, - order_type: orderType, - fee_bps: feeBps, - tx_fee_usd_est: txFeeUsdEst, - expected_reprices_per_entry: expectedReprices, - }, - dlob: best - ? { - table: usedTable, - size_usd: best.size_usd, - side: best.side, - mid_price: best.mid_price, - vwap_price: best.vwap_price, - impact_bps: best.impact_bps, - fill_pct: best.fill_pct, - updated_at: best.updated_at, - } - : null, - breakdown: { - trade_fee_usd: tradeFeeUsd, - slippage_usd: slippageUsd, - tx_fee_usd: txFeeUsdEst, - expected_modify_usd: modifyCostUsd, - total_usd: totalUsd, - total_bps: totalBps, - breakeven_bps: totalBps, - }, - }, - cfg.corsOrigin - ); - } catch (err) { - sendJson(res, 500, { ok: false, error: String(err?.message || err) }, cfg.corsOrigin); - } - return; - } - - // Monitor a pushed contract (read scope). - // Best-effort: if bot tables are not present yet in Hasura schema, return a clear error. - const monitorMatch = pathname.match(/^\/v1\/contracts\/([0-9a-f-]{36})\/monitor$/i); - if (req.method === 'GET' && monitorMatch) { - const auth = await requireValidToken(cfg, req, 'read'); - if (!auth.ok) { - sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); - return; - } - - const contractId = monitorMatch[1]; - const eventsLimit = clampSeriesLimit(url.searchParams.get('eventsLimit') || '2000', 2000); - const wantSeries = isTruthy(url.searchParams.get('series') || ''); - const seriesMax = clampSeriesLimit(url.searchParams.get('seriesMax') || '600', 600); - - const pickNum = (obj, keys) => { - if (!obj || typeof obj !== 'object') return 0; - for (const k of keys) { - const v = obj[k]; - const n = parseNumeric(v); - if (n != null) return n; - } - return 0; - }; - - try { - const q = ` - query ContractMonitor($id: uuid!, $limit: Int!) { - bot_contracts_by_pk(id: $id) { id } - bot_events(where: {contract_id: {_eq: $id}}, order_by: {created_at: asc}, limit: $limit) { - id - created_at - type - payload - } - } - `; - const data = await hasuraRequest(cfg, { admin: true }, q, { id: contractId, limit: eventsLimit }); - const contract = data?.bot_contracts_by_pk || null; - const events = Array.isArray(data?.bot_events) ? data.bot_events : []; - - const costs = { - tradeFeeUsd: 0, - txFeeUsd: 0, - slippageUsd: 0, - fundingUsd: 0, - realizedPnlUsd: 0, - totalCostsUsd: 0, - netPnlUsd: 0, - txCount: 0, - fillCount: 0, - cancelCount: 0, - modifyCount: 0, - errorCount: 0, - }; - - const series = []; - let cumTrade = 0; - let cumTx = 0; - let cumSlip = 0; - let cumFunding = 0; - let cumRealized = 0; - - for (const ev of events) { - const payload = ev?.payload && typeof ev.payload === 'object' ? ev.payload : {}; - const type = String(ev?.type || '').toLowerCase(); - - const tradeFeeUsd = pickNum(payload, ['trade_fee_usd', 'tradeFeeUsd', 'fee_usd', 'feeUsd']); - const txFeeUsd = pickNum(payload, ['tx_fee_usd', 'txFeeUsd', 'tx_usd', 'txUsd']); - const slippageUsd = pickNum(payload, ['slippage_usd', 'slippageUsd', 'impact_usd', 'impactUsd']); - const fundingUsd = pickNum(payload, ['funding_usd', 'fundingUsd']); - const realizedPnlUsd = pickNum(payload, ['realized_pnl_usd', 'realizedPnlUsd', 'pnl_usd', 'pnlUsd']); - - const txCount = Math.max( - 0, - Math.round( - pickNum(payload, ['tx_count', 'txCount', 'txs', 'txs_count']) || - (txFeeUsd > 0 ? 1 : 0) - ) - ); - - cumTrade += tradeFeeUsd; - cumTx += txFeeUsd; - cumSlip += slippageUsd; - cumFunding += fundingUsd; - cumRealized += realizedPnlUsd; - - costs.tradeFeeUsd = cumTrade; - costs.txFeeUsd = cumTx; - costs.slippageUsd = cumSlip; - costs.fundingUsd = cumFunding; - costs.realizedPnlUsd = cumRealized; - costs.txCount += txCount; - - const isFill = type.includes('fill'); - const isCancel = type.includes('cancel'); - const isModify = type.includes('modify') || type.includes('replace') || type.includes('reprice'); - const isError = type.includes('error') || type.includes('panic'); - - if (isFill) costs.fillCount += 1; - if (isCancel) costs.cancelCount += 1; - if (isModify) costs.modifyCount += 1; - if (isError) costs.errorCount += 1; - - if (wantSeries) { - const totalCostsUsd = cumTrade + cumTx + cumSlip + cumFunding; - const netPnlUsd = cumRealized - totalCostsUsd; - series.push({ - ts: ev?.created_at || getIsoNow(), - tradeFeeUsd: cumTrade, - txFeeUsd: cumTx, - slippageUsd: cumSlip, - fundingUsd: cumFunding, - totalCostsUsd, - realizedPnlUsd: cumRealized, - netPnlUsd, - }); - } - } - - costs.totalCostsUsd = costs.tradeFeeUsd + costs.txFeeUsd + costs.slippageUsd + costs.fundingUsd; - costs.netPnlUsd = costs.realizedPnlUsd - costs.totalCostsUsd; - - // Downsample to seriesMax if needed (keep last point). - let outSeries = null; - if (wantSeries) { - if (series.length <= seriesMax) { - outSeries = series; - } else { - const step = Math.max(1, Math.floor(series.length / seriesMax)); - const sampled = []; - for (let i = 0; i < series.length; i += step) sampled.push(series[i]); - if (sampled[sampled.length - 1] !== series[series.length - 1]) sampled.push(series[series.length - 1]); - outSeries = sampled.slice(-seriesMax); - } - } - - sendJson( - res, - 200, - { - ok: true, - contract, - eventsCount: events.length, - costs, - series: outSeries, - }, - cfg.corsOrigin - ); - } catch (err) { - const msg = String(err?.message || err); - if (msg.includes('bot_contracts') || msg.includes('bot_events')) { - sendJson(res, 501, { ok: false, error: 'bot_tables_missing' }, cfg.corsOrigin); - return; - } - sendJson(res, 500, { ok: false, error: msg }, cfg.corsOrigin); - } - return; - } - if (req.method === 'POST' && pathname === '/v1/ingest/tick') { const auth = await requireValidToken(cfg, req, 'write'); if (!auth.ok) { @@ -1115,6 +1688,429 @@ async function handler(cfg, req, res) { return; } + // Bot control-plane (store desired state in DB; executor reads from DB). + // Endpoints intentionally DO NOT "sign & send tx". + if (segments[0] === 'v1' && segments[1] === 'bots') { + // GET /v1/bots + if (req.method === 'GET' && segments.length === 2) { + const auth = await requireValidToken(cfg, req, 'bots:read'); + if (!auth.ok) { + sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); + return; + } + + const limit = clampInt(url.searchParams.get('limit') || '100', 1, 500); + const offset = clampInt(url.searchParams.get('offset') || '0', 0, 50_000); + + const query = ` + query Bots($limit: Int!, $offset: Int!) { + bot_config(order_by: {updated_at: desc}, limit: $limit, offset: $offset) { + id + name + market_name + market_type + mode + kill_switch + params + created_at + updated_at + } + } + `; + + try { + const data = await hasuraRequest(cfg, { admin: true }, query, { limit, offset }); + sendJson(res, 200, { ok: true, bots: data?.bot_config || [] }, cfg.corsOrigin); + } catch (err) { + if (isBotControlPlaneSchemaMissing(err)) { + sendJson( + res, + 200, + { + ok: true, + bots: [], + warning: 'bot_control_plane_schema_missing', + }, + cfg.corsOrigin + ); + return; + } + sendJson(res, 500, { ok: false, error: String(err?.message || err) }, cfg.corsOrigin); + } + return; + } + + // POST /v1/bots (create) + if (req.method === 'POST' && segments.length === 2) { + const auth = await requireValidToken(cfg, req, 'bots:write'); + if (!auth.ok) { + sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); + return; + } + + let body; + try { + body = await readBodyJson(req, { maxBytes: 1024 * 1024 }); + } catch { + sendJson(res, 400, { ok: false, error: 'invalid_json' }, cfg.corsOrigin); + return; + } + + const name = (body?.name || '')?.toString?.().trim(); + const market_name = (body?.market_name || body?.marketName || '')?.toString?.().trim(); + const market_type = (body?.market_type || body?.marketType || 'perp')?.toString?.().trim() || 'perp'; + const kill_switch = Boolean(body?.kill_switch ?? body?.killSwitch ?? false); + let mode; + try { + mode = normalizeBotMode(body?.mode) || 'off'; + } catch (err) { + sendJson(res, 400, { ok: false, error: String(err?.message || err) }, cfg.corsOrigin); + return; + } + + if (!name) { + sendJson(res, 400, { ok: false, error: 'missing_name' }, cfg.corsOrigin); + return; + } + if (!market_name) { + sendJson(res, 400, { ok: false, error: 'missing_market_name' }, cfg.corsOrigin); + return; + } + + const now = new Date().toISOString(); + const object = { + name, + market_name, + market_type, + mode, + kill_switch, + params: body?.params && typeof body.params === 'object' ? body.params : null, + updated_at: now, + }; + + const mutation = ` + mutation CreateBot($object: bot_config_insert_input!) { + insert_bot_config_one(object: $object) { + id + name + market_name + market_type + mode + kill_switch + created_at + updated_at + } + } + `; + + try { + const data = await hasuraRequest(cfg, { admin: true }, mutation, { object }); + const row = data?.insert_bot_config_one; + if (!row?.id) throw new Error('bot_create_failed'); + sendJson(res, 200, { ok: true, bot: row }, cfg.corsOrigin); + } catch (err) { + sendJson(res, 500, { ok: false, error: String(err?.message || err) }, cfg.corsOrigin); + } + return; + } + + // Endpoints with bot id. + const botId = segments[2]; + if (segments.length >= 3 && !isUuidLike(botId)) { + sendJson(res, 400, { ok: false, error: 'invalid_bot_id' }, cfg.corsOrigin); + return; + } + + // GET /v1/bots/:id + if (req.method === 'GET' && segments.length === 3) { + const auth = await requireValidToken(cfg, req, 'bots:read'); + if (!auth.ok) { + sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); + return; + } + + const query = ` + query Bot($id: uuid!) { + bot_config_by_pk(id: $id) { + id + name + market_name + market_type + mode + kill_switch + params + created_at + updated_at + } + } + `; + + try { + const data = await hasuraRequest(cfg, { admin: true }, query, { id: botId }); + const bot = data?.bot_config_by_pk; + if (!bot?.id) { + sendJson(res, 404, { ok: false, error: 'bot_not_found' }, cfg.corsOrigin); + return; + } + sendJson(res, 200, { ok: true, bot }, cfg.corsOrigin); + } catch (err) { + if (isBotControlPlaneSchemaMissing(err)) { + sendJson(res, 404, { ok: false, error: 'bot_control_plane_schema_missing' }, cfg.corsOrigin); + return; + } + sendJson(res, 500, { ok: false, error: String(err?.message || err) }, cfg.corsOrigin); + } + return; + } + + // PATCH /v1/bots/:id (partial update) (POST also supported for non-rest clients) + if ((req.method === 'POST' || req.method === 'PATCH') && segments.length === 3) { + const auth = await requireValidToken(cfg, req, 'bots:write'); + if (!auth.ok) { + sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); + return; + } + + let body; + try { + body = await readBodyJson(req, { maxBytes: 1024 * 1024 }); + } catch { + sendJson(res, 400, { ok: false, error: 'invalid_json' }, cfg.corsOrigin); + return; + } + + const set = {}; + if (body?.name != null) set.name = String(body.name).trim(); + if (body?.market_name != null || body?.marketName != null) { + set.market_name = String(body.market_name ?? body.marketName).trim(); + } + if (body?.market_type != null || body?.marketType != null) { + set.market_type = String(body.market_type ?? body.marketType).trim() || 'perp'; + } + if (body?.mode != null) { + try { + set.mode = normalizeBotMode(body.mode); + } catch (err) { + sendJson(res, 400, { ok: false, error: String(err?.message || err) }, cfg.corsOrigin); + return; + } + } + if (body?.kill_switch != null || body?.killSwitch != null) { + set.kill_switch = Boolean(body.kill_switch ?? body.killSwitch); + } + if (body?.params != null) { + set.params = body.params && typeof body.params === 'object' ? body.params : null; + } + + // Clean up empty strings. + if (typeof set.name === 'string' && !set.name) delete set.name; + if (typeof set.market_name === 'string' && !set.market_name) delete set.market_name; + + if (!Object.keys(set).length) { + sendJson(res, 400, { ok: false, error: 'no_changes' }, cfg.corsOrigin); + return; + } + + set.updated_at = new Date().toISOString(); + + const mutation = ` + mutation UpdateBot($id: uuid!, $set: bot_config_set_input!) { + update_bot_config_by_pk(pk_columns: {id: $id}, _set: $set) { + id + name + market_name + market_type + mode + kill_switch + params + created_at + updated_at + } + } + `; + + try { + const data = await hasuraRequest(cfg, { admin: true }, mutation, { id: botId, set }); + const bot = data?.update_bot_config_by_pk; + if (!bot?.id) { + sendJson(res, 404, { ok: false, error: 'bot_not_found' }, cfg.corsOrigin); + return; + } + sendJson(res, 200, { ok: true, bot }, cfg.corsOrigin); + } catch (err) { + sendJson(res, 500, { ok: false, error: String(err?.message || err) }, cfg.corsOrigin); + } + return; + } + + // POST /v1/bots/:id/mode + if (req.method === 'POST' && segments.length === 4 && segments[3] === 'mode') { + const auth = await requireValidToken(cfg, req, 'bots:write'); + if (!auth.ok) { + sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); + return; + } + + let body; + try { + body = await readBodyJson(req, { maxBytes: 1024 * 1024 }); + } catch { + sendJson(res, 400, { ok: false, error: 'invalid_json' }, cfg.corsOrigin); + return; + } + + let mode; + try { + mode = normalizeBotMode(body?.mode); + } catch (err) { + sendJson(res, 400, { ok: false, error: String(err?.message || err) }, cfg.corsOrigin); + return; + } + if (!mode) { + sendJson(res, 400, { ok: false, error: 'missing_mode' }, cfg.corsOrigin); + return; + } + + const mutation = ` + mutation SetMode($id: uuid!, $mode: String!, $ts: timestamptz!) { + update_bot_config_by_pk(pk_columns: {id: $id}, _set: {mode: $mode, updated_at: $ts}) { + id + mode + updated_at + } + } + `; + + try { + const ts = new Date().toISOString(); + const data = await hasuraRequest(cfg, { admin: true }, mutation, { id: botId, mode, ts }); + const bot = data?.update_bot_config_by_pk; + if (!bot?.id) { + sendJson(res, 404, { ok: false, error: 'bot_not_found' }, cfg.corsOrigin); + return; + } + sendJson(res, 200, { ok: true, bot }, cfg.corsOrigin); + } catch (err) { + sendJson(res, 500, { ok: false, error: String(err?.message || err) }, cfg.corsOrigin); + } + return; + } + + // POST /v1/bots/:id/kill-switch + if (req.method === 'POST' && segments.length === 4 && segments[3] === 'kill-switch') { + const auth = await requireValidToken(cfg, req, 'bots:write'); + if (!auth.ok) { + sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); + return; + } + + let body; + try { + body = await readBodyJson(req, { maxBytes: 1024 * 1024 }); + } catch { + sendJson(res, 400, { ok: false, error: 'invalid_json' }, cfg.corsOrigin); + return; + } + + const enabled = body?.enabled; + if (typeof enabled !== 'boolean') { + sendJson(res, 400, { ok: false, error: 'missing_enabled' }, cfg.corsOrigin); + return; + } + + const mutation = ` + mutation SetKillSwitch($id: uuid!, $kill: Boolean!, $ts: timestamptz!) { + update_bot_config_by_pk(pk_columns: {id: $id}, _set: {kill_switch: $kill, updated_at: $ts}) { + id + kill_switch + updated_at + } + } + `; + + try { + const ts = new Date().toISOString(); + const data = await hasuraRequest(cfg, { admin: true }, mutation, { id: botId, kill: enabled, ts }); + const bot = data?.update_bot_config_by_pk; + if (!bot?.id) { + sendJson(res, 404, { ok: false, error: 'bot_not_found' }, cfg.corsOrigin); + return; + } + sendJson(res, 200, { ok: true, bot }, cfg.corsOrigin); + } catch (err) { + sendJson(res, 500, { ok: false, error: String(err?.message || err) }, cfg.corsOrigin); + } + return; + } + + // GET /v1/bots/:id/state + if (req.method === 'GET' && segments.length === 4 && segments[3] === 'state') { + const auth = await requireValidToken(cfg, req, 'bots:read'); + if (!auth.ok) { + sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); + return; + } + + const query = ` + query BotState($botId: uuid!) { + bot_state_by_pk(bot_id: $botId) { + bot_id + last_heartbeat_at + last_action_at + last_error + state + updated_at + } + } + `; + + try { + const data = await hasuraRequest(cfg, { admin: true }, query, { botId }); + sendJson(res, 200, { ok: true, state: data?.bot_state_by_pk ?? null }, cfg.corsOrigin); + } catch (err) { + if (isBotControlPlaneSchemaMissing(err)) { + sendJson(res, 200, { ok: true, state: null, warning: 'bot_control_plane_schema_missing' }, cfg.corsOrigin); + return; + } + sendJson(res, 500, { ok: false, error: String(err?.message || err) }, cfg.corsOrigin); + } + return; + } + + // GET /v1/bots/:id/events + if (req.method === 'GET' && segments.length === 4 && segments[3] === 'events') { + const auth = await requireValidToken(cfg, req, 'bots:read'); + if (!auth.ok) { + sendJson(res, auth.status, { ok: false, error: auth.error }, cfg.corsOrigin); + return; + } + + const limit = clampInt(url.searchParams.get('limit') || '200', 1, 1000); + const query = ` + query BotEvents($botId: uuid!, $limit: Int!) { + bot_events(where: {bot_id: {_eq: $botId}}, order_by: {ts: desc}, limit: $limit) { + id + ts + type + payload + } + } + `; + + try { + const data = await hasuraRequest(cfg, { admin: true }, query, { botId, limit }); + sendJson(res, 200, { ok: true, events: data?.bot_events || [] }, cfg.corsOrigin); + } catch (err) { + if (isBotControlPlaneSchemaMissing(err)) { + sendJson(res, 200, { ok: true, events: [], warning: 'bot_control_plane_schema_missing' }, cfg.corsOrigin); + return; + } + sendJson(res, 500, { ok: false, error: String(err?.message || err) }, cfg.corsOrigin); + } + return; + } + } + if (req.method === 'POST' && pathname === '/v1/admin/tokens') { if (!isAdmin(cfg, req)) { sendJson(res, 401, { ok: false, error: 'admin_unauthorized' }, cfg.corsOrigin);