diff --git a/.gitignore b/.gitignore index 5ef6a52..ab3f448 100644 --- a/.gitignore +++ b/.gitignore @@ -39,3 +39,6 @@ yarn-error.log* # typescript *.tsbuildinfo next-env.d.ts + +# local runtime data +/data/ diff --git a/README.md b/README.md index e215bc4..5f63396 100644 --- a/README.md +++ b/README.md @@ -1,36 +1,56 @@ -This is a [Next.js](https://nextjs.org) project bootstrapped with [`create-next-app`](https://nextjs.org/docs/app/api-reference/cli/create-next-app). +# sling-ui -## Getting Started +一个本地运行的 Web UI，把「测试环境 → 本地 dev 库」的数据同步流程自动化： -First, run the development server: +- **Schema 迁移**：[Atlas](https://atlasgo.io)（表 / 索引 / 主外键约束） +- **数据迁移**：[sling](https://slingdata.io)（full-refresh / truncate / incremental / snapshot） + +第一阶段支持 MySQL / MariaDB / TiDB。 + +## 前置依赖 + +- Node.js 20+ 与 pnpm +- `sling` CLI（`curl -LO 'https://github.com/slingdata-io/sling-cli/releases/latest/download/sling_darwin_arm64.tar.gz' ...`，见官方文档；需要官方构建，`conns discover` 在 OSS 自编译版不可用） +- `atlas` CLI：`brew install ariga/tap/atlas` + +启动后应用会在顶部 banner 检测这两个二进制是否可用。 + +## 启动 ```bash -npm run dev -# or -yarn dev -# or -pnpm dev -# or -bun dev +pnpm install +pnpm dev # 或 pnpm build && pnpm start ``` -Open [http://localhost:3000](http://localhost:3000) with your browser to see the result. +打开 http://localhost:3000 。 -You can start editing the page by modifying `app/page.tsx`. The page auto-updates as you edit the file. +## 使用流程 -This project uses [`next/font`](https://nextjs.org/docs/app/building-your-application/optimizing/fonts) to automatically optimize and load [Geist](https://vercel.com/font), a new font family for Vercel. +1. **Connections**：新建源（测试环境）和目标（本地 dev）连接，点「测试连接」确认可用。 +2. **Pipelines → 新建流水线**：选源/目标连接、mode、是否开启 schema sync，「加载表」勾选要迁移的表。 +3. **流水线详情**： + - 「Schema Diff 预览」查看 Atlas dry-run 生成的 DDL； + - 「运行」执行迁移：先 `atlas schema apply`（同步表结构，含索引/约束），再 `sling run`（灌数据），日志实时流式展示。 +4. **Runs**：全部运行历史与日志回放。 -## Learn More +### mode 说明 -To learn more about Next.js, take a look at the following resources: +- `schema_sync` 开启时，Atlas 负责目标端表结构；此时即使选择 `full-refresh`，实际执行会翻译为 `truncate`（保留 Atlas 建的表结构重灌全量数据，避免 sling drop+重建丢掉索引/约束）。 +- `incremental` 需要为每张表配置 `primary_key` 和 `update_key`。 +- sling 连接 URL 自动带 `foreign_key_checks=0`（会话级），避免外键顺序问题；约束定义本身保留。 -- [Next.js Documentation](https://nextjs.org/docs) - learn about Next.js features and API. -- [Learn Next.js](https://nextjs.org/learn) - an interactive Next.js tutorial. +## 数据与安全说明 -You can check out [the Next.js GitHub repository](https://github.com/vercel/next.js) - your feedback and contributions are welcome! +- 所有应用数据在项目根 `data/` 目录（已 gitignore）：SQLite（`sling-ui.db`）、生成的 sling `env.yaml`、replication yaml、运行日志。 +- **连接密码以明文存储在本地 SQLite**；运行日志中的命令行已做密码脱敏，但 sling/atlas 自身的输出可能仍含连接信息。本工具定位为本地单人使用，请勿暴露到网络。 +- Atlas 需要一个 dev database 做 schema 规范化：实现为在**目标服务器**上创建专用 scratch schema `sling_ui_atlas_dev`（自动创建，保持干净），不依赖 docker。 +- sling 使用应用自管的 home 目录（`data/sling-home`），不会读写你的 `~/.sling`。 -## Deploy on Vercel +## 技术栈 -The easiest way to deploy your Next.js app is to use the [Vercel Platform](https://vercel.com/new?utm_medium=default-template&filter=next.js&utm_source=create-next-app&utm_campaign=create-next-app-readme) from the creators of Next.js. +Next.js 16（App Router）+ TypeScript + Tailwind 4 + shadcn/ui + better-sqlite3；通过 shell 调用 sling / atlas 二进制，`SLING_LOGGING=JSON` 解析结构化日志，SSE 实时推送。 -Check out our [Next.js deployment documentation](https://nextjs.org/docs/app/building-your-application/deploying) for more details. +## 路线图 + +- PostgreSQL 支持（连接类型 + pg_dump/Atlas 均已抽象） +- 流水线编辑 UI、定时调度、运行并发队列 diff --git a/app/api/connections/[id]/route.ts b/app/api/connections/[id]/route.ts new file mode 100644 index 0000000..60c25b9 --- /dev/null +++ b/app/api/connections/[id]/route.ts @@ -0,0 +1,34 @@ +import { deleteConnection, updateConnection } from "@/lib/db"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +type Ctx = { params: Promise<{ id: string }> }; + +export async function PUT(request: Request, { params }: Ctx) { + try { + const { id } = await params; + const body = await request.json(); + const conn = updateConnection(Number(id), body); + if (!conn) { + return Response.json({ error: "connection not found" }, { status: 404 }); + } + return Response.json(conn); + } catch (err) { + const message = (err as Error).message; + const status = message.includes("UNIQUE") ? 409 : 500; + return Response.json({ error: message }, { status }); + } +} + +export async function DELETE(_request: Request, { params }: Ctx) { + try { + const { id } = await params; + if (!deleteConnection(Number(id))) { + return Response.json({ error: "connection not found" }, { status: 404 }); + } + return Response.json({ ok: true }); + } catch (err) { + return Response.json({ error: (err as Error).message }, { status: 500 }); + } +} diff --git a/app/api/connections/[id]/streams/route.ts b/app/api/connections/[id]/streams/route.ts new file mode 100644 index 0000000..46821f0 --- /dev/null +++ b/app/api/connections/[id]/streams/route.ts @@ -0,0 +1,16 @@ +import { discoverStreams } from "@/lib/sling"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +export async function GET( + _request: Request, + { params }: { params: Promise<{ id: string }> } +) { + try { + const { id } = await params; + return Response.json(discoverStreams(Number(id))); + } catch (err) { + return Response.json({ error: (err as Error).message }, { status: 500 }); + } +} diff --git a/app/api/connections/[id]/test/route.ts b/app/api/connections/[id]/test/route.ts new file mode 100644 index 0000000..c816381 --- /dev/null +++ b/app/api/connections/[id]/test/route.ts @@ -0,0 +1,17 @@ +import { testConnection } from "@/lib/sling"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +export async function POST( + _request: Request, + { params }: { params: Promise<{ id: string }> } +) { + try { + const { id } = await params; + const result = testConnection(Number(id)); + return Response.json(result); + } catch (err) { + return Response.json({ error: (err as Error).message }, { status: 500 }); + } +} diff --git a/app/api/connections/route.ts b/app/api/connections/route.ts new file mode 100644 index 0000000..4e000a4 --- /dev/null +++ b/app/api/connections/route.ts @@ -0,0 +1,38 @@ +import { createConnection, listConnections } from "@/lib/db"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +export async function GET() { + try { + return Response.json(listConnections()); + } catch (err) { + return Response.json({ error: (err as Error).message }, { status: 500 }); + } +} + +export async function POST(request: Request) { + try { + const body = await request.json(); + if (!body.name || !body.type || !body.host) { + return Response.json( + { error: "name, type and host are required" }, + { status: 400 } + ); + } + const conn = createConnection({ + name: String(body.name), + type: String(body.type), + host: String(body.host), + port: Number(body.port ?? 3306), + user: String(body.user ?? ""), + password: String(body.password ?? ""), + database: String(body.database ?? ""), + }); + return Response.json(conn, { status: 201 }); + } catch (err) { + const message = (err as Error).message; + const status = message.includes("UNIQUE") ? 409 : 500; + return Response.json({ error: message }, { status }); + } +} diff --git a/app/api/health/route.ts b/app/api/health/route.ts new file mode 100644 index 0000000..eb1cf16 --- /dev/null +++ b/app/api/health/route.ts @@ -0,0 +1,12 @@ +import { checkBinaries } from "@/lib/binaries"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +export async function GET() { + try { + return Response.json(checkBinaries()); + } catch (err) { + return Response.json({ error: (err as Error).message }, { status: 500 }); + } +} diff --git a/app/api/pipelines/[id]/diff/route.ts b/app/api/pipelines/[id]/diff/route.ts new file mode 100644 index 0000000..bfb4bad --- /dev/null +++ b/app/api/pipelines/[id]/diff/route.ts @@ -0,0 +1,25 @@ +import { schemaDiff } from "@/lib/atlas"; +import { getPipeline } from "@/lib/db"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +export async function GET( + _request: Request, + { params }: { params: Promise<{ id: string }> } +) { + try { + const { id } = await params; + const pipeline = getPipeline(Number(id)); + if (!pipeline) { + return Response.json({ error: "pipeline not found" }, { status: 404 }); + } + const result = await schemaDiff( + pipeline.source_conn_id, + pipeline.target_conn_id + ); + return Response.json(result); + } catch (err) { + return Response.json({ error: (err as Error).message }, { status: 500 }); + } +} diff --git a/app/api/pipelines/[id]/route.ts b/app/api/pipelines/[id]/route.ts new file mode 100644 index 0000000..7b2f78b --- /dev/null +++ b/app/api/pipelines/[id]/route.ts @@ -0,0 +1,63 @@ +import { deletePipeline, getPipeline, updatePipeline } from "@/lib/db"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +type Ctx = { params: Promise<{ id: string }> }; + +function normalizeStreams(raw: unknown): string { + if (typeof raw === "string") return raw; + return JSON.stringify(raw ?? []); +} + +export async function GET(_request: Request, { params }: Ctx) { + try { + const { id } = await params; + const pipeline = getPipeline(Number(id)); + if (!pipeline) { + return Response.json({ error: "pipeline not found" }, { status: 404 }); + } + return Response.json(pipeline); + } catch (err) { + return Response.json({ error: (err as Error).message }, { status: 500 }); + } +} + +export async function PUT(request: Request, { params }: Ctx) { + try { + const { id } = await params; + const body = await request.json(); + const patch: Record = {}; + for (const key of [ + "name", + "source_conn_id", + "target_conn_id", + "schema_sync", + "mode", + ]) { + if (body[key] !== undefined) patch[key] = body[key]; + } + if (body.streams !== undefined) { + patch.streams = normalizeStreams(body.streams); + } + const pipeline = updatePipeline(Number(id), patch); + if (!pipeline) { + return Response.json({ error: "pipeline not found" }, { status: 404 }); + } + return Response.json(pipeline); + } catch (err) { + return Response.json({ error: (err as Error).message }, { status: 500 }); + } +} + +export async function DELETE(_request: Request, { params }: Ctx) { + try { + const { id } = await params; + if (!deletePipeline(Number(id))) { + return Response.json({ error: "pipeline not found" }, { status: 404 }); + } + return Response.json({ ok: true }); + } catch (err) { + return Response.json({ error: (err as Error).message }, { status: 500 }); + } +} diff --git a/app/api/pipelines/[id]/run/route.ts b/app/api/pipelines/[id]/run/route.ts new file mode 100644 index 0000000..06f8b8f --- /dev/null +++ b/app/api/pipelines/[id]/run/route.ts @@ -0,0 +1,22 @@ +import { AlreadyRunningError, jobManager } from "@/lib/jobs"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +export async function POST( + _request: Request, + { params }: { params: Promise<{ id: string }> } +) { + try { + const { id } = await params; + const run = jobManager.startRun(Number(id)); + return Response.json(run, { status: 201 }); + } catch (err) { + if (err instanceof AlreadyRunningError) { + return Response.json({ error: err.message }, { status: 409 }); + } + const message = (err as Error).message; + const status = message.includes("not found") ? 404 : 500; + return Response.json({ error: message }, { status }); + } +} diff --git a/app/api/pipelines/route.ts b/app/api/pipelines/route.ts new file mode 100644 index 0000000..0821fc7 --- /dev/null +++ b/app/api/pipelines/route.ts @@ -0,0 +1,40 @@ +import { createPipeline, listPipelines } from "@/lib/db"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +function normalizeStreams(raw: unknown): string { + if (typeof raw === "string") return raw; + return JSON.stringify(raw ?? []); +} + +export async function GET() { + try { + return Response.json(listPipelines()); + } catch (err) { + return Response.json({ error: (err as Error).message }, { status: 500 }); + } +} + +export async function POST(request: Request) { + try { + const body = await request.json(); + if (!body.name || !body.source_conn_id || !body.target_conn_id) { + return Response.json( + { error: "name, source_conn_id and target_conn_id are required" }, + { status: 400 } + ); + } + const pipeline = createPipeline({ + name: String(body.name), + source_conn_id: Number(body.source_conn_id), + target_conn_id: Number(body.target_conn_id), + schema_sync: body.schema_sync === undefined ? 1 : Number(body.schema_sync), + mode: body.mode ? String(body.mode) : "full-refresh", + streams: normalizeStreams(body.streams), + }); + return Response.json(pipeline, { status: 201 }); + } catch (err) { + return Response.json({ error: (err as Error).message }, { status: 500 }); + } +} diff --git a/app/api/runs/[id]/events/route.ts b/app/api/runs/[id]/events/route.ts new file mode 100644 index 0000000..5f65a92 --- /dev/null +++ b/app/api/runs/[id]/events/route.ts @@ -0,0 +1,93 @@ +import fs from "fs"; +import { getRun } from "@/lib/db"; +import { jobManager } from "@/lib/jobs"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +const encoder = new TextEncoder(); + +function sseLine(line: string, ts: number): Uint8Array { + return encoder.encode(`data: ${JSON.stringify({ line, ts })}\n\n`); +} + +function sseDone(status: string): Uint8Array { + return encoder.encode(`event: done\ndata: ${JSON.stringify({ status })}\n\n`); +} + +export async function GET( + _request: Request, + { params }: { params: Promise<{ id: string }> } +) { + try { + const { id } = await params; + const runId = Number(id); + const run = getRun(runId); + if (!run) { + return Response.json({ error: "run not found" }, { status: 404 }); + } + + let cleanup: () => void = () => {}; + + const stream = new ReadableStream({ + start(controller) { + let unsubscribe: (() => void) | null = null; + let closed = false; + const close = (status: string) => { + if (closed) return; + closed = true; + try { + controller.enqueue(sseDone(status)); + controller.close(); + } catch { + // client already gone + } + }; + + const replayFile = () => { + if (run.log_path && fs.existsSync(run.log_path)) { + const content = fs.readFileSync(run.log_path, "utf8"); + for (const line of content.split("\n")) { + if (line) controller.enqueue(sseLine(line, 0)); + } + } + close(run.status); + }; + + if (run.status === "running" || run.status === "queued") { + unsubscribe = jobManager.subscribe( + runId, + (ev) => { + if (!closed) controller.enqueue(sseLine(ev.line, ev.ts)); + }, + (status) => close(status) + ); + if (!unsubscribe) { + // Not active in this process (e.g. after HMR): fall back to file. + replayFile(); + return; + } + cleanup = () => { + closed = true; + unsubscribe?.(); + }; + } else { + replayFile(); + } + }, + cancel() { + cleanup(); + }, + }); + + return new Response(stream, { + headers: { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + Connection: "keep-alive", + }, + }); + } catch (err) { + return Response.json({ error: (err as Error).message }, { status: 500 }); + } +} diff --git a/app/api/runs/[id]/route.ts b/app/api/runs/[id]/route.ts new file mode 100644 index 0000000..2ec70a5 --- /dev/null +++ b/app/api/runs/[id]/route.ts @@ -0,0 +1,25 @@ +import fs from "fs"; +import { getRun } from "@/lib/db"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +export async function GET( + _request: Request, + { params }: { params: Promise<{ id: string }> } +) { + try { + const { id } = await params; + const run = getRun(Number(id)); + if (!run) { + return Response.json({ error: "run not found" }, { status: 404 }); + } + let log = ""; + if (run.log_path && fs.existsSync(run.log_path)) { + log = fs.readFileSync(run.log_path, "utf8"); + } + return Response.json({ ...run, log }); + } catch (err) { + return Response.json({ error: (err as Error).message }, { status: 500 }); + } +} diff --git a/app/api/runs/route.ts b/app/api/runs/route.ts new file mode 100644 index 0000000..04673de --- /dev/null +++ b/app/api/runs/route.ts @@ -0,0 +1,12 @@ +import { listRuns } from "@/lib/db"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +export async function GET() { + try { + return Response.json(listRuns()); + } catch (err) { + return Response.json({ error: (err as Error).message }, { status: 500 }); + } +} diff --git a/app/connections/page.tsx b/app/connections/page.tsx new file mode 100644 index 0000000..3f66dc9 --- /dev/null +++ b/app/connections/page.tsx @@ -0,0 +1,205 @@ +"use client"; + +import { Fragment, useCallback, useEffect, useState } from "react"; +import { toast } from "sonner"; +import { Plus, Trash2, Pencil, PlugZap } from "lucide-react"; +import { api, type Connection } from "@/components/api"; +import { fmtTime } from "@/components/format"; +import { ConnectionDialog } from "@/components/connection-dialog"; +import { ConfirmDialog } from "@/components/confirm-dialog"; +import { Button } from "@/components/ui/button"; +import { + Table, + TableBody, + TableCell, + TableHead, + TableHeader, + TableRow, +} from "@/components/ui/table"; + +interface TestResult { + ok: boolean; + output: string; +} + +export default function ConnectionsPage() { + const [list, setList] = useState([]); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(null); + + const [dialogOpen, setDialogOpen] = useState(false); + const [editing, setEditing] = useState(null); + const [deleting, setDeleting] = useState(null); + const [deleteLoading, setDeleteLoading] = useState(false); + + const [testingId, setTestingId] = useState(null); + const [testResults, setTestResults] = useState>({}); + const [expandedId, setExpandedId] = useState(null); + + const load = useCallback(() => { + api("/api/connections") + .then(setList) + .catch((e) => setError(e instanceof Error ? e.message : "加载失败")) + .finally(() => setLoading(false)); + }, []); + + useEffect(load, [load]); + + async function testConn(c: Connection) { + setTestingId(c.id); + try { + const res = await api(`/api/connections/${c.id}/test`, { + method: "POST", + }); + setTestResults((m) => ({ ...m, [c.id]: res })); + setExpandedId(c.id); + if (res.ok) toast.success(`连接「${c.name}」测试成功`); + else toast.error(`连接「${c.name}」测试失败`); + } catch (e) { + toast.error(e instanceof Error ? e.message : "测试失败"); + } finally { + setTestingId(null); + } + } + + async function doDelete() { + if (!deleting) return; + setDeleteLoading(true); + try { + await api(`/api/connections/${deleting.id}`, { method: "DELETE" }); + toast.success(`已删除连接「${deleting.name}」`); + setDeleting(null); + load(); + } catch (e) { + toast.error(e instanceof Error ? e.message : "删除失败"); + } finally { + setDeleteLoading(false); + } + } + + return ( +
+
+

Connections

+ +
+ + {loading ? ( +
加载中…
+ ) : error ? ( +
加载失败：{error}
+ ) : list.length === 0 ? ( +
+ 还没有连接，点击右上角「新建连接」添加 MySQL / MariaDB / TiDB 数据源。 +
+ ) : ( +
+ + + + 名称 + 类型 + Host + 数据库 + 创建时间 + 操作 + + + + {list.map((c) => ( + + + {c.name} + {c.type} + + {c.host}:{c.port} + + {c.database} + + {fmtTime(c.created_at)} + + +
+ + + +
+ + + {expandedId === c.id && testResults[c.id] ? ( + + + 
+                          {testResults[c.id].output ||
+                            (testResults[c.id].ok ? "连接成功" : "连接失败")}
+
+ + + ) : null} + + ))} + +
+
+ )} + + {dialogOpen ? ( + + ) : null} + !o && setDeleting(null)} + title={`删除连接「${deleting?.name ?? ""}」？`} + description="删除后引用该连接的流水线将无法运行。" + loading={deleteLoading} + onConfirm={doDelete} + /> +
+ ); +} diff --git a/app/globals.css b/app/globals.css index a2dc41e..c56032b 100644 --- a/app/globals.css +++ b/app/globals.css @@ -1,26 +1,130 @@ @import "tailwindcss"; +@import "tw-animate-css"; +@import "shadcn/tailwind.css"; -:root { - --background: #ffffff; - --foreground: #171717; -} +@custom-variant dark (&:is(.dark *)); @theme inline { --color-background: var(--background); --color-foreground: var(--foreground); - --font-sans: var(--font-geist-sans); + --font-sans: var(--font-sans); --font-mono: var(--font-geist-mono); + --font-heading: var(--font-sans); + --color-sidebar-ring: var(--sidebar-ring); + --color-sidebar-border: var(--sidebar-border); + --color-sidebar-accent-foreground: var(--sidebar-accent-foreground); + --color-sidebar-accent: var(--sidebar-accent); + --color-sidebar-primary-foreground: var(--sidebar-primary-foreground); + --color-sidebar-primary: var(--sidebar-primary); + --color-sidebar-foreground: var(--sidebar-foreground); + --color-sidebar: var(--sidebar); + --color-chart-5: var(--chart-5); + --color-chart-4: var(--chart-4); + --color-chart-3: var(--chart-3); + --color-chart-2: var(--chart-2); + --color-chart-1: var(--chart-1); + --color-ring: var(--ring); + --color-input: var(--input); + --color-border: var(--border); + --color-destructive: var(--destructive); + --color-accent-foreground: var(--accent-foreground); + --color-accent: var(--accent); + --color-muted-foreground: var(--muted-foreground); + --color-muted: var(--muted); + --color-secondary-foreground: var(--secondary-foreground); + --color-secondary: var(--secondary); + --color-primary-foreground: var(--primary-foreground); + --color-primary: var(--primary); + --color-popover-foreground: var(--popover-foreground); + --color-popover: var(--popover); + --color-card-foreground: var(--card-foreground); + --color-card: var(--card); + --radius-sm: calc(var(--radius) * 0.6); + --radius-md: calc(var(--radius) * 0.8); + --radius-lg: var(--radius); + --radius-xl: calc(var(--radius) * 1.4); + --radius-2xl: calc(var(--radius) * 1.8); + --radius-3xl: calc(var(--radius) * 2.2); + --radius-4xl: calc(var(--radius) * 2.6); } -@media (prefers-color-scheme: dark) { - :root { - --background: #0a0a0a; - --foreground: #ededed; +:root { + --background: oklch(1 0 0); + --foreground: oklch(0.145 0 0); + --card: oklch(1 0 0); + --card-foreground: oklch(0.145 0 0); + --popover: oklch(1 0 0); + --popover-foreground: oklch(0.145 0 0); + --primary: oklch(0.205 0 0); + --primary-foreground: oklch(0.985 0 0); + --secondary: oklch(0.97 0 0); + --secondary-foreground: oklch(0.205 0 0); + --muted: oklch(0.97 0 0); + --muted-foreground: oklch(0.556 0 0); + --accent: oklch(0.97 0 0); + --accent-foreground: oklch(0.205 0 0); + --destructive: oklch(0.577 0.245 27.325); + --border: oklch(0.922 0 0); + --input: oklch(0.922 0 0); + --ring: oklch(0.708 0 0); + --chart-1: oklch(0.87 0 0); + --chart-2: oklch(0.556 0 0); + --chart-3: oklch(0.439 0 0); + --chart-4: oklch(0.371 0 0); + --chart-5: oklch(0.269 0 0); + --radius: 0.625rem; + --sidebar: oklch(0.985 0 0); + --sidebar-foreground: oklch(0.145 0 0); + --sidebar-primary: oklch(0.205 0 0); + --sidebar-primary-foreground: oklch(0.985 0 0); + --sidebar-accent: oklch(0.97 0 0); + --sidebar-accent-foreground: oklch(0.205 0 0); + --sidebar-border: oklch(0.922 0 0); + --sidebar-ring: oklch(0.708 0 0); +} + +.dark { + --background: oklch(0.145 0 0); + --foreground: oklch(0.985 0 0); + --card: oklch(0.205 0 0); + --card-foreground: oklch(0.985 0 0); + --popover: oklch(0.205 0 0); + --popover-foreground: oklch(0.985 0 0); + --primary: oklch(0.922 0 0); + --primary-foreground: oklch(0.205 0 0); + --secondary: oklch(0.269 0 0); + --secondary-foreground: oklch(0.985 0 0); + --muted: oklch(0.269 0 0); + --muted-foreground: oklch(0.708 0 0); + --accent: oklch(0.269 0 0); + --accent-foreground: oklch(0.985 0 0); + --destructive: oklch(0.704 0.191 22.216); + --border: oklch(1 0 0 / 10%); + --input: oklch(1 0 0 / 15%); + --ring: oklch(0.556 0 0); + --chart-1: oklch(0.87 0 0); + --chart-2: oklch(0.556 0 0); + --chart-3: oklch(0.439 0 0); + --chart-4: oklch(0.371 0 0); + --chart-5: oklch(0.269 0 0); + --sidebar: oklch(0.205 0 0); + --sidebar-foreground: oklch(0.985 0 0); + --sidebar-primary: oklch(0.488 0.243 264.376); + --sidebar-primary-foreground: oklch(0.985 0 0); + --sidebar-accent: oklch(0.269 0 0); + --sidebar-accent-foreground: oklch(0.985 0 0); + --sidebar-border: oklch(1 0 0 / 10%); + --sidebar-ring: oklch(0.556 0 0); +} + +@layer base { + * { + @apply border-border outline-ring/50; } -} - -body { - background: var(--background); - color: var(--foreground); - font-family: Arial, Helvetica, sans-serif; -} + body { + @apply bg-background text-foreground; + } + html { + @apply font-sans; + } +} \ No newline at end of file diff --git a/app/layout.tsx b/app/layout.tsx index 9852c15..1197574 100644 --- a/app/layout.tsx +++ b/app/layout.tsx @@ -1,6 +1,9 @@ import type { Metadata } from "next"; import { Geist, Geist_Mono } from "next/font/google"; import "./globals.css"; +import { Toaster } from "@/components/ui/sonner"; +import { AppSidebar } from "@/components/app-sidebar"; +import { HealthBanner } from "@/components/health-banner"; const geistSans = Geist({ variable: "--font-geist-sans", @@ -13,17 +16,26 @@ const geistMono = Geist_Mono({ }); export const metadata: Metadata = { - title: "Create Next App", - description: "Generated by create next app", + title: "sling-ui", + description: "Atlas + Sling 数据同步控制台", }; export default function RootLayout({ children }: LayoutProps<"/">) { return ( - {children} + +
+ +
+ +
{children}
+
+
+ + ); } diff --git a/app/page.tsx b/app/page.tsx index c887311..624dd95 100644 --- a/app/page.tsx +++ b/app/page.tsx @@ -1,69 +1,125 @@ -import Image from "next/image"; +"use client"; + +import { useEffect, useState } from "react"; +import Link from "next/link"; +import { + api, + type Connection, + type Pipeline, + type RunWithPipeline, +} from "@/components/api"; +import { StatusBadge } from "@/components/status-badge"; +import { fmtTime, fmtDuration } from "@/components/format"; +import { + Card, + CardContent, + CardHeader, + CardTitle, +} from "@/components/ui/card"; +import { + Table, + TableBody, + TableCell, + TableHead, + TableHeader, + TableRow, +} from "@/components/ui/table"; +import { Button } from "@/components/ui/button"; + +export default function DashboardPage() { + const [connections, setConnections] = useState([]); + const [pipelines, setPipelines] = useState([]); + const [runs, setRuns] = useState([]); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(null); + + useEffect(() => { + Promise.all([ + api("/api/connections"), + api("/api/pipelines"), + api("/api/runs"), + ]) + .then(([c, p, r]) => { + setConnections(c); + setPipelines(p); + setRuns(r); + }) + .catch((e) => setError(e instanceof Error ? e.message : "加载失败")) + .finally(() => setLoading(false)); + }, []); + + if (loading) return
加载中…
; + if (error) return
加载失败：{error}
; + + const recent = runs.slice(0, 5); + const stats = [ + { label: "连接数", value: connections.length, href: "/connections" }, + { label: "流水线数", value: pipelines.length, href: "/pipelines" }, + { label: "运行总数", value: runs.length, href: "/runs" }, + ]; -export default function Home() { return ( -
-
- Next.js logo -
-

- To get started, edit the{" "} - - page.tsx - {" "} - file. -

-

- Looking for a starting point or more instructions? Head over to{" "} - - Templates - {" "} - or the{" "} - - Learning - {" "} - center. -

-
- -
+
+

Dashboard

+ +
+ {stats.map((s) => ( + + + + {s.label} + + + + + {s.value} + + + + ))} +
+ + + + 最近运行 + + + {recent.length === 0 ? ( +
还没有运行记录。
+ ) : ( + + + + ID + 流水线 + 状态 + 开始时间 + 耗时 + + + + + {recent.map((r) => ( + + #{r.id} + {r.pipeline_name ?? `(已删除 #${r.pipeline_id})`} + + + + {fmtTime(r.started_at)} + {fmtDuration(r.started_at, r.finished_at)} + + + + + ))} + +
+ )} + +
); } diff --git a/app/pipelines/[id]/page.tsx b/app/pipelines/[id]/page.tsx new file mode 100644 index 0000000..d03b1c7 --- /dev/null +++ b/app/pipelines/[id]/page.tsx @@ -0,0 +1,309 @@ +"use client"; + +import { useCallback, useEffect, useMemo, useState } from "react"; +import Link from "next/link"; +import { useParams } from "next/navigation"; +import { toast } from "sonner"; +import { Play, GitCompareArrows } from "lucide-react"; +import { + api, + ApiError, + type Connection, + type Pipeline, + type RunWithPipeline, + type StreamConfig, +} from "@/components/api"; +import { LogViewer } from "@/components/log-viewer"; +import { StatusBadge } from "@/components/status-badge"; +import { fmtTime, fmtDuration } from "@/components/format"; +import { Badge } from "@/components/ui/badge"; +import { Button } from "@/components/ui/button"; +import { + Card, + CardContent, + CardDescription, + CardHeader, + CardTitle, +} from "@/components/ui/card"; +import { Textarea } from "@/components/ui/textarea"; +import { + Table, + TableBody, + TableCell, + TableHead, + TableHeader, + TableRow, +} from "@/components/ui/table"; + +export default function PipelineDetailPage() { + const params = useParams<{ id: string }>(); + const id = Number(params.id); + + const [pipeline, setPipeline] = useState(null); + const [connMap, setConnMap] = useState>({}); + const [runs, setRuns] = useState([]); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(null); + + const [runLoading, setRunLoading] = useState(false); + const [diffLoading, setDiffLoading] = useState(false); + const [diffSql, setDiffSql] = useState(null); + + const loadRuns = useCallback(() => { + api("/api/runs") + .then((all) => + setRuns(all.filter((r) => r.pipeline_id === id)) + ) + .catch(() => {}); + }, [id]); + + useEffect(() => { + Promise.all([ + api(`/api/pipelines/${id}`), + api("/api/connections"), + ]) + .then(([p, conns]) => { + setPipeline(p); + setConnMap(Object.fromEntries(conns.map((c) => [c.id, c.name]))); + }) + .catch((e) => setError(e instanceof Error ? e.message : "加载失败")) + .finally(() => setLoading(false)); + loadRuns(); + }, [id, loadRuns]); + + // 有运行中的任务时轮询刷新历史列表 + const hasActive = runs.some( + (r) => r.status === "running" || r.status === "queued" + ); + useEffect(() => { + if (!hasActive) return; + const t = setInterval(loadRuns, 4000); + return () => clearInterval(t); + }, [hasActive, loadRuns]); + + const streams = useMemo(() => { + if (!pipeline) return []; + try { + const v = JSON.parse(pipeline.streams); + return Array.isArray(v) ? v : []; + } catch { + return []; + } + }, [pipeline]); + + const latestRun = runs[0] ?? null; + + async function runNow() { + setRunLoading(true); + try { + await api(`/api/pipelines/${id}/run`, { method: "POST" }); + toast.success("已开始运行"); + loadRuns(); + } catch (e) { + if (e instanceof ApiError && e.status === 409) { + toast.info("该流水线正在运行中，请等待当前运行结束"); + } else { + toast.error(e instanceof Error ? e.message : "启动失败"); + } + } finally { + setRunLoading(false); + } + } + + async function previewDiff() { + setDiffLoading(true); + setDiffSql(null); + try { + const res = await api<{ ok: boolean; sql: string }>( + `/api/pipelines/${id}/diff` + ); + setDiffSql(res.sql || "-- 无差异，schema 已一致"); + if (!res.ok) toast.info("Atlas 返回了非零状态，请检查 SQL 内容"); + } catch (e) { + toast.error(e instanceof Error ? e.message : "获取 diff 失败"); + } finally { + setDiffLoading(false); + } + } + + if (loading) return
加载中…
; + if (error || !pipeline) + return ( +
+ {error ? `加载失败：${error}` : "流水线不存在"} +
+ ); + + const connName = (cid: number) => connMap[cid] ?? `#${cid}`; + + return ( +
+
+

{pipeline.name}

+
+ + +
+
+ + + + 配置 + + +
+
+
源 → 目标
+
+ {connName(pipeline.source_conn_id)} →{" "} + {connName(pipeline.target_conn_id)} +
+
+
+
同步模式
+
{pipeline.mode}
+
+
+
表数量
+
{streams.length}
+
+
+
Schema 同步
+
+ {pipeline.schema_sync ? ( + + 开 + + ) : ( + + 关 + + )} +
+
+
+
创建时间
+
{fmtTime(pipeline.created_at)}
+
+
+
更新时间
+
{fmtTime(pipeline.updated_at)}
+
+
+ {streams.length > 0 ? ( +
+ {streams.map((s) => ( + + {s.name} + + ))} +
+ ) : null} + + + + {diffLoading || diffSql !== null ? ( + + + Schema Diff（Atlas dry-run） + + 目标库需要执行的 DDL。计算可能需要数十秒。 + + + + {diffLoading ? ( +
+ 正在计算 diff，请稍候… +
+ ) : ( +