package server

import (
	"context"
	"log/slog"
	"net"
	"time"

	"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/kas2agentk_tunnel"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/module/modserver"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/logz"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/syncz"
	"gitlab.com/gitlab-org/cluster-integration/tunnel/tool/retry"
	"golang.org/x/sync/errgroup"
	"google.golang.org/grpc"
)

type module struct {
	log                 *slog.Logger
	api                 modserver.API
	gitLabClient        gitlab.ClientInterface
	getAgentsPollConfig retry.PollConfig

	agentInMemServer     *grpc.Server
	agentInMemListener   net.Listener
	agentInMemServerConn *grpc.ClientConn
	agentInMemAuxCancel  context.CancelFunc

	wf *connWorkerFactory
}

func (m *module) Run(ctx context.Context) error {
	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		return m.agentInMemServer.Serve(m.agentInMemListener)
	})
	g.Go(func() error {
		wm := syncz.NewProtoWorkerManager(m.log, logz.AgentKey, m.wf)

		m.runWorkers(ctx, wm)

		wm.Stop() // Tell all workers to stop
		// We first want gRPC server to send GOAWAY and only then return from the RPC handlers.
		// So we delay signaling the handlers.
		// See https://github.com/grpc/grpc-go/issues/6830 for more background.
		// Start a goroutine in a second and...
		time.AfterFunc(time.Second, func() {
			m.agentInMemAuxCancel() // Signal RPC handlers to cleanly return ASAP
		})
		m.agentInMemServer.GracefulStop()     // Close the listener, send GOAWAY, and wait for all handlers to return
		wm.Wait()                             // Wait for all workers to stop
		return m.agentInMemServerConn.Close() // Close the client since nothing will use it
	})
	return g.Wait()
}

func (m *module) runWorkers(ctx context.Context, wm *syncz.WorkerManager[api.AgentKey, *gapi.ReceptiveAgent]) {
	_ = retry.PollWithBackoff(ctx, m.getAgentsPollConfig, func(ctx context.Context) (error, retry.AttemptResult) { //nolint:staticcheck
		resp, err := gapi.GetReceptiveAgents(ctx, m.gitLabClient)
		if err != nil {
			if gitlab.IsNotFound(err) {
				// Receptive agents is a paid feature but this GitLab instance doesn't have a license.
				m.log.Debug("No license for receptive agents")
				return nil, retry.Backoff
			}
			if ctx.Err() == nil { // not a context error
				m.api.HandleProcessingError(ctx, m.log, "Failed to get receptive agents", err)
			}
			return nil, retry.Backoff
		}
		err = wm.ApplyConfiguration(agentsToWorkSources(resp.Agents))
		if err != nil {
			m.api.HandleProcessingError(ctx, m.log, "ApplyConfiguration error", err)
			// continue with existing configuration
		}
		return nil, retry.Continue
	})
}

func (m *module) Name() string {
	return kas2agentk_tunnel.ModuleName
}

func agentsToWorkSources(agents []*gapi.ReceptiveAgent) []syncz.WorkSource[api.AgentKey, *gapi.ReceptiveAgent] {
	cfgs := make([]syncz.WorkSource[api.AgentKey, *gapi.ReceptiveAgent], 0, len(agents))

	for _, agent := range agents {
		cfgs = append(cfgs, syncz.WorkSource[api.AgentKey, *gapi.ReceptiveAgent]{
			ID:            api.AgentKey{ID: agent.Id, Type: api.AgentTypeKubernetes},
			Configuration: agent,
		})
	}

	return cfgs
}
