1"use client";2import { useMemo, useState, type FormEvent } from "react";3import { Plus, RefreshCw, RotateCcw, XCircle } from "lucide-react";4import { api, type Dataset, type HealthReport, type Job } from "@/lib/api";5import { useAsync, useInterval } from "@/lib/hooks";6import { useToast } from "@/lib/toast";7import { useAuth } from "@/lib/auth";8import { daysAgo, fmtDuration, fmtInt, fmtTs, splitList, today } from "@/lib/format";9import { Banner, Card, Empty, ErrorState, Field, PageHeader, StatusBadge } from "@/components/ui";10import { DataTable } from "@/components/DataTable";11import { JsonView } from "@/components/JsonView";12import { ConfirmDialog, Modal } from "@/components/ConfirmDialog";1314const KINDS = ["incremental", "metadata", "backfill", "fundamentals", "derive", "compaction", "quality", "backup", "gc"] as const;15type Kind = (typeof KINDS)[number];16const STATUSES = ["", "queued", "running", "succeeded", "failed", "dead", "cancelled"];1718const KIND_HELP: Record<Kind, string> = {19 incremental: "Fetch the vendor's latest day or week of bars and append (idempotent).",20 metadata: "Refresh the instrument master, corporate actions and vendor checkpoints.",21 backfill: "Full-history download for a ticker range (first letter). Heavy on disk and bandwidth.",22 fundamentals: "Ingest point-in-time fundamentals for the given tickers (or the tracked list).",23 derive: "Rebuild derived datasets (5m / 1d) from 1-minute bars.",24 compaction: "Merge delta files into base files and publish a new dataset version.",25 quality: "Run the quality checks and record anomalies.",26 backup: "pg_dump + rsync of the lake metadata to the backup target.",27 gc: "Garbage-collect retired files and expired exports.",28};2930export default function IngestionPage() {31 const { isOwner } = useAuth();32 const toast = useToast();33 const [status, setStatus] = useState("");34 const [kind, setKind] = useState("");35 const jobs = useAsync((s) => api.get<{ data: Job[] }>("/admin/jobs", { status, kind, limit: 300 }, s), [status, kind], { enabled: isOwner });36 const dead = useAsync((s) => api.get<{ data: Job[] }>("/admin/jobs", { status: "dead", limit: 100 }, s), [], { enabled: isOwner });37 const failed = useAsync((s) => api.get<{ data: Job[] }>("/admin/jobs", { status: "failed", limit: 100 }, s), [], { enabled: isOwner });38 const datasets = useAsync((s) => api.get<{ data: Dataset[] }>("/catalog/datasets", undefined, s), []);39 const health = useAsync((s) => api.get<HealthReport>("/health", undefined, s), []);40 const [inspect, setInspect] = useState<Job | null>(null);41 const [showForm, setShowForm] = useState(false);4243 const reloadAll = () => {44 jobs.reload();45 dead.reload();46 failed.reload();47 };48 const hasLive = (jobs.data?.data ?? []).some((j) => j.status === "queued" || j.status === "running");49 useInterval(reloadAll, hasLive ? 5000 : 30000);5051 async function act(j: Job, action: "retry" | "cancel") {52 try {53 await api.post(`/admin/jobs/${j.id}/${action}`);54 toast.success(`Job ${action === "retry" ? "re-queued" : "cancelled"}`, j.id);55 reloadAll();56 } catch (e) {57 toast.error(e);58 }59 }6061 const counts = useMemo(() => {62 const m: Record<string, number> = {};63 for (const j of jobs.data?.data ?? []) m[j.status] = (m[j.status] ?? 0) + 1;64 return m;65 }, [jobs.data]);6667 if (!isOwner) return <Banner tone="warn">Ingestion controls require the owner role.</Banner>;6869 const columns = [70 { key: "kind", header: "Kind", render: (j: Job) => <span className="font-medium">{j.kind}</span> },71 { key: "dataset_id", header: "Dataset", mono: true, hideSm: true, render: (j: Job) => <span className="font-mono text-xs">{j.dataset_id ?? "—"}</span> },72 { key: "status", header: "Status", render: (j: Job) => <StatusBadge status={j.status} /> },73 { key: "attempts", header: "Att.", num: true, hideSm: true, render: (j: Job) => `${j.attempts ?? 0}/${j.max_attempts ?? "—"}` },74 { key: "created_at", header: "Created", hideSm: true, render: (j: Job) => <span className="tnum text-xs">{fmtTs(j.created_at)}</span> },75 { key: "duration", header: "Duration", render: (j: Job) => <span className="tnum text-xs">{j.started_at ? fmtDuration(j.started_at, j.finished_at) : "—"}</span> },76 { key: "progress", header: "Progress", render: (j: Job) => (j.progress ? <JsonView value={j.progress} collapsed /> : <span className="text-ink-3">—</span>) },77 { key: "last_error", header: "Error", render: (j: Job) => (j.last_error ? <span className="line-clamp-2 text-xs text-crit-ink">{j.last_error}</span> : <span className="text-ink-3">—</span>) },78 {79 key: "actions",80 header: "",81 render: (j: Job) => (82 <div className="flex gap-1" onClick={(e) => e.stopPropagation()}>83 {(j.status === "failed" || j.status === "dead" || j.status === "cancelled") && (84 <button className="btn btn-sm" onClick={() => void act(j, "retry")} title="Retry">85 <RotateCcw className="h-3.5 w-3.5" /> Retry86 </button>87 )}88 {j.status === "queued" && (89 <button className="btn btn-sm" onClick={() => void act(j, "cancel")} title="Cancel">90 <XCircle className="h-3.5 w-3.5" /> Cancel91 </button>92 )}93 </div>94 ),95 },96 ];9798 return (99 <>100 <PageHeader101 title="Ingestion"102 subtitle={<>Job queue · {Object.entries(counts).map(([k, v]) => `${k} ${v}`).join(" · ") || "no jobs"} · refresh every {hasLive ? 5 : 30} s</>}103 actions={104 <>105 <button className="btn btn-sm" onClick={reloadAll}>106 <RefreshCw className="h-3.5 w-3.5" /> Refresh107 </button>108 <button className="btn btn-sm btn-primary" onClick={() => setShowForm(true)}>109 <Plus className="h-3.5 w-3.5" /> Enqueue job110 </button>111 </>112 }113 />114115 {health.data?.components.ingest_worker && health.data.components.ingest_worker !== "running" && (116 <Banner tone="crit" className="mb-4">117 The ingest worker is <strong>{health.data.components.ingest_worker}</strong> — queued jobs will not start until it holds the lake lock.118 </Banner>119 )}120121 {(dead.data?.data.length ?? 0) + (failed.data?.data.length ?? 0) > 0 && (122 <Card title={`Dead-letter & failed (${(dead.data?.data.length ?? 0) + (failed.data?.data.length ?? 0)})`} className="mb-4 border-crit/40">123 <DataTable rows={[...(dead.data?.data ?? []), ...(failed.data?.data ?? [])]} rowKey={(j) => j.id} mobile="scroll" columns={columns} onRowClick={setInspect} />124 </Card>125 )}126127 <Card128 title="Jobs"129 actions={130 <div className="flex gap-2">131 <select className="input w-auto py-1 text-xs" value={status} onChange={(e) => setStatus(e.target.value)} aria-label="Status filter">132 {STATUSES.map((s) => (133 <option key={s} value={s}>134 {s || "any status"}135 </option>136 ))}137 </select>138 <select className="input w-auto py-1 text-xs" value={kind} onChange={(e) => setKind(e.target.value)} aria-label="Kind filter">139 <option value="">any kind</option>140 {KINDS.map((k) => (141 <option key={k} value={k}>142 {k}143 </option>144 ))}145 </select>146 </div>147 }148 >149 {jobs.error && <ErrorState error={jobs.error} retry={jobs.reload} />}150 <DataTable rows={jobs.data?.data} loading={jobs.loading} rowKey={(j) => j.id} mobile="scroll" columns={columns} onRowClick={setInspect} empty="No jobs match." footer={<span>{fmtInt(jobs.data?.data.length ?? 0)} jobs (max 300)</span>} />151 </Card>152153 <Modal open={!!inspect} title={inspect ? `${inspect.kind} · ${inspect.id}` : ""} onClose={() => setInspect(null)} wide>154 {inspect && (155 <div className="grid gap-3 md:grid-cols-2">156 <div className="md:col-span-2 flex flex-wrap items-center gap-2 text-sm">157 <StatusBadge status={inspect.status} />158 <span className="font-mono text-xs">{inspect.dataset_id}</span>159 <span className="text-xs text-ink-3">priority {inspect.priority} · attempts {inspect.attempts}/{inspect.max_attempts} · by {inspect.created_by ?? "—"}</span>160 </div>161 <div className="text-xs">162 <div className="text-ink-3">created</div>163 {fmtTs(inspect.created_at)}164 <div className="mt-1 text-ink-3">started</div>165 {fmtTs(inspect.started_at)}166 <div className="mt-1 text-ink-3">finished</div>167 {fmtTs(inspect.finished_at)}168 <div className="mt-1 text-ink-3">next run</div>169 {fmtTs(inspect.next_run_at)}170 </div>171 <div>172 <div className="mb-1 text-xs text-ink-3">params</div>173 <JsonView value={inspect.params ?? {}} />174 </div>175 <div>176 <div className="mb-1 text-xs text-ink-3">progress</div>177 <JsonView value={inspect.progress ?? {}} />178 </div>179 <div>180 <div className="mb-1 text-xs text-ink-3">result</div>181 <JsonView value={inspect.result ?? {}} />182 </div>183 {inspect.last_error && (184 <div className="md:col-span-2">185 <div className="mb-1 text-xs text-ink-3">last_error</div>186 <pre className="overflow-auto rounded-md border border-crit/40 bg-crit-soft p-2 text-xs text-crit-ink whitespace-pre-wrap">{inspect.last_error}</pre>187 </div>188 )}189 </div>190 )}191 </Modal>192193 <Modal open={showForm} title="Enqueue a job" onClose={() => setShowForm(false)}>194 <EnqueueForm195 datasets={datasets.data?.data ?? []}196 lake={health.data?.components.lake}197 onDone={() => {198 setShowForm(false);199 reloadAll();200 }}201 />202 </Modal>203 </>204 );205}206207function EnqueueForm({ datasets, lake, onDone }: { datasets: Dataset[]; lake?: HealthReport["components"]["lake"]; onDone: () => void }) {208 const toast = useToast();209 const [kind, setKind] = useState<Kind>("incremental");210 const [datasetId, setDatasetId] = useState("market_bars.stock.1m.raw");211 const [period, setPeriod] = useState<"day" | "week">("day");212 const [letter, setLetter] = useState("A");213 const [tickers, setTickers] = useState("");214 const [start, setStart] = useState(daysAgo(30));215 const [end, setEnd] = useState(today());216 const [priority, setPriority] = useState(50);217 const [extra, setExtra] = useState("{}");218 const [confirm, setConfirm] = useState(false);219 const [busy, setBusy] = useState(false);220221 const needsDataset = ["incremental", "backfill", "derive", "compaction", "quality"].includes(kind);222223 function buildParams(): Record<string, unknown> {224 let base: Record<string, unknown> = {};225 try {226 base = extra.trim() ? (JSON.parse(extra) as Record<string, unknown>) : {};227 } catch {228 throw new Error("Extra params must be valid JSON");229 }230 if (kind === "incremental") base = { period, ...base };231 if (kind === "backfill") base = { ticker_range: letter.toUpperCase(), ...base };232 if (kind === "fundamentals") {233 const t = splitList(tickers);234 if (t.length) base = { tickers: t, ...base };235 }236 if (kind === "quality" || kind === "derive") base = { start, end, ...base };237 return base;238 }239240 async function submit(e?: FormEvent) {241 e?.preventDefault();242 if (kind === "backfill" && !confirm) {243 setConfirm(true);244 return;245 }246 setBusy(true);247 try {248 const params = buildParams();249 const out = await api.post<{ id: string }>("/admin/jobs", { kind, dataset_id: needsDataset || kind === "fundamentals" ? datasetId || null : null, params, priority });250 toast.success("Job enqueued", `${kind} · ${out.id}`);251 onDone();252 } catch (err) {253 toast.error(err);254 } finally {255 setBusy(false);256 setConfirm(false);257 }258 }259260 return (261 <form onSubmit={submit} className="flex flex-col gap-3">262 <Field label="Kind">263 <select className="input" value={kind} onChange={(e) => setKind(e.target.value as Kind)}>264 {KINDS.map((k) => (265 <option key={k} value={k}>266 {k}267 </option>268 ))}269 </select>270 </Field>271 <p className="text-xs text-ink-2">{KIND_HELP[kind]}</p>272 {needsDataset && (273 <Field label="Dataset">274 <select className="input font-mono text-xs" value={datasetId} onChange={(e) => setDatasetId(e.target.value)}>275 {!datasets.some((d) => d.id === datasetId) && <option value={datasetId}>{datasetId}</option>}276 {datasets.map((d) => (277 <option key={d.id} value={d.id}>278 {d.id}279 </option>280 ))}281 </select>282 </Field>283 )}284 {kind === "incremental" && (285 <Field label="Period">286 <select className="input" value={period} onChange={(e) => setPeriod(e.target.value as "day" | "week")}>287 <option value="day">latest day</option>288 <option value="week">latest week</option>289 </select>290 </Field>291 )}292 {kind === "backfill" && (293 <Field label="Ticker range (first letter)" hint="One letter A–Z or 0 (numeric tickers). One backfill per letter.">294 <select className="input" value={letter} onChange={(e) => setLetter(e.target.value)}>295 {"ABCDEFGHIJKLMNOPQRSTUVWXYZ0".split("").map((l) => (296 <option key={l} value={l}>297 {l}298 </option>299 ))}300 </select>301 </Field>302 )}303 {kind === "fundamentals" && (304 <Field label="Tickers" hint="Comma / newline separated. Leave empty to use the tracked list from Settings.">305 <textarea className="input min-h-20 font-mono uppercase" value={tickers} onChange={(e) => setTickers(e.target.value)} placeholder="AAPL, MSFT" />306 </Field>307 )}308 {(kind === "quality" || kind === "derive") && (309 <div className="grid grid-cols-2 gap-2">310 <Field label="Start">311 <input type="date" className="input" value={start} onChange={(e) => setStart(e.target.value)} />312 </Field>313 <Field label="End">314 <input type="date" className="input" value={end} onChange={(e) => setEnd(e.target.value)} />315 </Field>316 </div>317 )}318 <div className="grid grid-cols-2 gap-2">319 <Field label="Priority" hint="lower runs first (default 50)">320 <input type="number" className="input" min={1} max={100} value={priority} onChange={(e) => setPriority(Number(e.target.value) || 50)} />321 </Field>322 <Field label="Extra params (JSON)">323 <input className="input font-mono text-xs" value={extra} onChange={(e) => setExtra(e.target.value)} />324 </Field>325 </div>326 <div className="flex justify-end gap-2 pt-1">327 <button className="btn btn-primary" disabled={busy}>328 {busy ? "Enqueuing…" : kind === "backfill" ? "Review & enqueue" : "Enqueue"}329 </button>330 </div>331 <ConfirmDialog332 open={confirm}333 title={`Start backfill for tickers “${letter.toUpperCase()}…”?`}334 confirmLabel="Start backfill"335 danger336 busy={busy}337 onCancel={() => setConfirm(false)}338 onConfirm={() => void submit()}339 >340 <p>A backfill downloads the full 1-minute history for every ticker starting with <strong>{letter.toUpperCase()}</strong>. This can take hours and add tens of GB.</p>341 {lake && (342 <p className="mt-2 rounded-md border border-warn/50 bg-warn-soft p-2 text-xs text-warn-ink">343 Disk budget: <strong>{lake.free_gb} GB free</strong> · curated {lake.curated_gb} GB of a {lake.soft_limit_gb} GB soft limit · hard minimum {lake.hard_min_free_gb} GB free. The API refuses backfills under {(lake.hard_min_free_gb * 1.5).toFixed(0)} GB free.344 </p>345 )}346 </ConfirmDialog>347 </form>348 );349}350351