worker_manager.go

// SPDX-License-Identifier: Apache-2.0

package git

import (
	"context"
	"fmt"
	"sync"

	"github.com/go-logr/logr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3"
	"github.com/ConfigButler/gitops-reverser/internal/types"
	"github.com/ConfigButler/gitops-reverser/internal/typeset"
)

// DefaultBranchBufferMaxBytes is the default cap on a worker's combined event
// buffer + unpushed-events memory. Operators override this via
// --branch-buffer-max-size (8Mi by default).
const DefaultBranchBufferMaxBytes int64 = 8 * 1024 * 1024

// WorkerManager manages BranchWorkers.
// Creates workers per (repo, branch), shared by multiple GitDestinations.
// Implements controller-runtime's Runnable interface for lifecycle management.
type WorkerManager struct {
	Client client.Client
	Log    logr.Logger

	branchBufferMaxBytes int64
	sensitiveResources   types.SensitiveResourcePolicy

	mu      sync.RWMutex
	workers map[BranchKey]*BranchWorker
	ctx     context.Context
	// mapper is the GVK->GVR resolver injected into every worker so store scans build a
	// resource-identity inventory. It is set once at startup (SetMapper) before any
	// worker is created; a nil mapper keeps workers structure-only.
	mapper typeset.Lookup

	// sshHostKeys configures SSH host-key resolution for every worker's credential reads. Set
	// once at startup (SetSSHHostKeyConfig) before any worker is created.
	sshHostKeys SSHHostKeyConfig
}

// NewWorkerManager creates a new worker manager.
// branchBufferMaxBytes bounds each worker's combined buffer + unpushed-events
// memory. Pass 0 (or a negative value) to use DefaultBranchBufferMaxBytes.
func NewWorkerManager(
	client client.Client,
	log logr.Logger,
	branchBufferMaxBytes int64,
	sensitiveResources types.SensitiveResourcePolicy,
) *WorkerManager {
	if branchBufferMaxBytes <= 0 {
		branchBufferMaxBytes = DefaultBranchBufferMaxBytes
	}
	return &WorkerManager{
		Client:               client,
		Log:                  log,
		branchBufferMaxBytes: branchBufferMaxBytes,
		sensitiveResources:   sensitiveResources,
		workers:              make(map[BranchKey]*BranchWorker),
	}
}

// SetMapper injects the GVK->GVR resolver used by every worker's store scan. It is
// called once at startup, before any GitTarget registers a worker, so each worker
// created by EnsureWorker carries it.
func (m *WorkerManager) SetMapper(mapper typeset.Lookup) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.mapper = mapper
}

// SetSSHHostKeyConfig injects the SSH host-key resolution config used by every worker's credential
// reads. Like SetMapper, it is called once at startup before any worker is created.
func (m *WorkerManager) SetSSHHostKeyConfig(cfg SSHHostKeyConfig) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.sshHostKeys = cfg
}

// RegisterTarget ensures a worker exists for the target's (provider, branch)
// and registers the target with that worker.
// This is called by GitTarget controller when a target becomes Ready.
func (m *WorkerManager) RegisterTarget(
	ctx context.Context,
	targetName, targetNamespace string,
	providerName, providerNamespace string,
	branch, path string,
) error {
	if err := m.EnsureWorker(ctx, providerName, providerNamespace, branch); err != nil {
		return err
	}

	m.Log.Info("GitTarget registered with branch worker",
		"target", fmt.Sprintf("%s/%s", targetNamespace, targetName),
		"workerKey", BranchKey{
			RepoNamespace: providerNamespace,
			RepoName:      providerName,
			Branch:        branch,
		}.String(),
		"path", path)

	return nil
}

// EnsureWorker ensures a worker exists for the given (provider, branch).
// Worker creation/start is protected by the manager lock.
func (m *WorkerManager) EnsureWorker(
	_ context.Context,
	providerName, providerNamespace string,
	branch string,
) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	key := BranchKey{
		RepoNamespace: providerNamespace,
		RepoName:      providerName,
		Branch:        branch,
	}

	if _, exists := m.workers[key]; !exists {
		m.Log.Info("Creating new branch worker", "key", key.String())
		worker := NewBranchWorker(
			m.Client,
			m.Log.WithName("branch-worker"),
			providerName,
			providerNamespace,
			branch,
			newContentWriter(m.sensitiveResources),
			m.branchBufferMaxBytes,
		)
		// Inject the resolver before Start: the field is read only by the event-loop
		// goroutine Start spawns, so setting it here (under m.mu, before that goroutine
		// exists) is race-free.
		worker.mapper = m.mapper
		worker.sshHostKeys = m.sshHostKeys

		if err := worker.Start(m.ctx); err != nil {
			return fmt.Errorf("failed to start worker for %s: %w", key.String(), err)
		}

		m.workers[key] = worker
	}

	return nil
}

// UnregisterTarget removes a GitTarget from its worker.
// Destroys the worker if it was the last target using it.
// This is called by GitTarget controller when a target is deleted.
func (m *WorkerManager) UnregisterTarget(
	_, _ string,
	providerName, providerNamespace string,
	branch string,
) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	key := BranchKey{
		RepoNamespace: providerNamespace,
		RepoName:      providerName,
		Branch:        branch,
	}

	worker, exists := m.workers[key]
	if !exists {
		return nil
	}

	// Worker no longer tracks targets internally - always destroy worker
	// since WorkerManager handles all lifecycle decisions
	m.Log.Info("Unregistering target, destroying worker", "key", key.String())
	worker.Stop()
	delete(m.workers, key)

	return nil
}

// GetWorkerForTarget finds the worker for a target's (provider, branch).
// Returns the worker and true if found, nil and false otherwise.
// This is used by EventRouter to dispatch events to the correct worker.
func (m *WorkerManager) GetWorkerForTarget(
	providerName, providerNamespace string,
	branch string,
) (*BranchWorker, bool) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	key := BranchKey{
		RepoNamespace: providerNamespace,
		RepoName:      providerName,
		Branch:        branch,
	}

	worker, exists := m.workers[key]
	return worker, exists
}

// ReconcileWorkers checks active GitTargets and cleans up orphaned workers.
// This ensures workers are removed when their GitTargets are deleted.
func (m *WorkerManager) ReconcileWorkers(ctx context.Context) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Get all GitTargets
	var targetList configv1alpha3.GitTargetList
	if err := m.Client.List(ctx, &targetList); err != nil {
		return fmt.Errorf("failed to list GitTargets: %w", err)
	}

	// Build set of needed workers from active GitTargets
	neededWorkers := make(map[BranchKey]bool)
	for _, target := range targetList.Items {
		// Skip deleted targets
		if !target.DeletionTimestamp.IsZero() {
			continue
		}

		// Determine namespace (Provider is always in same namespace as Target)
		providerNS := target.Namespace

		key := BranchKey{
			RepoNamespace: providerNS,
			RepoName:      target.Spec.ProviderRef.Name,
			Branch:        target.Spec.Branch,
		}
		neededWorkers[key] = true
	}

	// Cleanup orphaned workers
	for key, worker := range m.workers {
		if !neededWorkers[key] {
			m.Log.Info("Cleaning up orphaned worker", "key", key.String())
			worker.Stop()
			delete(m.workers, key)
		}
	}

	m.Log.V(1).Info("Worker reconciliation complete",
		"activeWorkers", len(m.workers),
		"neededWorkers", len(neededWorkers))

	return nil
}

// Start implements manager.Runnable interface.
// This is called by controller-runtime when the manager starts.
func (m *WorkerManager) Start(ctx context.Context) error {
	// Publish the context under m.mu: EnsureWorker reads m.ctx under the same lock,
	// so guarding the write makes that read race-free regardless of call ordering.
	m.mu.Lock()
	m.ctx = ctx
	m.mu.Unlock()
	m.Log.Info("WorkerManager started")

	<-ctx.Done()

	m.Log.Info("WorkerManager shutting down")
	m.mu.Lock()
	defer m.mu.Unlock()

	// Stop all workers gracefully
	for key, worker := range m.workers {
		m.Log.Info("Stopping worker for shutdown", "key", key.String())
		worker.Stop()
	}

	m.workers = make(map[BranchKey]*BranchWorker)
	m.Log.Info("WorkerManager stopped")
	return nil
}

// NeedLeaderElection ensures only the elected leader manages workers.
// This prevents multiple pods from managing the same workers.
func (m *WorkerManager) NeedLeaderElection() bool {
	return true
}