package queue

import (
	"context"
	"fmt"
	"sync"
	"time"

	"security-center-backend/internal/pkg/logger"
)

// SyncExecutor is a function that executes a sync job
type SyncExecutor func(ctx context.Context, job *SyncJobPayload) error

// WorkerPool manages a pool of workers that process jobs from the queue
type WorkerPool struct {
	queue              JobQueueInterface
	executor           SyncExecutor
	admission          AdmissionController
	eventReporter      JobEventReporter
	logger             *logger.Logger
	workerCount        int
	stopChan           chan struct{}
	wg                 sync.WaitGroup
	mu                 sync.Mutex
	isRunning          bool
	activeJobs         map[string]bool // track active jobs for metrics
	inProgressAccounts map[uint64]bool // track accounts currently being synced (for rate limiting)
	activeCancels      map[uint64]context.CancelFunc
}

// NewWorkerPool creates a new worker pool
func NewWorkerPool(
	queue JobQueueInterface,
	workerCount int,
	log *logger.Logger,
	executor SyncExecutor,
) *WorkerPool {
	if workerCount <= 0 {
		workerCount = 5 // default
	}

	return &WorkerPool{
		queue:              queue,
		executor:           executor,
		logger:             log,
		workerCount:        workerCount,
		stopChan:           make(chan struct{}),
		activeJobs:         make(map[string]bool),
		inProgressAccounts: make(map[uint64]bool),
		activeCancels:      make(map[uint64]context.CancelFunc),
	}
}

func (wp *WorkerPool) SetAdmissionController(controller AdmissionController) {
	wp.mu.Lock()
	defer wp.mu.Unlock()
	wp.admission = controller
}

func (wp *WorkerPool) SetEventReporter(reporter JobEventReporter) {
	wp.mu.Lock()
	defer wp.mu.Unlock()
	wp.eventReporter = reporter
}

// Start spawns worker goroutines and begins processing jobs
func (wp *WorkerPool) Start() {
	wp.mu.Lock()
	if wp.isRunning {
		wp.mu.Unlock()
		return
	}
	wp.isRunning = true
	if wp.stopChan == nil {
		wp.stopChan = make(chan struct{})
	} else {
		select {
		case <-wp.stopChan:
			wp.stopChan = make(chan struct{})
		default:
		}
	}
	wp.mu.Unlock()

	if wp.logger != nil {
		wp.logger.Info("starting worker pool", "worker_count", wp.workerCount)
	}

	for i := 0; i < wp.workerCount; i++ {
		wp.wg.Add(1)
		go wp.worker(fmt.Sprintf("worker-%d", i))
	}

	// Start cleanup goroutine for stuck jobs
	wp.wg.Add(1)
	go wp.cleanupLoop()
}

// Stop gracefully shuts down the worker pool
func (wp *WorkerPool) Stop() {
	wp.mu.Lock()
	if !wp.isRunning {
		wp.mu.Unlock()
		return
	}
	wp.isRunning = false
	stopChan := wp.stopChan
	wp.mu.Unlock()

	if wp.logger != nil {
		wp.logger.Info("stopping worker pool, waiting for in-flight jobs to complete")
	}
	close(stopChan)
	wp.wg.Wait()
	if wp.logger != nil {
		wp.logger.Info("worker pool stopped")
	}
}

// worker is the main worker loop that processes jobs
func (wp *WorkerPool) worker(consumerName string) {
	defer wp.wg.Done()

	if wp.logger != nil {
		wp.logger.Info("worker started", "consumer", consumerName)
	}

	for {
		select {
		case <-wp.stopChan:
			if wp.logger != nil {
				wp.logger.Info("worker received stop signal", "consumer", consumerName)
			}
			return
		default:
			// Try to dequeue a job
			job, streamID, err := wp.queue.Dequeue(context.Background(), consumerName)
			if err != nil {
				if wp.logger != nil {
					wp.logger.Error("failed to dequeue job", "consumer", consumerName, "error", err)
				}
				time.Sleep(5 * time.Second)
				continue
			}

			if job == nil {
				// No job available, continue loop
				continue
			}

			// Process the job
			wp.processJob(job, streamID, consumerName)
		}
	}
}

// processJob handles a single job execution
func (wp *WorkerPool) processJob(job *SyncJobPayload, streamID, consumerName string) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	wp.trackCancel(job, cancel)
	defer wp.clearCancel(job)

	// Check if account is already being synced (rate limiting)
	if !wp.acquireAccountLock(job.CloudAccountID) {
		wp.deferJob(ctx, job, streamID, "account_overlap", 30*time.Second, map[string]interface{}{
			"slowdown_source":    "coordination_guard",
			"coordination_state": "account_overlap",
			"consumer":           consumerName,
		})
		return
	}
	if decision := wp.evaluateAdmission(ctx, job); !decision.Allowed {
		wp.releaseAccountLock(job.CloudAccountID)
		wp.deferJob(ctx, job, streamID, decision.Reason, decision.Delay, decision.Metadata)
		return
	}
	defer wp.releaseAccountLock(job.CloudAccountID)

	// Mark job as active
	wp.trackJob(job.JobID, true)
	defer wp.trackJob(job.JobID, false)

	if wp.logger != nil {
		wp.logger.Info("processing job",
			"job_id", job.JobID,
			"account_id", job.CloudAccountID,
			"job_type", job.JobType,
			"consumer", consumerName,
		)
	}

	// Execute the sync using the executor callback
	startTime := time.Now()
	err := wp.executor(ctx, job)
	duration := time.Since(startTime)

	if err != nil {
		if wp.logger != nil {
			wp.logger.Error("job failed",
				"job_id", job.JobID,
				"account_id", job.CloudAccountID,
				"duration", duration,
				"error", err,
			)
		}

		// Handle failure (retry or DLQ)
		if failErr := wp.queue.Fail(ctx, job, streamID, err); failErr != nil {
			if wp.logger != nil {
				wp.logger.Error("failed to handle job failure", "job_id", job.JobID, "error", failErr)
			}
		}
		return
	}

	// Job succeeded
	if wp.logger != nil {
		wp.logger.Info("job completed successfully",
			"job_id", job.JobID,
			"account_id", job.CloudAccountID,
			"duration", duration,
		)
	}

	// Acknowledge the job
	if err := wp.queue.Acknowledge(ctx, streamID); err != nil {
		if wp.logger != nil {
			wp.logger.Error("failed to acknowledge job", "job_id", job.JobID, "error", err)
		}
	}
}

func (wp *WorkerPool) CancelSyncJob(syncJobID uint64) bool {
	wp.mu.Lock()
	cancel, ok := wp.activeCancels[syncJobID]
	wp.mu.Unlock()
	if !ok || cancel == nil {
		return false
	}
	cancel()
	return true
}

func (wp *WorkerPool) trackCancel(job *SyncJobPayload, cancel context.CancelFunc) {
	if job == nil || job.SyncJobID == nil || cancel == nil {
		return
	}
	wp.mu.Lock()
	defer wp.mu.Unlock()
	wp.activeCancels[*job.SyncJobID] = cancel
}

func (wp *WorkerPool) clearCancel(job *SyncJobPayload) {
	if job == nil || job.SyncJobID == nil {
		return
	}
	wp.mu.Lock()
	defer wp.mu.Unlock()
	delete(wp.activeCancels, *job.SyncJobID)
}

func (wp *WorkerPool) evaluateAdmission(ctx context.Context, job *SyncJobPayload) AdmissionDecision {
	wp.mu.Lock()
	admission := wp.admission
	activeLocalJobs := len(wp.activeJobs)
	wp.mu.Unlock()
	if admission == nil {
		return AdmissionDecision{Allowed: true}
	}
	decision := admission(ctx, job, activeLocalJobs)
	if decision.Allowed {
		return decision
	}
	if decision.Delay <= 0 {
		decision.Delay = 15 * time.Second
	}
	if decision.Metadata == nil {
		decision.Metadata = map[string]interface{}{}
	}
	if _, ok := decision.Metadata["slowdown_source"]; !ok {
		decision.Metadata["slowdown_source"] = "coordination_guard"
	}
	if _, ok := decision.Metadata["coordination_state"]; !ok && decision.Reason != "" {
		decision.Metadata["coordination_state"] = decision.Reason
	}
	return decision
}

func (wp *WorkerPool) deferJob(ctx context.Context, job *SyncJobPayload, streamID, reason string, delay time.Duration, metadata map[string]interface{}) {
	if wp.logger != nil {
		wp.logger.Info("deferring job",
			"job_id", job.JobID,
			"account_id", job.CloudAccountID,
			"reason", reason,
			"delay", delay,
		)
	}
	if metadata == nil {
		metadata = map[string]interface{}{}
	}
	metadata["reason"] = reason
	if delay > 0 {
		metadata["retry_after_sec"] = int(delay.Seconds())
	}
	if job.SyncJobID != nil && wp.eventReporter != nil {
		wp.eventReporter(ctx, *job.SyncJobID, "warning", "Job deferred by runtime coordination guard", metadata)
	}
	if delay > 0 {
		timer := time.NewTimer(delay)
		defer timer.Stop()
		select {
		case <-timer.C:
		case <-ctx.Done():
		case <-wp.stopChan:
		}
	}
	if err := wp.queue.Enqueue(ctx, job); err != nil {
		if wp.logger != nil {
			wp.logger.Error("failed to requeue deferred job", "job_id", job.JobID, "error", err)
		}
		return
	}
	if err := wp.queue.Acknowledge(ctx, streamID); err != nil {
		if wp.logger != nil {
			wp.logger.Error("failed to acknowledge deferred job", "job_id", job.JobID, "error", err)
		}
	}
}

// acquireAccountLock attempts to acquire a lock for syncing the given account
// Returns true if lock acquired, false if account is already being synced
func (wp *WorkerPool) acquireAccountLock(accountID uint64) bool {
	wp.mu.Lock()
	defer wp.mu.Unlock()

	if wp.inProgressAccounts[accountID] {
		return false
	}

	wp.inProgressAccounts[accountID] = true
	return true
}

// releaseAccountLock releases the lock for the given account
func (wp *WorkerPool) releaseAccountLock(accountID uint64) {
	wp.mu.Lock()
	defer wp.mu.Unlock()
	delete(wp.inProgressAccounts, accountID)
}

// trackJob marks a job as active or inactive for metrics
func (wp *WorkerPool) trackJob(jobID string, active bool) {
	wp.mu.Lock()
	defer wp.mu.Unlock()

	if active {
		wp.activeJobs[jobID] = true
	} else {
		delete(wp.activeJobs, jobID)
	}
}

// GetActiveJobCount returns the number of currently processing jobs
func (wp *WorkerPool) GetActiveJobCount() int {
	wp.mu.Lock()
	defer wp.mu.Unlock()
	return len(wp.activeJobs)
}

func (wp *WorkerPool) WorkerCount() int {
	wp.mu.Lock()
	defer wp.mu.Unlock()
	return wp.workerCount
}

// cleanupLoop periodically checks for and handles stuck jobs
func (wp *WorkerPool) cleanupLoop() {
	defer wp.wg.Done()

	ticker := time.NewTicker(5 * time.Minute)
	defer ticker.Stop()

	for {
		select {
		case <-wp.stopChan:
			return
		case <-ticker.C:
			if err := wp.queue.CleanupStuckJobs(context.Background()); err != nil {
				if wp.logger != nil {
					wp.logger.Error("failed to cleanup stuck jobs", "error", err)
				}
			}
		}
	}
}