//go:build e2e
// +build e2e

package utils

import (
	"context"
	"fmt"

	"gopkg.in/yaml.v3"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

type AgentConfig struct {
	Api       AgentApi             `yaml:"api,omitempty"`
	Sources   map[string]Source    `yaml:"sources,omitempty"`
	TransForm map[string]TransForm `yaml:"transforms,omitempty"`
	Sinks     map[string]Sink      `yaml:"sinks,omitempty"`
}

type AgentApi struct {
	Address    string `yaml:"address,omitempty"`
	Enabled    bool   `yaml:"enabled"`
	Playground bool   `yaml:"playground"`
}

type Source struct {
	Type     string `yaml:"type"`
	Format   string `yaml:"format"`
	Interval int    `yaml:"interval"`
}

type TransForm struct {
	Type    string   `yaml:"type"`
	Inputs  []string `yaml:"inputs"`
	Metrics []Metric `yaml:"metrics,omitempty"`
	Source  string   `yaml:"source,omitempty"`
}
type Metric struct {
	Type      string            `yaml:"type"`
	Field     string            `yaml:"field"`
	Name      string            `yaml:"name"`
	NameSpace string            `yaml:"namespace,omitempty"`
	Tags      map[string]string `yaml:"tags,omitempty"`
}

type Sink struct {
	Type             string   `yaml:"type"`
	Inputs           []string `yaml:"inputs"`
	Target           string   `yaml:"target,omitempty"`
	Encoding         Encode   `yaml:"encoding,omitempty"`
	EndPoint         string   `yaml:"endpoint,omitempty"`
	Healthcheck      bool     `yaml:"healthcheck"`
	Request          Reqs     `yaml:"request,omitempty"`
	Acknowledgements string   `yaml:"acknowledgements,omitempty"`
	Region           string   `yaml:"region,omitempty"`
	Bucket           string   `yaml:"bucket,omitempty"`
	KeyPrefix        string   `yaml:"key_prefix,omitempty"`
	Compression      string   `yaml:"compression,omitempty"`
	Auth             Role     `yaml:"auth,omitempty"`
}

type Role struct {
	AssumeRole string `yaml:"assume_role,omitempty"`
}

type Reqs struct {
	RetryAttempts int `yaml:"retry_attempts,omitempty"`
}

type Encode struct {
	Codec string `yaml:"codec,omitempty"`
}

type StartWriteParam struct {
	ClusterID  string
	NameSpace  string //agent namespace
	DNS        string
	Name       string //agent name
	TenantID   string
	AssumeRole string
	Buket      string
}

const (
	sinkName     = "my_sink_id"
	configMapKey = "agent.yaml"
)

func StartPrometheusWrite(param StartWriteParam) {
	configmap, err := kubeClient.CoreV1().ConfigMaps(param.NameSpace).Get(context.TODO(), param.Name, metav1.GetOptions{})
	if err != nil {
		panic(err)
	}
	//var out vector.
	var vectorConfig AgentConfig
	err = yaml.Unmarshal([]byte(configmap.Data["agent.yaml"]), &vectorConfig)
	if err != nil {
		panic(err)
	}
	//ep := "http://" + param.DNS + "/" + param.ClusterID + "/api/v1/write"
	//if config.Service.Vendor != "aws" {
	ep := "http://" + param.DNS + "/api/v1/write"
	//}
	writeSink := make(map[string]Sink)
	writeSink[sinkName] = Sink{
		Type: "prometheus_remote_write",
		Inputs: []string{
			"my_transform",
		},
		EndPoint:    ep,
		Healthcheck: false,
		Request: Reqs{
			RetryAttempts: 1,
		},
	}

	tfrom := make(map[string]TransForm)
	tfrom["remap_transform"] = TransForm{
		Type: "remap",
		Inputs: []string{
			"my_source_id",
		},
		Source: ". = parse_json!(.message)",
	}

	tfrom["my_transform"] = TransForm{
		Type: "log_to_metric",
		Inputs: []string{
			"remap_transform",
		},
		Metrics: []Metric{
			{
				Type:      "counter",
				Field:     "method",
				Name:      "http_reqpest_method_total",
				NameSpace: "demo",
				Tags: map[string]string{
					"method": "{{method}}",
					"tag":    "counter",
				},
			},
			{
				Type:      "gauge",
				Field:     "status",
				Name:      "status_gauge",
				NameSpace: "demo",
				Tags: map[string]string{
					"method":   "{{method}}",
					"protocol": "{{protocol}}",
					"host":     "{{host}}",
					"tag":      "gauge",
				},
			},
			{
				Type:      "histogram",
				Field:     "bytes",
				Name:      "bytes_histogram",
				NameSpace: "demo",
				Tags: map[string]string{
					"status": "{{status}}",
					"host":   "{{host}}",
					"tag":    "histogram",
				},
			},
			{
				Type:      "summary",
				Field:     "bytes",
				Name:      "bytes_summary",
				NameSpace: "demo",
				Tags: map[string]string{
					"status": "{{status}}",
					"host":   "{{host}}",
					"tag":    "summary",
				},
			},
		},
	}
	vectorConfig.TransForm = tfrom
	vectorConfig.Sinks = writeSink
	res, err := yaml.Marshal(vectorConfig)
	if err != nil {
		panic(err)
	}
	configmap.Data[configMapKey] = string(res)

	kubeClient.CoreV1().ConfigMaps(param.NameSpace).Update(context.TODO(), configmap, metav1.UpdateOptions{})
}

func StartLogWrite(param StartWriteParam) bool {
	switch config.Service.Vendor {
	case "aws":
		StartS3Write(param)
		return true
	case "gcp":
		StartGCSWrite(param)
		return true
	}
	return false
}

func StartS3Write(param StartWriteParam) {
	configmap, err := kubeClient.CoreV1().ConfigMaps(param.NameSpace).Get(context.TODO(), param.Name, metav1.GetOptions{})
	if err != nil {
		panic(err)
	}
	//var out vector.
	var vectorConfig AgentConfig
	err = yaml.Unmarshal([]byte(configmap.Data["agent.yaml"]), &vectorConfig)
	if err != nil {
		panic(err)
	}
	logKeyPrefix := fmt.Sprintf(config.DataLink.LogKeyPrefix, param.TenantID)
	writeSink := make(map[string]Sink)
	writeSink[sinkName] = Sink{
		Type: "aws_s3",
		Inputs: []string{
			"my_transform",
		},

		Healthcheck: false,
		Region:      config.Service.Region,
		Bucket:      param.Buket,
		KeyPrefix:   logKeyPrefix,
		Compression: "gzip",
		Auth: Role{
			AssumeRole: param.AssumeRole,
		},
		Encoding: Encode{
			Codec: "json",
		},
	}

	tfrom := make(map[string]TransForm)
	tfrom["my_transform"] = TransForm{
		Type: "remap",
		Inputs: []string{
			"my_source_id",
		},
		Source: ". = parse_json!(.message)",
	}
	vectorConfig.Sinks = writeSink
	vectorConfig.TransForm = tfrom
	res, err := yaml.Marshal(vectorConfig)
	if err != nil {
		panic(err)
	}
	configmap.Data[configMapKey] = string(res)

	kubeClient.CoreV1().ConfigMaps(param.NameSpace).Update(context.TODO(), configmap, metav1.UpdateOptions{})
}

func StartGCSWrite(param StartWriteParam) {
	configmap, err := kubeClient.CoreV1().ConfigMaps(param.NameSpace).Get(context.TODO(), param.Name, metav1.GetOptions{})
	if err != nil {
		panic(err)
	}
	//var out vector.
	var vectorConfig AgentConfig
	err = yaml.Unmarshal([]byte(configmap.Data["agent.yaml"]), &vectorConfig)
	if err != nil {
		panic(err)
	}
	logKeyPrefix := fmt.Sprintf(config.DataLink.LogKeyPrefix, param.TenantID)
	writeSink := make(map[string]Sink)
	writeSink[sinkName] = Sink{
		Type: "gcp_cloud_storage",
		Inputs: []string{
			"my_transform",
		},

		Healthcheck: false,
		//Region:      config.Service.Region,
		Bucket:      param.Buket,
		KeyPrefix:   logKeyPrefix,
		Compression: "gzip",
		//Auth: Role{
		//	AssumeRole: config.AwsS3.AssumeRole,
		//},
		Encoding: Encode{
			Codec: "json",
		},
	}

	tfrom := make(map[string]TransForm)
	tfrom["my_transform"] = TransForm{
		Type: "remap",
		Inputs: []string{
			"my_source_id",
		},
		Source: ". = parse_json!(.message)",
	}
	vectorConfig.Sinks = writeSink
	vectorConfig.TransForm = tfrom
	res, err := yaml.Marshal(vectorConfig)
	if err != nil {
		panic(err)
	}
	configmap.Data[configMapKey] = string(res)

	kubeClient.CoreV1().ConfigMaps(param.NameSpace).Update(context.TODO(), configmap, metav1.UpdateOptions{})
}
func StopAgentWrite(ns, name string) {
	configmap, err := kubeClient.CoreV1().ConfigMaps(ns).Get(context.TODO(), name, metav1.GetOptions{})
	if err != nil {
		panic(err)
	}
	//var out vector.
	var vectorConfig AgentConfig
	err = yaml.Unmarshal([]byte(configmap.Data["agent.yaml"]), &vectorConfig)
	if err != nil {
		panic(err)
	}
	writeSink := make(map[string]Sink)
	writeSink[sinkName] = Sink{
		Type: "console",
		Inputs: []string{
			"my_transform",
		},
		Target: "stdout",
		Encoding: Encode{
			Codec: "json",
		},
	}

	vectorConfig.Sinks = writeSink
	res, err := yaml.Marshal(vectorConfig)
	if err != nil {
		panic(err)
	}
	configmap.Data[configMapKey] = string(res)
	kubeClient.CoreV1().ConfigMaps(ns).Update(context.TODO(), configmap, metav1.UpdateOptions{})
}