TsDB2/cmd/vrtxsqlweb/main.go

// Command vrtxsqlweb 给 migrate 库套一个本地 Web 界面。
//
// 和 cmd/vrtxsql 是同一套能力, 只是把 flag 换成浏览器里的表单:
// 填两个 DSN、看一眼源库结构、勾选要迁的表、点一下, 迁移过程以日志流实时回显,
// 结束给出每张表的报告 (行数 / 列数 / 索引数 / 耗时)。
//
//	go run ./cmd/vrtxsqlweb -addr :8099
//
// 实现只用了标准库 + 一个内嵌的 index.html, 没有前端工具链, 也没有外部依赖:
// 它把 migrate 这个库直接暴露给 HTTP, 不复制任何迁移逻辑。
package main

import (
	_ "embed"
	"encoding/json"
	"flag"
	"fmt"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"runtime"
	"cmp"
	"slices"
	"strings"
	"time"

	"git.ycl.pub/yl32940/TsDB2/migrate"
)

// index.html 是这份单页界面, 编译期嵌进二进制 —— 运行时不需要任何静态文件。
//
//go:embed index.html
var indexHTML []byte

func main() {
	// 默认只听回环: 这是个本地工具, 而且 /api/browse 能列本机任意目录。
	addr := flag.String("addr", "127.0.0.1:8099", "监听地址, 如 127.0.0.1:8099")
	flag.Parse()

	mux := http.NewServeMux()
	mux.HandleFunc("/", handleIndex)
	mux.HandleFunc("/api/browse", handleBrowse)
	mux.HandleFunc("/api/stat", handleStat)
	mux.HandleFunc("/api/list", handleList)
	mux.HandleFunc("/api/databases", handleDatabases)
	mux.HandleFunc("/api/migrate", handleMigrate)

	url := "http://" + *addr
	if strings.HasPrefix(*addr, ":") {
		url = "http://localhost" + *addr
	}
	log.Printf("vrtxsqlweb 已启动: %s", url)
	log.Printf("提示: 服务运行的当前目录就是相对路径的基准目录")
	// 迁移大表可能跑很久, 这里刻意不设 WriteTimeout。
	if err := http.ListenAndServe(*addr, mux); err != nil {
		log.Fatal(err)
	}
}

func handleIndex(w http.ResponseWriter, r *http.Request) {
	if r.URL.Path != "/" {
		http.NotFound(w, r)
		return
	}
	w.Header().Set("Content-Type", "text/html; charset=utf-8")
	_, _ = w.Write(indexHTML)
}

// ============================ /api/browse ============================

// browseReq.Path 用指针是为了区分「字段没传」和「传了空串」:
//
//	nil  → 从服务进程的当前目录起步 (前端首次打开弹窗)
//	""   → 虚拟根 (Windows 列出所有盘符; Unix 就是 /)
//	其他 → 该目录; 若给的是文件, 自动退到它所在目录
type browseReq struct {
	Path *string `json:"path"`
}

type browseEntry struct {
	Name  string `json:"name"`
	Path  string `json:"path"`
	IsDir bool   `json:"isDir"`
	Size  int64  `json:"size"`
	Ext   string `json:"ext"`
}

type browseResp struct {
	Path   string        `json:"path"`
	Parent string        `json:"parent"`
	Root   bool          `json:"root"` // 已到顶 (此电脑 / 文件系统根)
	Dirs   []browseEntry `json:"dirs"`
	Files  []browseEntry `json:"files"`
	Error  string        `json:"error,omitempty"`
}

// handleBrowse 给前端的「浏览…」弹窗列目录。
// 这是本地工具且默认只监听回环地址, 所以不做沙箱限制。
func handleBrowse(w http.ResponseWriter, r *http.Request) {
	var req browseReq
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeJSON(w, browseResp{Error: "请求体解析失败: " + err.Error()})
		return
	}

	var dir string
	switch {
	case req.Path == nil:
		wd, err := os.Getwd()
		if err != nil {
			writeJSON(w, browseResp{Error: "取当前目录失败: " + err.Error()})
			return
		}
		dir = wd
	case *req.Path == "":
		if runtime.GOOS == "windows" {
			writeJSON(w, browseResp{Dirs: windowsDrives(), Root: true})
			return
		}
		dir = string(filepath.Separator)
	default:
		dir = *req.Path
	}

	abs, err := filepath.Abs(dir)
	if err != nil {
		writeJSON(w, browseResp{Error: "路径非法: " + err.Error()})
		return
	}
	if st, err := os.Stat(abs); err == nil && !st.IsDir() {
		abs = filepath.Dir(abs)
	}

	resp := browseResp{Path: abs, Dirs: []browseEntry{}, Files: []browseEntry{}}
	switch parent := filepath.Dir(abs); {
	case parent == abs && runtime.GOOS == "windows":
		resp.Parent = "" // 盘根再往上一级 = 此电脑
	case parent == abs:
		resp.Root = true // Unix 的 /
	default:
		resp.Parent = parent
	}

	ents, err := os.ReadDir(abs)
	if err != nil {
		resp.Error = "读取目录失败: " + err.Error()
		writeJSON(w, resp)
		return
	}
	for _, e := range ents {
		name := e.Name()
		if strings.HasPrefix(name, ".") || strings.HasPrefix(name, "$") {
			continue // 隐藏项和系统目录不展示
		}
		full := filepath.Join(abs, name)
		if e.IsDir() {
			resp.Dirs = append(resp.Dirs, browseEntry{Name: name, Path: full, IsDir: true})
			continue
		}
		var size int64
		if info, err := e.Info(); err == nil {
			size = info.Size()
		}
		resp.Files = append(resp.Files, browseEntry{
			Name: name,
			Path: full,
			Size: size,
			Ext:  strings.ToLower(filepath.Ext(name)),
		})
	}
	slices.SortFunc(resp.Dirs, func(a, b browseEntry) int { return cmp.Compare(a.Name, b.Name) })
	slices.SortFunc(resp.Files, func(a, b browseEntry) int { return cmp.Compare(a.Name, b.Name) })
	writeJSON(w, resp)
}

func windowsDrives() []browseEntry {
	out := make([]browseEntry, 0, 8)
	for c := 'A'; c <= 'Z'; c++ {
		root := string(c) + ":\\"
		if st, err := os.Stat(root); err == nil && st.IsDir() {
			out = append(out, browseEntry{Name: root, Path: root, IsDir: true})
		}
	}
	return out
}

// ============================ /api/stat ============================

type statReq struct {
	Path string `json:"path"`
}

type statResp struct {
	Path         string `json:"path"`
	Exists       bool   `json:"exists"`
	IsDir        bool   `json:"isDir"`
	Size         int64  `json:"size"`
	Parent       string `json:"parent"`
	ParentExists bool   `json:"parentExists"`
	Error        string `json:"error,omitempty"`
}

// handleStat 让前端能显示「文件已存在 / 不存在(将新建) / 父目录不存在」。
// 目标库一般是新建的, 所以要能区分「还没这个文件」和「路径写错了」。
func handleStat(w http.ResponseWriter, r *http.Request) {
	var req statReq
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeJSON(w, statResp{Error: "请求体解析失败: " + err.Error()})
		return
	}
	p := strings.TrimSpace(req.Path)
	if p == "" {
		writeJSON(w, statResp{})
		return
	}
	abs, err := filepath.Abs(p)
	if err != nil {
		writeJSON(w, statResp{Error: "路径非法: " + err.Error()})
		return
	}
	resp := statResp{Path: abs, Parent: filepath.Dir(abs)}
	if st, err := os.Stat(abs); err == nil {
		resp.Exists = true
		resp.IsDir = st.IsDir()
		resp.Size = st.Size()
	}
	if st, err := os.Stat(resp.Parent); err == nil && st.IsDir() {
		resp.ParentExists = true
	}
	writeJSON(w, resp)
}

// ============================ /api/list ============================

type listReq struct {
	DSN    string   `json:"dsn"`
	Tables []string `json:"tables"`
}

type colInfo struct {
	Name      string `json:"name"`
	Type      string `json:"type"`
	Size      int    `json:"size"`
	Dynamic   bool   `json:"dynamic"`
	IsPK      bool   `json:"isPK"`
	Synthetic bool   `json:"synthetic"`
}

type idxInfo struct {
	Name     string   `json:"name"`
	Type     string   `json:"type"`
	Columns  []string `json:"columns"`
	RowStore bool     `json:"rowStore"`
}

type tableInfo struct {
	Name        string    `json:"name"`
	Columns     []colInfo `json:"columns"`
	PKColumn    string    `json:"pkColumn"`
	PKIndexType string    `json:"pkIndexType"`
	RowStore    []string  `json:"rowStore"`
	Indexes     []idxInfo `json:"indexes"`
	Rows        uint64    `json:"rows"`
}

type listResp struct {
	Kind     string      `json:"kind"`
	Identity string      `json:"identity"`
	Tables   []tableInfo `json:"tables"`
	Error    string      `json:"error,omitempty"`
}

type dbListReq struct {
	Host string `json:"host"`
	Port string `json:"port"`
	User string `json:"user"`
	Pass string `json:"pass"`
}

type dbListResp struct {
	Databases []string `json:"databases,omitempty"`
	Error     string   `json:"error,omitempty"`
}

// sysSchemas MySQL 自带的库, 迁它们没有意义, 从候选里去掉。
var sysSchemas = map[string]bool{
	"information_schema": true,
	"performance_schema": true,
	"mysql":              true,
	"sys":                true,
}

// handleDatabases 只回答"这台 MySQL 上有哪些库", 给连接表单把库名做成候选。
// 库名不必先知道 —— 能不能列出库只取决于账号权限。
func handleDatabases(w http.ResponseWriter, r *http.Request) {
	var req dbListReq
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeJSON(w, dbListResp{Error: "请求体解析失败: " + err.Error()})
		return
	}
	user := strings.TrimSpace(req.User)
	if user == "" {
		writeJSON(w, dbListResp{Error: "请先填写用户名"})
		return
	}
	host := strings.TrimSpace(req.Host)
	if host == "" {
		host = "127.0.0.1"
	}
	port := strings.TrimSpace(req.Port)
	if port == "" {
		port = "3306"
	}

	// 库名留空: / 后面只挂连接参数, 顺序不能反过来
	dsn := fmt.Sprintf("%s:%s@tcp(%s:%s)/?charset=utf8mb4", user, req.Pass, host, port)
	names, err := migrate.MySQLDatabases(dsn)
	if err != nil {
		writeJSON(w, dbListResp{Error: friendlyDBErr(err, host, port)})
		return
	}
	keep := make([]string, 0, len(names))
	for _, n := range names {
		if !sysSchemas[n] {
			keep = append(keep, n)
		}
	}
	writeJSON(w, dbListResp{Databases: keep})
}

// friendlyDBErr 把驱动的报错翻成"接下来该查什么"。
// 密码错、权限不足这类原文已经够清楚, 原样带上; 麻烦的是连不上 ——
// 那是一句英文 TCP 报错, 看着不知道从哪儿下手。
func friendlyDBErr(err error, host, port string) string {
	msg := err.Error()
	switch {
	case strings.Contains(msg, "refused"), strings.Contains(msg, "connectex"),
		strings.Contains(msg, "no such host"), strings.Contains(msg, "i/o timeout"),
		strings.Contains(msg, "deadline exceeded"):
		return fmt.Sprintf("连不上 %s:%s —— 检查 MySQL 是否在运行、主机名和端口是否写对 (%v)",
			host, port, err)
	case strings.Contains(msg, "Access denied"):
		return "账号或密码不对：" + msg
	}
	return msg
}

// handleList 只读源库结构, 不碰目标库 —— 对应 CLI 的 -list。
func handleList(w http.ResponseWriter, r *http.Request) {
	var req listReq
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeJSON(w, listResp{Error: "请求体解析失败: " + err.Error()})
		return
	}
	if req.DSN == "" {
		writeJSON(w, listResp{Error: "请先填写源库 DSN"})
		return
	}

	ep, err := migrate.ParseDSN(req.DSN, migrate.DefaultOptions())
	if err != nil {
		writeJSON(w, listResp{Error: err.Error()})
		return
	}
	defer ep.Close()

	resp := listResp{Kind: string(ep.Kind), Identity: ep.Identity()}
	names, err := ep.Src.Tables()
	if err != nil {
		resp.Error = "列出表失败: " + err.Error()
		writeJSON(w, resp)
		return
	}

	want := map[string]bool{}
	for _, t := range req.Tables {
		want[t] = true
	}
	for _, name := range names {
		if name == migrate.MetaTable {
			continue // 保真表是迁移的随身行李, 不当作业务表展示
		}
		if len(want) > 0 && !want[name] {
			continue
		}
		sc, err := ep.Src.Schema(name)
		if err != nil {
			resp.Tables = append(resp.Tables, tableInfo{
				Name:        name,
				PKIndexType: "读结构失败: " + err.Error(),
			})
			continue
		}
		resp.Tables = append(resp.Tables, toTableInfo(sc))
	}
	writeJSON(w, resp)
}

func toTableInfo(sc *migrate.TableSchema) tableInfo {
	ti := tableInfo{
		Name:        sc.Name,
		PKColumn:    sc.PKColumn,
		PKIndexType: migrate.IndexTypeName(sc.PKIndexType),
		RowStore:    sc.RowStore(),
		Rows:        sc.Rows,
		Columns:     make([]colInfo, 0, len(sc.Columns)),
	}
	for _, c := range sc.Columns {
		ti.Columns = append(ti.Columns, colInfo{
			Name:      c.Name,
			Type:      migrate.TypeName(c.Type),
			Size:      c.Size(),
			Dynamic:   c.IsDynamic(),
			IsPK:      c.Name == sc.PKColumn,
			Synthetic: sc.IsSynthetic(c.Name),
		})
	}
	for _, idx := range sc.Indexes {
		ti.Indexes = append(ti.Indexes, idxInfo{
			Name:     idx.Name,
			Type:     migrate.IndexTypeName(idx.Type),
			Columns:  idx.Columns,
			RowStore: idx.IsRowStore(),
		})
	}
	return ti
}

// ============================ /api/migrate ============================

type migrateReq struct {
	From     string   `json:"from"`
	To       string   `json:"to"`
	Tables   []string `json:"tables"`
	Recreate bool     `json:"recreate"`
	Batch    int      `json:"batch"`
	Limit    uint64   `json:"limit"`
	DryRun   bool     `json:"dryRun"`
	NoMeta   bool     `json:"noMeta"`
	AutoPK   bool     `json:"autoPk"`
	Loose    bool     `json:"loose"`
	// TruncOversize 值超过引擎单值上限 (16360 字节) 时截断并逐行报出 ——
	// 与 CLI 的 -truncate-oversize 同一个开关, 明确的数据丢失, 默认关。
	TruncOversize bool `json:"truncateOversize"`
	// KeepGoing 单张表失败时跳过它继续迁其余表 (前端默认勾上)。
	KeepGoing bool `json:"keepGoing"`
	// EachRow 每读一行数据就推一条事件, 界面能看着数据一行行滚过去。
	// 开销随行数线性上涨 (百万行 = 百万条事件), 默认关。
	EachRow bool `json:"eachRow"`
}

// streamEvent 迁移过程以 NDJSON 一行一条推送 (前端按行解析):
//
//	{"type":"log",      "at":"15:04:05","msg":"...","level":"stage"}
//	{"type":"progress", "table":"orders","rows":12000,"elapsed":"1.2s"}
//	{"type":"row",      "table":"orders","rows":12001,"msg":"id=1, name=..."}
//	{"type":"done",     "report":{...}}
//	{"type":"error",    "msg":"..."}
type streamEvent struct {
	Type    string      `json:"type"`
	Msg     string      `json:"msg,omitempty"`
	At      string      `json:"at,omitempty"`
	Level   string      `json:"level,omitempty"` // 空=普通; "stage"=阶段分隔行
	Table   string      `json:"table,omitempty"`
	Rows    uint64      `json:"rows,omitempty"`
	Elapsed string      `json:"elapsed,omitempty"`
	Report  *reportInfo `json:"report,omitempty"`
}

type reportInfo struct {
	From    string        `json:"from"`
	To      string        `json:"to"`
	Rows    uint64        `json:"rows"`
	Elapsed string        `json:"elapsed"`
	Tables  []tableReport `json:"tables"`
}

type tableReport struct {
	Table   string `json:"table"`
	Rows    uint64 `json:"rows"`
	Columns int    `json:"columns"`
	Indexes int    `json:"indexes"`
	Elapsed string `json:"elapsed"`
	Skipped bool   `json:"skipped"`
	Note    string `json:"note"`
	Error   string `json:"error"`
}

func handleMigrate(w http.ResponseWriter, r *http.Request) {
	var req migrateReq
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, "请求体解析失败: "+err.Error(), http.StatusBadRequest)
		return
	}

	flusher, _ := w.(http.Flusher)
	w.Header().Set("Content-Type", "application/x-ndjson; charset=utf-8")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("X-Accel-Buffering", "no") // 走反代时也别缓冲
	w.WriteHeader(http.StatusOK)

	enc := json.NewEncoder(w)
	emit := func(ev streamEvent) {
		_ = enc.Encode(ev)
		if flusher != nil {
			flusher.Flush()
		}
	}
	now := func() string { return time.Now().Format("15:04:05") }

	if req.From == "" || req.To == "" {
		emit(streamEvent{Type: "error", Msg: "请同时填写源库与目标库 DSN"})
		return
	}
	if req.From == req.To {
		emit(streamEvent{Type: "error", Msg: "源库与目标库不能是同一个"})
		return
	}

	opts := migrate.DefaultOptions()
	opts.Create = true
	opts.Recreate = req.Recreate
	opts.EmbedMeta = !req.NoMeta
	opts.StrictValue = !req.Loose
	opts.TruncateOversize = req.TruncOversize
	opts.AutoPK = req.AutoPK
	opts.ContinueOnError = req.KeepGoing
	opts.DryRun = req.DryRun
	opts.Limit = req.Limit
	opts.Tables = req.Tables
	if req.Batch > 0 {
		opts.Batch = req.Batch
	}
	// 把库的逐表进度接到 HTTP 流上: Run 内部每完成一张表就回调这里。
	opts.Verbose = true
	opts.Log = func(format string, args ...any) {
		emit(streamEvent{Type: "log", At: now(), Msg: fmt.Sprintf(format, args...)})
	}
	// 大表"已写 N 行"走结构化事件: 前端原地刷一行, 不往日志里灌流水
	// (灌进去会把日志顶到看不见, 滚动条也就一直在跳)。
	opts.Progress = func(table string, rows uint64, elapsed time.Duration) {
		emit(streamEvent{Type: "progress", Table: table, Rows: rows,
			Elapsed: elapsed.Round(time.Millisecond).String()})
	}
	// 逐行事件 (可选): 让人眼看着数据一行行滚过去。日志区有行数上限, 不会无限长。
	if req.EachRow {
		opts.EachRow = func(table string, n uint64, row migrate.Row) {
			emit(streamEvent{Type: "row", Table: table, Rows: n, Msg: formatRowValues(row)})
		}
	}

	src, err := migrate.ParseDSN(req.From, opts)
	if err != nil {
		emit(streamEvent{Type: "error", Msg: "打开源库: " + err.Error()})
		return
	}
	defer src.Close()

	dst, err := migrate.ParseDSN(req.To, opts)
	if err != nil {
		emit(streamEvent{Type: "error", Msg: "打开目标库: " + err.Error()})
		return
	}
	defer dst.Close()

	emit(streamEvent{Type: "log", At: now(),
		Msg: fmt.Sprintf("开始: %s → %s", src.Identity(), dst.Identity())})

	// 两阶段: 先把所有表的结构立起来, 再灌数据。结构阶段是秒级的 —— 点完按钮
	// 很快就能看到 52 张表全部就位, 不用盯着一张表猜是卡住还是死锁。
	var rep1 *migrate.Report
	if !req.DryRun {
		schemaOpts := *opts
		schemaOpts.SchemaOnly = true
		// 结构阶段尽量往前推: 一张表建不出来不该挡住后面几十张, 明细留给数据阶段报。
		schemaOpts.ContinueOnError = true
		emit(streamEvent{Type: "log", At: now(), Level: "stage",
			Msg: "阶段 1/2: 同步表结构 (建表 + 建索引, 不搬数据)"})
		rep1, _ = migrate.Run(src.Src, dst.Dst, &schemaOpts)
		if rep1 != nil {
			ok, bad := 0, 0
			for _, tr := range rep1.Tables {
				if tr.Error != "" {
					bad++
				} else {
					ok++
				}
			}
			msg := fmt.Sprintf("阶段 1/2 完成: %d 张表结构就绪", ok)
			if bad > 0 {
				msg += fmt.Sprintf(", %d 张建表失败 (具体错误在数据阶段报)", bad)
			}
			emit(streamEvent{Type: "log", At: now(), Msg: msg})
		}
	}

	dataOpts := *opts
	// 表已经在阶段 1 建好了, 这里再 Recreate 一次纯属白干 (还会把刚建的表删掉)。
	dataOpts.Recreate = false
	if !req.DryRun {
		emit(streamEvent{Type: "log", At: now(), Level: "stage", Msg: "阶段 2/2: 迁移数据"})
	}
	rep, runErr := migrate.Run(src.Src, dst.Dst, &dataOpts)
	if rep != nil {
		mergeSchemaNotes(rep, rep1)
		emit(streamEvent{Type: "done", Report: toReportInfo(rep)})
	}
	if runErr != nil {
		emit(streamEvent{Type: "error", Msg: runErr.Error()})
	}
}

// mergeSchemaNotes 把阶段 1 才知道的事补进数据阶段的报告。
//
// 为什么需要: 数据阶段面对的已经是建好的表, CreateTable 直接返回, 于是"哪个索引
// 被跳过"这类只发生在建表时刻的信息在阶段 2 的报告里是空的 —— 不补上, 用户看
// 报告会以为索引都建全了。
func mergeSchemaNotes(rep, schemaRep *migrate.Report) {
	if rep == nil || schemaRep == nil {
		return
	}
	notes := make(map[string]string, len(schemaRep.Tables))
	for _, tr := range schemaRep.Tables {
		n := strings.TrimSuffix(strings.TrimSuffix(tr.Note, "仅结构"), "; ")
		if n != "" {
			notes[tr.Table] = n
		}
	}
	for i := range rep.Tables {
		n, ok := notes[rep.Tables[i].Table]
		if !ok {
			continue
		}
		if rep.Tables[i].Note == "" {
			rep.Tables[i].Note = n
		} else {
			rep.Tables[i].Note += "; " + n
		}
	}
}

// formatRowValues 把一行数据拼成 "值, 值, ..." 供逐行滚动显示。
// 截断长值: 这些行是给人看"数据在动"的, 不是给人读全文的。
func formatRowValues(row migrate.Row) string {
	var b strings.Builder
	for i, v := range row.Values {
		if i > 0 {
			b.WriteString(", ")
		}
		s := fmt.Sprintf("%v", v)
		if r := []rune(s); len(r) > 32 {
			s = string(r[:32]) + "…"
		}
		b.WriteString(s)
	}
	s := b.String()
	if r := []rune(s); len(r) > 220 {
		s = string(r[:220]) + "…"
	}
	return s
}

func toReportInfo(rep *migrate.Report) *reportInfo {
	ri := &reportInfo{
		From:    string(rep.From),
		To:      string(rep.To),
		Rows:    rep.Rows,
		Elapsed: rep.Elapsed.Round(time.Millisecond).String(),
		Tables:  make([]tableReport, 0, len(rep.Tables)),
	}
	for _, tr := range rep.Tables {
		ri.Tables = append(ri.Tables, tableReport{
			Table:   tr.Table,
			Rows:    tr.Rows,
			Columns: tr.Columns,
			Indexes: tr.Indexes,
			Elapsed: tr.Elapsed.Round(time.Millisecond).String(),
			Skipped: tr.Skipped,
			Note:    tr.Note,
			Error:   tr.Error,
		})
	}
	return ri
}

func writeJSON(w http.ResponseWriter, v any) {
	w.Header().Set("Content-Type", "application/json; charset=utf-8")
	_ = json.NewEncoder(w).Encode(v)
}