package events

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"hash/fnv"
	"log"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/nexuscrm/backend/internal/domain/ports"
	"github.com/nexuscrm/backend/internal/infrastructure/persistence"
	apperrors "github.com/nexuscrm/backend/pkg/errors"
	"github.com/nexuscrm/backend/pkg/utils"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/events"
)

// OutboxEvent status constants
const (
	OutboxStatusPending   = "pending"
	OutboxStatusProcessed = "processed"
	OutboxStatusFailed    = "failed"
	MaxRetryAttempts      = 5

	// defaultOutboxParallelism bounds how many events in a single
	// ProcessOutbox batch are dispatched concurrently. Each event uses its
	// own transaction and ClaimEvent atomically transitions the row from
	// 'pending' → 'processing', so concurrent dispatch of distinct events
	// is safe; this cap just limits goroutine / DB-connection fan-out.
	defaultOutboxParallelism = 8
	// minOutboxParallelism / maxOutboxParallelism clamp the env-override
	// so a typo in OUTBOX_WORKER_PARALLELISM can't collapse throughput
	// to zero or exhaust the DB pool.
	minOutboxParallelism = 1
	maxOutboxParallelism = 64

	// outboxParallelismEnv is the env var that overrides
	// defaultOutboxParallelism. Parsed once at service construction.
	outboxParallelismEnv = "OUTBOX_WORKER_PARALLELISM"
)

// outboxNoSubscriberObjects names objects whose record events have no
// useful subscriber and would only consume worker bandwidth if enqueued.
//
// FlowInterview is the canonical case. The executor inserts a
// FlowInterview row on every persistent flow run; that INSERT goes
// through PersistenceService and would normally enqueue a
// record.created event. The dispatcher then matches zero
// record-triggered flows for FlowInterview and returns nil — the row
// was pure overhead. Under sustained flow load on sandbox
// new-prod-essential (2026-04-22) this self-amplification produced
// 4500+ pending events in <30 minutes, blocking real CRM events from
// being processed (issue #1658).
//
// Map keys are derived from the table-name constants (see
// constants.TableFlowInterview) and lowercased so the lookup matches
// any caller-side casing. Sourcing from the constant means a future
// table rename moves the key automatically — no silent un-filtering.
// Membership is intentionally narrow: only objects whose record
// events are provably pure noise belong here.
var outboxNoSubscriberObjects = map[string]bool{
	strings.ToLower(constants.TableFlowInterview): true,
}

// OutboxMetadataLookup is the minimal MetadataService surface that the
// outbox needs to validate the dynamic arm of outboxNoSubscriberObjects.
// Defined as an interface (not a hard dependency on *MetadataService) so
// production wiring stays in service_manager_init.go and tests can mock
// the lookup without dragging in the whole metadata layer.
type OutboxMetadataLookup interface {
	HasActiveRecordTriggeredFlow(ctx context.Context, objectAPIName string) bool
}

// OutboxService handles transactional event storage and async publishing.
// It implements the Outbox Pattern for guaranteed event delivery.
type OutboxService struct {
	repo      ports.OutboxRepository
	eventBus  ports.EventPublisher
	txManager ports.TransactionManager

	// metadata is OPTIONAL. When nil, the outboxNoSubscriberObjects
	// filter falls through to the enqueue path (fail-safe: do not
	// silently drop events when we cannot prove there is no subscriber).
	// Production wires this via WithMetadataLookup; most unit tests
	// leave it nil because they do not exercise the noise path.
	metadata OutboxMetadataLookup

	// parallelism bounds the number of goroutines dispatching events in
	// a single ProcessOutbox batch. Read from OUTBOX_WORKER_PARALLELISM
	// at construction, clamped to [minOutboxParallelism, maxOutboxParallelism].
	parallelism int

	// Worker control
	stopCh   chan struct{}
	stopOnce sync.Once
	wg       sync.WaitGroup
}

// NewOutboxService creates a new OutboxService
func NewOutboxService(repo ports.OutboxRepository, eventBus ports.EventPublisher, txManager ports.TransactionManager) *OutboxService {
	return &OutboxService{
		repo:        repo,
		eventBus:    eventBus,
		txManager:   txManager,
		parallelism: resolveOutboxParallelism(),
		stopCh:      make(chan struct{}),
	}
}

// WithMetadataLookup wires the OutboxMetadataLookup that arms the
// dynamic side of the outboxNoSubscriberObjects filter. Returns the
// receiver so callers can chain it onto NewOutboxService.
//
// When unset, the filter degrades to the safe fallback (enqueue, never
// drop). Production must call this with the live MetadataService so the
// filter actually fires; tests can leave it unset OR pass a stub when
// they specifically exercise the drop path.
func (os *OutboxService) WithMetadataLookup(m OutboxMetadataLookup) *OutboxService {
	os.metadata = m
	return os
}

// resolveOutboxParallelism reads the env override, clamps it, and falls
// back to defaultOutboxParallelism on parse failure. Centralised so the
// constructor stays short and the clamp logic is easy to unit-test.
func resolveOutboxParallelism() int {
	raw := os.Getenv(outboxParallelismEnv)
	if raw == "" {
		return defaultOutboxParallelism
	}
	n, err := strconv.Atoi(raw)
	if err != nil {
		log.Printf("⚠️ [Outbox] invalid %s=%q, falling back to %d: %v", outboxParallelismEnv, raw, defaultOutboxParallelism, err)
		return defaultOutboxParallelism
	}
	if n < minOutboxParallelism {
		return minOutboxParallelism
	}
	if n > maxOutboxParallelism {
		return maxOutboxParallelism
	}
	return n
}

// EnqueueEvent stores an event in the outbox table within the current transaction.
// The transaction is extracted from the context if present.
//
// Events for objects in outboxNoSubscriberObjects (e.g. FlowInterview)
// are dropped silently when no active record-triggered flow is wired
// against the object — see that variable's doc + the dynamic check
// below for why both arms are required.
func (os *OutboxService) EnqueueEvent(ctx context.Context, eventType events.EventType, payload RecordEventPayload) error {
	if outboxNoSubscriberObjects[strings.ToLower(payload.ObjectAPIName)] {
		// Dynamic arm: even though this object is on the noise candidate
		// list, the frontend TriggerConfigPanel exposes every schema as
		// a record-trigger choice, so a user could legitimately wire a
		// flow against (e.g.) FlowInterview. Drop ONLY when the
		// metadata lookup confirms no active flow is listening.
		//
		// Fail-safe: if the lookup is unwired (most unit tests) or the
		// cache cannot answer, we treat the event as needed and let it
		// through — silently dropping a wired user flow's events is a
		// far worse outcome than carrying a few noise rows through the
		// outbox.
		if os.metadata != nil && !os.metadata.HasActiveRecordTriggeredFlow(ctx, payload.ObjectAPIName) {
			// Per-event volume is bounded: the filter only matches
			// pure-noise objects, so this fires at the writer's rate
			// (typically scheduler / REST flow invokes). The log is the
			// breadcrumb operators need when debugging "why didn't my
			// FlowInterview audit row publish to WS / CDC?".
			log.Printf("⏭ [Outbox] dropped %s for object=%s (no active record-triggered flow; see outboxNoSubscriberObjects)", eventType, payload.ObjectAPIName)
			return nil
		}
	}

	// Inject Tracing Metadata (Envelope Pattern)
	if payload.Metadata == nil {
		payload.Metadata = make(map[string]string)
	}

	// Capture Trace Context using traceparent only
	if val := ctx.Value(constants.ContextKeyTraceparent); val != nil {
		if str, ok := val.(string); ok {
			payload.Metadata[string(constants.ContextKeyTraceparent)] = str
		}
	}

	// Repository automatically handles transaction extraction from context
	id, err := os.repo.Enqueue(ctx, string(eventType), payload)
	if err != nil {
		return err
	}
	log.Printf("✅ [Outbox] Enqueued event %s (Type: %s, ID: %s)", eventType, string(eventType), id)
	return nil
}

// StartWorker starts the background worker that processes pending outbox events.
// The worker polls with the specified interval.
func (os *OutboxService) StartWorker(interval time.Duration) {
	os.wg.Add(1)
	go func() {
		defer os.wg.Done()
		// Panic Recovery
		defer func() {
			if r := recover(); r != nil {
				log.Printf("🔥 CRITICAL: Outbox worker panicked: %v", r)
			}
		}()

		ticker := time.NewTicker(interval)
		defer ticker.Stop()

		log.Printf("📤 Outbox worker started with %v interval (parallelism=%d)", interval, os.parallelism)

		for {
			select {
			case <-os.stopCh:
				log.Printf("📤 Outbox worker stopping...")
				return
			case <-ticker.C:
				func() {
					// Protected execution to keep worker alive
					defer func() {
						if r := recover(); r != nil {
							log.Printf("🔥 Outbox iteration panic: %v", r)
						}
					}()
					if err := os.ProcessOutbox(context.Background()); err != nil {
						log.Printf("⚠️ Outbox worker error: %v", err)
					}
				}()
			}
		}
	}()
}

// StopWorker stops the background worker gracefully
func (os *OutboxService) StopWorker() {
	os.stopOnce.Do(func() {
		close(os.stopCh)
	})
	os.wg.Wait()
	log.Printf("📤 Outbox worker stopped")
}

// ProcessOutbox processes pending events in the outbox table.
//
// Events are partitioned across `parallelism` worker goroutines by a hash
// of their record identifier (ObjectAPIName + "/" + Record.Id when those
// are present, otherwise the event ID itself). This preserves per-record
// FIFO order — two events for the same record always land on the same
// slot and are processed in CreatedDate ASC order (the order in which
// GetPendingEvents returns them). Cross-record events can interleave
// freely, which is fine because the pre-fix serial loop never gave any
// stronger guarantee across records either.
//
// Each event is processed in its own transaction via processEventAtomic.
// Claim safety is carried by the atomic UPDATE in ClaimEvent; see
// persistence/outbox_repository.go for why that replaces the old
// SELECT ... FOR UPDATE SKIP LOCKED.
func (os *OutboxService) ProcessOutbox(ctx context.Context) error {
	// Short batch-correlation id so every log line emitted inside this
	// invocation can be filtered together. With N=8 goroutines and a
	// 500ms ticker, logs from adjacent batches would otherwise interleave
	// unreadably.
	batchID := utils.GenerateID()
	if len(batchID) >= 8 {
		batchID = batchID[:8]
	}

	pending, err := os.repo.GetPendingEvents(ctx, 100)
	if err != nil {
		return err
	}

	if len(pending) == 0 {
		return nil
	}

	log.Printf("🔄 [Outbox][%s] Processing %d pending events (parallelism=%d)", batchID, len(pending), os.parallelism)

	// Per-slot worker goroutines. Each slot reads from its own channel
	// and processes events sequentially, preserving per-record order
	// within the slot. Same-record events MUST land on the same slot.
	slotChans := make([]chan OutboxEventEnvelope, os.parallelism)
	for i := range slotChans {
		// Larger buffer absorbs bursts when the slot handler briefly
		// slows (DB round-trip, publish retry) so the dispatch loop
		// doesn't block cross-slot fan-out. Correctness does not
		// depend on the buffer size; 16 is a pragmatic compromise
		// between memory and throughput under p99 latency spikes.
		slotChans[i] = make(chan OutboxEventEnvelope, 16)
	}

	var wg sync.WaitGroup
	wg.Add(os.parallelism)
	for i, ch := range slotChans {
		go func(slot int, in <-chan OutboxEventEnvelope) {
			defer wg.Done()
			// Recover is placed INSIDE the per-event func so a single
			// handler panic drops only that event; the slot worker keeps
			// draining the channel for its siblings. If recover lived
			// outside the range loop (the prior Round 2 placement), a
			// panic would unwind the whole goroutine and strand every
			// remaining same-slot event until the next ProcessOutbox
			// tick — a correctness regression for per-record FIFO that
			// also manifested as silent event loss under the 5-minute
			// stale-processing window.
			for env := range in {
				func(env OutboxEventEnvelope) {
					defer func() {
						if r := recover(); r != nil {
							log.Printf("🔥 [Outbox][%s] slot %d event %s panicked: %v", batchID, slot, env.ID, r)
						}
					}()
					if err := os.processEventAtomic(ctx, env.ID, env.EventType, env.Payload, env.RetryCount); err != nil {
						log.Printf("⚠️ [Outbox][%s] slot %d failed event %s: %v", batchID, slot, env.ID, err)
					}
				}(env)
			}
		}(i, ch)
	}

	// Dispatch events to slots. Abort on ctx.Done() so Stop() (and any
	// future caller-cancellable ctx) does not have to wait for the whole
	// batch to dispatch before the worker can exit.
	dispatchErr := func() error {
		for i := range pending {
			ev := pending[i]
			slot := slotForEvent(ev.ID, ev.Payload, os.parallelism)
			env := OutboxEventEnvelope{
				ID:         ev.ID,
				EventType:  ev.EventType,
				Payload:    ev.Payload,
				RetryCount: ev.RetryCount,
			}
			select {
			case slotChans[slot] <- env:
			case <-ctx.Done():
				return ctx.Err()
			}
		}
		return nil
	}()

	// Close all slot channels so workers finish after draining. Do this
	// regardless of dispatchErr so goroutines always terminate; events
	// already buffered in the slot channels still get processed.
	for _, ch := range slotChans {
		close(ch)
	}
	wg.Wait()

	if dispatchErr != nil {
		log.Printf("⚠️ [Outbox][%s] dispatch interrupted: %v", batchID, dispatchErr)
		return dispatchErr
	}
	return nil
}

// OutboxEventEnvelope carries the subset of fields processEventAtomic needs.
// Exported only so the per-slot goroutine signatures stay readable; it
// has no callers outside this file.
type OutboxEventEnvelope struct {
	ID         string
	EventType  string
	Payload    string
	RetryCount int
}

// slotForEvent picks a worker slot deterministically from the event's
// record identity so every event touching the same record lands on the
// same slot and is therefore processed in insertion order. Falls back
// to the event ID (always present) when the payload doesn't parse or
// lacks an ObjectAPIName/Record.Id pair — those are still assigned a
// stable slot, just not grouped with anything else.
//
// Trade-off: same-record FIFO is preserved; cross-record events can
// interleave freely across slots. The pre-fix serial loop never offered
// a stronger cross-record guarantee either — it just happened to be
// CreatedDate-ordered by accident of single-threaded dispatch.
func slotForEvent(id, payloadJSON string, parallelism int) int {
	if parallelism <= 1 {
		return 0
	}
	key := recordKeyFromPayload(payloadJSON)
	if key == "" {
		key = id
	}
	h := fnv.New32a()
	_, _ = h.Write([]byte(key))
	return int(h.Sum32() % uint32(parallelism))
}

// recordKeyFromPayload extracts "ObjectAPIName/Record.Id" if both are
// present and non-empty; otherwise returns "". The payload JSON is
// already unmarshalled a second time inside processEventAtomic — the
// double-decode cost here is negligible compared to the dispatch
// latency we're parallelising around.
func recordKeyFromPayload(payloadJSON string) string {
	if payloadJSON == "" {
		return ""
	}
	var probe struct {
		ObjectAPIName string `json:"object_api_name"`
		Record        struct {
			ID string `json:"Id"`
		} `json:"record"`
	}
	if err := json.Unmarshal([]byte(payloadJSON), &probe); err != nil {
		return ""
	}
	if probe.ObjectAPIName == "" || probe.Record.ID == "" {
		return ""
	}
	return probe.ObjectAPIName + "/" + probe.Record.ID
}

// processEventAtomic claims an event, publishes it, and updates status atomically
func (os *OutboxService) processEventAtomic(ctx context.Context, id, eventType, payloadJSON string, retryCount int) error {
	return os.txManager.WithTransaction(ctx, func(txCtx context.Context) error {
		// Try to claim this specific event. Under the new atomic-UPDATE
		// implementation, ClaimEvent returns ErrOutboxEventAlreadyClaimed
		// when the row is no longer 'pending' — that is a benign race
		// outcome, not a bug, and must NOT abort the tx (which would
		// rollback any status side effect if we ever add one).
		claimedID, err := os.repo.ClaimEvent(txCtx, id)
		if err != nil {
			if errors.Is(err, persistence.ErrOutboxEventAlreadyClaimed) {
				return nil
			}
			return apperrors.NewInternalError("failed to claim event", err)
		}
		if claimedID == "" {
			return nil // Legacy "already claimed" signal; retained for safety
		}

		// Parse payload
		var payload RecordEventPayload
		if err := json.Unmarshal([]byte(payloadJSON), &payload); err != nil {
			log.Printf("❌ [Outbox] Event %s failed payload unmarshal: %v", id, err)
			if markErr := os.repo.UpdateStatus(txCtx, id, OutboxStatusFailed, fmt.Sprintf("invalid payload: %v", err)); markErr != nil {
				return apperrors.NewInternalError("failed to mark event as failed", markErr)
			}
			return nil // Commit the failure status
		}

		// Restore Context from Metadata (Continuity)
		pubCtx := txCtx
		if payload.Metadata != nil {
			for k, v := range payload.Metadata {
				pubCtx = context.WithValue(pubCtx, k, v)
			}
		}

		// Publish via EventBus
		if err := os.eventBus.Publish(pubCtx, events.EventType(eventType), payload); err != nil {
			// Increment retry count
			newRetryCount := retryCount + 1
			if newRetryCount >= MaxRetryAttempts {
				if markErr := os.repo.UpdateStatus(txCtx, id, OutboxStatusFailed, fmt.Sprintf("max retries exceeded: %v", err)); markErr != nil {
					return apperrors.NewInternalError("failed to mark event as failed", markErr)
				}
				return nil // Commit failure
			}

			// Update retry count (and release lock by commit)
			if updateErr := os.repo.IncrementRetry(txCtx, id, newRetryCount, err.Error()); updateErr != nil {
				return apperrors.NewInternalError("failed to update retry count", updateErr)
			}
			log.Printf("⚠️ [Outbox] Event %s failed (Attempt %d/%d). Error: %v", id, newRetryCount, MaxRetryAttempts, err)
			return nil // Commit increment
		}

		// Mark as processed
		if err := os.repo.UpdateStatus(txCtx, id, OutboxStatusProcessed, ""); err != nil {
			return apperrors.NewInternalError("failed to mark as processed", err)
		}

		log.Printf("✅ [Outbox] Successfully processed event %s (Type: %s)", id, eventType)
		return nil // Commit success
	})
}

// CleanupProcessed removes old processed events from the outbox.
// This should be called periodically (e.g., daily) to prevent table bloat.
func (os *OutboxService) CleanupProcessed(ctx context.Context, olderThan time.Duration) (int64, error) {
	cutoff := time.Now().Add(-olderThan)
	return os.repo.CleanupProcessed(ctx, cutoff)
}