// Package gateway 实现 Collector Gateway：接收 Agent 上报、提供指标/日志
// 查询 REST API，并维护内存存储与已确认序号。
package gateway

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

	"google.golang.org/protobuf/proto"

	"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/config"
	"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/model"
	"git.opencomputing.cn/yumoqing/host-metrics-collector/pkg/collectorpb"
	"git.opencomputing.cn/yumoqing/host-metrics-collector/pkg/convert"
)

const (
	headerAgentToken = "X-Agent-Token"
	headerHostID     = "X-Host-Id"

	contentTypeProtobuf = "application/x-protobuf"
	contentTypeJSON     = "application/json"
)

// Server 是 Gateway 的 HTTP 处理器，同时暴露：
//   - Agent 上报接口（protobuf）：/v1/agent/*
//   - REST 查询接口（JSON）：/api/v1/*
//   - 健康检查：/healthz、/readyz
type Server struct {
	cfg   config.GatewayConfig
	store *Store
}

// NewServer 构建 Gateway HTTP 处理器。
func NewServer(cfg config.GatewayConfig, store *Store) *Server {
	return &Server{cfg: cfg, store: store}
}

// Handler 返回路由处理器。
func (s *Server) Handler() http.Handler {
	mux := http.NewServeMux()

	// Agent 上报接口（protobuf，Token 认证）。
	mux.HandleFunc("POST /v1/agent/heartbeat", s.requireAgentToken(s.handleHeartbeat))
	mux.HandleFunc("POST /v1/agent/metrics", s.requireAgentToken(s.handleAgentMetrics))
	mux.HandleFunc("POST /v1/agent/logs", s.requireAgentToken(s.handleAgentLogs))
	mux.HandleFunc("GET /v1/agent/checkpoint/{host_id}", s.requireAgentToken(s.handleCheckpoint))

	// 健康检查。
	mux.HandleFunc("GET /healthz", s.handleHealthz)
	mux.HandleFunc("GET /readyz", s.handleReadyz)

	// REST 查询接口（JSON）。
	mux.HandleFunc("GET /api/v1/hosts/{host_id}/metrics", s.handleHostMetrics)
	mux.HandleFunc("GET /api/v1/hosts/{host_id}/logs", s.handleHostLogs)
	mux.HandleFunc("GET /api/v1/metrics/query/instant", s.handleMetricsInstant)
	mux.HandleFunc("GET /api/v1/metrics/query", s.handleMetricsQuery)
	mux.HandleFunc("POST /api/v1/logs/search", s.handleLogsSearch)

	return mux
}

func (s *Server) requireAgentToken(next http.HandlerFunc) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		if s.cfg.AgentToken != "" && r.Header.Get(headerAgentToken) != s.cfg.AgentToken {
			writeProto(w, http.StatusUnauthorized, &collectorpb.Ack{Ok: false, Error: "unauthorized"})
			return
		}
		next(w, r)
	}
}

func (s *Server) handleHeartbeat(w http.ResponseWriter, r *http.Request) {
	body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20))
	if err != nil {
		writeProto(w, http.StatusBadRequest, &collectorpb.HeartbeatResponse{Ok: false})
		return
	}
	req := &collectorpb.HeartbeatRequest{}
	if err := proto.Unmarshal(body, req); err != nil {
		writeProto(w, http.StatusBadRequest, &collectorpb.HeartbeatResponse{Ok: false})
		return
	}
	writeProto(w, http.StatusOK, &collectorpb.HeartbeatResponse{
		Ok:         true,
		ServerTime: time.Now().Unix(),
		Config:     map[string]string{},
	})
}

func (s *Server) handleAgentMetrics(w http.ResponseWriter, r *http.Request) {
	body, err := io.ReadAll(io.LimitReader(r.Body, 64<<20))
	if err != nil {
		writeProto(w, http.StatusBadRequest, &collectorpb.Ack{Ok: false, Error: "read body"})
		return
	}
	batch := &collectorpb.MetricBatch{}
	if err := proto.Unmarshal(body, batch); err != nil {
		writeProto(w, http.StatusBadRequest, &collectorpb.Ack{Ok: false, Error: "invalid protobuf"})
		return
	}
	if batch.HostId == "" {
		writeProto(w, http.StatusBadRequest, &collectorpb.Ack{Ok: false, Seq: batch.Seq, Error: "host_id is required"})
		return
	}

	samples := make([]model.Sample, 0, len(batch.Samples))
	for _, pb := range batch.Samples {
		samples = append(samples, convert.SampleFromPB(pb))
	}
	s.store.AddMetrics(batch.HostId, batch.Seq, samples)
	s.store.SetLastAckedSeq(batch.HostId, batch.Seq)

	writeProto(w, http.StatusOK, &collectorpb.Ack{Ok: true, Seq: batch.Seq})
}

func (s *Server) handleAgentLogs(w http.ResponseWriter, r *http.Request) {
	body, err := io.ReadAll(io.LimitReader(r.Body, 64<<20))
	if err != nil {
		writeProto(w, http.StatusBadRequest, &collectorpb.Ack{Ok: false, Error: "read body"})
		return
	}
	batch := &collectorpb.LogBatch{}
	if err := proto.Unmarshal(body, batch); err != nil {
		writeProto(w, http.StatusBadRequest, &collectorpb.Ack{Ok: false, Error: "invalid protobuf"})
		return
	}
	if batch.HostId == "" {
		writeProto(w, http.StatusBadRequest, &collectorpb.Ack{Ok: false, Seq: batch.Seq, Error: "host_id is required"})
		return
	}

	entries := make([]model.LogRecord, 0, len(batch.Entries))
	for _, pb := range batch.Entries {
		entries = append(entries, convert.LogFromPB(pb))
	}
	s.store.AddLogs(batch.HostId, batch.Seq, entries)
	s.store.SetLastAckedSeq(batch.HostId, batch.Seq)

	writeProto(w, http.StatusOK, &collectorpb.Ack{Ok: true, Seq: batch.Seq})
}

func (s *Server) handleCheckpoint(w http.ResponseWriter, r *http.Request) {
	hostID := r.PathValue("host_id")
	writeProto(w, http.StatusOK, &collectorpb.CheckpointResponse{LastAckedSeq: s.store.LastAckedSeq(hostID)})
}

func (s *Server) handleHealthz(w http.ResponseWriter, r *http.Request) {
	writeJSON(w, http.StatusOK, 0, "ok", map[string]string{"status": "ok"})
}

func (s *Server) handleReadyz(w http.ResponseWriter, r *http.Request) {
	writeJSON(w, http.StatusOK, 0, "ok", map[string]string{"status": "ready"})
}

func (s *Server) handleHostMetrics(w http.ResponseWriter, r *http.Request) {
	hostID := r.PathValue("host_id")
	q := r.URL.Query()

	from, to, err := parseRange(q.Get("from"), q.Get("to"))
	if err != nil {
		writeJSON(w, http.StatusBadRequest, 40001, err.Error(), nil)
		return
	}

	series, err := s.store.HostMetrics(hostID, q.Get("metric"), from, to)
	if err != nil {
		writeJSON(w, http.StatusInternalServerError, 50001, err.Error(), nil)
		return
	}
	writeJSON(w, http.StatusOK, 0, "ok", series)
}

func (s *Server) handleHostLogs(w http.ResponseWriter, r *http.Request) {
	hostID := r.PathValue("host_id")
	q := r.URL.Query()

	from, to, err := parseRange(q.Get("from"), q.Get("to"))
	if err != nil {
		writeJSON(w, http.StatusBadRequest, 40001, err.Error(), nil)
		return
	}
	page, pageSize, err := parsePagination(q.Get("page"), q.Get("page_size"))
	if err != nil {
		writeJSON(w, http.StatusBadRequest, 40001, err.Error(), nil)
		return
	}

	total, items, err := s.store.HostLogs(hostID, q.Get("level"), q.Get("keyword"), from, to, page, pageSize)
	if err != nil {
		writeJSON(w, http.StatusInternalServerError, 50001, err.Error(), nil)
		return
	}
	writeJSON(w, http.StatusOK, 0, "ok", map[string]any{"total": total, "items": items})
}

func (s *Server) handleMetricsQuery(w http.ResponseWriter, r *http.Request) {
	q := r.URL.Query()
	from, to, err := parseRange(q.Get("from"), q.Get("to"))
	if err != nil {
		writeJSON(w, http.StatusBadRequest, 40001, err.Error(), nil)
		return
	}
	labels, err := parseLabels(q.Get("labels"))
	if err != nil {
		writeJSON(w, http.StatusBadRequest, 40001, err.Error(), nil)
		return
	}

	series, err := s.store.QueryMetrics(q.Get("metric"), labels, from, to)
	if err != nil {
		writeJSON(w, http.StatusInternalServerError, 50001, err.Error(), nil)
		return
	}
	writeJSON(w, http.StatusOK, 0, "ok", map[string]any{"resultType": "matrix", "result": series})
}

func (s *Server) handleMetricsInstant(w http.ResponseWriter, r *http.Request) {
	q := r.URL.Query()
	labels, err := parseLabels(q.Get("labels"))
	if err != nil {
		writeJSON(w, http.StatusBadRequest, 40001, err.Error(), nil)
		return
	}

	points, err := s.store.InstantMetrics(q.Get("metric"), labels)
	if err != nil {
		writeJSON(w, http.StatusInternalServerError, 50001, err.Error(), nil)
		return
	}
	writeJSON(w, http.StatusOK, 0, "ok", map[string]any{"resultType": "vector", "result": points})
}

func (s *Server) handleLogsSearch(w http.ResponseWriter, r *http.Request) {
	var req struct {
		HostIDs  []string `json:"host_ids"`
		Level    string   `json:"level"`
		Keyword  string   `json:"keyword"`
		From     string   `json:"from"`
		To       string   `json:"to"`
		Page     int      `json:"page"`
		PageSize int      `json:"page_size"`
	}
	if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&req); err != nil {
		writeJSON(w, http.StatusBadRequest, 40001, "invalid json body", nil)
		return
	}

	from, to, err := parseRange(req.From, req.To)
	if err != nil {
		writeJSON(w, http.StatusBadRequest, 40001, err.Error(), nil)
		return
	}
	page, pageSize := req.Page, req.PageSize
	if page < 1 {
		page = 1
	}
	if pageSize < 1 {
		pageSize = 20
	}

	total, items, err := s.store.SearchLogs(req.HostIDs, req.Level, req.Keyword, from, to, page, pageSize)
	if err != nil {
		writeJSON(w, http.StatusInternalServerError, 50001, err.Error(), nil)
		return
	}
	writeJSON(w, http.StatusOK, 0, "ok", map[string]any{"total": total, "items": items})
}

// --- 序列化辅助 ---

func writeProto(w http.ResponseWriter, status int, msg proto.Message) {
	data, err := proto.Marshal(msg)
	if err != nil {
		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
		w.WriteHeader(http.StatusInternalServerError)
		_, _ = io.WriteString(w, "marshal response failed")
		return
	}
	w.Header().Set("Content-Type", contentTypeProtobuf)
	w.WriteHeader(status)
	_, _ = w.Write(data)
}

func writeJSON(w http.ResponseWriter, status, code int, msg string, data any) {
	w.Header().Set("Content-Type", contentTypeJSON)
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(map[string]any{
		"code":    code,
		"message": msg,
		"data":    data,
	})
}

func parseRange(fromStr, toStr string) (time.Time, time.Time, error) {
	from, err := parseTime(fromStr)
	if err != nil {
		return time.Time{}, time.Time{}, fmt.Errorf("invalid from: %w", err)
	}
	to, err := parseTime(toStr)
	if err != nil {
		return time.Time{}, time.Time{}, fmt.Errorf("invalid to: %w", err)
	}
	if !from.IsZero() && !to.IsZero() && from.After(to) {
		return time.Time{}, time.Time{}, fmt.Errorf("from must be before to")
	}
	return from, to, nil
}

func parseTime(s string) (time.Time, error) {
	if s == "" {
		return time.Time{}, nil
	}
	// 优先按 Unix 秒解析，便于与内部时间戳一致。
	if n, err := strconv.ParseInt(s, 10, 64); err == nil {
		return time.Unix(n, 0), nil
	}
	if t, err := time.Parse(time.RFC3339, s); err == nil {
		return t, nil
	}
	if t, err := time.Parse(time.RFC3339Nano, s); err == nil {
		return t, nil
	}
	return time.Time{}, fmt.Errorf("invalid time %q", s)
}

func parsePagination(pageStr, sizeStr string) (int, int, error) {
	page := 1
	if pageStr != "" {
		n, err := strconv.Atoi(pageStr)
		if err != nil || n < 1 {
			return 0, 0, fmt.Errorf("invalid page %q", pageStr)
		}
		page = n
	}
	pageSize := 20
	if sizeStr != "" {
		n, err := strconv.Atoi(sizeStr)
		if err != nil || n < 1 || n > 200 {
			return 0, 0, fmt.Errorf("invalid page_size %q", sizeStr)
		}
		pageSize = n
	}
	return page, pageSize, nil
}

func parseLabels(s string) (map[string]string, error) {
	if strings.TrimSpace(s) == "" {
		return nil, nil
	}
	out := map[string]string{}
	for _, part := range strings.Split(s, ",") {
		part = strings.TrimSpace(part)
		if part == "" {
			continue
		}
		kv := strings.SplitN(part, "=", 2)
		if len(kv) != 2 {
			return nil, fmt.Errorf("invalid labels %q, expect k=v", s)
		}
		out[strings.TrimSpace(kv[0])] = strings.TrimSpace(kv[1])
	}
	return out, nil
}