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"

	kasObservabilityPort = "8151/tcp"
	kasKubernetesAPIPort = "8154/tcp"
	kasAPIPort           = "8153/tcp"

	ciJobToken1 = "ci_job_token1"
)

type kasHolder struct {
	c                   testcontainers.Container
	agentServerURL      string
	apiServerURL        string // gRPC-compatible URL
	apiServerSecret     []byte
	apiServerClient     grpc.ClientConnInterface
	apiServerClientOnce sync.Once
}

type kasOptions struct {
	name       string
	gitalyPort int
	cfg        *kascfg.ConfigurationFile
}

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

	kasC, err := testcontainers.Run(s.ctx, s.agentCInspect.Image,
		testcontainers.WithHostPortAccess(s.urlPort(opts.cfg.Gitlab.Address), opts.gitalyPort),
		testcontainers.WithEntrypoint("/usr/bin/kas"),
		testcontainers.WithEnv(map[string]string{
			"OWN_PRIVATE_API_CIDR":     "172.16.0.0/12", // private subnet of class B
			"KAS_FF_GRAPH_API_ENABLED": "true",
		}),
		testcontainers.WithExposedPorts(kasObservabilityPort, kasKubernetesAPIPort, kasAPIPort),
		testcontainers.WithCmd("--configuration-file", kasConfigPath),
		testcontainers.WithWaitStrategyAndDeadline(
			time.Minute,
			wait.ForHTTP("/readiness").
				WithStartupTimeout(time.Minute).
				WithPort(kasObservabilityPort).
				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, kasAPIPort, "")
	s.Require().NoError(err)

	return &kasHolder{
		c: kasC,
		agentServerURL: (&url.URL{
			Scheme: "grpc",
			Host:   net.JoinHostPort(ip, kasAgentServerPort),
		}).String(),
		apiServerURL:    "dns:" + 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:           ":8150", // default port but all ips
				ListenGracePeriod: listenGrace,
			},
			KubernetesApi: &kascfg.KubernetesApiCF{
				Listen: &kascfg.ListenCF{
					Network:           ptr.To("tcp4"),
					Address:           ":8154",
					ListenGracePeriod: listenGrace,
				},
				WebsocketTokenSecretFile: kasWebsocketTokenSecretPath,
			},
		},
		Observability: &kascfg.ObservabilityCF{
			UsageReportingPeriod: durationpb.New(0),
			Listen: &kascfg.ObservabilityListenCF{
				Network: ptr.To("tcp4"),
				Address: ":8151", // 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:                  ":8153", // 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(kas.apiServerURL,
			grpc.WithDefaultServiceConfig(`{"loadBalancingConfig":[{"round_robin":{}}]}`),
			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, kasKubernetesAPIPort, "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: fmt.Sprintf("ci:%d:%s", agentKey.ID, ciJobToken1),
		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",
		fmt.Sprintf("Bearer ci:%d:%s", agentKey.ID, ciJobToken1), "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
}
