diff --git a/README.md b/README.md index 5f63396..27516b6 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,11 @@ pnpm dev # 或 pnpm build && pnpm start - `schema_sync` 开启时,Atlas 负责目标端表结构;此时即使选择 `full-refresh`,实际执行会翻译为 `truncate`(保留 Atlas 建的表结构重灌全量数据,避免 sling drop+重建丢掉索引/约束)。 - `incremental` 需要为每张表配置 `primary_key` 和 `update_key`。 - sling 连接 URL 自动带 `foreign_key_checks=0`(会话级),避免外键顺序问题;约束定义本身保留。 +- 连接支持自定义「连接参数」(URL query string,sling/Atlas 均生效),流水线支持自定义「环境变量」(写入 replication 的 `env:`)。 + +## 已知问题 + +- TiDB 目标库报 `Error 8048: isolation level 'SERIALIZABLE' is not supported`:见 [docs/tidb-serializable-isolation.md](docs/tidb-serializable-isolation.md),给连接加参数 `tidb_skip_isolation_level_check=1` 即可。 ## 数据与安全说明 diff --git a/app/api/connections/[id]/route.ts b/app/api/connections/[id]/route.ts index 60c25b9..f513dad 100644 --- a/app/api/connections/[id]/route.ts +++ b/app/api/connections/[id]/route.ts @@ -1,4 +1,4 @@ -import { deleteConnection, updateConnection } from "@/lib/db"; +import { deleteConnection, updateConnection, normalizeJsonColumn } from "@/lib/db"; export const runtime = "nodejs"; export const dynamic = "force-dynamic"; @@ -9,6 +9,9 @@ export async function PUT(request: Request, { params }: Ctx) { try { const { id } = await params; const body = await request.json(); + if (body.params !== undefined) { + body.params = normalizeJsonColumn(body.params); + } const conn = updateConnection(Number(id), body); if (!conn) { return Response.json({ error: "connection not found" }, { status: 404 }); diff --git a/app/api/connections/route.ts b/app/api/connections/route.ts index 4e000a4..aaf7dff 100644 --- a/app/api/connections/route.ts +++ b/app/api/connections/route.ts @@ -1,4 +1,4 @@ -import { createConnection, listConnections } from "@/lib/db"; +import { createConnection, listConnections, normalizeJsonColumn } from "@/lib/db"; export const runtime = "nodejs"; export const dynamic = "force-dynamic"; @@ -28,6 +28,7 @@ export async function POST(request: Request) { user: String(body.user ?? ""), password: String(body.password ?? ""), database: String(body.database ?? ""), + params: normalizeJsonColumn(body.params), }); return Response.json(conn, { status: 201 }); } catch (err) { diff --git a/app/api/pipelines/[id]/route.ts b/app/api/pipelines/[id]/route.ts index b13a7b9..09e47dd 100644 --- a/app/api/pipelines/[id]/route.ts +++ b/app/api/pipelines/[id]/route.ts @@ -1,4 +1,4 @@ -import { deletePipeline, getPipeline, updatePipeline } from "@/lib/db"; +import { deletePipeline, getPipeline, updatePipeline, normalizeJsonColumn } from "@/lib/db"; export const runtime = "nodejs"; export const dynamic = "force-dynamic"; @@ -42,6 +42,9 @@ export async function PUT(request: Request, { params }: Ctx) { if (body.streams !== undefined) { patch.streams = normalizeStreams(body.streams); } + if (body.env !== undefined) { + patch.env = normalizeJsonColumn(body.env); + } const pipeline = updatePipeline(Number(id), patch); if (!pipeline) { return Response.json({ error: "pipeline not found" }, { status: 404 }); diff --git a/app/api/pipelines/route.ts b/app/api/pipelines/route.ts index cbd5136..6e6d3a6 100644 --- a/app/api/pipelines/route.ts +++ b/app/api/pipelines/route.ts @@ -1,4 +1,4 @@ -import { createPipeline, listPipelines } from "@/lib/db"; +import { createPipeline, listPipelines, normalizeJsonColumn } from "@/lib/db"; export const runtime = "nodejs"; export const dynamic = "force-dynamic"; @@ -34,6 +34,7 @@ export async function POST(request: Request) { include_fk: body.include_fk === undefined ? 0 : Number(body.include_fk), mode: body.mode ? String(body.mode) : "full-refresh", streams: normalizeStreams(body.streams), + env: normalizeJsonColumn(body.env), }); return Response.json(pipeline, { status: 201 }); } catch (err) { diff --git a/app/connections/page.tsx b/app/connections/page.tsx index 3f66dc9..e3bc374 100644 --- a/app/connections/page.tsx +++ b/app/connections/page.tsx @@ -117,7 +117,18 @@ export default function ConnectionsPage() { {list.map((c) => ( - {c.name} + + {c.name} + {c.params && c.params !== "{}" ? ( +
+ {Object.entries( + JSON.parse(c.params) as Record + ) + .map(([k, v]) => `${k}=${v}`) + .join(" ")} +
+ ) : null} +
{c.type} {c.host}:{c.port} diff --git a/app/pipelines/[id]/page.tsx b/app/pipelines/[id]/page.tsx index 4a4cd5b..2422ef0 100644 --- a/app/pipelines/[id]/page.tsx +++ b/app/pipelines/[id]/page.tsx @@ -91,6 +91,22 @@ export default function PipelineDetailPage() { } }, [pipeline]); + const envEntries = useMemo<[string, string][]>(() => { + if (!pipeline) return []; + try { + const v = JSON.parse(pipeline.env || "{}") as unknown; + if (v && typeof v === "object" && !Array.isArray(v)) { + return Object.entries(v as Record).map(([k, val]) => [ + k, + String(val), + ]); + } + } catch { + // fall through + } + return []; + }, [pipeline]); + const latestRun = runs[0] ?? null; async function runNow() { @@ -203,6 +219,14 @@ export default function PipelineDetailPage() {
包含外键
{pipeline.include_fk ? "是" : "否"}
+ {envEntries.length > 0 ? ( +
+
环境变量
+
+ {envEntries.map(([k, v]) => `${k}=${v}`).join(" ")} +
+
+ ) : null}
创建时间
{fmtTime(pipeline.created_at)}
diff --git a/components/connection-dialog.tsx b/components/connection-dialog.tsx index ecb944e..e512cb7 100644 --- a/components/connection-dialog.tsx +++ b/components/connection-dialog.tsx @@ -3,6 +3,7 @@ import { useState } from "react"; import { toast } from "sonner"; import { api, type Connection } from "@/components/api"; +import { keyValueLinesFromJson, parseKeyValueLines } from "@/components/kv"; import { Button } from "@/components/ui/button"; import { Dialog, @@ -20,6 +21,7 @@ import { SelectTrigger, SelectValue, } from "@/components/ui/select"; +import { Textarea } from "@/components/ui/textarea"; interface ConnectionDialogProps { open: boolean; @@ -59,6 +61,9 @@ export function ConnectionDialog({ : EMPTY ); const [saving, setSaving] = useState(false); + const [paramsText, setParamsText] = useState(() => + initial ? keyValueLinesFromJson(initial.params) : "" + ); const set = (k: keyof typeof EMPTY) => (e: React.ChangeEvent) => setForm((f) => ({ ...f, [k]: e.target.value })); @@ -79,6 +84,7 @@ export function ConnectionDialog({ user: form.user.trim(), password: form.password, database: form.database.trim(), + params: parseKeyValueLines(paramsText), }); if (initial) { await api(`/api/connections/${initial.id}`, { @@ -174,6 +180,21 @@ export function ConnectionDialog({ onChange={set("database")} />
+
+ +