package server

import (
	"context"
	"crypto/tls"
	"fmt"
	"net"
	"net/http"
	"net/netip"
	"strings"
	"time"

	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/api"
	gapi "gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/gitlab/api"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/module/autoflow"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/module/autoflow/encryption"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/module/autoflow/engine"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/module/autoflow/rpc"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/module/modserver"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/module/modshared"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/cache"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/httpz"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/ioz"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/prototool"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/redistool"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/tlstool"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/pkg/kascfg"
	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
	tclient "go.temporal.io/sdk/client"
	totel "go.temporal.io/sdk/contrib/opentelemetry"
	"go.temporal.io/sdk/converter"
	tinterceptor "go.temporal.io/sdk/interceptor"
	"go.temporal.io/sdk/workflow"
	"google.golang.org/grpc"
	"google.golang.org/grpc/stats"
)

const (
	kasWorkerTaskQueue = "kas-worker"
)

type Factory struct {
	CSH              stats.Handler
	StreamClientProm grpc.StreamClientInterceptor
	UnaryClientProm  grpc.UnaryClientInterceptor
}

func (f *Factory) New(config *modserver.Config) (modserver.Module, error) {
	cfg := config.Config.Autoflow
	if cfg == nil {
		config.Log.Info("AutoFlow module is not enabled in config")
		return nil, nil
	}

	var tlsConfig *tls.Config
	if cfg.Temporal.EnableTls {
		var err error
		tlsConfig, err = tlstool.ClientConfigWithCACertKeyPair(
			cfg.Temporal.GetCaCertificateFile(),
			cfg.Temporal.GetCertificateFile(),
			cfg.Temporal.GetKeyFile(),
		)
		if err != nil {
			return nil, err
		}
	}

	dataConverter := converter.GetDefaultDataConverter()
	if cfg.Temporal.WorkflowDataEncryption != nil {
		secretKey, err := ioz.ReadSecretAES256Key(cfg.Temporal.WorkflowDataEncryption.SecretKeyFile)
		if err != nil {
			return nil, err
		}

		dataConverter = encryption.NewDataConverter(
			dataConverter,
			encryption.DataConverterOptions{SecretKey: secretKey},
		)
	}

	tracer := config.TraceProvider.Tracer(autoflow.ModuleName)
	tracingInterceptor, err := totel.NewTracingInterceptor(totel.TracerOptions{
		Tracer: tracer,
	})
	if err != nil {
		return nil, err
	}

	tc, err := tclient.Dial(tclient.Options{
		HostPort:  cfg.Temporal.HostPort,
		Namespace: cfg.Temporal.GetNamespace(), // empty string means default value, which is "default"
		Logger:    config.Log,
		MetricsHandler: totel.NewMetricsHandler(totel.MetricsHandlerOptions{
			Meter: config.Meter,
			// InitialAttributes: attribute.Set{},
			OnError: func(err error) {
				config.API.HandleProcessingError(context.Background(), config.Log, "Temporal SDK received error from configured OTEL meter", err)
			},
		}),
		//Identity:           "",
		DataConverter: dataConverter,
		//FailureConverter:   nil,
		ContextPropagators: []workflow.ContextPropagator{newContextPropagator(dataConverter)},
		ConnectionOptions: tclient.ConnectionOptions{
			TLS: tlsConfig,
			//Authority:                    "",
			//EnableKeepAliveCheck:         false,
			//KeepAliveTime:                0,
			//KeepAliveTimeout:             0,
			//KeepAlivePermitWithoutStream: false,
			//MaxPayloadSize:               0,
			DialOptions: []grpc.DialOption{
				grpc.WithStatsHandler(otelgrpc.NewClientHandler(
					otelgrpc.WithTracerProvider(config.MaybeTraceProvider(config.GRPCClientTracing)),
					otelgrpc.WithMeterProvider(config.MeterProvider),
					otelgrpc.WithPropagators(config.TracePropagator),
					otelgrpc.WithMessageEvents(otelgrpc.ReceivedEvents, otelgrpc.SentEvents),
				)),
				grpc.WithStatsHandler(f.CSH),
				grpc.WithChainStreamInterceptor(
					f.StreamClientProm,
				),
				grpc.WithChainUnaryInterceptor(
					f.UnaryClientProm,
				),
				grpc.WithSharedWriteBuffer(true),
				grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(api.GRPCMaxMessageSize)),
			},
		},
		//HeadersProvider:   nil,
		//TrafficController: nil,
		Interceptors: []tinterceptor.ClientInterceptor{tracingInterceptor},
	})
	if err != nil {
		return nil, err
	}

	safeNetDialerControl, err := newSafeNetDialerControlFromConfig(cfg.HttpClient)
	if err != nil {
		return nil, err
	}

	flowEng, err := engine.NewEngine(engine.Options{
		Log:          config.Log,
		TC:           tc,
		GitalyPool:   config.Gitaly,
		GitLabClient: config.GitLabClient,
		RepoInfoCache: cache.NewWithError[engine.RepoInfoCacheKey, *api.ProjectInfo](
			cfg.ProjectInfoCacheTtl.AsDuration(),
			cfg.ProjectInfoCacheErrorTtl.AsDuration(),
			&redistool.ErrCacher[engine.RepoInfoCacheKey]{
				Log:          config.Log,
				ErrRep:       modshared.APIToErrReporter(config.API),
				Client:       config.RedisClient,
				ErrMarshaler: prototool.ProtoErrMarshaler{},
				KeyToRedisKey: func(cacheKey engine.RepoInfoCacheKey) string {
					return config.Config.Redis.KeyPrefix +
						":repo_info_errs:" +
						cacheKey.ProjectID
				},
			},
			tracer,
			gapi.IsCacheableError,
		),
		HTTPRT: otelhttp.NewTransport(
			&http.Transport{
				Proxy: http.ProxyFromEnvironment,
				DialContext: (&net.Dialer{
					Timeout:   30 * time.Second,
					KeepAlive: 30 * time.Second,
					Control:   safeNetDialerControl,
				}).DialContext,
				TLSClientConfig:       tlstool.ClientConfig(),
				MaxIdleConns:          10,
				IdleConnTimeout:       90 * time.Second,
				TLSHandshakeTimeout:   10 * time.Second,
				ResponseHeaderTimeout: 20 * time.Second,
				ForceAttemptHTTP2:     true,
			},
			otelhttp.WithPropagators(config.TracePropagator),
			otelhttp.WithTracerProvider(config.MaybeTraceProvider(config.HTTPClientTracing)),
			otelhttp.WithMeterProvider(config.MeterProvider),
		),
		TaskQueue:      kasWorkerTaskQueue,
		GitLabURL:      strings.TrimSuffix(config.Config.Gitlab.ExternalURL(), "/"),
		RedisClient:    config.RedisClient,
		RedisKeyPrefix: config.Config.Redis.KeyPrefix,
		HandleProcessingError: func(msg string, err error) {
			config.API.HandleProcessingError(context.Background(), config.Log, msg, err)
		},
		Tracer:                                  tracer,
		Meter:                                   config.Meter,
		WorkflowsPerProjectIDEventTypePerMinute: cfg.WorkflowsPerProjectIdEventTypePerMinute,
		WorkerInterceptors:                      []tinterceptor.WorkerInterceptor{&loggerInterceptor{}},
	})
	if err != nil {
		return nil, err
	}
	srv := &server{
		log:       config.Log,
		serverAPI: config.API,
		flowEng:   flowEng,
	}
	rpc.RegisterAutoFlowServer(config.APIServer, srv)
	return &module{
		flowEng: flowEng,
		tc:      tc,
	}, nil
}

func (f *Factory) Name() string {
	return autoflow.ModuleName
}

func newSafeNetDialerControlFromConfig(cfg *kascfg.HTTPClientCF) (httpz.ControlFunc, error) {
	allowedPorts := make([]uint16, len(cfg.AllowedPorts))
	for i, p := range cfg.AllowedPorts {
		allowedPorts[i] = uint16(p) //nolint:gosec
	}

	allowedIPs, err := parseIPs(cfg.AllowedIps)
	if err != nil {
		return nil, err
	}
	allowedIPCIDRs, err := parseCIDRs(cfg.AllowedIpCidrs)
	if err != nil {
		return nil, err
	}

	blockedIPs, err := parseIPs(cfg.BlockedIps)
	if err != nil {
		return nil, err
	}
	blockedIPCIDRs, err := parseCIDRs(cfg.BlockedIpCidrs)
	if err != nil {
		return nil, err
	}

	return httpz.NewSafeNetDialerControl(&httpz.SafeNetControlConfig{
		AllowedPorts:   allowedPorts,
		AllowedIPs:     allowedIPs,
		AllowedIPCIDRs: allowedIPCIDRs,
		BlockedIPs:     blockedIPs,
		BlockedIPCIDRs: blockedIPCIDRs,
	}), nil
}

func parseIPs(ss []string) ([]netip.Addr, error) {
	ips := make([]netip.Addr, len(ss))
	for i, s := range ss {
		ip, err := netip.ParseAddr(s)
		if err != nil {
			return nil, fmt.Errorf("configured IP %q in AutoFlow HTTP client config cannot be parsed: %w", s, err)
		}
		ips[i] = ip
	}
	return ips, nil
}

func parseCIDRs(ss []string) ([]netip.Prefix, error) {
	cidrs := make([]netip.Prefix, len(ss))
	for i, s := range ss {
		cidr, err := netip.ParsePrefix(s)
		if err != nil {
			return nil, fmt.Errorf("configured CIDR %q in AutoFlow HTTP client config cannot be parsed: %w", s, err)
		}
		cidrs[i] = cidr
	}
	return cidrs, nil
}
