package server

import (
	"context"
	"errors"
	"fmt"
	"log/slog"

	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/api"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/gitlab"
	gapi "gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/gitlab/api"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/module/flux/rpc"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/module/modserver"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/module/usage_metrics"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/server_api"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/cache"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/logz"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/pkg/event"
	"gitlab.com/gitlab-org/cluster-integration/tunnel/tool/grpcz"
	"gitlab.com/gitlab-org/cluster-integration/tunnel/tool/retry"
	otelmetric "go.opentelemetry.io/otel/metric"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

const (
	maxBufferedNotifications = 10
)

type server struct {
	rpc.UnsafeGitLabFluxServer
	serverAPI               modserver.API
	notifiedCounter         otelmetric.Int64Counter
	notifiedUsageCounter    usage_metrics.Counter
	notifiedProjectsCounter usage_metrics.UniqueCounter
	droppedCounter          otelmetric.Int64Counter
	pollConfig              retry.PollConfig
	projectAccessClient     *projectAccessClient
}

func (s *server) ReconcileProjects(req *rpc.ReconcileProjectsRequest, server grpc.ServerStreamingServer[rpc.ReconcileProjectsResponse]) error {
	ctx := server.Context()
	rpcAPI := modserver.AgentRPCAPIFromContext(ctx)
	log := rpcAPI.Log()
	var agentInfo server_api.AgentInfo
	var err error

	err = rpcAPI.PollWithBackoff(s.pollConfig, func() (error, retry.AttemptResult) { //nolint:staticcheck
		agentInfo, err = rpcAPI.AgentInfo(ctx, log)
		if err != nil {
			if status.Code(err) == codes.Unavailable {
				return nil, retry.Backoff
			}
			return err, retry.Done // no wrap
		}
		return nil, retry.Done
	})
	if agentInfo == nil {
		return err // ctx done, err may be nil but must return
	}

	log = log.With(logz.AgentKey(agentInfo.AgentKey()))
	log.Debug("Started reconcile projects ...")
	defer log.Debug("Stopped reconcile projects ...")

	pipe := make(chan *event.Project, maxBufferedNotifications)

	projects := req.ToProjectSet()
	stop := s.serverAPI.OnGitPushEvent(func(e *event.GitPushEvent) {
		if !projects.Has(e.Project.FullPath) {
			// NOTE: it's probably not a good idea to log here as we'd get one for every event,
			// which on GitLab.com is thousands per minute.
			return
		}

		select {
		case pipe <- e.Project:
		default:
			s.droppedCounter.Add(context.Background(), 1)
			// NOTE: if for whatever reason the other goroutine isn't able to keep up with the events,
			// we just drop them for now.
			log.Debug("Dropping git push event", logz.ProjectID(e.Project.FullPath))
		}
	})
	defer stop()

	// Stop listening for push events when:
	// - server is shutting down
	// - the RPC connection is done
	// - max connection age is reached
	ageCtx := grpcz.MaxConnectionAgeContextFromStreamContext(ctx)
	done := ageCtx.Done()
	for {
		select {
		case <-done:
			return nil
		case project := <-pipe:
			hasAccess, err := s.verifyProjectAccess(ctx, log, rpcAPI, agentInfo.AgentKey(), project.FullPath)
			if err != nil {
				rpcAPI.HandleProcessingError(log, fmt.Sprintf("Failed to check if project %s is accessible by agent", project.FullPath), err, logz.AgentKey(agentInfo.AgentKey()))
				continue
			}
			if !hasAccess {
				log.Info("Agent doesn't have access to the project. Not notifying it about a git push", logz.ProjectID(project.FullPath))
				continue
			}

			// increase Flux git push event notification counter
			s.notifiedCounter.Add(context.Background(), 1)
			s.notifiedUsageCounter.Inc()
			s.notifiedProjectsCounter.Add(project.Id)

			err = server.Send(&rpc.ReconcileProjectsResponse{
				Project: &rpc.Project{Id: project.FullPath},
			})
			if err != nil {
				_ = rpcAPI.HandleIOError(log, fmt.Sprintf("failed to send reconcile message for project %s", project.FullPath), err)
			}
		}
	}
}

// verifyProjectAccess verifies if the given agent has access to the given project.
// If this is not the case `false` is returned, otherwise `true`.
// If the error has the code Unavailable a caller my retry.
func (s *server) verifyProjectAccess(ctx context.Context, log *slog.Logger, rpcAPI modserver.AgentRPCAPI, agentKey api.AgentKey,
	projectID string) (bool, error) {
	hasAccess, err := s.projectAccessClient.VerifyProjectAccess(ctx, rpcAPI.AgentTokenWithType(), projectID)
	switch {
	case err == nil:
		return hasAccess, nil
	case errors.Is(err, context.Canceled):
		err = status.Error(codes.Canceled, err.Error())
	case errors.Is(err, context.DeadlineExceeded):
		err = status.Error(codes.DeadlineExceeded, err.Error())
	default:
		rpcAPI.HandleProcessingError(log, "VerifyProjectAccess()", err, logz.AgentKey(agentKey))
		err = status.Error(codes.Unavailable, "unavailable")
	}
	return false, err
}

type projectAccessClient struct {
	gitLabClient       gitlab.ClientInterface
	projectAccessCache *cache.CacheWithErr[projectAccessCacheKey, bool]
}

func (c *projectAccessClient) VerifyProjectAccess(ctx context.Context, agentTokenWithType api.AgentTokenWithType, projectID string) (bool, error) {
	key := projectAccessCacheKey{agentToken: agentTokenWithType.Token, projectID: projectID}
	return c.projectAccessCache.GetItem(ctx, key, func() (bool, error) {
		return gapi.VerifyProjectAccess(ctx, c.gitLabClient, agentTokenWithType, projectID, gitlab.WithoutRetries())
	})
}

type projectAccessCacheKey struct {
	agentToken api.AgentToken
	projectID  string
}
