package agentk

import (
	"context"
	"fmt"
	"io"
	"log/slog"
	"time"

	notificationv1 "github.com/fluxcd/notification-controller/api/v1"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/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/modagent"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/grpctool"
	"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"
	"google.golang.org/grpc"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/util/sets"
	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
	"k8s.io/client-go/tools/cache"
)

const (
	// projectReceiverIndex the name for the index on the FluxCD notification/Receiver objects
	// that maps them to GitLab project paths.
	projectReceiverIndex                  = "project"
	defaultReconciliationDebounceDuration = 10 * time.Second
)

// client represents the client part of the Flux agent module.
type client struct {
	log                            *slog.Logger
	agentAPI                       modagent.API
	agentKey                       api.AgentKey
	fluxGitLabClient               rpc.GitLabFluxClient
	pollConfig                     retry.PollConfig
	receiverIndexer                cache.Indexer
	reconcileTrigger               reconcileTrigger
	updateProjectsToReconcileC     chan []string
	reconciliationDebounceDuration time.Duration
	corev1APIClient                corev1.CoreV1Interface
}

type clientFactory func(ctx context.Context, url string, receiverIndexer cache.Indexer) (*client, error)

// projectReconciler is an interface to start reconciliation
// of projects available to an underlying indexer
type projectReconciler interface {
	// ReconcileIndexedProjects starts the reconciliation of whatever
	// projects are in the underlying index.
	// This method only starts the reconciliation - which is an asynchronous process
	// and effectively is run in another goroutine.
	ReconcileIndexedProjects(ctx context.Context)
}

// newClient adds an index on the given indexer and returns a new client.
// The added index maps GitLab project paths to receiver objects, see createProjectIndex
func newClient(log *slog.Logger, agentAPI modagent.API, agentKey api.AgentKey, fluxGitLabClient rpc.GitLabFluxClient, pollConfig retry.PollConfig, receiverIndexer cache.Indexer, reconcileTrigger reconcileTrigger, corev1APIClient corev1.CoreV1Interface) (*client, error) {
	err := addProjectIndex(receiverIndexer)
	if err != nil {
		return nil, err
	}
	updateProjectsToReconcileC := make(chan []string)
	return &client{
		log:                            log,
		agentAPI:                       agentAPI,
		agentKey:                       agentKey,
		fluxGitLabClient:               fluxGitLabClient,
		pollConfig:                     pollConfig,
		receiverIndexer:                receiverIndexer,
		reconcileTrigger:               reconcileTrigger,
		updateProjectsToReconcileC:     updateProjectsToReconcileC,
		reconciliationDebounceDuration: defaultReconciliationDebounceDuration,
		corev1APIClient:                corev1APIClient,
	}, nil
}

// addProjectIndex adds a new index to the given indexer.
// The added index maps GitLab project paths (like `gitlab-org/gitlab`) to FluxCD notification/Receiver objects,
// which in turn trace back to FluxCD source/GitRepository objects.
// The index is created based on the projectAnnotationKey annotation on the notification/Receiver object
// which are being indexed.
func addProjectIndex(receiverIndexer cache.Indexer) error {
	err := receiverIndexer.AddIndexers(map[string]cache.IndexFunc{
		projectReceiverIndex: func(obj any) ([]string, error) {
			u, ok := obj.(*unstructured.Unstructured)
			if !ok {
				return nil, fmt.Errorf("failed to cast object of type %T into *unstructured.Unstructured", obj)
			}

			project, ok := u.GetAnnotations()[api.ProjectKey]
			if !ok {
				// NOTE: this is not an issue at this point, because it may very well be that this
				// receiver doesn't (yet) have the annotation. This function will eventually
				// be called again once it has it.
				return nil, nil
			}

			return []string{project}, nil
		},
	})
	if err != nil {
		return fmt.Errorf("unable to add %s indexer: %w", projectReceiverIndex, err)
	}
	return nil
}

// RunProjectReconciliation runs a new reconciliation process
// for the latest projects.
// A reconciliation for a new set of projects can be started using
// the ReconcileIndexedProjects method.
// There is only ever one reconciliation process running at the same
// time and a call to ReconcileIndexedProjects will terminate
// a potentially previously started reconciliation process.
// The c.reconciliationDebounceDuration controls how long to debounce
// before starting a new reconciliation for received projects.
func (c *client) RunProjectReconciliation(ctx context.Context) {
	done := ctx.Done()

	wh := syncz.NewWorkerHolder[sets.Set[string]](
		func(projectsToReconcile sets.Set[string]) syncz.Worker {
			return syncz.WorkerFunc(func(ctx context.Context) {
				c.reconcileProjects(ctx, projectsToReconcile)
			})
		},
		isEqualProjectSets,
	)
	defer wh.StopAndWait()

	debounceTimer := time.NewTimer(time.Hour)
	debounceTimer.Stop()
	defer debounceTimer.Stop()

	var lastProjects []string

	for {
		select {
		case <-done:
			// Shutdown
			return
		case projects := <-c.updateProjectsToReconcileC:
			lastProjects = projects
			if !debounceTimer.Stop() {
				select {
				case <-debounceTimer.C:
				default:
				}
			}
			debounceTimer.Reset(c.reconciliationDebounceDuration)
		case <-debounceTimer.C:
			newProjects := sets.New(lastProjects...)
			c.log.Debug("Reconcile project update", logz.ProjectsToReconcile(newProjects))
			wh.ApplyConfig(ctx, newProjects)
		}
	}
}

// ReconcileIndexedProjects starts the reconciliation of the latest indexed projects.
// The index must be created using createProjectIndex using the newClient factory function.
// This method only *starts* the reconciliation, but actual process runs within
// RunProjectReconciliation.
func (c *client) ReconcileIndexedProjects(ctx context.Context) {
	projects := c.receiverIndexer.ListIndexFuncValues(projectReceiverIndex)

	select {
	case <-ctx.Done():
	case c.updateProjectsToReconcileC <- projects:
	}
}

// isEqualProjectSets returns true if the given project sets are equal.
// The order and possible duplicates don't matter.
func isEqualProjectSets(projects1, projects2 sets.Set[string]) bool {
	return projects1.Equal(projects2)
}

// reconcileProjects makes an API call to the server to wait for reconciliation updates of a set of projects.
// Once one of these projects is updated it triggers the associated FluxCD notification/Receiver webhook.
func (c *client) reconcileProjects(ctx context.Context, projects sets.Set[string]) {
	c.log.Debug("Started watching projects for reconciliation", logz.ProjectsToReconcile(projects))
	defer c.log.Debug("Stopped watching projects for reconciliation", logz.ProjectsToReconcile(projects))

	req := &rpc.ReconcileProjectsRequest{Project: rpc.ReconcileProjectsFromSet(projects)}

	_ = retry.PollWithBackoff(ctx, c.pollConfig, func(ctx context.Context) (error, retry.AttemptResult) { //nolint:staticcheck
		ctx, cancel := context.WithCancel(ctx)
		defer cancel() // ensure streaming call is canceled
		rpcClient, err := c.fluxGitLabClient.ReconcileProjects(ctx, req, grpc.WaitForReady(true))
		if err != nil {
			c.agentAPI.HandleProcessingError(ctx, c.log, "Failed to reconcile projects", err, logz.AgentKey(c.agentKey))
			return nil, retry.Backoff
		}

		for {
			resp, err := rpcClient.Recv()
			if err != nil {
				if err == io.EOF { //nolint:errorlint
					// server closed connection, retrying
					return nil, retry.ContinueImmediately
				}
				if grpctool.RequestCanceled(err) {
					// request was canceled
					c.log.Debug("ReconcileProjects request has been canceled, backing off and awaiting cancellation")
					return nil, retry.Backoff
				}
				c.agentAPI.HandleProcessingError(ctx, c.log, "Failed to receive project to reconcile", err, logz.AgentKey(c.agentKey))
				return nil, retry.Backoff
			}

			c.reconcileProject(ctx, resp.Project.Id)
		}
	})
}

// reconcileProject reconciles a single project by triggering the FluxCD notification/Receiver webhooks
// indexed with this project.
// The Receiver object must have:
// - a projectAnnotationKey that matches the given project
// - a webhook path
// ... in order to be triggered.
func (c *client) reconcileProject(ctx context.Context, project string) {
	log := c.log.With(logz.ProjectID(project))
	objs, err := c.receiverIndexer.ByIndex(projectReceiverIndex, project)
	if err != nil {
		log.Error("Unable to get Receivers for project", logz.Error(err))
		return
	}

	for _, obj := range objs {
		u := obj.(*unstructured.Unstructured)
		var nr notificationv1.Receiver
		err = runtime.DefaultUnstructuredConverter.FromUnstructured(u.UnstructuredContent(), &nr)
		if err != nil {
			log.Error("Unable to convert unstructured object to Receiver", logz.Error(err))
			continue
		}

		if p := nr.Annotations[api.ProjectKey]; project != p || nr.Status.WebhookPath == "" {
			continue
		}

		secretName := nr.Spec.SecretRef.Name
		ns := nr.Namespace
		secret, err := c.corev1APIClient.Secrets(ns).Get(ctx, secretName, metav1.GetOptions{})

		if err != nil {
			log.Error("Unable to get secret value for Receiver", logz.Error(err))
			return
		}

		err = c.reconcileTrigger.reconcile(ctx, nr.Status.WebhookPath, secret.Data["token"])
		if err != nil {
			log.Error("Unable to trigger Receiver", logz.Error(err))
			continue
		}
	}
}
