package chlog

import (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"hash/fnv"
	"io"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"
)

const chTimeLayout = "2006-01-02 15:04:05.999"

type Row struct {
	Timestamp string            `json:"timestamp"`
	Host      string            `json:"host"`
	Source    string            `json:"source"`
	Namespace string            `json:"namespace"`
	Pod       string            `json:"pod"`
	Container string            `json:"container"`
	Stream    string            `json:"stream"`
	Severity  string            `json:"severity"`
	Message   string            `json:"message"`
	Labels    map[string]string `json:"labels"`
	Fields    map[string]string `json:"fields"`
}

func (r Row) Time() time.Time {
	t, err := time.Parse(chTimeLayout, r.Timestamp)
	if err != nil {
		return time.Time{}
	}
	return t.UTC()
}

// Key identifies a row for overlap dedupe during paging and tailing.
func (r Row) Key() uint64 {
	h := fnv.New64a()
	for _, s := range []string{r.Timestamp, r.Host, r.Source, r.Namespace, r.Pod, r.Container, r.Stream, r.Message} {
		_, _ = io.WriteString(h, s)
		h.Write([]byte{0})
	}
	return h.Sum64()
}

type Client struct {
	cfg  Config
	http *http.Client
}

func NewClient(cfg Config) *Client {
	return &Client{cfg: cfg, http: &http.Client{Timeout: 130 * time.Second}}
}

// Run executes the query and streams each result row to fn. The SQL travels
// in the request body; every value goes as a param_* HTTP parameter.
func (c *Client) Run(ctx context.Context, q Query, fn func(Row) error) error {
	v := url.Values{}
	v.Set("default_format", "JSONEachRow")
	for name, value := range q.Params {
		v.Set("param_"+name, value)
	}
	u := strings.TrimRight(c.cfg.URL, "/") + "/?" + v.Encode()

	req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, strings.NewReader(q.SQL))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "text/plain")
	req.Header.Set("X-ClickHouse-User", c.cfg.User)
	if c.cfg.Password != "" {
		req.Header.Set("X-ClickHouse-Key", c.cfg.Password)
	}

	resp, err := c.http.Do(req)
	if err != nil {
		return fmt.Errorf("clickhouse request: %w", err)
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
		return fmt.Errorf("clickhouse HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
	}

	sc := bufio.NewScanner(resp.Body)
	sc.Buffer(make([]byte, 0, 64*1024), 16*1024*1024)
	line := 0
	for sc.Scan() {
		line++
		b := sc.Bytes()
		if len(b) == 0 {
			continue
		}
		var r Row
		if err := json.Unmarshal(b, &r); err != nil {
			return fmt.Errorf("parse result row %s: %w", strconv.Itoa(line), err)
		}
		if err := fn(r); err != nil {
			return err
		}
	}
	return sc.Err()
}