package it

import (
	"bytes"
	"context"
	"crypto/rand"
	"encoding/base64"
	"fmt"
	"net"
	"net/http"
	"net/http/httptest"
	"net/url"
	"strconv"
	"sync"
	"time"

	"github.com/coder/websocket"
	"github.com/golang-jwt/jwt/v5"
	"github.com/testcontainers/testcontainers-go"
	"github.com/testcontainers/testcontainers-go/log"
	"github.com/testcontainers/testcontainers-go/wait"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/api"
	agent_tracker_rpc "gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/module/agent_tracker/rpc"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/grpctool"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/httpz"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/testing/testhelpers"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/pkg/kascfg"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/types/known/durationpb"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/utils/ptr"
)

const (
	kasConfigPath                = "/etc/kas/config.json"
	kasGitLabAuthnSecretPath     = "/etc/kas/gitlab-authn-secret"
	kasPrivateAPIAuthnSecretPath = "/etc/kas/private-api-authn-secret"
	kasAPIAuthnSecretPath        = "/etc/kas/api-authn-secret"
	kasWebsocketTokenSecretPath  = "/etc/kas/websocket-token-secret"

	kasAgentServerPort        = "8150"
	kasAgentServerPortProto   = kasAgentServerPort + "/tcp"
	kasObservabilityPort      = "8151"
	kasObservabilityPortProto = kasObservabilityPort + "/tcp"
	kasAPIPort                = "8153"
	kasAPIPortProto           = kasAPIPort + "/tcp"
	kasKubernetesAPIPort      = "8154"
	kasKubernetesAPIPortProto = "8154/tcp"

	ciJobToken1 = "ci_job_token1"
)

type kasHolder struct {
	c testcontainers.Container
	// agentServerURL is URL of the agent server, suitable for access from inside the Docker host, e.g. from agentk container.
	agentServerURL string
	// apiServerEndpoint is ip:port of the API server endpoint, suitable for access from outside the Docker host, e.g. from the test.
	apiServerEndpoint   string
	apiServerSecret     []byte
	apiServerClient     grpc.ClientConnInterface
	apiServerClientOnce sync.Once
}

type kasOptions struct {
	name       string
	gitalyPort int
	cfg        *kascfg.ConfigurationFile
	maxProcs   string
	noRace     bool // race detector is enabled by default
}

func (s *integrationSuite) startKAS(opts kasOptions) *kasHolder {
	if opts.name == "" {
		opts.name = "kas"
	}
	cfgBytes, err := protojson.Marshal(opts.cfg)
	s.Require().NoError(err)

	hostPortAccess := []int{s.urlPort(opts.cfg.Gitlab.Address)}
	if opts.gitalyPort > 0 {
		hostPortAccess = append(hostPortAccess, opts.gitalyPort)
	}
	entrypoint := "/usr/local/bin/kas_race"
	if opts.noRace {
		entrypoint = "/usr/bin/kas"
	}
	kasC, err := testcontainers.Run(s.ctx, s.agentCInspect.Image,
		testcontainers.WithHostPortAccess(hostPortAccess...),
		testcontainers.WithEntrypoint(entrypoint),
		testcontainers.WithEnv(map[string]string{
			"OWN_PRIVATE_API_CIDR": "172.16.0.0/12", // private subnet of class B
			"GOMAXPROCS":           opts.maxProcs,
		}),
		testcontainers.WithExposedPorts(kasObservabilityPortProto, kasKubernetesAPIPortProto, kasAPIPortProto),
		testcontainers.WithCmd("--configuration-file", kasConfigPath),
		testcontainers.WithWaitStrategyAndDeadline(
			time.Minute,
			wait.ForHTTP("/readiness").
				WithStartupTimeout(time.Minute).
				WithPort(kasObservabilityPortProto).
				WithForcedIPv4LocalHost(),
		),
		testcontainers.WithFiles(
			testcontainers.ContainerFile{
				Reader:            bytes.NewReader(cfgBytes),
				ContainerFilePath: kasConfigPath,
				FileMode:          0o644,
			},
			testcontainers.ContainerFile{
				Reader:            bytes.NewReader(base64.StdEncoding.AppendEncode(nil, s.gitLabAccessSecret)),
				ContainerFilePath: kasGitLabAuthnSecretPath,
				FileMode:          0o644,
			},
			testcontainers.ContainerFile{
				Reader:            bytes.NewReader(base64.StdEncoding.AppendEncode(nil, s.kasPrivateAPISecret)),
				ContainerFilePath: kasPrivateAPIAuthnSecretPath,
				FileMode:          0o644,
			},
			testcontainers.ContainerFile{
				Reader:            bytes.NewReader(base64.StdEncoding.AppendEncode(nil, s.kasAPIServerSecret)),
				ContainerFilePath: kasAPIAuthnSecretPath,
				FileMode:          0o644,
			},
			testcontainers.ContainerFile{
				Reader:            bytes.NewReader(base64.StdEncoding.AppendEncode(nil, s.kasWebsocketTokenSecret)),
				ContainerFilePath: kasWebsocketTokenSecretPath,
				FileMode:          0o644,
			},
		),
		testcontainers.WithLogConsumers(&testLogConsumer{t: s.T(), prefix: opts.name + ":"}),
		testcontainers.WithLogger(log.TestLogger(s.T())),
	)
	s.Require().NoError(err)
	s.containerToTerminate(kasC)

	ip, err := kasC.ContainerIP(s.ctx)
	s.Require().NoError(err)

	apiServerEndpoint, err := kasC.PortEndpoint(s.ctx, kasAPIPortProto, "")
	s.Require().NoError(err)

	return &kasHolder{
		c: kasC,
		agentServerURL: (&url.URL{
			Scheme: "grpc",
			Host:   net.JoinHostPort(ip, kasAgentServerPort),
		}).String(),
		apiServerEndpoint: apiServerEndpoint,
		apiServerSecret:   s.kasAPIServerSecret,
	}
}

// baseKASConfig should only be called after Redis has been started.
// A single config base should be used because it sets the Redis prefix that should be the same
// for all kas instances in a test.
func (s *integrationSuite) baseKASConfig(gitLabSrv *httptest.Server) *kascfg.ConfigurationFile {
	redisIP, err := s.redisC.ContainerIP(s.ctx)
	s.Require().NoError(err)

	redisPrefix := make([]byte, 8)
	_, err = rand.Read(redisPrefix)
	s.Require().NoError(err)

	u, err := url.Parse(gitLabSrv.URL)
	s.Require().NoError(err)

	p := u.Port()
	if p == "" {
		u.Host = testcontainers.HostInternal
	} else {
		u.Host = net.JoinHostPort(testcontainers.HostInternal, p)
	}

	listenGrace := durationpb.New(100 * time.Millisecond)
	return &kascfg.ConfigurationFile{
		Gitlab: &kascfg.GitLabCF{
			Address:                  u.String(),
			AuthenticationSecretFile: kasGitLabAuthnSecretPath,
		},
		Agent: &kascfg.AgentCF{
			Listen: &kascfg.ListenAgentCF{
				Network:           ptr.To("tcp4"),
				Address:           ":" + kasAgentServerPort, // default port but all ips
				ListenGracePeriod: listenGrace,
			},
			KubernetesApi: &kascfg.KubernetesApiCF{
				Listen: &kascfg.ListenCF{
					Network:           ptr.To("tcp4"),
					Address:           ":" + kasKubernetesAPIPort,
					ListenGracePeriod: listenGrace,
				},
				WebsocketTokenSecretFile: kasWebsocketTokenSecretPath,
			},
		},
		Observability: &kascfg.ObservabilityCF{
			UsageReportingPeriod: durationpb.New(0),
			Listen: &kascfg.ObservabilityListenCF{
				Network: ptr.To("tcp4"),
				Address: ":" + kasObservabilityPort, // default port but all ips
			},
			Logging: &kascfg.LoggingCF{
				Level:     kascfg.LogLevelEnum_debug,
				GrpcLevel: ptr.To(kascfg.LogLevelEnum_debug),
			},
			EventReportingPeriod: durationpb.New(0),
		},
		Redis: &kascfg.RedisCF{
			RedisConfig: &kascfg.RedisCF_Server{
				Server: &kascfg.RedisServerCF{
					Address: net.JoinHostPort(redisIP, "6379"),
				},
			},
			KeyPrefix: fmt.Sprintf("%x", redisPrefix),
		},
		Api: &kascfg.ApiCF{
			Listen: &kascfg.ListenApiCF{
				Network:                  ptr.To("tcp4"),
				Address:                  ":" + kasAPIPort, // default port but all ips
				AuthenticationSecretFile: kasAPIAuthnSecretPath,
				ListenGracePeriod:        listenGrace,
			},
		},
		PrivateApi: &kascfg.PrivateApiCF{
			Listen: &kascfg.ListenPrivateApiCF{
				Network:                  ptr.To("tcp4"),
				Address:                  ":8155", // default port but all ips
				AuthenticationSecretFile: kasPrivateAPIAuthnSecretPath,
				ListenGracePeriod:        listenGrace,
			},
		},
	}
}

func (s *integrationSuite) urlPort(someURL string) int {
	u, err := url.Parse(someURL)
	s.Require().NoError(err)
	p := u.Port()
	if p != "" {
		intP, parseErr := strconv.ParseUint(p, 10, 16)
		s.Require().NoError(parseErr)
		return int(intP)
	}
	switch u.Scheme {
	case "http":
		return 80
	case "https":
		return 443
	default:
		s.T().Fatalf("unknown URL scheme in URL %s: %s", someURL, u.Scheme)
		return 0
	}
}

func (s *integrationSuite) waitForAgentToRegister(kas *kasHolder) {
	s.T().Log("Waiting for agentk to register")
	apiClient := agent_tracker_rpc.NewAgentTrackerClient(s.clientForKASAPI(kas))
	s.Require().Eventually(func() bool {
		resp, err := apiClient.GetConnectedAgentksByAgentIDs(s.ctx, &agent_tracker_rpc.GetConnectedAgentksByAgentIDsRequest{
			AgentIds: []int64{testhelpers.AgentkKey1.ID},
		})
		s.Require().NoError(err)
		return len(resp.Agents) == 1
	}, 3*time.Minute, 50*time.Millisecond, "Agentk not registered")
	s.T().Log("Agentk registered")
}

// clientForKASAPI returns a gRPC client for the API server endpoint.
func (s *integrationSuite) clientForKASAPI(kas *kasHolder) grpc.ClientConnInterface {
	kas.apiServerClientOnce.Do(func() {
		conn, err := grpc.NewClient("dns:"+kas.apiServerEndpoint,
			grpc.WithTransportCredentials(insecure.NewCredentials()),
			grpc.WithPerRPCCredentials(&grpctool.JWTCredentials{
				SigningMethod: jwt.SigningMethodHS256,
				SigningKey:    kas.apiServerSecret,
				Audience:      api.JWTKAS,
				Issuer:        "it",
				Insecure:      true, // We may or may not have TLS setup, so always say creds don't need TLS.
			}),
		)
		s.Require().NoError(err)
		s.cleanup(func() {
			s.NoError(conn.Close())
		})
		kas.apiServerClient = conn
	})
	return kas.apiServerClient
}

func (s *integrationSuite) k8sProxyURL(kas *kasHolder) string {
	k8sAPIURL, err := kas.c.PortEndpoint(s.ctx, kasKubernetesAPIPortProto, "http")
	s.Require().NoError(err)

	return k8sAPIURL
}

func (s *integrationSuite) k8sClientForAgentKey(kas *kasHolder, agentKey api.AgentKey) kubernetes.Interface {
	c, err := kubernetes.NewForConfig(&rest.Config{
		Host:        s.k8sProxyURL(kas),
		BearerToken: ciJobTokenHeaderValue(agentKey),
		QPS:         -1,
		Burst:       -1,
	})
	s.Require().NoError(err)
	return c
}

func (s *integrationSuite) dialWS(kas *kasHolder, agentKey api.AgentKey) *websocket.Conn {
	conn, err := dialWS(
		s.ctx,
		s.k8sProxyURL(kas)+"/graph",
		"Bearer "+ciJobTokenHeaderValue(agentKey),
		"gitlab-agent-graph-api",
	)
	s.Require().NoError(err)
	return conn
}

func dialWS(ctx context.Context, dialURL, credentials string, subprotocols ...string) (*websocket.Conn, error) {
	conn, _, err := websocket.Dial(ctx, dialURL, &websocket.DialOptions{ //nolint:bodyclose
		HTTPClient: &http.Client{
			Transport: &http.Transport{
				TLSHandshakeTimeout:   10 * time.Second,
				ResponseHeaderTimeout: 20 * time.Second,
			},
			CheckRedirect: func(req *http.Request, via []*http.Request) error {
				return http.ErrUseLastResponse
			},
		},
		HTTPHeader: http.Header{
			httpz.AuthorizationHeader: []string{credentials},
		},
		Subprotocols: subprotocols,
	})
	return conn, err
}

func ciJobTokenHeaderValue(agentKey api.AgentKey) string {
	return fmt.Sprintf("ci:%d:%s", agentKey.ID, ciJobToken1)
}
