main.go 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699
  1. // Command vrtxsqlweb 给 migrate 库套一个本地 Web 界面。
  2. //
  3. // 和 cmd/vrtxsql 是同一套能力, 只是把 flag 换成浏览器里的表单:
  4. // 填两个 DSN、看一眼源库结构、勾选要迁的表、点一下, 迁移过程以日志流实时回显,
  5. // 结束给出每张表的报告 (行数 / 列数 / 索引数 / 耗时)。
  6. //
  7. // go run ./cmd/vrtxsqlweb -addr :8099
  8. //
  9. // 实现只用了标准库 + 一个内嵌的 index.html, 没有前端工具链, 也没有外部依赖:
  10. // 它把 migrate 这个库直接暴露给 HTTP, 不复制任何迁移逻辑。
  11. package main
  12. import (
  13. _ "embed"
  14. "encoding/json"
  15. "flag"
  16. "fmt"
  17. "log"
  18. "net/http"
  19. "os"
  20. "path/filepath"
  21. "runtime"
  22. "cmp"
  23. "slices"
  24. "strings"
  25. "time"
  26. "git.ycl.pub/yl32940/TsDB2/migrate"
  27. )
  28. // index.html 是这份单页界面, 编译期嵌进二进制 —— 运行时不需要任何静态文件。
  29. //
  30. //go:embed index.html
  31. var indexHTML []byte
  32. func main() {
  33. // 默认只听回环: 这是个本地工具, 而且 /api/browse 能列本机任意目录。
  34. addr := flag.String("addr", "127.0.0.1:8099", "监听地址, 如 127.0.0.1:8099")
  35. flag.Parse()
  36. mux := http.NewServeMux()
  37. mux.HandleFunc("/", handleIndex)
  38. mux.HandleFunc("/api/browse", handleBrowse)
  39. mux.HandleFunc("/api/stat", handleStat)
  40. mux.HandleFunc("/api/list", handleList)
  41. mux.HandleFunc("/api/databases", handleDatabases)
  42. mux.HandleFunc("/api/migrate", handleMigrate)
  43. url := "http://" + *addr
  44. if strings.HasPrefix(*addr, ":") {
  45. url = "http://localhost" + *addr
  46. }
  47. log.Printf("vrtxsqlweb 已启动: %s", url)
  48. log.Printf("提示: 服务运行的当前目录就是相对路径的基准目录")
  49. // 迁移大表可能跑很久, 这里刻意不设 WriteTimeout。
  50. if err := http.ListenAndServe(*addr, mux); err != nil {
  51. log.Fatal(err)
  52. }
  53. }
  54. func handleIndex(w http.ResponseWriter, r *http.Request) {
  55. if r.URL.Path != "/" {
  56. http.NotFound(w, r)
  57. return
  58. }
  59. w.Header().Set("Content-Type", "text/html; charset=utf-8")
  60. _, _ = w.Write(indexHTML)
  61. }
  62. // ============================ /api/browse ============================
  63. // browseReq.Path 用指针是为了区分「字段没传」和「传了空串」:
  64. //
  65. // nil → 从服务进程的当前目录起步 (前端首次打开弹窗)
  66. // "" → 虚拟根 (Windows 列出所有盘符; Unix 就是 /)
  67. // 其他 → 该目录; 若给的是文件, 自动退到它所在目录
  68. type browseReq struct {
  69. Path *string `json:"path"`
  70. }
  71. type browseEntry struct {
  72. Name string `json:"name"`
  73. Path string `json:"path"`
  74. IsDir bool `json:"isDir"`
  75. Size int64 `json:"size"`
  76. Ext string `json:"ext"`
  77. }
  78. type browseResp struct {
  79. Path string `json:"path"`
  80. Parent string `json:"parent"`
  81. Root bool `json:"root"` // 已到顶 (此电脑 / 文件系统根)
  82. Dirs []browseEntry `json:"dirs"`
  83. Files []browseEntry `json:"files"`
  84. Error string `json:"error,omitempty"`
  85. }
  86. // handleBrowse 给前端的「浏览…」弹窗列目录。
  87. // 这是本地工具且默认只监听回环地址, 所以不做沙箱限制。
  88. func handleBrowse(w http.ResponseWriter, r *http.Request) {
  89. var req browseReq
  90. if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
  91. writeJSON(w, browseResp{Error: "请求体解析失败: " + err.Error()})
  92. return
  93. }
  94. var dir string
  95. switch {
  96. case req.Path == nil:
  97. wd, err := os.Getwd()
  98. if err != nil {
  99. writeJSON(w, browseResp{Error: "取当前目录失败: " + err.Error()})
  100. return
  101. }
  102. dir = wd
  103. case *req.Path == "":
  104. if runtime.GOOS == "windows" {
  105. writeJSON(w, browseResp{Dirs: windowsDrives(), Root: true})
  106. return
  107. }
  108. dir = string(filepath.Separator)
  109. default:
  110. dir = *req.Path
  111. }
  112. abs, err := filepath.Abs(dir)
  113. if err != nil {
  114. writeJSON(w, browseResp{Error: "路径非法: " + err.Error()})
  115. return
  116. }
  117. if st, err := os.Stat(abs); err == nil && !st.IsDir() {
  118. abs = filepath.Dir(abs)
  119. }
  120. resp := browseResp{Path: abs, Dirs: []browseEntry{}, Files: []browseEntry{}}
  121. switch parent := filepath.Dir(abs); {
  122. case parent == abs && runtime.GOOS == "windows":
  123. resp.Parent = "" // 盘根再往上一级 = 此电脑
  124. case parent == abs:
  125. resp.Root = true // Unix 的 /
  126. default:
  127. resp.Parent = parent
  128. }
  129. ents, err := os.ReadDir(abs)
  130. if err != nil {
  131. resp.Error = "读取目录失败: " + err.Error()
  132. writeJSON(w, resp)
  133. return
  134. }
  135. for _, e := range ents {
  136. name := e.Name()
  137. if strings.HasPrefix(name, ".") || strings.HasPrefix(name, "$") {
  138. continue // 隐藏项和系统目录不展示
  139. }
  140. full := filepath.Join(abs, name)
  141. if e.IsDir() {
  142. resp.Dirs = append(resp.Dirs, browseEntry{Name: name, Path: full, IsDir: true})
  143. continue
  144. }
  145. var size int64
  146. if info, err := e.Info(); err == nil {
  147. size = info.Size()
  148. }
  149. resp.Files = append(resp.Files, browseEntry{
  150. Name: name,
  151. Path: full,
  152. Size: size,
  153. Ext: strings.ToLower(filepath.Ext(name)),
  154. })
  155. }
  156. slices.SortFunc(resp.Dirs, func(a, b browseEntry) int { return cmp.Compare(a.Name, b.Name) })
  157. slices.SortFunc(resp.Files, func(a, b browseEntry) int { return cmp.Compare(a.Name, b.Name) })
  158. writeJSON(w, resp)
  159. }
  160. func windowsDrives() []browseEntry {
  161. out := make([]browseEntry, 0, 8)
  162. for c := 'A'; c <= 'Z'; c++ {
  163. root := string(c) + ":\\"
  164. if st, err := os.Stat(root); err == nil && st.IsDir() {
  165. out = append(out, browseEntry{Name: root, Path: root, IsDir: true})
  166. }
  167. }
  168. return out
  169. }
  170. // ============================ /api/stat ============================
  171. type statReq struct {
  172. Path string `json:"path"`
  173. }
  174. type statResp struct {
  175. Path string `json:"path"`
  176. Exists bool `json:"exists"`
  177. IsDir bool `json:"isDir"`
  178. Size int64 `json:"size"`
  179. Parent string `json:"parent"`
  180. ParentExists bool `json:"parentExists"`
  181. Error string `json:"error,omitempty"`
  182. }
  183. // handleStat 让前端能显示「文件已存在 / 不存在(将新建) / 父目录不存在」。
  184. // 目标库一般是新建的, 所以要能区分「还没这个文件」和「路径写错了」。
  185. func handleStat(w http.ResponseWriter, r *http.Request) {
  186. var req statReq
  187. if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
  188. writeJSON(w, statResp{Error: "请求体解析失败: " + err.Error()})
  189. return
  190. }
  191. p := strings.TrimSpace(req.Path)
  192. if p == "" {
  193. writeJSON(w, statResp{})
  194. return
  195. }
  196. abs, err := filepath.Abs(p)
  197. if err != nil {
  198. writeJSON(w, statResp{Error: "路径非法: " + err.Error()})
  199. return
  200. }
  201. resp := statResp{Path: abs, Parent: filepath.Dir(abs)}
  202. if st, err := os.Stat(abs); err == nil {
  203. resp.Exists = true
  204. resp.IsDir = st.IsDir()
  205. resp.Size = st.Size()
  206. }
  207. if st, err := os.Stat(resp.Parent); err == nil && st.IsDir() {
  208. resp.ParentExists = true
  209. }
  210. writeJSON(w, resp)
  211. }
  212. // ============================ /api/list ============================
  213. type listReq struct {
  214. DSN string `json:"dsn"`
  215. Tables []string `json:"tables"`
  216. }
  217. type colInfo struct {
  218. Name string `json:"name"`
  219. Type string `json:"type"`
  220. Size int `json:"size"`
  221. Dynamic bool `json:"dynamic"`
  222. IsPK bool `json:"isPK"`
  223. Synthetic bool `json:"synthetic"`
  224. }
  225. type idxInfo struct {
  226. Name string `json:"name"`
  227. Type string `json:"type"`
  228. Columns []string `json:"columns"`
  229. RowStore bool `json:"rowStore"`
  230. }
  231. type tableInfo struct {
  232. Name string `json:"name"`
  233. Columns []colInfo `json:"columns"`
  234. PKColumn string `json:"pkColumn"`
  235. PKIndexType string `json:"pkIndexType"`
  236. RowStore []string `json:"rowStore"`
  237. Indexes []idxInfo `json:"indexes"`
  238. Rows uint64 `json:"rows"`
  239. }
  240. type listResp struct {
  241. Kind string `json:"kind"`
  242. Identity string `json:"identity"`
  243. Tables []tableInfo `json:"tables"`
  244. Error string `json:"error,omitempty"`
  245. }
  246. type dbListReq struct {
  247. Host string `json:"host"`
  248. Port string `json:"port"`
  249. User string `json:"user"`
  250. Pass string `json:"pass"`
  251. }
  252. type dbListResp struct {
  253. Databases []string `json:"databases,omitempty"`
  254. Error string `json:"error,omitempty"`
  255. }
  256. // sysSchemas MySQL 自带的库, 迁它们没有意义, 从候选里去掉。
  257. var sysSchemas = map[string]bool{
  258. "information_schema": true,
  259. "performance_schema": true,
  260. "mysql": true,
  261. "sys": true,
  262. }
  263. // handleDatabases 只回答"这台 MySQL 上有哪些库", 给连接表单把库名做成候选。
  264. // 库名不必先知道 —— 能不能列出库只取决于账号权限。
  265. func handleDatabases(w http.ResponseWriter, r *http.Request) {
  266. var req dbListReq
  267. if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
  268. writeJSON(w, dbListResp{Error: "请求体解析失败: " + err.Error()})
  269. return
  270. }
  271. user := strings.TrimSpace(req.User)
  272. if user == "" {
  273. writeJSON(w, dbListResp{Error: "请先填写用户名"})
  274. return
  275. }
  276. host := strings.TrimSpace(req.Host)
  277. if host == "" {
  278. host = "127.0.0.1"
  279. }
  280. port := strings.TrimSpace(req.Port)
  281. if port == "" {
  282. port = "3306"
  283. }
  284. // 库名留空: / 后面只挂连接参数, 顺序不能反过来
  285. dsn := fmt.Sprintf("%s:%s@tcp(%s:%s)/?charset=utf8mb4", user, req.Pass, host, port)
  286. names, err := migrate.MySQLDatabases(dsn)
  287. if err != nil {
  288. writeJSON(w, dbListResp{Error: friendlyDBErr(err, host, port)})
  289. return
  290. }
  291. keep := make([]string, 0, len(names))
  292. for _, n := range names {
  293. if !sysSchemas[n] {
  294. keep = append(keep, n)
  295. }
  296. }
  297. writeJSON(w, dbListResp{Databases: keep})
  298. }
  299. // friendlyDBErr 把驱动的报错翻成"接下来该查什么"。
  300. // 密码错、权限不足这类原文已经够清楚, 原样带上; 麻烦的是连不上 ——
  301. // 那是一句英文 TCP 报错, 看着不知道从哪儿下手。
  302. func friendlyDBErr(err error, host, port string) string {
  303. msg := err.Error()
  304. switch {
  305. case strings.Contains(msg, "refused"), strings.Contains(msg, "connectex"),
  306. strings.Contains(msg, "no such host"), strings.Contains(msg, "i/o timeout"),
  307. strings.Contains(msg, "deadline exceeded"):
  308. return fmt.Sprintf("连不上 %s:%s —— 检查 MySQL 是否在运行、主机名和端口是否写对 (%v)",
  309. host, port, err)
  310. case strings.Contains(msg, "Access denied"):
  311. return "账号或密码不对:" + msg
  312. }
  313. return msg
  314. }
  315. // handleList 只读源库结构, 不碰目标库 —— 对应 CLI 的 -list。
  316. func handleList(w http.ResponseWriter, r *http.Request) {
  317. var req listReq
  318. if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
  319. writeJSON(w, listResp{Error: "请求体解析失败: " + err.Error()})
  320. return
  321. }
  322. if req.DSN == "" {
  323. writeJSON(w, listResp{Error: "请先填写源库 DSN"})
  324. return
  325. }
  326. ep, err := migrate.ParseDSN(req.DSN, migrate.DefaultOptions())
  327. if err != nil {
  328. writeJSON(w, listResp{Error: err.Error()})
  329. return
  330. }
  331. defer ep.Close()
  332. resp := listResp{Kind: string(ep.Kind), Identity: ep.Identity()}
  333. names, err := ep.Src.Tables()
  334. if err != nil {
  335. resp.Error = "列出表失败: " + err.Error()
  336. writeJSON(w, resp)
  337. return
  338. }
  339. want := map[string]bool{}
  340. for _, t := range req.Tables {
  341. want[t] = true
  342. }
  343. for _, name := range names {
  344. if name == migrate.MetaTable {
  345. continue // 保真表是迁移的随身行李, 不当作业务表展示
  346. }
  347. if len(want) > 0 && !want[name] {
  348. continue
  349. }
  350. sc, err := ep.Src.Schema(name)
  351. if err != nil {
  352. resp.Tables = append(resp.Tables, tableInfo{
  353. Name: name,
  354. PKIndexType: "读结构失败: " + err.Error(),
  355. })
  356. continue
  357. }
  358. resp.Tables = append(resp.Tables, toTableInfo(sc))
  359. }
  360. writeJSON(w, resp)
  361. }
  362. func toTableInfo(sc *migrate.TableSchema) tableInfo {
  363. ti := tableInfo{
  364. Name: sc.Name,
  365. PKColumn: sc.PKColumn,
  366. PKIndexType: migrate.IndexTypeName(sc.PKIndexType),
  367. RowStore: sc.RowStore(),
  368. Rows: sc.Rows,
  369. Columns: make([]colInfo, 0, len(sc.Columns)),
  370. }
  371. for _, c := range sc.Columns {
  372. ti.Columns = append(ti.Columns, colInfo{
  373. Name: c.Name,
  374. Type: migrate.TypeName(c.Type),
  375. Size: c.Size(),
  376. Dynamic: c.IsDynamic(),
  377. IsPK: c.Name == sc.PKColumn,
  378. Synthetic: sc.IsSynthetic(c.Name),
  379. })
  380. }
  381. for _, idx := range sc.Indexes {
  382. ti.Indexes = append(ti.Indexes, idxInfo{
  383. Name: idx.Name,
  384. Type: migrate.IndexTypeName(idx.Type),
  385. Columns: idx.Columns,
  386. RowStore: idx.IsRowStore(),
  387. })
  388. }
  389. return ti
  390. }
  391. // ============================ /api/migrate ============================
  392. type migrateReq struct {
  393. From string `json:"from"`
  394. To string `json:"to"`
  395. Tables []string `json:"tables"`
  396. Recreate bool `json:"recreate"`
  397. Batch int `json:"batch"`
  398. Limit uint64 `json:"limit"`
  399. DryRun bool `json:"dryRun"`
  400. NoMeta bool `json:"noMeta"`
  401. AutoPK bool `json:"autoPk"`
  402. Loose bool `json:"loose"`
  403. // TruncOversize 值超过引擎单值上限 (16360 字节) 时截断并逐行报出 ——
  404. // 与 CLI 的 -truncate-oversize 同一个开关, 明确的数据丢失, 默认关。
  405. TruncOversize bool `json:"truncateOversize"`
  406. // KeepGoing 单张表失败时跳过它继续迁其余表 (前端默认勾上)。
  407. KeepGoing bool `json:"keepGoing"`
  408. // EachRow 每读一行数据就推一条事件, 界面能看着数据一行行滚过去。
  409. // 开销随行数线性上涨 (百万行 = 百万条事件), 默认关。
  410. EachRow bool `json:"eachRow"`
  411. }
  412. // maxBatchRows 批量接口一次请求允许的最大批大小。batch 直接控制 migrate 包
  413. // make([]Row, 0, Batch) 的预分配 (见 migrate/run.go), 不设上限一次请求就能
  414. // 预分配出超大切片把服务打 OOM。
  415. const maxBatchRows = 50000
  416. // streamEvent 迁移过程以 NDJSON 一行一条推送 (前端按行解析):
  417. //
  418. // {"type":"log", "at":"15:04:05","msg":"...","level":"stage"}
  419. // {"type":"progress", "table":"orders","rows":12000,"elapsed":"1.2s"}
  420. // {"type":"row", "table":"orders","rows":12001,"msg":"id=1, name=..."}
  421. // {"type":"done", "report":{...}}
  422. // {"type":"error", "msg":"..."}
  423. type streamEvent struct {
  424. Type string `json:"type"`
  425. Msg string `json:"msg,omitempty"`
  426. At string `json:"at,omitempty"`
  427. Level string `json:"level,omitempty"` // 空=普通; "stage"=阶段分隔行
  428. Table string `json:"table,omitempty"`
  429. Rows uint64 `json:"rows,omitempty"`
  430. Elapsed string `json:"elapsed,omitempty"`
  431. Report *reportInfo `json:"report,omitempty"`
  432. }
  433. type reportInfo struct {
  434. From string `json:"from"`
  435. To string `json:"to"`
  436. Rows uint64 `json:"rows"`
  437. Elapsed string `json:"elapsed"`
  438. Tables []tableReport `json:"tables"`
  439. }
  440. type tableReport struct {
  441. Table string `json:"table"`
  442. Rows uint64 `json:"rows"`
  443. Columns int `json:"columns"`
  444. Indexes int `json:"indexes"`
  445. Elapsed string `json:"elapsed"`
  446. Skipped bool `json:"skipped"`
  447. Note string `json:"note"`
  448. Error string `json:"error"`
  449. }
  450. func handleMigrate(w http.ResponseWriter, r *http.Request) {
  451. var req migrateReq
  452. if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
  453. http.Error(w, "请求体解析失败: "+err.Error(), http.StatusBadRequest)
  454. return
  455. }
  456. if req.Batch > maxBatchRows {
  457. http.Error(w, fmt.Sprintf("batch 超过上限 %d", maxBatchRows), http.StatusBadRequest)
  458. return
  459. }
  460. flusher, _ := w.(http.Flusher)
  461. w.Header().Set("Content-Type", "application/x-ndjson; charset=utf-8")
  462. w.Header().Set("Cache-Control", "no-cache")
  463. w.Header().Set("X-Accel-Buffering", "no") // 走反代时也别缓冲
  464. w.WriteHeader(http.StatusOK)
  465. enc := json.NewEncoder(w)
  466. emit := func(ev streamEvent) {
  467. _ = enc.Encode(ev)
  468. if flusher != nil {
  469. flusher.Flush()
  470. }
  471. }
  472. now := func() string { return time.Now().Format("15:04:05") }
  473. if req.From == "" || req.To == "" {
  474. emit(streamEvent{Type: "error", Msg: "请同时填写源库与目标库 DSN"})
  475. return
  476. }
  477. if req.From == req.To {
  478. emit(streamEvent{Type: "error", Msg: "源库与目标库不能是同一个"})
  479. return
  480. }
  481. opts := migrate.DefaultOptions()
  482. opts.Create = true
  483. opts.Recreate = req.Recreate
  484. opts.EmbedMeta = !req.NoMeta
  485. opts.StrictValue = !req.Loose
  486. opts.TruncateOversize = req.TruncOversize
  487. opts.AutoPK = req.AutoPK
  488. opts.ContinueOnError = req.KeepGoing
  489. opts.DryRun = req.DryRun
  490. opts.Limit = req.Limit
  491. opts.Tables = req.Tables
  492. if req.Batch > 0 {
  493. opts.Batch = req.Batch
  494. }
  495. // 把库的逐表进度接到 HTTP 流上: Run 内部每完成一张表就回调这里。
  496. opts.Verbose = true
  497. opts.Log = func(format string, args ...any) {
  498. emit(streamEvent{Type: "log", At: now(), Msg: fmt.Sprintf(format, args...)})
  499. }
  500. // 大表"已写 N 行"走结构化事件: 前端原地刷一行, 不往日志里灌流水
  501. // (灌进去会把日志顶到看不见, 滚动条也就一直在跳)。
  502. opts.Progress = func(table string, rows uint64, elapsed time.Duration) {
  503. emit(streamEvent{Type: "progress", Table: table, Rows: rows,
  504. Elapsed: elapsed.Round(time.Millisecond).String()})
  505. }
  506. // 逐行事件 (可选): 让人眼看着数据一行行滚过去。日志区有行数上限, 不会无限长。
  507. if req.EachRow {
  508. opts.EachRow = func(table string, n uint64, row migrate.Row) {
  509. emit(streamEvent{Type: "row", Table: table, Rows: n, Msg: formatRowValues(row)})
  510. }
  511. }
  512. src, err := migrate.ParseDSN(req.From, opts)
  513. if err != nil {
  514. emit(streamEvent{Type: "error", Msg: "打开源库: " + err.Error()})
  515. return
  516. }
  517. defer src.Close()
  518. dst, err := migrate.ParseDSN(req.To, opts)
  519. if err != nil {
  520. emit(streamEvent{Type: "error", Msg: "打开目标库: " + err.Error()})
  521. return
  522. }
  523. defer dst.Close()
  524. emit(streamEvent{Type: "log", At: now(),
  525. Msg: fmt.Sprintf("开始: %s → %s", src.Identity(), dst.Identity())})
  526. // 两阶段: 先把所有表的结构立起来, 再灌数据。结构阶段是秒级的 —— 点完按钮
  527. // 很快就能看到 52 张表全部就位, 不用盯着一张表猜是卡住还是死锁。
  528. var rep1 *migrate.Report
  529. if !req.DryRun {
  530. schemaOpts := *opts
  531. schemaOpts.SchemaOnly = true
  532. // 结构阶段尽量往前推: 一张表建不出来不该挡住后面几十张, 明细留给数据阶段报。
  533. schemaOpts.ContinueOnError = true
  534. emit(streamEvent{Type: "log", At: now(), Level: "stage",
  535. Msg: "阶段 1/2: 同步表结构 (建表 + 建索引, 不搬数据)"})
  536. rep1, _ = migrate.Run(src.Src, dst.Dst, &schemaOpts)
  537. if rep1 != nil {
  538. ok, bad := 0, 0
  539. for _, tr := range rep1.Tables {
  540. if tr.Error != "" {
  541. bad++
  542. } else {
  543. ok++
  544. }
  545. }
  546. msg := fmt.Sprintf("阶段 1/2 完成: %d 张表结构就绪", ok)
  547. if bad > 0 {
  548. msg += fmt.Sprintf(", %d 张建表失败 (具体错误在数据阶段报)", bad)
  549. }
  550. emit(streamEvent{Type: "log", At: now(), Msg: msg})
  551. }
  552. }
  553. dataOpts := *opts
  554. // 表已经在阶段 1 建好了, 这里再 Recreate 一次纯属白干 (还会把刚建的表删掉)。
  555. dataOpts.Recreate = false
  556. if !req.DryRun {
  557. emit(streamEvent{Type: "log", At: now(), Level: "stage", Msg: "阶段 2/2: 迁移数据"})
  558. }
  559. rep, runErr := migrate.Run(src.Src, dst.Dst, &dataOpts)
  560. if rep != nil {
  561. mergeSchemaNotes(rep, rep1)
  562. emit(streamEvent{Type: "done", Report: toReportInfo(rep)})
  563. }
  564. if runErr != nil {
  565. emit(streamEvent{Type: "error", Msg: runErr.Error()})
  566. }
  567. }
  568. // mergeSchemaNotes 把阶段 1 才知道的事补进数据阶段的报告。
  569. //
  570. // 为什么需要: 数据阶段面对的已经是建好的表, CreateTable 直接返回, 于是"哪个索引
  571. // 被跳过"这类只发生在建表时刻的信息在阶段 2 的报告里是空的 —— 不补上, 用户看
  572. // 报告会以为索引都建全了。
  573. func mergeSchemaNotes(rep, schemaRep *migrate.Report) {
  574. if rep == nil || schemaRep == nil {
  575. return
  576. }
  577. notes := make(map[string]string, len(schemaRep.Tables))
  578. for _, tr := range schemaRep.Tables {
  579. n := strings.TrimSuffix(strings.TrimSuffix(tr.Note, "仅结构"), "; ")
  580. if n != "" {
  581. notes[tr.Table] = n
  582. }
  583. }
  584. for i := range rep.Tables {
  585. n, ok := notes[rep.Tables[i].Table]
  586. if !ok {
  587. continue
  588. }
  589. if rep.Tables[i].Note == "" {
  590. rep.Tables[i].Note = n
  591. } else {
  592. rep.Tables[i].Note += "; " + n
  593. }
  594. }
  595. }
  596. // formatRowValues 把一行数据拼成 "值, 值, ..." 供逐行滚动显示。
  597. // 截断长值: 这些行是给人看"数据在动"的, 不是给人读全文的。
  598. func formatRowValues(row migrate.Row) string {
  599. var b strings.Builder
  600. for i, v := range row.Values {
  601. if i > 0 {
  602. b.WriteString(", ")
  603. }
  604. s := fmt.Sprintf("%v", v)
  605. if r := []rune(s); len(r) > 32 {
  606. s = string(r[:32]) + "…"
  607. }
  608. b.WriteString(s)
  609. }
  610. s := b.String()
  611. if r := []rune(s); len(r) > 220 {
  612. s = string(r[:220]) + "…"
  613. }
  614. return s
  615. }
  616. func toReportInfo(rep *migrate.Report) *reportInfo {
  617. ri := &reportInfo{
  618. From: string(rep.From),
  619. To: string(rep.To),
  620. Rows: rep.Rows,
  621. Elapsed: rep.Elapsed.Round(time.Millisecond).String(),
  622. Tables: make([]tableReport, 0, len(rep.Tables)),
  623. }
  624. for _, tr := range rep.Tables {
  625. ri.Tables = append(ri.Tables, tableReport{
  626. Table: tr.Table,
  627. Rows: tr.Rows,
  628. Columns: tr.Columns,
  629. Indexes: tr.Indexes,
  630. Elapsed: tr.Elapsed.Round(time.Millisecond).String(),
  631. Skipped: tr.Skipped,
  632. Note: tr.Note,
  633. Error: tr.Error,
  634. })
  635. }
  636. return ri
  637. }
  638. func writeJSON(w http.ResponseWriter, v any) {
  639. w.Header().Set("Content-Type", "application/json; charset=utf-8")
  640. _ = json.NewEncoder(w).Encode(v)
  641. }