opsxdb/apps/server/internal/db/influx.go

package db

// influx.go —— InfluxDB 2.x 原生接入（HTTP API，零第三方依赖）。
//
// 字段映射（连接表单）：
//
//	主机/端口 → 服务地址（默认 :8086）
//	用户      → org（如 opsx）
//	密码      → admin token（如 opsxdb-demo-token）
//	库名      → bucket（如 demo）
//
// 能力：对象树 bucket → measurement（Flux schema.measurements）；
// SQL 编辑器直写 Flux（from(bucket: ...) |> range(...) ...），结果按注解 CSV 解析为表格。
// Flux 只读（写走 /write 专用通道），危险语句拦截在 ops.go 入口统一过一遍。

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strings"
	"time"
)

func isInflux(d string) bool { return driverKey(d) == "influxdb" }

func ifxBase(c ConnReq) string {
	host := strings.TrimSpace(c.Host)
	if host == "" {
		host = "127.0.0.1"
	}
	port := c.Port
	if port == 0 {
		port = 8086
	}
	return fmt.Sprintf("http://%s:%d", host, port)
}

func ifxToken(c ConnReq) string { return c.Password }

// ifxDo 统一请求：Token 鉴权 + org 查询参数。
func ifxDo(ctx context.Context, c ConnReq, method, path string, body io.Reader, out any) error {
	req, err := http.NewRequestWithContext(ctx, method, ifxBase(c)+path, body)
	if err != nil {
		return err
	}
	if ifxToken(c) != "" {
		req.Header.Set("Authorization", "Token "+ifxToken(c))
	}
	if body != nil {
		req.Header.Set("Content-Type", "application/json")
	}
	resp, err := (&http.Client{Timeout: 30 * time.Second}).Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode >= 400 {
		msg, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
		return fmt.Errorf("InfluxDB HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(msg)))
	}
	if out != nil {
		return json.NewDecoder(resp.Body).Decode(out)
	}
	_, _ = io.Copy(io.Discard, resp.Body)
	return nil
}

func ifxOrg(c ConnReq) (string, error) {
	org := strings.TrimSpace(c.User)
	if org == "" {
		return "", fmt.Errorf("InfluxDB 2.x 需要填「用户」= org 名（如 opsx）")
	}
	return org, nil
}

func ifxPing(c ConnReq) error {
	org, err := ifxOrg(c)
	if err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
	defer cancel()
	var out struct {
		Buckets []struct {
			Name string `json:"name"`
		} `json:"buckets"`
	}
	return ifxDo(ctx, c, http.MethodGet, "/api/v2/buckets?org="+org, nil, &out)
}

func ifxListDatabases(ctx context.Context, c ConnReq) ([]string, error) {
	org, err := ifxOrg(c)
	if err != nil {
		return nil, err
	}
	var out struct {
		Buckets []struct {
			Name string `json:"name"`
		} `json:"buckets"`
	}
	if err := ifxDo(ctx, c, http.MethodGet, "/api/v2/buckets?org="+org, nil, &out); err != nil {
		return nil, err
	}
	names := []string{}
	for _, b := range out.Buckets {
		names = append(names, b.Name)
	}
	return names, nil
}

// ifxQueryFlux 跑 Flux 并把注解 CSV 解析为表格。
func ifxQueryFlux(ctx context.Context, c ConnReq, flux string, maxRows int) (ExecResult, error) {
	org, err := ifxOrg(c)
	if err != nil {
		return ExecResult{}, err
	}
	body := map[string]string{"query": flux, "type": "flux"}
	b, err := json.Marshal(body)
	if err != nil {
		return ExecResult{}, err
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, ifxBase(c)+"/api/v2/query?org="+org, strings.NewReader(string(b)))
	if err != nil {
		return ExecResult{}, err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Accept", "application/csv")
	if ifxToken(c) != "" {
		req.Header.Set("Authorization", "Token "+ifxToken(c))
	}
	resp, err := (&http.Client{Timeout: 60 * time.Second}).Do(req)
	if err != nil {
		return ExecResult{}, err
	}
	defer resp.Body.Close()
	raw, err := io.ReadAll(resp.Body)
	if err != nil {
		return ExecResult{}, err
	}
	if resp.StatusCode >= 400 {
		return ExecResult{}, fmt.Errorf("InfluxDB HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(raw)))
	}
	return parseAnnotatedCSV(string(raw), maxRows)
}

// parseAnnotatedCSV 解析 InfluxDB 注解 CSV：
// #group/#datatype/#default 注释行 + 空行 + header 行 + 数据行。
func parseAnnotatedCSV(text string, maxRows int) (ExecResult, error) {
	lines := strings.Split(strings.ReplaceAll(text, "\r\n", "\n"), "\n")
	headerIdx := -1
	var header []string
	for i, ln := range lines {
		s := strings.TrimSpace(ln)
		if s == "" || strings.HasPrefix(s, "#") {
			continue
		}
		headerIdx = i
		header = strings.Split(ln, ",")
		break
	}
	if headerIdx < 0 {
		return ExecResult{Columns: []string{}, Rows: []map[string]interface{}{}}, nil
	}
	res := ExecResult{Columns: header, Rows: []map[string]interface{}{}}
	for _, ln := range lines[headerIdx+1:] {
		if strings.TrimSpace(ln) == "" {
			continue
		}
		cells := strings.Split(ln, ",")
		row := make(map[string]interface{}, len(header))
		for i, col := range header {
			v := ""
			if i < len(cells) {
				v = cells[i]
			}
			row[col] = v
		}
		res.Rows = append(res.Rows, row)
		if len(res.Rows) >= maxRows {
			res.Truncated = true
			break
		}
	}
	res.RowCount = len(res.Rows)
	return res, nil
}

func ifxListTables(ctx context.Context, c ConnReq) ([]string, error) {
	bucket := strings.TrimSpace(c.Database)
	if bucket == "" {
		return nil, fmt.Errorf("InfluxDB 列表需要库名（bucket）")
	}
	flux := fmt.Sprintf("import \"influxdata/influxdb/schema\"\nschema.measurements(bucket: \"%s\")", strings.ReplaceAll(bucket, "\"", ""))
	res, err := ifxQueryFlux(ctx, c, flux, 10000)
	if err != nil {
		return nil, err
	}
	out := []string{}
	if len(res.Columns) == 0 {
		return out, nil
	}
	// schema.measurements 输出列为 [,result,table,_value]：优先 _value 列，退化为最后一列
	idx := len(res.Columns) - 1
	for i, col := range res.Columns {
		if col == "_value" {
			idx = i
			break
		}
	}
	col := res.Columns[idx]
	for _, row := range res.Rows {
		if v, ok := row[col].(string); ok && v != "" {
			out = append(out, v)
		}
	}
	return out, nil
}

// ifxExecSQL 编辑器内容按 Flux 处理（只读）。
func ifxExecSQL(ctx context.Context, c ConnReq, start time.Time) ExecResult {
	if c.Offset > 0 {
		return ExecResult{Columns: []string{}, Rows: []map[string]interface{}{}, Error: "InfluxDB Flux 结果暂不支持翻页（可缩小 range 重查）", TimeMs: time.Since(start).Milliseconds()}
	}
	limit := rowLimit(c)
	if limit > MaxRowLimit {
		limit = MaxRowLimit
	}
	res, err := ifxQueryFlux(ctx, c, strings.TrimSpace(c.SQL), limit)
	if err != nil {
		return execFail(start, err)
	}
	res.TimeMs = time.Since(start).Milliseconds()
	return res
}

// ifxListColumns measurement 的字段探查：Flux schema.fieldKeys + tagKeys 合并。
func ifxListColumns(ctx context.Context, c ConnReq) ([]ColumnInfo, error) {
	bucket := strings.TrimSpace(c.Database)
	table := strings.TrimSpace(c.Table)
	if bucket == "" || table == "" {
		return nil, fmt.Errorf("InfluxDB 列探查需要库名与 measurement")
	}
	safeB := strings.ReplaceAll(bucket, "\"", "")
	safeM := strings.ReplaceAll(table, "\"", "")
	flux := fmt.Sprintf(`import "influxdata/influxdb/schema"
union(tables: [
  schema.fieldKeys(bucket: "%s", predicate: (r) => r._measurement == "%s"),
  schema.tagKeys(bucket: "%s", predicate: (r) => r._measurement == "%s")
])`, safeB, safeM, safeB, safeM)
	res, err := ifxQueryFlux(ctx, c, flux, 1000)
	if err != nil {
		return nil, err
	}
	cols := []ColumnInfo{}
	seen := map[string]bool{}
	if len(res.Columns) == 0 {
		return cols, nil
	}
	last := res.Columns[len(res.Columns)-1]
	for _, row := range res.Rows {
		name, _ := row[last].(string)
		if name == "" || seen[name] {
			continue
		}
		seen[name] = true
		cols = append(cols, ColumnInfo{Name: name, Type: "tag/field", Nullable: true})
	}
	return cols, nil
}