opsxdb/server/internal/db/agent.go

package db

// agent.go —— JDBC agent 桥（Go 侧）。
//
// 架构：每种 agent 驱动（如达梦）由本机 JVM 进程承载官方 JDBC 驱动，
// Go 服务端通过 stdin/stdout 的 JSON Lines 协议与其通信（协议见 agents/README.md）。
// 目标：不引入 CGO/JNI，商用驱动 jar 由用户自备（许可合规），产品侧只做发现与编排。
//
// 安全约定（与 Java 侧一致）：
//   - query 必带 maxRows 硬截断（JDBC Statement.setMaxRows），深分页超过 agentMaxRows 直接拒绝
//   - 每条语句 setQueryTimeout 兜底，timeoutSec 取自请求 ctx deadline
//   - v1 限制：ctx 取消不会真正掐断 JDBC 执行（靠 timeoutSec 兜底），见 agents/README.md 已知限制

import (
	"archive/zip"
	"bufio"
	"context"
	"crypto/sha1"
	"crypto/sha256"
	"encoding/hex"
	"hash"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

const (
	agentProtocolVersion = 1
	agentHelloTimeout    = 30 * time.Second
	agentMaxRows         = 20000 // 与 Java 侧 setMaxRows 上限一致
	// gson 2.12.1 固定版本 + sha1 钉死（下载自 Maven Central / 阿里云镜像，校验失败即拒用）
	gsonVersion = "2.12.1"
	gsonSHA1    = "4e773a317740b83b43cfc3d652962856041697cb"

	// agent jar 与开源驱动 jar：首次连接时按需下载到 ~/.opsxdb/（DMG 不打包任何 Java 产物）。
	// 下载源双托管（Gitee 主 / GitHub 备），sha256 钉版本；商用驱动（达梦）不代下。
	agentJarName   = "opsx-agent-jdbc.jar"
	agentJarSHA256 = "6efc3c71965729f1fe09e6912bcdcd713d5e9d7a4ca87227eaa7a13af2b14227"
	agentJarURLs   = "https://gitee.com/attacker/opsxdb-driver/raw/main/dist/opsx-agent-jdbc.jar,https://raw.githubusercontent.com/attacker/opsxdb-driver/main/dist/opsx-agent-jdbc.jar"
)

// agentArtifact 驱动 jar 清单（开源、可再分发）。Maven 坐标 + sha256 与 opsxdb-driver
// 仓库 drivers/*.yaml 一致；商用驱动（达梦）不在此表，连接时给 manualSource 指引。
type agentArtifact struct {
	File   string // 落盘文件名
	Maven  string // group:artifact:version；空 = 不自动下载
	SHA256 string
}

var agentDriverArtifacts = map[string][]agentArtifact{
	"mongodb": {
		{"mongodb-driver-sync-5.2.1.jar", "org.mongodb:mongodb-driver-sync:5.2.1", "c10ad97ab8a148c53ab8377a3fa89e49c2f9be277f1216c7f69967b7cc1e1699"},
		{"mongodb-driver-core-5.2.1.jar", "org.mongodb:mongodb-driver-core:5.2.1", "5ae1d49fc6b1b42046cb16c1ce9f6952e36d0203f085d9fe045266460c51da19"},
		{"bson-5.2.1.jar", "org.mongodb:bson:5.2.1", "5bca0022f4a933795788244b0ba3d7d92d6969730cf3b279bb37f9f7ffba5f35"},
	},
	"zookeeper": {
		{"zookeeper-3.9.3.jar", "org.apache.zookeeper:zookeeper:3.9.3", "ad6570105768d18382a762325b4460368b7089a5a77c053e682b17942e0e72b5"},
		{"zookeeper-jute-3.9.3.jar", "org.apache.zookeeper:zookeeper-jute:3.9.3", "be72e2c9099413a9878ce830b65bb22e0be755dc2a22b290550ab9e03b5b4cfa"},
		{"slf4j-api-2.0.16.jar", "org.slf4j:slf4j-api:2.0.16", "a12578dde1ba00bd9b816d388a0b879928d00bab3c83c240f7013bf4196c579a"},
		{"netty-common-4.1.113.Final.jar", "io.netty:netty-common:4.1.113.Final", "c0fb22d47111cb06aac2af67fe55e2e216a49fd00e767f4acb7488b280f8c327"},
		{"netty-buffer-4.1.113.Final.jar", "io.netty:netty-buffer:4.1.113.Final", "a85c198180a8de997e8f64a62e054946a39af0708466c1bd67747d393d2feee1"},
		{"netty-transport-4.1.113.Final.jar", "io.netty:netty-transport:4.1.113.Final", "cb8b97ff77d7c5f1c591c84d2dee3389a0eaa63a3137b7b8c0c64e1dbada6688"},
		{"netty-resolver-4.1.113.Final.jar", "io.netty:netty-resolver:4.1.113.Final", "b2c110a547e7dc6e2b27017ea4ef98416d7d832d2cf40625d0273b90e61df8ab"},
		{"netty-codec-4.1.113.Final.jar", "io.netty:netty-codec:4.1.113.Final", "c7669cbeda2b5c6284627b7ffafd6f1014a96639013b26ebb9d6bcb828f76542"},
		{"netty-handler-4.1.113.Final.jar", "io.netty:netty-handler:4.1.113.Final", "7a583c4fe5880504d1257a4a6bfe6a635cd34ffe18aab0e4caabbf17d104f172"},
	},
}

// AgentSpec 描述一种 agent 驱动的接入方式。
type AgentSpec struct {
	DriverKey   string // 与前端 DriverId 一致（dameng、elasticsearch…）
	DriverClass string // JDBC 驱动类名（报错定位；Java 侧按 META-INF/services 自动注册）
	URLTemplate string // jdbc url 模板，占位 {{.Host}}/{{.Port}}/{{.Database}}
	JarsDirEnv  string // 用户放置官方驱动 jar 的目录环境变量；缺省 ~/.opsxdb/agents/jdbc/drivers/<key>
	MainClass   string // agent 主类，缺省 OpsxJdbcAgent（JDBC 桥）；非 JDBC 驱动（如 ZK）给专属主类
	// TablesSQL 非空时：列表不走 DatabaseMetaData，改跑这条 SQL 并取首列（如 ES 的 SHOW TABLES）。
	TablesSQL string
	// DefaultDatabase 非空时：对象树只挂这一个库节点（ES 无多库概念，恒为 default）。
	DefaultDatabase string
	// JVMArgs 追加给 agent 进程的 JVM 参数（如 ZK 用 NIO socket 避开 netty 依赖）。
	JVMArgs []string
}

// agentDriverSpecs —— agent 驱动注册表。新增驱动在此登记即可（达梦先行）。
// 注意：仅注册"官方只提供 JDBC、且 JDBC 不锁付费许可"的驱动。Elasticsearch 曾评估走
// x-pack-sql-jdbc，但其 JDBC 客户端要求付费许可（basic 下报 non-compliant），故 ES 走原生 HTTP（es.go）。
var agentDriverSpecs = map[string]AgentSpec{
	"dameng": {
		DriverKey:   "dameng",
		DriverClass: "dm.jdbc.driver.DmDriver",
		URLTemplate: "jdbc:dm://{{.Host}}:{{.Port}}",
		JarsDirEnv:  "OPSXDB_DAMENG_JARS",
	},
	"zookeeper": {
		DriverKey:   "zookeeper",
		DriverClass: "org.apache.zookeeper.ZooKeeper",
		URLTemplate: "{{.Host}}:{{.Port}}",
		JarsDirEnv:  "OPSXDB_ZK_JARS",
		MainClass:   "OpsxZkAgent",
		// 3.9 默认 Netty socket 需额外 6 个 netty jar；NIO 实现仍内置，零额外依赖
		JVMArgs: []string{"-Dzookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNIO"},
	},
	"mongodb": {
		DriverKey:   "mongodb",
		DriverClass: "com.mongodb.client.MongoClients",
		URLTemplate: "mongodb://{{.Host}}:{{.Port}}",
		JarsDirEnv:  "OPSXDB_MONGO_JARS",
		MainClass:   "OpsxMongoAgent",
	},
}

func agentSpec(d string) (AgentSpec, bool) {
	s, ok := agentDriverSpecs[driverKey(d)]
	return s, ok
}

// IsAgentDriver 该驱动是否走 JDBC agent 桥（前端据此展示"未就绪/需配置"）。
func IsAgentDriver(d string) bool {
	_, ok := agentSpec(d)
	return ok
}

// isAgentDriver 包内简写（ops/open/session 调用点使用）。
func isAgentDriver(d string) bool { return IsAgentDriver(d) }

// AgentConfigured 驱动 jar 目录已配置且存在（未配置时给出许可友好的提示）。
func AgentConfigured(d string) (bool, string) {
	spec, ok := agentSpec(d)
	if !ok {
		return false, ""
	}
	dir := agentJarsDir(spec)
	entries, err := os.ReadDir(dir)
	if err != nil {
		return false, dir
	}
	for _, e := range entries {
		if !e.IsDir() && strings.HasSuffix(strings.ToLower(e.Name()), ".jar") {
			return true, dir
		}
	}
	return false, dir
}

// ---------- 协议信封 ----------

type agentRequest struct {
	ID   uint64         `json:"id"`
	Op   string         `json:"op"`
	Args map[string]any `json:"args"`
}

type agentResponse struct {
	ID     uint64          `json:"id"`
	OK     bool            `json:"ok"`
	Result json.RawMessage `json:"result"`
	Error  string          `json:"error"`
}

// ---------- AgentClient（一个 JVM 进程一个实例，多连接复用） ----------

type pendingCall struct {
	ch chan agentResponse
}

// AgentClient 是与单个 JVM agent 进程的会话。
type AgentClient struct {
	spec      AgentSpec
	cmd       *exec.Cmd
	stdin     io.WriteCloser
	pendingMu sync.Mutex
	pending   map[uint64]pendingCall
	idGen     atomic.Uint64

	stderrTail *cappedBuffer // 排障用：JVM stderr 末尾若干 KB

	readerDone chan struct{}
	closeOnce  sync.Once
}

// cappedBuffer 并发安全的环形缓冲，只保留末尾 cap 字节。
type cappedBuffer struct {
	mu   sync.Mutex
	buf  []byte
	capN int
}

func newCappedBuffer(n int) *cappedBuffer { return &cappedBuffer{capN: n} }

func (b *cappedBuffer) Write(p []byte) (int, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.buf = append(b.buf, p...)
	if len(b.buf) > b.capN {
		b.buf = append([]byte(nil), b.buf[len(b.buf)-b.capN:]...)
	}
	return len(p), nil
}

func (b *cappedBuffer) String() string {
	b.mu.Lock()
	defer b.mu.Unlock()
	return string(b.buf)
}

// StartAgent 按 spec 启动（或复用）JVM agent 进程并完成握手。
// agent jar 与开源驱动 jar 在首次连接时按需下载（DMG 不打包 Java 产物，保持安装包瘦身）。
func StartAgent(spec AgentSpec) (*AgentClient, error) {
	javaBin := os.Getenv("OPSXDB_JAVA_BIN")
	if javaBin == "" {
		p, err := exec.LookPath("java")
		if err != nil {
			return nil, errors.New("未找到 java：请安装 JDK 17+ 或设置 OPSXDB_JAVA_BIN")
		}
		javaBin = p
	}
	if err := ensureAgentRuntime(spec); err != nil {
		return nil, err
	}
	home := agentHome()
	gsonJar, err := ensureGson(home)
	if err != nil {
		return nil, err
	}
	agentJar := os.Getenv("OPSXDB_AGENT_JAR")
	if agentJar == "" {
		agentJar = filepath.Join(agentDir(), agentJarName)
	}
	jarsDir := agentJarsDir(spec)
	mainClass := spec.MainClass
	if mainClass == "" {
		mainClass = "OpsxJdbcAgent"
	}
	classPath := agentJar + string(os.PathListSeparator) + gsonJar
	if mainClass != "OpsxJdbcAgent" {
		// 专属 agent（如 OpsxZkAgent）的主类直接依赖厂商 jar，须上应用类路径；
		// JDBC 桥则用 --jars 隔离加载驱动，避免与 gson 等类冲突。
		if jars, _ := filepath.Glob(filepath.Join(jarsDir, "*.jar")); len(jars) > 0 {
			classPath += string(os.PathListSeparator) + strings.Join(jars, string(os.PathListSeparator))
		}
	}
	args := append([]string{}, spec.JVMArgs...)
	args = append(args, "-cp", classPath, mainClass, "--jars", jarsDir)
	cmd := exec.Command(javaBin, args...)
	return StartAgentFromCmd(cmd, spec)
}

// StartAgentFromCmd 以自定义命令启动 agent（测试注入 stub 进程用）。
func StartAgentFromCmd(cmd *exec.Cmd, spec AgentSpec) (*AgentClient, error) {
	stdin, err := cmd.StdinPipe()
	if err != nil {
		return nil, err
	}
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return nil, err
	}
	cli := &AgentClient{
		spec:       spec,
		cmd:        cmd,
		stdin:      stdin,
		pending:    map[uint64]pendingCall{},
		stderrTail: newCappedBuffer(16 * 1024),
		readerDone: make(chan struct{}),
	}
	cmd.Stderr = cli.stderrTail
	if err := cmd.Start(); err != nil {
		return nil, fmt.Errorf("启动 agent 进程失败: %w", err)
	}
	go cli.readerLoop(bufio.NewReader(stdout))

	// 握手：Java 侧启动即推送一行 hello 公告
	type hello struct {
		Agent    string `json:"agent"`
		Version  string `json:"version"`
		Protocol int    `json:"protocol"`
	}
	ctx, cancel := context.WithTimeout(context.Background(), agentHelloTimeout)
	defer cancel()
	var h hello
	if err := cli.call(ctx, "hello", nil, &h); err != nil {
		cli.Kill()
		return nil, fmt.Errorf("agent 握手失败: %w（stderr: %s）", err, cli.stderrTail.String())
	}
	if h.Agent != "opsx-agent-jdbc" || h.Protocol != agentProtocolVersion {
		cli.Kill()
		return nil, fmt.Errorf("agent 协议不匹配: agent=%s protocol=%d（期望 protocol=%d）", h.Agent, h.Protocol, agentProtocolVersion)
	}
	return cli, nil
}

// readerLoop 单 goroutine 读 stdout，按 id 路由响应。
func (c *AgentClient) readerLoop(r *bufio.Reader) {
	defer close(c.readerDone)
	for {
		line, err := r.ReadBytes('\n')
		if len(line) > 0 {
			var resp agentResponse
			if uerr := json.Unmarshal(line, &resp); uerr == nil {
				c.pendingMu.Lock()
				pc, ok := c.pending[resp.ID]
				if ok {
					delete(c.pending, resp.ID)
				}
				c.pendingMu.Unlock()
				if ok {
					pc.ch <- resp
				}
			}
		}
		if err != nil {
			c.pendingMu.Lock()
			for id, pc := range c.pending {
				delete(c.pending, id)
				pc.ch <- agentResponse{ID: id, OK: false, Error: "agent 进程已退出"}
			}
			c.pendingMu.Unlock()
			return
		}
	}
}

// call 串行写请求 + 等待响应；读由 readerLoop 并发分发。
func (c *AgentClient) call(ctx context.Context, op string, args map[string]any, out any) error {
	id := c.idGen.Add(1)
	ch := make(chan agentResponse, 1)
	c.pendingMu.Lock()
	c.pending[id] = pendingCall{ch: ch}
	c.pendingMu.Unlock()

	payload, err := json.Marshal(agentRequest{ID: id, Op: op, Args: args})
	if err != nil {
		return err
	}
	writeMu.Lock()
	_, werr := c.stdin.Write(append(payload, '\n'))
	writeMu.Unlock()
	if werr != nil {
		c.pendingMu.Lock()
		delete(c.pending, id)
		c.pendingMu.Unlock()
		return fmt.Errorf("写入 agent 失败: %w", werr)
	}

	select {
	case resp := <-ch:
		if !resp.OK {
			return fmt.Errorf("agent: %s", resp.Error)
		}
		if out != nil && len(resp.Result) > 0 {
			if uerr := json.Unmarshal(resp.Result, out); uerr != nil {
				return fmt.Errorf("解析 agent 响应失败: %w", uerr)
			}
		}
		return nil
	case <-ctx.Done():
		c.pendingMu.Lock()
		delete(c.pending, id)
		c.pendingMu.Unlock()
		return ctx.Err()
	}
}

// 全进程共享的写锁：多个 AgentClient（不同 JVM）并发写各自 stdin 不需要互斥，
// 但同一客户端的并发 call 必须串行写，避免行交错。
var writeMu sync.Mutex

// Kill 立即终止 JVM（不优雅 shutdown；句柄 close 由 withAgent 负责）。
func (c *AgentClient) Kill() {
	c.closeOnce.Do(func() {
		_ = c.stdin.Close()
		if c.cmd.Process != nil {
			_ = c.cmd.Process.Kill()
		}
		<-c.readerDone
	})
}

// Shutdown 优雅退出（op shutdown 后进程自退）。
func (c *AgentClient) Shutdown(ctx context.Context) {
	ctx2, cancel := context.WithTimeout(ctx, 3*time.Second)
	defer cancel()
	_ = c.call(ctx2, "shutdown", nil, nil)
	c.Kill()
}

// ---------- agent 进程池（每驱动一个 JVM，长驻复用） ----------

var agentPool = struct {
	sync.Mutex
	m map[string]*AgentClient
}{m: map[string]*AgentClient{}}

func getAgentClient(d string) (*AgentClient, error) {
	spec, ok := agentSpec(d)
	if !ok {
		return nil, unsupportedDriverErr(ConnReq{Driver: d})
	}
	agentPool.Lock()
	defer agentPool.Unlock()
	if cli, ok := agentPool.m[spec.DriverKey]; ok {
		return cli, nil
	}
	cli, err := StartAgent(spec)
	if err != nil {
		return nil, err
	}
	agentPool.m[spec.DriverKey] = cli
	return cli, nil
}

// ---------- URL 与连接打开 ----------

// renderAgentURL 按模板渲染 jdbc url（达梦: jdbc:dm://host:port）。
func renderAgentURL(spec AgentSpec, c ConnReq) (string, error) {
	repl := strings.NewReplacer(
		"{{.Host}}", c.Host,
		"{{.Port}}", fmt.Sprintf("%d", c.Port),
		"{{.Database}}", c.Database,
	)
	url := repl.Replace(spec.URLTemplate)
	if c.Host == "" {
		return "", fmt.Errorf("%s 需要主机地址", spec.DriverKey)
	}
	if c.Port == 0 {
		return "", fmt.Errorf("%s 需要端口", spec.DriverKey)
	}
	return url, nil
}

// agentOpen 在 agent 进程内打开一个 JDBC 连接，返回句柄。
func agentOpen(ctx context.Context, cli *AgentClient, c ConnReq) (string, error) {
	spec := cli.spec
	url, err := renderAgentURL(spec, c)
	if err != nil {
		return "", err
	}
	args := map[string]any{
		"url":      url,
		"user":     c.User,
		"password": c.Password,
		"readOnly": false,
		"db":       c.Database, // 非 JDBC 桥（Mongo DSL）用它路由库；JDBC 桥忽略
	}
	var res struct {
		Handle string `json:"handle"`
	}
	if err := cli.call(ctx, "open", args, &res); err != nil {
		return "", err
	}
	if res.Handle == "" {
		return "", errors.New("agent 未返回句柄")
	}
	return res.Handle, nil
}

// ---------- 与 withDB 对称的 agent 执行入口 ----------

// withAgent 取（或建）agent 连接并执行 fn；会话内按 database 缓存句柄，直连模式用完即关。
func withAgent(c ConnReq, fn func(cli *AgentClient, handle string) error) error {
	if c.SessionID != "" {
		s, err := GetSession(c.SessionID)
		if err != nil {
			return err
		}
		cli, handle, err := s.agentConn(c)
		if err != nil {
			return err
		}
		return fn(cli, handle)
	}
	cli, err := getAgentClient(c.Driver)
	if err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	handle, err := agentOpen(ctx, cli, c)
	if err != nil {
		return err
	}
	defer func() {
		ctx2, cancel2 := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel2()
		_ = cli.call(ctx2, "close", map[string]any{"handle": handle}, nil)
	}()
	return fn(cli, handle)
}

// agentTimeoutSec 从 ctx deadline 推算语句超时（兜底，防慢查询挂死业务库）。
func agentTimeoutSec(ctx context.Context, def int) int {
	if dl, ok := ctx.Deadline(); ok {
		sec := int(time.Until(dl).Seconds()) + 1
		if sec > 0 && sec < 3600 {
			return sec
		}
	}
	return def
}

// ---------- agent 版元数据 / 查询（ops.go 分支调用） ----------

func agentListDatabases(ctx context.Context, c ConnReq) ([]string, error) {
	spec, _ := agentSpec(c.Driver)
	// 无多库概念的驱动（ES）：对象树只挂一个固定库节点
	if spec.DefaultDatabase != "" {
		return []string{spec.DefaultDatabase}, nil
	}
	var out []string
	err := withAgent(c, func(cli *AgentClient, handle string) error {
		var res struct {
			Names []string `json:"names"`
		}
		if err := cli.call(ctx, "metaDatabases", map[string]any{"handle": handle}, &res); err != nil {
			return err
		}
		out = res.Names
		return nil
	})
	if out == nil {
		out = []string{}
	}
	return out, err
}

func agentListTables(ctx context.Context, c ConnReq) ([]string, error) {
	spec, _ := agentSpec(c.Driver)
	// 专属列表 SQL（ES SHOW TABLES）：跑查询取首列
	if spec.TablesSQL != "" {
		var out []string
		err := withAgent(c, func(cli *AgentClient, handle string) error {
			var res struct {
				Columns []string                 `json:"columns"`
				Rows    []map[string]interface{} `json:"rows"`
			}
			if err := cli.call(ctx, "query", map[string]any{
				"handle": handle, "sql": spec.TablesSQL, "maxRows": 10000, "timeoutSec": agentTimeoutSec(ctx, 30),
			}, &res); err != nil {
				return err
			}
			if len(res.Columns) == 0 {
				return nil
			}
			first := res.Columns[0]
			for _, row := range res.Rows {
				if v, ok := row[first].(string); ok && v != "" {
					out = append(out, v)
				}
			}
			return nil
		})
		if out == nil {
			out = []string{}
		}
		return out, err
	}
	var out []string
	err := withAgent(c, func(cli *AgentClient, handle string) error {
		var res struct {
			Names []string `json:"names"`
		}
		if err := cli.call(ctx, "metaTables", map[string]any{"handle": handle, "schema": c.Database}, &res); err != nil {
			return err
		}
		out = res.Names
		return nil
	})
	if out == nil {
		out = []string{}
	}
	return out, err
}

func agentListColumns(ctx context.Context, c ConnReq) ([]ColumnInfo, error) {
	var out []ColumnInfo
	err := withAgent(c, func(cli *AgentClient, handle string) error {
		var res struct {
			Columns []ColumnInfo `json:"columns"`
		}
		if err := cli.call(ctx, "metaColumns", map[string]any{"handle": handle, "schema": c.Database, "table": c.Table}, &res); err != nil {
			return err
		}
		out = res.Columns
		return nil
	})
	if out == nil {
		out = []ColumnInfo{}
	}
	return out, err
}

// agentVerb 提取语句动词：ZK 风格 "GET /path" 取首词；Mongo DSL "db.c.find({...})" 取第二段。
func agentVerb(sql string) string {
	first := strings.TrimSpace(sql)
	if fields := strings.Fields(first); len(fields) > 0 {
		first = fields[0]
	}
	if strings.HasPrefix(first, "db.") {
		parts := strings.SplitN(first, ".", 3)
		if len(parts) == 3 {
			v := parts[2]
			if i := strings.IndexAny(v, "({"); i >= 0 {
				v = v[:i]
			}
			return strings.ToUpper(v)
		}
	}
	if i := strings.IndexAny(first, "({"); i >= 0 {
		first = first[:i]
	}
	return strings.ToUpper(first)
}

// agentExecSQL 对应 execOnDB：查询带 maxRows 硬截断 + offset 分页；非查询返回 affectedRows。
func agentExecSQL(ctx context.Context, c ConnReq, start time.Time) ExecResult {
	var res ExecResult
	err := withAgent(c, func(cli *AgentClient, handle string) error {
		timeout := agentTimeoutSec(ctx, 60)
		// ZK 专属动词路由：GET/LS 走 query（返回行列），CREATE/SET/DELETE 走 exec
		useQuery := isQuerySQL(c.SQL)
		if driverKey(c.Driver) == "zookeeper" {
			verb := agentVerb(c.SQL)
			useQuery = verb == "GET" || verb == "LS"
		}
		if driverKey(c.Driver) == "mongodb" {
			verb := agentVerb(c.SQL)
			useQuery = verb == "FIND" || verb == "COUNT"
		}
		if !useQuery {
			var r struct {
				AffectedRows int64 `json:"affectedRows"`
			}
			if err := cli.call(ctx, "exec", map[string]any{"handle": handle, "sql": c.SQL, "timeoutSec": timeout, "db": c.Database}, &r); err != nil {
				return err
			}
			res = ExecResult{
				Columns:  []string{"affected_rows"},
				Rows:     []map[string]interface{}{{"affected_rows": r.AffectedRows}},
				RowCount: 1,
				TimeMs:   time.Since(start).Milliseconds(),
			}
			return nil
		}
		limit := rowLimit(c)
		offset := c.Offset
		if offset < 0 {
			offset = 0
		}
		if offset+limit > agentMaxRows {
			return fmt.Errorf("agent 驱动暂不支持深分页（offset+limit ≤ %d）", agentMaxRows)
		}
		var r struct {
			Columns   []string                 `json:"columns"`
			Rows      []map[string]interface{} `json:"rows"`
			Truncated bool                     `json:"truncated"`
		}
		if err := cli.call(ctx, "query", map[string]any{
			"handle": handle, "sql": c.SQL,
			"maxRows": offset + limit, "timeoutSec": timeout, "db": c.Database,
		}, &r); err != nil {
			return err
		}
		rows := r.Rows
		truncated := r.Truncated
		if offset > 0 {
			if len(rows) > offset {
				rows = rows[offset:]
			} else {
				rows = []map[string]interface{}{}
			}
		}
		if rows == nil {
			rows = []map[string]interface{}{}
		}
		res = ExecResult{
			Columns:   r.Columns,
			Rows:      rows,
			RowCount:  len(rows),
			TimeMs:    time.Since(start).Milliseconds(),
			Truncated: truncated,
			Limit:     limit,
			Offset:    offset,
		}
		return nil
	})
	if err != nil {
		return execFail(start, err)
	}
	return res
}

func agentExplain(ctx context.Context, c ConnReq, sqlText string) ExplainResult {
	// v1：通用 EXPLAIN 前缀（达梦支持）；方言化计划格式后续按驱动补。
	cc := c
	cc.SQL = "EXPLAIN " + sqlText
	cc.Limit = 200
	var res ExplainResult
	err := withAgent(cc, func(cli *AgentClient, handle string) error {
		var r struct {
			Columns []string                 `json:"columns"`
			Rows    []map[string]interface{} `json:"rows"`
		}
		if err := cli.call(ctx, "query", map[string]any{
			"handle": handle, "sql": cc.SQL, "maxRows": 200, "timeoutSec": agentTimeoutSec(ctx, 30),
		}, &r); err != nil {
			return err
		}
		res = ExplainResult{Columns: r.Columns, Rows: r.Rows}
		return nil
	})
	if err != nil {
		return ExplainResult{Error: errMsg(err)}
	}
	return res
}

// agentPing 供 TestConnection / OpenSession 探活。
func agentPing(c ConnReq) error {
	return withAgent(c, func(cli *AgentClient, handle string) error {
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		return cli.call(ctx, "ping", map[string]any{"handle": handle}, nil)
	})
}

// ---------- agent 运行环境（全部在用户家目录，连接时按需落盘） ----------

// agentHome 根目录（~/.opsxdb）
func agentHome() string {
	if h := os.Getenv("OPSXDB_HOME"); h != "" {
		return h
	}
	home, err := os.UserHomeDir()
	if err != nil {
		return ".opsxdb"
	}
	return filepath.Join(home, ".opsxdb")
}

// agentDir agent jar 目录：~/.opsxdb/agent
func agentDir() string {
	return filepath.Join(agentHome(), "agent")
}

// agentJarsDir 厂商驱动 jar 目录：~/.opsxdb/drivers/<key>（用户家目录，DMG 不打包）
func agentJarsDir(spec AgentSpec) string {
	if v := os.Getenv(spec.JarsDirEnv); v != "" {
		return v
	}
	return filepath.Join(agentHome(), "drivers", spec.DriverKey)
}

// ensureAgentRuntime 连接时按需下载：agent jar（双托管源 + sha256）与开源驱动 jar（Maven 镜像 + sha256）。
// 商用驱动不在下载范围，由后续 AgentConfigured 给出 manualSource 指引。
// 网络失败时回退离线驱动包（~/.opsxdb/offline-driver-pack.zip，pack.sh 产物）。
func ensureAgentRuntime(spec AgentSpec) error {
	if err := ensureAgentJar(); err != nil {
		return err
	}
	artifacts, ok := agentDriverArtifacts[spec.DriverKey]
	if !ok {
		return nil // 商用/未知驱动：不代下，由 AgentConfigured 校验并给指引
	}
	dir := agentJarsDir(spec)
	if err := os.MkdirAll(dir, 0o755); err != nil {
		return err
	}
	for _, a := range artifacts {
		dest := filepath.Join(dir, a.File)
		if _, err := os.Stat(dest); err == nil {
			continue
		}
		if a.Maven == "" || a.SHA256 == "" {
			continue
		}
		if err := downloadMavenAllMirrors(a, dest); err != nil {
			// 离线/内网：从离线驱动包导入
			if impErr := importFromOfflinePack(map[string]string{
				"drivers/" + spec.DriverKey + "/" + a.File: dest,
			}); impErr == nil {
				continue
			}
			return fmt.Errorf("%s 驱动 jar 下载失败: %w（离线环境可放置离线包 ~/.opsxdb/offline-driver-pack.zip 或手动放 %s）",
				spec.DriverKey, err, dir)
		}
	}
	return nil
}

// downloadMavenAllMirrors 按镜像顺序下载并校验；全部失败返回最后一个错误。
func downloadMavenAllMirrors(a agentArtifact, dest string) error {
	mirrors := []string{
		os.Getenv("MAVEN_MIRROR"),
		"https://maven.aliyun.com/repository/central",
		"https://repo1.maven.org/maven2",
	}
	var lastErr error
	for _, base := range mirrors {
		if base == "" {
			continue
		}
		if err := downloadPinned(base+"/"+mavenPath(a.Maven), dest, a.SHA256); err == nil {
			return nil
		} else {
			lastErr = err
		}
	}
	return lastErr
}

// offlinePackPath 离线驱动包位置（OPSXDB_OFFLINE_ZIP 可覆盖）。
func offlinePackPath() string {
	if v := os.Getenv("OPSXDB_OFFLINE_ZIP"); v != "" {
		return v
	}
	return filepath.Join(agentHome(), "offline-driver-pack.zip")
}

// importFromOfflinePack 从离线包按 manifest 校验后提取文件。
// files: zip 内相对路径 → 本地目标路径。
func importFromOfflinePack(files map[string]string) error {
	zr, err := zip.OpenReader(offlinePackPath())
	if err != nil {
		return err
	}
	defer zr.Close()
	// 先读 manifest 建 hash 表
	hashes := map[string]string{}
	for _, f := range zr.File {
		if f.Name != "pack/manifest.json" {
			continue
		}
		rc, err := f.Open()
		if err != nil {
			return err
		}
		var m struct {
			Files []struct {
				Path   string `json:"path"`
				SHA256 string `json:"sha256"`
			} `json:"files"`
		}
		err = json.NewDecoder(rc).Decode(&m)
		rc.Close()
		if err != nil {
			return err
		}
		for _, e := range m.Files {
			hashes["pack/"+e.Path] = e.SHA256
		}
		break
	}
	for rel, dest := range files {
		rel = "pack/" + strings.TrimPrefix(rel, "pack/")
		want, ok := hashes[rel]
		if !ok {
			return fmt.Errorf("离线包 %s 中不存在 %s", offlinePackPath(), rel)
		}
		for _, f := range zr.File {
			if f.Name != rel {
				continue
			}
			rc, err := f.Open()
			if err != nil {
				return err
			}
			data, err := io.ReadAll(io.LimitReader(rc, 64<<20))
			rc.Close()
			if err != nil {
				return err
			}
			if got := fmt.Sprintf("%x", sha256.Sum256(data)); got != want {
				return fmt.Errorf("离线包 %s 校验失败: %s", rel, got)
			}
			if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
				return err
			}
			if err := os.WriteFile(dest, data, 0o644); err != nil {
				return err
			}
			return nil
		}
		return fmt.Errorf("离线包 %s 中找不到 %s", offlinePackPath(), rel)
	}
	return nil
}

// ensureAgentJar agent jar：本地缺时先网络下载，失败回退离线包。
func ensureAgentJar() error {
	agentJar := os.Getenv("OPSXDB_AGENT_JAR")
	if agentJar != "" {
		return nil // 显式指定，由调用方负责存在性
	}
	agentJar = filepath.Join(agentDir(), agentJarName)
	if _, err := os.Stat(agentJar); err == nil {
		return nil
	}
	if err := os.MkdirAll(agentDir(), 0o755); err != nil {
		return err
	}
	var lastErr error
	for _, u := range strings.Split(agentJarURLs, ",") {
		if err := downloadPinned(u, agentJar, agentJarSHA256); err == nil {
			return nil
		} else {
			lastErr = err
		}
	}
	if err := importFromOfflinePack(map[string]string{"pack/agent/" + agentJarName: agentJar}); err == nil {
		return nil
	}
	return fmt.Errorf("agent jar 获取失败: %w（联网下载与离线包均未成功；可手动放置到 %s 或设置 OPSXDB_AGENT_JAR）", lastErr, agentJar)
}

// mavenPath group:artifact:version → org/mongodb/bson/5.2.1/bson-5.2.1.jar
func mavenPath(gav string) string {
	parts := strings.Split(gav, ":")
	if len(parts) != 3 {
		return ""
	}
	return strings.ReplaceAll(parts[0], ".", "/") + "/" + parts[1] + "/" + parts[2] + "/" + parts[1] + "-" + parts[2] + ".jar"
}

// ensureGson 下载并校验 gson（不存在时）。多源回退：Maven Central → 阿里云镜像。
func ensureGson(home string) (string, error) {
	libs := filepath.Join(home, "libs")
	if err := os.MkdirAll(libs, 0o755); err != nil {
		return "", err
	}
	name := fmt.Sprintf("gson-%s.jar", gsonVersion)
	dest := filepath.Join(libs, name)
	if _, err := os.Stat(dest); err == nil {
		return dest, nil
	}
	urls := []string{
		fmt.Sprintf("https://repo1.maven.org/maven2/com/google/code/gson/gson/%s/%s", gsonVersion, name),
		fmt.Sprintf("https://maven.aliyun.com/repository/central/com/google/code/gson/gson/%s/%s", gsonVersion, name),
	}
	var lastErr error
	for _, u := range urls {
		if err := downloadPinned(u, dest, gsonSHA1); err != nil {
			lastErr = err
			continue
		}
		return dest, nil
	}
	return "", fmt.Errorf("gson 下载失败（已按 sha1 钉版本校验）: %w", lastErr)
}

// downloadPinned 下载到临时文件并校验 sha1，通过后原子改名。
func downloadPinned(url, dest, wantHash string) error {
	client := &http.Client{Timeout: 60 * time.Second}
	resp, err := client.Get(url)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("%s → HTTP %d", url, resp.StatusCode)
	}
	tmp := dest + ".tmp"
	f, err := os.Create(tmp)
	if err != nil {
		return err
	}
	// 按期望哈希长度自动选算法：64 位 hex = sha256（驱动/agent 清单），40 位 = sha1（gson 的 .sha1 惯例）
	var h hash.Hash
	if len(wantHash) == 64 {
		h = sha256.New()
	} else {
		h = sha1.New()
	}
	_, copyErr := io.Copy(io.MultiWriter(f, h), resp.Body)
	closeErr := f.Close()
	if copyErr != nil {
		_ = os.Remove(tmp)
		return copyErr
	}
	if closeErr != nil {
		_ = os.Remove(tmp)
		return closeErr
	}
	if got := hex.EncodeToString(h.Sum(nil)); got != wantHash {
		_ = os.Remove(tmp)
		return fmt.Errorf("%s 校验失败: got=%s 期望=%s", url, got, wantHash)
	}
	return os.Rename(tmp, dest)
}

// mongoVerbAsSQL 把 Mongo DSL 写动词映射为 SQL 危险词，复用既有词表与确认流：
// deleteOne/deleteMany → DELETE（无 WHERE 必触发确认），updateOne/updateMany → UPDATE。
func mongoVerbAsSQL(sql string) string {
	repl := strings.NewReplacer(
		"deleteOne", "DELETE",
		"deleteMany", "DELETE",
		"updateOne", "UPDATE",
		"updateMany", "UPDATE",
	)
	return repl.Replace(sql)
}

// ---------- 驱动管理（UI 状态/预装/离线导入） ----------

type JarStatus struct {
	File      string `json:"file"`
	Size      int64  `json:"size"`
	Installed bool   `json:"installed"`
}

type DriverStatus struct {
	ID             string      `json:"id"`
	Label          string      `json:"label"`
	Runtime        string      `json:"runtime"`        // agent
	License        string      `json:"license"`        // oss / proprietary
	AutoFetch      bool       `json:"autoFetch"`      // 开源可自动下载
	Jars           []JarStatus `json:"jars"`
	AllInstalled   bool        `json:"allInstalled"`
	JDKRequired    bool       `json:"jdkRequired"`    // agent 驱动恒 true
	JDKOK          bool        `json:"jdkOk"`
	JavaVersion    string      `json:"javaVersion"`    // 探测到的版本（如 21.0.12）
	AgentInstalled bool       `json:"agentInstalled"` // agent jar 是否就位
	OfflinePack    bool       `json:"offlinePack"`    // 离线驱动包是否就位
}

// DriversStatus 汇总所有 agent 驱动的安装状态，供驱动管理界面展示。
func DriversStatus() []DriverStatus {
	javaBin, _ := exec.LookPath("java")
	javaVersion := ""
	if javaBin != "" {
		if out, err := exec.Command(javaBin, "-version").CombinedOutput(); err == nil {
			javaVersion = parseJavaVersion(string(out))
		}
	}
	_, offlineErr := os.Stat(offlinePackPath())

	out := []DriverStatus{}
	// 按注册表顺序输出（map 遍历乱序，用固定顺序）
	keys := []string{"mongodb", "zookeeper", "dameng"}
	for _, k := range keys {
		spec, ok := agentDriverSpecs[k]
		if !ok {
			continue
		}
		artifacts, auto := agentDriverArtifacts[k]
		st := DriverStatus{
			ID:             spec.DriverKey,
			Label:          spec.DriverKey,
			Runtime:        "agent",
			AutoFetch:      auto,
			JDKRequired:    true,
			JDKOK:          javaBin != "",
			JavaVersion:    javaVersion,
			OfflinePack:    offlineErr == nil,
			AgentInstalled: fileExists(filepath.Join(agentDir(), agentJarName)),
		}
		if os.Getenv("OPSXDB_AGENT_JAR") != "" {
			st.AgentInstalled = fileExists(os.Getenv("OPSXDB_AGENT_JAR"))
		}
		if st.AutoFetch {
			st.License = "oss"
			dir := agentJarsDir(spec)
			for _, a := range artifacts {
				p := filepath.Join(dir, a.File)
				js := JarStatus{File: a.File, Installed: fileExists(p)}
				if fi, err := os.Stat(p); err == nil {
					js.Size = fi.Size()
				}
				st.Jars = append(st.Jars, js)
			}
			st.AllInstalled = st.AgentInstalled && everyJar(st.Jars)
		} else {
			st.License = "proprietary"
			// 商用：只报告目录里是否已有 jar
			dir := agentJarsDir(spec)
			entries, _ := os.ReadDir(dir)
			n := 0
			for _, e := range entries {
				if !e.IsDir() && strings.HasSuffix(strings.ToLower(e.Name()), ".jar") {
					n++
				}
			}
			st.License = "proprietary"
			st.Jars = []JarStatus{}
			st.AllInstalled = st.AgentInstalled && n > 0
		}
		out = append(out, st)
	}
	return out
}

// InstallDriver 预装指定驱动的运行时（agent jar + 开源驱动 jar），供驱动管理界面"安装"按钮。
func InstallDriver(driverKey string) error {
	spec, ok := agentDriverSpecs[driverKey]
	if !ok {
		return fmt.Errorf("未知 agent 驱动: %s", driverKey)
	}
	if _, err := exec.LookPath(os.Getenv("OPSXDB_JAVA_BIN")); err != nil && os.Getenv("OPSXDB_JAVA_BIN") == "" {
		if _, err := exec.LookPath("java"); err != nil {
			return fmt.Errorf("未找到 java，请先安装 JDK 17+")
		}
	}
	return ensureAgentRuntime(spec)
}

// ImportOfflinePack 全量导入离线驱动包（上传 zip → 校验 manifest → 落盘 ~/.opsxdb/）。
func ImportOfflinePack(zipPath string) (int, error) {
	zr, err := zip.OpenReader(zipPath)
	if err != nil {
		return 0, err
	}
	defer zr.Close()
	hashes := map[string]string{}
	var entries []*zip.File
	for _, f := range zr.File {
		entries = append(entries, f)
		if f.Name != "pack/manifest.json" {
			continue
		}
		rc, err := f.Open()
		if err != nil {
			return 0, err
		}
		var m struct {
			Files []struct {
				Path   string `json:"path"`
				SHA256 string `json:"sha256"`
			} `json:"files"`
		}
		err = json.NewDecoder(rc).Decode(&m)
		rc.Close()
		if err != nil {
			return 0, err
		}
		for _, e := range m.Files {
			hashes["pack/"+e.Path] = e.SHA256
		}
	}
	if len(hashes) == 0 {
		return 0, fmt.Errorf("离线包缺 manifest.json")
	}
	home := agentHome()
	imported := 0
	for _, f := range entries {
		want, ok := hashes[f.Name]
		if !ok || strings.HasSuffix(f.Name, "/") {
			continue
		}
		rel := strings.TrimPrefix(f.Name, "pack/")
		// 只允许写入白名单前缀，防 zip 路径穿越
		if !strings.HasPrefix(rel, "agent/") && !strings.HasPrefix(rel, "libs/") && !strings.HasPrefix(rel, "drivers/") {
			continue
		}
		rc, err := f.Open()
		if err != nil {
			return imported, err
		}
		data, err := io.ReadAll(io.LimitReader(rc, 64<<20))
		rc.Close()
		if err != nil {
			return imported, err
		}
		if got := fmt.Sprintf("%x", sha256.Sum256(data)); got != want {
			return imported, fmt.Errorf("%s 校验失败", rel)
		}
		dest := filepath.Join(home, filepath.FromSlash(rel))
		if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
			return imported, err
		}
		if err := os.WriteFile(dest, data, 0o644); err != nil {
			return imported, err
		}
		imported++
	}
	return imported, nil
}

func fileExists(p string) bool {
	_, err := os.Stat(p)
	return err == nil
}

func everyJar(js []JarStatus) bool {
	for _, j := range js {
		if !j.Installed {
			return false
		}
	}
	return true
}

// parseJavaVersion 从 `java -version` 输出提取版本号（输出在 stderr/stdout 第一行）。
func parseJavaVersion(out string) string {
	for _, line := range strings.Split(out, "\n") {
		if i := strings.Index(line, "version \""); i >= 0 {
			rest := line[i+len("version \""):]
			if j := strings.Index(rest, "\""); j > 0 {
				return rest[:j]
			}
		}
	}
	return ""
}