package prom

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime/schema"
	utilnet "k8s.io/apimachinery/pkg/util/net"
	"k8s.io/client-go/kubernetes"
	k8srest "k8s.io/client-go/rest"

	"github.com/sophotechlabs/spinoza/internal/api"
)

const (
	rangePath        = "api/v1/query_range"
	instantPath      = "api/v1/query"
	defaultPort      = "9090"
	maxPort          = 65535
	callTimeout      = 15 * time.Second
	buildPath        = "api/v1/status/buildinfo"
	promNameKey      = "app.kubernetes.io/name"
	promNameVal      = "prometheus"
	maxNamed         = 3
	maxResponseBytes = 8 << 20
)

var selectors = []string{
	"operated-prometheus=true",
	promNameKey + "=" + promNameVal,
}

var ErrUnavailable = errors.New("prometheus is unavailable")

type Target struct {
	Namespace string
	Service   string
	Port      string
	Scheme    string
}

func (t Target) String() string {
	return fmt.Sprintf("%s/%s:%s (%s)", t.Namespace, t.Service, t.Port, t.Scheme)
}

type Proxy interface {
	Get(ctx context.Context, target Target, path string, params map[string]string) ([]byte, error)
}

type serviceProxy struct {
	cs kubernetes.Interface
}

func (s *serviceProxy) Get(ctx context.Context, target Target, path string, params map[string]string) ([]byte, error) {
	bounded, cancel := context.WithTimeout(ctx, callTimeout)
	defer cancel()
	core := s.cs.CoreV1()
	client, ok := core.RESTClient().(*k8srest.RESTClient)
	if !ok || client == nil {
		return nil, errors.New("the kubernetes client cannot safely proxy prometheus")
	}
	proxyRequest := client.Get().
		Namespace(target.Namespace).
		Resource("services").
		SubResource("proxy").
		Name(utilnet.JoinSchemeNamePort(target.Scheme, target.Service, target.Port)).
		Suffix(path)
	for key, value := range params {
		proxyRequest = proxyRequest.Param(key, value)
	}
	endpoint := proxyRequest.URL()
	request, err := http.NewRequestWithContext(bounded, http.MethodGet, endpoint.String(), http.NoBody)
	if err != nil {
		return nil, err
	}
	request.Header.Set("Accept", "application/json")
	httpClient := client.Client
	if httpClient == nil {
		httpClient = http.DefaultClient
	}
	response, err := httpClient.Do(request)
	if err != nil {
		return nil, err
	}
	defer func() { _ = response.Body.Close() }()
	body, err := io.ReadAll(io.LimitReader(response.Body, maxResponseBytes+1))
	if err != nil {
		return nil, err
	}
	if len(body) > maxResponseBytes {
		return nil, fmt.Errorf("prometheus response is larger than %d bytes", maxResponseBytes)
	}
	if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices {
		return nil, apierrors.NewGenericServerResponse(
			response.StatusCode,
			http.MethodGet,
			schema.GroupResource{Resource: "services/proxy"},
			target.Service,
			string(body),
			0,
			false,
		)
	}
	return body, nil
}

type Client struct {
	cs       kubernetes.Interface
	proxy    Proxy
	override Target
	mu       sync.Mutex
	resolved *Target
}

func NewClient(cs kubernetes.Interface, override Target) *Client {
	return &Client{cs: cs, proxy: &serviceProxy{cs: cs}, override: override}
}

func NewClientWithProxy(cs kubernetes.Interface, proxy Proxy, override Target) *Client {
	return &Client{cs: cs, proxy: proxy, override: override}
}

func ParseTarget(spec string) (Target, error) {
	if spec == "" {
		return Target{}, nil
	}
	namespace, rest, found := strings.Cut(spec, "/")
	if !found {
		return Target{}, fmt.Errorf("expected namespace/service:port, got %q", spec)
	}
	service, port, hasPort := strings.Cut(rest, ":")
	if !hasPort {
		port = defaultPort
	}
	if namespace == "" || service == "" {
		return Target{}, fmt.Errorf("expected namespace/service:port, got %q", spec)
	}
	portErr := validPort(port)
	if portErr != nil {
		return Target{}, fmt.Errorf("expected namespace/service:port, got %q: %w", spec, portErr)
	}
	return Target{Namespace: namespace, Service: service, Port: port}, nil
}

func validPort(port string) error {
	number, err := strconv.Atoi(port)
	if err != nil {
		return fmt.Errorf("port %q is not a number", port)
	}
	if number < 1 {
		return fmt.Errorf("port %d is out of range", number)
	}
	if number > maxPort {
		return fmt.Errorf("port %d is out of range", number)
	}
	return nil
}

func (c *Client) Target(ctx context.Context) (Target, error) {
	c.mu.Lock()
	cached := c.resolved
	c.mu.Unlock()
	if cached != nil {
		return *cached, nil
	}

	target, err := c.discover(ctx)
	if err != nil {
		return Target{}, err
	}
	scheme, err := c.probe(ctx, target)
	if err != nil {
		return Target{}, err
	}
	target.Scheme = scheme

	c.mu.Lock()
	c.resolved = &target
	c.mu.Unlock()
	return target, nil
}

func (c *Client) forget(stale Target) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.resolved == nil {
		return
	}
	if *c.resolved != stale {
		return
	}
	c.resolved = nil
}

func (c *Client) discover(ctx context.Context) (Target, error) {
	if c.override.Service != "" {
		return c.override, nil
	}
	for _, selector := range selectors {
		found, err := c.onlyMatch(ctx, selector)
		if err != nil {
			return Target{}, err
		}
		if found.Service != "" {
			return found, nil
		}
	}
	return Target{}, fmt.Errorf("%w: no service matched %s; set --prometheus to namespace/service:port", ErrUnavailable, strings.Join(selectors, " or "))
}

func (c *Client) onlyMatch(ctx context.Context, selector string) (Target, error) {
	list, err := c.cs.CoreV1().Services("").List(ctx, metav1.ListOptions{LabelSelector: selector})
	if err != nil {
		return Target{}, err
	}
	found := make([]Target, 0, len(list.Items))
	for i := range list.Items {
		service := &list.Items[i]
		port := webPort(service.Spec.Ports)
		if port == "" {
			continue
		}
		found = append(found, Target{Namespace: service.Namespace, Service: service.Name, Port: port})
	}
	if len(found) == 0 {
		return Target{}, nil
	}
	if len(found) > 1 {
		return Target{}, fmt.Errorf("%w: %s matches %d services (%s); set --prometheus to the one to read", ErrUnavailable, selector, len(found), nameList(found))
	}
	return found[0], nil
}

func nameList(found []Target) string {
	shown := found
	suffix := ""
	if len(shown) > maxNamed {
		shown = shown[:maxNamed]
		suffix = ", ..."
	}
	names := make([]string, 0, len(shown))
	for _, target := range shown {
		names = append(names, target.Namespace+"/"+target.Service)
	}
	return strings.Join(names, ", ") + suffix
}

func webPort(ports []corev1.ServicePort) string {
	for _, port := range ports {
		if port.Port == 9090 {
			return "9090"
		}
	}
	for _, port := range ports {
		if strings.Contains(port.Name, "web") || strings.Contains(port.Name, "http") {
			return strconv.Itoa(int(port.Port))
		}
	}
	return ""
}

func (c *Client) probe(ctx context.Context, target Target) (string, error) {
	var lastErr error
	for _, scheme := range []string{"https", "http"} {
		probed := target
		probed.Scheme = scheme
		_, err := c.proxy.Get(ctx, probed, buildPath, nil)
		if err == nil {
			return scheme, nil
		}
		if apierrors.IsForbidden(err) {
			return "", fmt.Errorf("%w: this account may not proxy services (verb get on services/proxy): %w", ErrUnavailable, err)
		}
		lastErr = err
	}
	return "", fmt.Errorf("%w: %s did not answer over https or http: %w", ErrUnavailable, target, lastErr)
}

func (c *Client) Range(ctx context.Context, query string, start, end time.Time, step time.Duration) ([]api.MetricPoint, error) {
	target, err := c.Target(ctx)
	if err != nil {
		return nil, err
	}
	params := map[string]string{
		"query": query,
		"start": strconv.FormatInt(start.Unix(), 10),
		"end":   strconv.FormatInt(end.Unix(), 10),
		"step":  strconv.Itoa(int(step.Seconds())),
	}
	raw, err := c.proxy.Get(ctx, target, rangePath, params)
	if err != nil {
		c.forget(target)
		return nil, err
	}
	return decodeRange(raw)
}

type Sample struct {
	Labels map[string]string
	Value  float64
}

func (c *Client) Instant(ctx context.Context, query string, at time.Time) ([]Sample, error) {
	target, err := c.Target(ctx)
	if err != nil {
		return nil, err
	}
	params := map[string]string{
		"query": query,
		"time":  strconv.FormatInt(at.Unix(), 10),
	}
	raw, err := c.proxy.Get(ctx, target, instantPath, params)
	if err != nil {
		c.forget(target)
		return nil, err
	}
	return decodeInstant(raw)
}

type instantResponse struct {
	Status string `json:"status"`
	Error  string `json:"error"`
	Data   struct {
		ResultType string `json:"resultType"`
		Result     []struct {
			Metric map[string]string `json:"metric"`
			Value  []any             `json:"value"`
		} `json:"result"`
	} `json:"data"`
}

func decodeInstant(raw []byte) ([]Sample, error) {
	var body instantResponse
	err := json.Unmarshal(raw, &body)
	if err != nil {
		return nil, fmt.Errorf("prometheus returned an unreadable response: %w", err)
	}
	if body.Status != "success" {
		return nil, fmt.Errorf("prometheus rejected the query: %s", body.Error)
	}
	if body.Data.ResultType != "vector" {
		return nil, fmt.Errorf("prometheus answered with %q, which is not a vector", body.Data.ResultType)
	}
	samples := make([]Sample, 0, len(body.Data.Result))
	for _, entry := range body.Data.Result {
		point, ok := pointOf(entry.Value)
		if !ok {
			continue
		}
		samples = append(samples, Sample{Labels: entry.Metric, Value: point.Value})
	}
	return samples, nil
}

type rangeResponse struct {
	Status string `json:"status"`
	Error  string `json:"error"`
	Data   struct {
		Result []struct {
			Values [][]any `json:"values"`
		} `json:"result"`
	} `json:"data"`
}

func decodeRange(raw []byte) ([]api.MetricPoint, error) {
	var body rangeResponse
	err := json.Unmarshal(raw, &body)
	if err != nil {
		return nil, fmt.Errorf("prometheus returned an unreadable response: %w", err)
	}
	if body.Status != "success" {
		return nil, fmt.Errorf("prometheus rejected the query: %s", body.Error)
	}
	if len(body.Data.Result) == 0 {
		return []api.MetricPoint{}, nil
	}
	if len(body.Data.Result) > 1 {
		return nil, fmt.Errorf(
			"the query matched %d series and a range answers with one; aggregate it, "+
				"with sum() or max() over the labels you do not want",
			len(body.Data.Result),
		)
	}
	values := body.Data.Result[0].Values
	points := make([]api.MetricPoint, 0, len(values))
	for _, pair := range values {
		point, ok := pointOf(pair)
		if !ok {
			continue
		}
		points = append(points, point)
	}
	return points, nil
}

func pointOf(pair []any) (api.MetricPoint, bool) {
	if len(pair) != 2 {
		return api.MetricPoint{}, false
	}
	at, ok := pair[0].(float64)
	if !ok {
		return api.MetricPoint{}, false
	}
	text, ok := pair[1].(string)
	if !ok {
		return api.MetricPoint{}, false
	}
	value, err := strconv.ParseFloat(text, 64)
	if err != nil {
		return api.MetricPoint{}, false
	}
	if math.IsNaN(value) || math.IsInf(value, 0) {
		return api.MetricPoint{}, false
	}
	return api.MetricPoint{At: int64(at), Value: value}, true
}