package functional

import (
	"context"
	"fmt"
	"strings"

	log "github.com/ViaQ/logerr/v2/log/static"
	logging "github.com/openshift/cluster-logging-operator/apis/logging/v1"
	"github.com/openshift/cluster-logging-operator/internal/url"
	"github.com/openshift/cluster-logging-operator/internal/utils"
	"github.com/openshift/cluster-logging-operator/test"
	"github.com/openshift/cluster-logging-operator/test/helpers/cmd"
	"github.com/openshift/cluster-logging-operator/test/helpers/kafka"
	"github.com/openshift/cluster-logging-operator/test/helpers/types"
	"k8s.io/apimachinery/pkg/util/wait"
)

type Option struct {
	Name  string
	Value string
}

func OptionsInclude(name string, options []Option) (bool, Option) {
	for _, o := range options {
		if o.Name == name {
			return true, o
		}
	}
	return false, Option{}
}

func (f *CollectorFunctionalFramework) ReadApplicationLogsFrom(outputName string) ([]types.ApplicationLog, error) {
	raw, err := f.ReadLogsFrom(outputName, applicationLog)
	if err != nil {
		return nil, err
	}

	var logs []types.ApplicationLog
	if err = types.StrictlyParseLogs(utils.ToJsonLogs(raw), &logs); err != nil {
		return nil, err
	}
	return logs, nil
}

func (f *CollectorFunctionalFramework) ReadRawApplicationLogsFrom(outputName string) ([]string, error) {
	return f.ReadLogsFrom(outputName, applicationLog)
}

func (f *CollectorFunctionalFramework) ReadInfrastructureLogsFrom(outputName string) ([]string, error) {
	return f.ReadLogsFrom(outputName, logging.InputNameInfrastructure)
}
func (f *CollectorFunctionalFramework) ReadApplicationLogsFromKafka(topic string, brokerlistener string, consumercontainername string) (results []string, err error) {
	//inter broker zookeeper connect is plaintext so use plaintext port to check on sent messages from kafka producer ie. fluent-kafka plugin
	//till this step it must be ensured that a topic is created and published in kafka-consumer-clo-app-topic container within functional pod
	var result string
	outputFilename := "/shared/consumed.logs"
	cmd := fmt.Sprintf("tail -1 %s", outputFilename)
	err = wait.PollImmediate(defaultRetryInterval, maxDuration, func() (done bool, err error) {
		result, err = f.RunCommand(consumercontainername, "/bin/bash", "-c", cmd)
		if result != "" && err == nil {
			return true, nil
		}
		log.V(4).Error(err, "Polling logs from kafka")
		return false, nil
	})
	if err == nil {
		results = strings.Split(strings.TrimSpace(result), "\n")
	}
	log.V(4).Info("Returning", "logs", result)
	return results, err
}

func (f *CollectorFunctionalFramework) ReadAuditLogsFrom(outputName string) ([]string, error) {
	return f.ReadLogsFrom(outputName, auditLog)
}

func (f *CollectorFunctionalFramework) Readk8sAuditLogsFrom(outputName string) ([]string, error) {
	return f.ReadLogsFrom(outputName, k8sAuditLog)
}

func (f *CollectorFunctionalFramework) ReadOvnAuditLogsFrom(outputName string) ([]string, error) {
	return f.ReadLogsFrom(outputName, ovnAuditLog)
}

func (f *CollectorFunctionalFramework) ReadLogsFrom(outputName, sourceType string) (results []string, err error) {
	outputSpecs := f.Forwarder.Spec.OutputMap()
	outputType := outputName
	var outputSpec *logging.OutputSpec
	if output, found := outputSpecs[outputName]; found {
		outputType = output.Type
		outputSpec = output
	}
	var readLogs func() ([]string, error)

	switch outputType {
	case logging.OutputTypeKafka:
		readLogs = func() ([]string, error) {
			switch sourceType {
			case logging.InputNameAudit:
				sourceType = kafka.AuditLogsTopic
			case logging.InputNameInfrastructure:
				sourceType = kafka.InfraLogsTopic
			default:
				sourceType = kafka.AppLogsTopic
			}
			container := kafka.ConsumerNameForTopic(sourceType)
			return f.ReadApplicationLogsFromKafka(sourceType, "localhost:9092", container)
		}
	case logging.OutputTypeElasticsearch:
		readLogs = func() ([]string, error) {
			option := Option{"port", "9200"}
			if outputSpec != nil {
				if esurl, err := url.Parse(outputSpec.URL); err == nil {
					option.Value = esurl.Port()
				}
			}
			return f.GetLogsFromElasticSearch(outputName, sourceType, option)
		}
	default:
		readLogs = func() ([]string, error) {
			var result string
			outputFiles, ok := outputLogFile[outputType]
			if !ok {
				return nil, fmt.Errorf(fmt.Sprintf("cant find output of type %s in outputSpec %v", outputName, outputSpecs))
			}
			file, ok := outputFiles[sourceType]
			if !ok {
				return nil, fmt.Errorf(fmt.Sprintf("can't find log of type %s", sourceType))
			}

			result, err = f.ReadFileFrom(outputName, file)
			if err == nil {
				results = strings.Split(strings.TrimSpace(result), "\n")
			}
			log.V(4).Info("Returning", "logs", result)
			return results, err
		}
	}
	return readLogs()
}

func (f *CollectorFunctionalFramework) ReadFileFrom(outputName, filePath string) (result string, err error) {
	err = wait.PollImmediate(defaultRetryInterval, f.GetMaxReadDuration(), func() (done bool, err error) {
		result, err = f.RunCommand(strings.ToLower(outputName), "cat", filePath)
		if result != "" && err == nil {
			return true, nil
		}
		log.V(4).Error(err, "Polling logs")
		return false, nil
	})
	log.V(3).Info("Returning", "content", result)
	return result, err
}

func (f *CollectorFunctionalFramework) ReadNApplicationLogsFrom(n uint64, outputName string) ([]string, error) {
	lines := []string{}
	ctx, cancel := context.WithTimeout(context.Background(), test.SuccessTimeout())
	defer cancel()
	reader, err := cmd.TailReaderForContainer(f.Pod, outputName, ApplicationLogFile)
	if err != nil {
		log.V(3).Error(err, "Error creating tail reader")
		return nil, err
	}
	for {
		line, err := reader.ReadLineContext(ctx)
		if err != nil {
			log.V(3).Error(err, "Error readlinecontext")
			return nil, err
		}
		lines = append(lines, line)
		n--
		if n == 0 {
			break
		}
	}
	return lines, err
}
