diff --git a/app/api/pipelines/[id]/diff/route.ts b/app/api/pipelines/[id]/diff/route.ts index bfb4bad..e8b2654 100644 --- a/app/api/pipelines/[id]/diff/route.ts +++ b/app/api/pipelines/[id]/diff/route.ts @@ -14,10 +14,7 @@ export async function GET( if (!pipeline) { return Response.json({ error: "pipeline not found" }, { status: 404 }); } - const result = await schemaDiff( - pipeline.source_conn_id, - pipeline.target_conn_id - ); + const result = await schemaDiff(pipeline); 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 index 7b2f78b..b13a7b9 100644 --- a/app/api/pipelines/[id]/route.ts +++ b/app/api/pipelines/[id]/route.ts @@ -33,6 +33,8 @@ export async function PUT(request: Request, { params }: Ctx) { "source_conn_id", "target_conn_id", "schema_sync", + "schema_scope", + "include_fk", "mode", ]) { if (body[key] !== undefined) patch[key] = body[key]; diff --git a/app/api/pipelines/route.ts b/app/api/pipelines/route.ts index 0821fc7..cbd5136 100644 --- a/app/api/pipelines/route.ts +++ b/app/api/pipelines/route.ts @@ -30,6 +30,8 @@ export async function POST(request: Request) { 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), + schema_scope: body.schema_scope ? String(body.schema_scope) : "selected", + include_fk: body.include_fk === undefined ? 0 : Number(body.include_fk), mode: body.mode ? String(body.mode) : "full-refresh", streams: normalizeStreams(body.streams), }); diff --git a/app/pipelines/[id]/edit/page.tsx b/app/pipelines/[id]/edit/page.tsx new file mode 100644 index 0000000..8e916ca --- /dev/null +++ b/app/pipelines/[id]/edit/page.tsx @@ -0,0 +1,32 @@ +"use client"; + +import { useEffect, useState } from "react"; +import { useParams } from "next/navigation"; +import { api, type Pipeline } from "@/components/api"; +import { PipelineForm } from "@/components/pipeline-form"; + +export default function EditPipelinePage() { + const params = useParams<{ id: string }>(); + const id = Number(params.id); + + const [pipeline, setPipeline] = useState(null); + const [error, setError] = useState(null); + const [loading, setLoading] = useState(true); + + useEffect(() => { + api(`/api/pipelines/${id}`) + .then(setPipeline) + .catch((e) => setError(e instanceof Error ? e.message : "加载失败")) + .finally(() => setLoading(false)); + }, [id]); + + if (loading) return
加载中…
; + if (error || !pipeline) + return ( +
+ {error ? `加载失败：${error}` : "流水线不存在"} +
+ ); + + return ; +} diff --git a/app/pipelines/[id]/page.tsx b/app/pipelines/[id]/page.tsx index d03b1c7..4a4cd5b 100644 --- a/app/pipelines/[id]/page.tsx +++ b/app/pipelines/[id]/page.tsx @@ -4,7 +4,7 @@ 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 { Play, GitCompareArrows, Pencil } from "lucide-react"; import { api, ApiError, @@ -141,6 +141,12 @@ export default function PipelineDetailPage() {

{pipeline.name}

+
+
+
同步范围
+
+ {pipeline.schema_scope === "all" ? "整个库" : "仅选中的表"} +
+
+
+
包含外键
+
{pipeline.include_fk ? "是" : "否"}
+
创建时间
{fmtTime(pipeline.created_at)}
diff --git a/app/pipelines/new/page.tsx b/app/pipelines/new/page.tsx index f0662de..3d3c130 100644 --- a/app/pipelines/new/page.tsx +++ b/app/pipelines/new/page.tsx @@ -1,314 +1,7 @@ "use client"; -import { useEffect, useMemo, useState } from "react"; -import { useRouter } from "next/navigation"; -import { toast } from "sonner"; -import { - api, - type Connection, - type Pipeline, - type StreamConfig, - type StreamInfo, -} from "@/components/api"; -import { Button } from "@/components/ui/button"; -import { - Card, - CardContent, - CardDescription, - CardHeader, - CardTitle, -} from "@/components/ui/card"; -import { Checkbox } from "@/components/ui/checkbox"; -import { Input } from "@/components/ui/input"; -import { Label } from "@/components/ui/label"; -import { - Select, - SelectContent, - SelectItem, - SelectTrigger, - SelectValue, -} from "@/components/ui/select"; -import { Switch } from "@/components/ui/switch"; - -const MODES = [ - { value: "full-refresh", label: "full-refresh（全量覆盖，默认）" }, - { value: "truncate", label: "truncate（清空后写入）" }, - { value: "incremental", label: "incremental（增量）" }, - { value: "snapshot", label: "snapshot（快照）" }, -]; +import { PipelineForm } from "@/components/pipeline-form"; export default function NewPipelinePage() { - const router = useRouter(); - const [connections, setConnections] = useState([]); - const [connError, setConnError] = useState(null); - - const [name, setName] = useState(""); - const [sourceId, setSourceId] = useState(""); - const [targetId, setTargetId] = useState(""); - const [mode, setMode] = useState("full-refresh"); - const [schemaSync, setSchemaSync] = useState(true); - - const [streamList, setStreamList] = useState(null); - const [streamsLoading, setStreamsLoading] = useState(false); - const [selected, setSelected] = useState>({}); - const [saving, setSaving] = useState(false); - - useEffect(() => { - api("/api/connections") - .then(setConnections) - .catch((e) => setConnError(e instanceof Error ? e.message : "加载失败")); - }, []); - - // 切换源连接时同时清空已加载的表（在 onValueChange 里处理） - - const selectedList = useMemo(() => Object.values(selected), [selected]); - - async function loadStreams() { - if (!sourceId) return; - setStreamsLoading(true); - try { - const list = await api( - `/api/connections/${sourceId}/streams` - ); - setStreamList(list); - if (list.length === 0) toast.info("该连接没有可同步的表"); - } catch (e) { - toast.error(e instanceof Error ? e.message : "加载表失败"); - } finally { - setStreamsLoading(false); - } - } - - function toggleStream(s: StreamInfo, checked: boolean) { - setSelected((m) => { - const next = { ...m }; - if (checked) next[s.name] = { name: s.name }; - else delete next[s.name]; - return next; - }); - } - - function selectAll(on: boolean) { - if (!streamList) return; - setSelected( - on - ? Object.fromEntries( - streamList.map((s) => [s.name, selected[s.name] ?? { name: s.name }]) - ) - : {} - ); - } - - function setStreamField( - name: string, - field: "primary_key" | "update_key", - value: string - ) { - setSelected((m) => ({ - ...m, - [name]: { ...m[name], name, [field]: value || undefined }, - })); - } - - async function submit() { - if (!name.trim()) return toast.error("请填写流水线名称"); - if (!sourceId || !targetId) return toast.error("请选择源连接和目标连接"); - if (sourceId === targetId) return toast.error("源连接和目标连接不能相同"); - if (selectedList.length === 0) return toast.error("请至少选择一张表"); - - setSaving(true); - try { - const p = await api("/api/pipelines", { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ - name: name.trim(), - source_conn_id: Number(sourceId), - target_conn_id: Number(targetId), - schema_sync: schemaSync ? 1 : 0, - mode, - streams: JSON.stringify(selectedList), - }), - }); - toast.success("流水线已创建"); - router.push(`/pipelines/${p.id}`); - } catch (e) { - toast.error(e instanceof Error ? e.message : "创建失败"); - setSaving(false); - } - } - - return ( -
-

新建流水线

- - {connError ? ( -
连接列表加载失败：{connError}
- ) : null} - - - - 基本信息 - - -
- - setName(e.target.value)} - placeholder="例如：测试库 → 本地 dev" - /> -
-
-
- - -
-
- - -
-
-
- - -
-
-
- - - 先用 Atlas 同步 schema（表/索引/约束），再迁移数据 - -
- setSchemaSync(!!c)} /> -
- - - - - - 选择要同步的表 - - 选择源连接后点击「加载表」，勾选需要同步的表。 - - - -
- - {streamList && streamList.length > 0 ? ( - <> - - - - 已选 {selectedList.length} / {streamList.length} - - - ) : null} -
- - {streamList === null ? null : streamList.length === 0 ? ( -
没有可同步的表。
- ) : ( -
- {streamList.map((s) => { - const checked = !!selected[s.name]; - return ( -
- toggleStream(s, !!c)} - /> - {s.name} - {mode === "incremental" && checked ? ( -
- - setStreamField(s.name, "primary_key", e.target.value) - } - /> - - setStreamField(s.name, "update_key", e.target.value) - } - /> -
- ) : null} -
- ); - })} -
- )} - - - -
- - -
-
- ); + return ; } diff --git a/app/pipelines/page.tsx b/app/pipelines/page.tsx index 9474fe1..cdd1787 100644 --- a/app/pipelines/page.tsx +++ b/app/pipelines/page.tsx @@ -122,6 +122,9 @@ export default function PipelinesPage() { + + {streamList && streamList.length > 0 ? ( + <> + + + + 已选 {selectedList.length} / {streamList.length} + + + ) : null} +
+ + {streamList === null ? ( + selectedList.length > 0 ? ( +
+ 当前已选 {selectedList.length} 张表（ + {selectedList.map((s) => s.name).join("、")}），点击「加载表」可修改。 +
+ ) : null + ) : streamList.length === 0 ? ( +
没有可同步的表。
+ ) : ( +
+ {streamList.map((s) => { + const checked = !!selected[s.name]; + return ( +
+ toggleStream(s, !!c)} + /> + {s.name} + {mode === "incremental" && checked ? ( +
+ + setStreamField(s.name, "primary_key", e.target.value) + } + /> + + setStreamField(s.name, "update_key", e.target.value) + } + /> +
+ ) : null} +
+ ); + })} +
+ )} + + + +
+ + +
+
+ ); +} diff --git a/lib/atlas.ts b/lib/atlas.ts index 970b0a7..16eb5f0 100644 --- a/lib/atlas.ts +++ b/lib/atlas.ts @@ -1,6 +1,6 @@ import { execFile } from "child_process"; -import { Connection, getConnection } from "./db"; -import { connectionUrl, execSql } from "./sling"; +import { Connection, Pipeline, StreamConfig, getConnection } from "./db"; +import { connectionUrl, discoverStreams, execSql } from "./sling"; import { resolveBinary } from "./binaries"; // Atlas needs an empty scratch database ("dev database") to plan diffs. @@ -23,8 +23,13 @@ export function prepareDevDatabase(tgt: Connection): string { return devDatabaseUrl(tgt); } -function baseApplyArgs(srcUrl: string, tgtUrl: string, devUrl: string): string[] { - return [ +function baseApplyArgs( + srcUrl: string, + tgtUrl: string, + devUrl: string, + excludes: string[] +): string[] { + const args = [ "schema", "apply", "--url", @@ -34,47 +39,112 @@ function baseApplyArgs(srcUrl: string, tgtUrl: string, devUrl: string): string[] "--dev-url", devUrl, ]; + // Community Atlas has no --include; --exclude globs match bare resource + // names (no schema prefix). Excluded objects are dropped from the diff + // entirely, so e.g. unsupported column types in them never surface. + for (const p of excludes) args.push("--exclude", p); + return args; } export function buildApplyArgs( srcUrl: string, tgtUrl: string, - devUrl: string + devUrl: string, + excludes: string[] = [] ): string[] { - return [...baseApplyArgs(srcUrl, tgtUrl, devUrl), "--auto-approve"]; + return [...baseApplyArgs(srcUrl, tgtUrl, devUrl, excludes), "--auto-approve"]; } // atlas rejects --dry-run combined with --auto-approve. export function buildDryRunArgs( srcUrl: string, tgtUrl: string, - devUrl: string + devUrl: string, + excludes: string[] = [] ): string[] { - return [...baseApplyArgs(srcUrl, tgtUrl, devUrl), "--dry-run"]; + return [...baseApplyArgs(srcUrl, tgtUrl, devUrl, excludes), "--dry-run"]; +} + +// Compute the --exclude patterns for a pipeline's schema step. Atlas diffs +// whole databases, and tables that exist only on the target would be planned +// as DROPs — so for 'selected' scope we must exclude every non-selected table +// from BOTH sides. Bare table names only: exclude globs match resource names +// without schema prefixes. +export async function computeSchemaExcludes( + pipeline: Pipeline +): Promise { + if ((pipeline.schema_scope || "selected") === "all") return []; + + const streams = JSON.parse(pipeline.streams || "[]") as StreamConfig[]; + const selected = new Set( + streams.map((s) => { + const i = s.name.lastIndexOf("."); + return i >= 0 ? s.name.slice(i + 1) : s.name; + }) + ); + + const tableNames = (connId: number, side: string): Set => { + try { + return new Set( + discoverStreams(connId).map((s) => { + const i = s.name.lastIndexOf("."); + return i >= 0 ? s.name.slice(i + 1) : s.name; + }) + ); + } catch (e) { + // Silently falling back would risk DROPping target-only tables. + throw new Error( + `无法获取${side}库表列表，已中止 schema 同步：${(e as Error).message}` + ); + } + }; + + const srcTables = tableNames(pipeline.source_conn_id, "源"); + const tgtTables = tableNames(pipeline.target_conn_id, "目标"); + + const excludes: string[] = []; + for (const t of new Set([...srcTables, ...tgtTables])) { + if (!selected.has(t)) excludes.push(t); + } + // FKs referencing excluded tables vanish from the desired state anyway; + // excluding all FKs additionally avoids touching constraints on the + // selected tables themselves. + if (!pipeline.include_fk) excludes.push("*[type=foreign-key]"); + return excludes; } function atlasBin(): string { return resolveBinary("atlas") ?? "atlas"; } -export function schemaDiff( - sourceConnId: number, - targetConnId: number +export async function schemaDiff( + pipeline: Pipeline ): Promise<{ ok: boolean; sql: string }> { - const src = getConnection(sourceConnId); - const tgt = getConnection(targetConnId); + const src = getConnection(pipeline.source_conn_id); + const tgt = getConnection(pipeline.target_conn_id); if (!src || !tgt) throw new Error("connection not found"); + let excludes: string[]; + try { + excludes = await computeSchemaExcludes(pipeline); + } catch (e) { + return { ok: false, sql: (e as Error).message }; + } let devUrl: string; try { devUrl = prepareDevDatabase(tgt); } catch (e) { const err = e as { stdout?: string; stderr?: string; message?: string }; - return Promise.resolve({ + return { ok: false, sql: [err.stdout, err.stderr, err.message].filter(Boolean).join("\n"), - }); + }; } - const args = buildDryRunArgs(connectionUrl(src), connectionUrl(tgt), devUrl); + const args = buildDryRunArgs( + connectionUrl(src), + connectionUrl(tgt), + devUrl, + excludes + ); return new Promise((resolve) => { execFile( /* turbopackIgnore: true */ atlasBin(), diff --git a/lib/db.ts b/lib/db.ts index 7cc67c4..20f57ac 100644 --- a/lib/db.ts +++ b/lib/db.ts @@ -41,6 +41,8 @@ export interface Pipeline { source_conn_id: number; target_conn_id: number; schema_sync: number; // 1 | 0 + schema_scope: string; // 'selected' | 'all' + include_fk: number; // 1 | 0 mode: string; streams: string; // JSON: StreamConfig[] created_at: string; @@ -52,6 +54,8 @@ export interface PipelineInput { source_conn_id: number; target_conn_id: number; schema_sync?: number; + schema_scope?: string; + include_fk?: number; mode?: string; streams?: string; } @@ -91,6 +95,8 @@ function createDb(): Database.Database { source_conn_id INTEGER NOT NULL REFERENCES connections(id), target_conn_id INTEGER NOT NULL REFERENCES connections(id), schema_sync INTEGER DEFAULT 1, + schema_scope TEXT NOT NULL DEFAULT 'selected', + include_fk INTEGER NOT NULL DEFAULT 0, mode TEXT DEFAULT 'full-refresh', streams TEXT NOT NULL DEFAULT '[]', created_at TEXT DEFAULT (datetime('now')), @@ -107,6 +113,30 @@ function createDb(): Database.Database { pid INTEGER ); `); + // Lightweight migrations for databases created before these columns existed. + // Several Next.js workers may open the DB concurrently during build, so a + // racing ALTER can fail with "duplicate column name" — safe to ignore. + const addColumnIfMissing = (sql: string, column: string) => { + const cols = new Set( + (db.pragma("table_info(pipelines)") as { name: string }[]).map( + (c) => c.name + ) + ); + if (cols.has(column)) return; + try { + db.exec(sql); + } catch (e) { + if (!(e as Error).message.includes("duplicate column name")) throw e; + } + }; + addColumnIfMissing( + "ALTER TABLE pipelines ADD COLUMN schema_scope TEXT NOT NULL DEFAULT 'selected'", + "schema_scope" + ); + addColumnIfMissing( + "ALTER TABLE pipelines ADD COLUMN include_fk INTEGER NOT NULL DEFAULT 0", + "include_fk" + ); return db; } @@ -186,14 +216,16 @@ export function getPipeline(id: number): Pipeline | null { export function createPipeline(input: PipelineInput): Pipeline { const res = db .prepare( - `INSERT INTO pipelines (name, source_conn_id, target_conn_id, schema_sync, mode, streams) - VALUES (@name, @source_conn_id, @target_conn_id, @schema_sync, @mode, @streams)` + `INSERT INTO pipelines (name, source_conn_id, target_conn_id, schema_sync, schema_scope, include_fk, mode, streams) + VALUES (@name, @source_conn_id, @target_conn_id, @schema_sync, @schema_scope, @include_fk, @mode, @streams)` ) .run({ name: input.name, source_conn_id: input.source_conn_id, target_conn_id: input.target_conn_id, schema_sync: input.schema_sync ?? 1, + schema_scope: input.schema_scope ?? "selected", + include_fk: input.include_fk ?? 0, mode: input.mode ?? "full-refresh", streams: input.streams ?? "[]", }); @@ -209,6 +241,8 @@ export function updatePipeline( "source_conn_id", "target_conn_id", "schema_sync", + "schema_scope", + "include_fk", "mode", "streams", ].filter((k) => input[k as keyof PipelineInput] !== undefined); diff --git a/lib/jobs.ts b/lib/jobs.ts index d1a4286..6037adf 100644 --- a/lib/jobs.ts +++ b/lib/jobs.ts @@ -12,7 +12,7 @@ import { updateRunStatus, } from "./db"; import { SLING_HOME, buildReplicationYaml, connectionUrl } from "./sling"; -import { buildApplyArgs, prepareDevDatabase } from "./atlas"; +import { buildApplyArgs, computeSchemaExcludes, prepareDevDatabase } from "./atlas"; import { resolveBinary } from "./binaries"; export interface JobEvent { @@ -137,11 +137,20 @@ class JobManager { const src = getConnection(pipeline.source_conn_id); const tgt = getConnection(pipeline.target_conn_id); if (!src || !tgt) throw new Error("pipeline connections not found"); + const excludes = await computeSchemaExcludes(pipeline); + const scope = pipeline.schema_scope || "selected"; + const tableExcludes = pipeline.include_fk + ? excludes.length + : excludes.filter((p) => p !== "*[type=foreign-key]").length; + this.emit( + state, + `[schema] scope=${scope}, excluding ${tableExcludes} objects, fk=${pipeline.include_fk ? "on" : "off"}` + ); const devUrl = prepareDevDatabase(tgt); const code = await this.runProcess( state, resolveBinary("atlas") ?? "atlas", - buildApplyArgs(connectionUrl(src), connectionUrl(tgt), devUrl), + buildApplyArgs(connectionUrl(src), connectionUrl(tgt), devUrl, excludes), {} ); if (code !== 0) {