//! Store-agnostic Engine with built-in settle loop.
//!
//! Engine<D> publishes events to a pluggable [`Store`](crate::store::Store)
//! backend and settles the full causal tree synchronously.

use std::sync::Arc;
use uuid::Uuid;

use anyhow::Result;
use tracing::info;

use crate::aggregator::{Aggregate, Aggregator, AggregatorRegistry, Apply};
use crate::event_store::event_type_short_name;
use crate::handler::{GlobalDlqMapper, Handler, Projection};
use crate::handler_registry::HandlerRegistry;
use crate::job_executor::{HandlerStatus, JobExecutor};
use crate::memory_store::MemoryStore;
use crate::process::{EmitFuture, ProcessHandle};
use crate::store::Store;
use crate::types::{
    EmittedEvent, EventOutcome, EventWorkerConfig, HandlerCompletion, HandlerDlq,
    HandlerResolution, HandlerWorkerConfig, NewEvent, QueuedEvent, QueuedHandler,
    Snapshot, NAMESPACE_SEESAW,
};
use crate::upcaster::{Upcaster, UpcasterRegistry};

/// Store-agnostic Engine with built-in settle loop.
///
/// Publishes events to a pluggable [`Store`] backend (defaults to an
/// in-memory store) and drives them to completion using `JobExecutor`.
///
/// Supply a custom store via [`with_store`](Engine::with_store) for
/// durability, crash recovery, or distributed workers.
pub struct Engine<D>
where
    D: Send + Sync + 'static,
{
    store: Arc<dyn Store>,
    deps: Arc<D>,
    handlers: Arc<HandlerRegistry<D>>,
    aggregators: Arc<AggregatorRegistry>,
    upcasters: Arc<UpcasterRegistry>,
    snapshot_every: Option<u64>,
    event_metadata: serde_json::Map<String, serde_json::Value>,
    global_dlq_mapper: Option<GlobalDlqMapper>,
}

impl<D> Engine<D>
where
    D: Send + Sync + 'static,
{
    /// Create new engine with dependencies.
    ///
    /// Defaults to an in-memory store. Use [`with_store`](Engine::with_store)
    /// to supply a Postgres or other durable backend.
    pub fn new(deps: D) -> Self {
        Self {
            store: Arc::new(MemoryStore::new()),
            deps: Arc::new(deps),
            handlers: Arc::new(HandlerRegistry::new()),
            aggregators: Arc::new(AggregatorRegistry::new()),
            upcasters: Arc::new(UpcasterRegistry::new()),
            snapshot_every: None,
            event_metadata: serde_json::Map::new(),
            global_dlq_mapper: None,
        }
    }

    /// Access the shared dependencies.
    pub fn deps(&self) -> &Arc<D> {
        &self.deps
    }

    /// Read aggregate state by ID. Returns `Arc::new(A::default())` if no state exists.
    pub fn aggregate<A: Aggregate + 'static>(&self, id: Uuid) -> Arc<A> {
        self.aggregators.get_transition_arc::<A>(id).1
    }

    /// Read singleton aggregate state. Returns `Arc::new(A::default())` if no state exists.
    pub fn singleton<A: Aggregate + 'static>(&self) -> Arc<A> {
        self.aggregators.get_singleton_arc::<A>().1
    }

    /// Invalidate cached aggregate state, forcing re-hydration from the Store.
    ///
    /// Use after ingesting foreign events (e.g. from another node) so the
    /// next settle loop rebuilds the aggregate from the persistent log.
    pub fn invalidate_aggregate<A: Aggregate>(&self, id: Uuid) {
        let key = format!("{}:{}", A::aggregate_type(), id);
        self.aggregators.remove_state(&key);
    }

    /// Enable auto-checkpoint snapshots every N events.
    ///
    /// Requires a snapshot store to be set. When both are configured,
    /// snapshots are saved automatically during the settle loop.
    pub fn snapshot_every(mut self, events: u64) -> Self {
        self.snapshot_every = Some(events);
        self
    }

    /// Set metadata to stamp on every persisted event.
    ///
    /// Metadata travels with the event through the Store, letting
    /// adapters pull application-level context (e.g. `run_id`, `schema_v`,
    /// `actor`) without holding state themselves.
    ///
    /// ```ignore
    /// let engine = Engine::new(deps)
    ///     .with_store(my_store)
    ///     .with_event_metadata(serde_json::json!({
    ///         "run_id": "scrape-abc123",
    ///         "schema_v": 1
    ///     }));
    /// ```
    pub fn with_event_metadata(mut self, metadata: serde_json::Value) -> Self {
        if let serde_json::Value::Object(map) = metadata {
            self.event_metadata = map;
        } else {
            panic!("with_event_metadata expects a JSON object");
        }
        self
    }

    /// Supply a custom store backend.
    ///
    /// Replaces the default in-memory store with the provided implementation.
    ///
    /// ```ignore
    /// let engine = Engine::new(deps)
    ///     .with_store(Arc::new(PostgresStore::new(pool)));
    /// ```
    pub fn with_store(mut self, store: Arc<dyn Store>) -> Self {
        self.store = store;
        self
    }

    /// Register a global DLQ callback that fires when any handler exhausts retries.
    ///
    /// The callback receives [`DlqTerminalInfo`] with handler and event metadata,
    /// and returns a serializable event that gets dispatched into the causal chain.
    ///
    /// Per-handler `on_failure` takes precedence when present.
    ///
    /// ```ignore
    /// let engine = Engine::new(deps)
    ///     .on_dlq(|info: DlqTerminalInfo| HandlerFailed {
    ///         handler_id: info.handler_id.clone(),
    ///         error: info.error.clone(),
    ///     });
    /// ```
    pub fn on_dlq<E, F>(mut self, mapper: F) -> Self
    where
        E: serde::Serialize + Send + Sync + 'static,
        F: Fn(crate::handler::DlqTerminalInfo) -> E + Send + Sync + 'static,
    {
        let event_type = std::any::type_name::<E>().to_string();
        self.global_dlq_mapper = Some(Arc::new(move |info| {
            let failed_handler_id = info.handler_id.clone();
            let event = mapper(info);
            Ok(crate::EmittedEvent {
                event_type: event_type.clone(),
                payload: serde_json::to_value(&event)?,
                batch_id: None,
                batch_index: None,
                batch_size: None,
                handler_id: Some(failed_handler_id),
                ephemeral: None,
            })
        }));
        self
    }

    /// Register a handler.
    pub fn with_handler(mut self, handler: Handler<D>) -> Self {
        Arc::get_mut(&mut self.handlers)
            .expect("Cannot add handler after cloning")
            .register(handler);
        self
    }

    /// Register a projection.
    ///
    /// Projections receive ALL events, return `Result<()>`, and run
    /// sequentially before other handlers.
    pub fn with_projection(mut self, projection: Projection<D>) -> Self {
        Arc::get_mut(&mut self.handlers)
            .expect("Cannot add projection after cloning")
            .register_projection(projection);
        self
    }

    /// Register multiple handlers.
    pub fn with_handlers<I>(mut self, handlers: I) -> Self
    where
        I: IntoIterator<Item = Handler<D>>,
    {
        let registry = Arc::get_mut(&mut self.handlers).expect("Cannot add handlers after cloning");
        for handler in handlers {
            registry.register(handler);
        }
        self
    }

    /// Register an aggregator: maintains live in-memory state for aggregate `A`.
    ///
    /// `extract_id` maps the event to the aggregate ID it belongs to.
    ///
    /// ```ignore
    /// let engine = Engine::new(deps)
    ///     .with_aggregator::<OrderPlaced, Order, _>(|e| e.order_id)
    ///     .with_aggregator::<OrderShipped, Order, _>(|e| e.order_id);
    /// ```
    pub fn with_aggregator<E, A, F>(mut self, extract_id: F) -> Self
    where
        E: Clone + Send + Sync + serde::Serialize + serde::de::DeserializeOwned + 'static,
        A: Aggregate + Apply<E> + serde::Serialize + serde::de::DeserializeOwned,
        F: Fn(&E) -> Uuid + Send + Sync + 'static,
    {
        // Register event codec so emitted events of this type can be serialized
        let codec = Arc::new(crate::event_codec::EventCodec {
            event_type: std::any::type_name::<E>().to_string(),
            type_id: std::any::TypeId::of::<E>(),
            decode: Arc::new(|payload| {
                let event: E = serde_json::from_value(payload.clone())?;
                Ok(Arc::new(event))
            }),
        });
        Arc::get_mut(&mut self.handlers)
            .expect("Cannot add aggregator after cloning")
            .register_codec(codec);

        Arc::get_mut(&mut self.aggregators)
            .expect("Cannot add aggregator after cloning")
            .register(Aggregator::new::<E, A, F>(extract_id));
        self
    }

    /// Register multiple aggregators at once.
    pub fn with_aggregators<I>(mut self, aggregators: I) -> Self
    where
        I: IntoIterator<Item = Aggregator>,
    {
        let registry = Arc::get_mut(&mut self.aggregators)
            .expect("Cannot add aggregators after cloning");
        for aggregator in aggregators {
            registry.register(aggregator);
        }
        self
    }

    /// Register an upcaster for event type `E`.
    ///
    /// Upcasters transform old event payloads to the current schema during decode.
    /// Chain multiple upcasters to evolve through versions: v1 → v2 → v3 → current.
    ///
    /// ```ignore
    /// let engine = Engine::new(deps)
    ///     .with_upcaster::<OrderPlaced>(1, |mut v: serde_json::Value| {
    ///         v["currency"] = serde_json::json!("USD");
    ///         Ok(v)
    ///     });
    /// ```
    pub fn with_upcaster<E, F>(mut self, from_version: u32, transform: F) -> Self
    where
        E: 'static,
        F: Fn(serde_json::Value) -> anyhow::Result<serde_json::Value> + Send + Sync + 'static,
    {
        let short_name = event_type_short_name(std::any::type_name::<E>()).to_string();
        Arc::get_mut(&mut self.upcasters)
            .expect("Cannot add upcaster after cloning")
            .register(Upcaster {
                event_type: short_name,
                from_version,
                transform: Arc::new(transform),
            });
        self
    }

    /// Emit an event into the engine (returns lazy future).
    ///
    /// Awaiting directly publishes the event (fire-and-forget).
    /// Chain `.settled()` to drive the full causal tree to completion:
    ///
    /// ```ignore
    /// // Fire-and-forget
    /// engine.emit(event).await?;
    ///
    /// // Synchronous settlement
    /// engine.emit(event).settled().await?;
    /// ```
    pub fn emit<E>(&self, event: E) -> EmitFuture
    where
        E: Clone + Send + Sync + serde::Serialize + serde::de::DeserializeOwned + 'static,
    {
        let engine = self.clone();
        let engine2 = self.clone();

        let publish: crate::process::PublishFn = Box::new(move |correlation_id| {
            Box::pin(async move { engine.publish_event(event, correlation_id).await })
        });

        let settle: crate::process::SettleFn = Box::new(move || {
            Box::pin(async move { engine2.settle().await })
        });

        EmitFuture::new(publish, settle)
    }

    /// Emit a type-erased `EventOutput` directly.
    ///
    /// Use when you have heterogeneous events from an `Events` bag
    /// and need to emit them without downcasting.
    ///
    /// ```ignore
    /// for output in events.into_outputs() {
    ///     engine.emit_output(output).settled().await?;
    /// }
    /// ```
    pub fn emit_output(&self, output: crate::handler::EventOutput) -> EmitFuture {
        let engine = self.clone();
        let engine2 = self.clone();

        let publish: crate::process::PublishFn = Box::new(move |correlation_id| {
            Box::pin(async move { engine.publish_output(output, correlation_id).await })
        });

        let settle: crate::process::SettleFn = Box::new(move || {
            Box::pin(async move { engine2.settle().await })
        });

        EmitFuture::new(publish, settle)
    }

    /// Cancel a running workflow by correlation ID.
    ///
    /// Best-effort "stop-at-next-checkpoint": handlers already mid-execution
    /// will complete, but their output events will be rejected. Side effects
    /// (API calls, emails) from in-flight handlers cannot be undone.
    pub async fn cancel(&self, correlation_id: Uuid) -> Result<()> {
        self.store.cancel_correlation(correlation_id).await
    }

    /// Return a summary of pending work for a correlation ID.
    pub async fn status(&self, correlation_id: Uuid) -> Result<crate::types::QueueStatus> {
        self.store.queue_status(correlation_id).await
    }

    /// Drive all pending events and handlers to completion.
    pub async fn settle(&self) -> Result<()> {
        let executor = JobExecutor::new(
            self.deps.clone(),
            self.store.clone(),
            self.handlers.clone(),
            self.aggregators.clone(),
            self.upcasters.clone(),
            self.global_dlq_mapper.clone(),
        );
        let event_config = EventWorkerConfig::default();
        let handler_config = HandlerWorkerConfig::default();

        loop {
            let mut processed_any = false;

            // Reclaim stale handlers (running longer than timeout)
            self.store.reclaim_stale().await?;

            // Drain event queue (cache cancellation status per correlation_id)
            let mut cancelled_cache = std::collections::HashSet::new();
            while let Some(event) = self.store.poll_next().await? {
                processed_any = true;

                // Cancellation checkpoint 1: reject cancelled events before
                // persisting to the event log or mutating aggregate state.
                let is_cancelled = if cancelled_cache.contains(&event.correlation_id) {
                    true
                } else if self.store.is_cancelled(event.correlation_id).await? {
                    cancelled_cache.insert(event.correlation_id);
                    true
                } else {
                    false
                };
                if is_cancelled {
                    info!(
                        correlation_id = %event.correlation_id,
                        event_id = %event.event_id,
                        event_type = %event.event_type,
                        "Rejecting event: workflow cancelled",
                    );
                    self.store
                        .complete_event(EventOutcome::Rejected {
                            event_row_id: event.id,
                            event_id: event.event_id,
                            error: "cancelled".to_string(),
                            reason: "cancelled".to_string(),
                        })
                        .await?;
                    continue;
                }

                // Persist every event + hydrate cold aggregates before state mutation.
                // Default Store no-ops make this safe for MemoryStore.
                self.persist_and_hydrate(&event).await?;

                // Apply event to live aggregator state before handlers run
                self.apply_to_aggregators(&event);

                // Auto-checkpoint snapshots if configured
                if let Some(threshold) = self.snapshot_every {
                    self.maybe_auto_snapshot(&event, threshold).await?;
                }

                match executor.execute_event(&event, &event_config).await {
                    Ok(commit) => {
                        if !commit.handler_descriptions.is_empty() {
                            if let Err(e) = self
                                .store
                                .set_handler_descriptions(
                                    commit.correlation_id,
                                    commit.handler_descriptions.clone(),
                                )
                                .await
                            {
                                tracing::warn!(
                                    correlation_id = %commit.correlation_id,
                                    "Failed to persist handler descriptions: {}",
                                    e
                                );
                            }
                        }
                        self.store
                            .complete_event(EventOutcome::Processed(commit))
                            .await?;
                    }
                    Err(e) => {
                        self.store
                            .complete_event(EventOutcome::Rejected {
                                event_row_id: event.id,
                                event_id: event.event_id,
                                error: e.to_string(),
                                reason: "settle_error".to_string(),
                            })
                            .await?;
                    }
                }
            }

            // Drain all ready handlers and run them in parallel
            let mut executions = Vec::new();
            while let Some(execution) = self.store.poll_next_handler().await? {
                executions.push(execution);
            }

            if !executions.is_empty() {
                processed_any = true;

                // Cancellation checkpoint 2: batch-check unique correlation IDs,
                // then DLQ cancelled handlers before execution.
                let unique_ids: std::collections::HashSet<Uuid> =
                    executions.iter().map(|e| e.correlation_id).collect();
                let mut cancelled_ids = std::collections::HashSet::new();
                for id in unique_ids {
                    if cancelled_cache.contains(&id)
                        || self.store.is_cancelled(id).await?
                    {
                        cancelled_ids.insert(id);
                    }
                }

                let mut active_executions = Vec::new();
                for execution in executions {
                    if cancelled_ids.contains(&execution.correlation_id) {
                        info!(
                            correlation_id = %execution.correlation_id,
                            event_id = %execution.event_id,
                            handler_id = %execution.handler_id,
                            "DLQ handler: workflow cancelled",
                        );
                        self.store
                            .resolve_handler(HandlerResolution::DeadLetter(HandlerDlq {
                                event_id: execution.event_id,
                                handler_id: execution.handler_id.clone(),
                                error: "cancelled".to_string(),
                                reason: "cancelled".to_string(),
                                attempts: execution.attempts,
                                events_to_publish: Vec::new(),
                                log_entries: Vec::new(),
                            }))
                            .await?;
                    } else {
                        active_executions.push(execution);
                    }
                }
                let executions = active_executions;

                // Hydrate cold aggregates before handler execution.
                // On resume (no events in the queue), aggregates are never
                // populated by persist_and_hydrate, so handlers would see
                // default state.
                //
                // We hydrate ALL registered aggregate types (not just those
                // matching the handler's event type) because handler filters
                // can read any aggregate — e.g. a singleton updated by a
                // completely different event type.
                //
                // For each aggregate type we hydrate:
                // 1. Singletons (Uuid::nil) — always checked
                // 2. Instance IDs extracted from handler event payloads
                for aggregate_type in self.aggregators.unique_aggregate_types() {
                    // Always hydrate singletons
                    let singleton_key = format!("{}:{}", aggregate_type, Uuid::nil());
                    if !self.aggregators.has_state(&singleton_key) {
                        self.hydrate_aggregate(aggregate_type, Uuid::nil(), &singleton_key).await?;
                    }
                }
                // Also hydrate instance aggregates matching handler event payloads
                for execution in &executions {
                    let matching = self.aggregators.find_by_event_type(&execution.event_type);
                    for agg in &matching {
                        let agg_id = match agg.extract_id_from_json(&execution.event_payload) {
                            Some(id) => id,
                            None => continue,
                        };
                        let key = format!("{}:{}", agg.aggregate_type, agg_id);
                        if !self.aggregators.has_state(&key) {
                            self.hydrate_aggregate(&agg.aggregate_type, agg_id, &key).await?;
                        }
                    }
                }

                let handler_futures: Vec<_> = executions
                    .iter()
                    .map(|execution| {
                        executor.execute_handler(execution.clone(), &handler_config)
                    })
                    .collect();

                let handler_results = futures::future::join_all(handler_futures).await;

                for (exec_result, execution_clone) in
                    handler_results.into_iter().zip(executions)
                {
                    let event_id = execution_clone.event_id;
                    let handler_id = execution_clone.handler_id.clone();

                    match exec_result {
                        Ok(result) => match result.status {
                            HandlerStatus::Success => {
                                let events_to_publish = Self::build_queued_events(
                                    result.emitted_events,
                                    event_id,
                                    &handler_id,
                                    execution_clone.correlation_id,
                                    execution_clone.hops,
                                    "",
                                );
                                self.store
                                    .resolve_handler(HandlerResolution::Complete(HandlerCompletion {
                                        event_id,
                                        handler_id,
                                        result: result.result,
                                        events_to_publish,
                                        log_entries: result.log_entries,
                                    }))
                                    .await?;
                            }
                            HandlerStatus::Failed { error, attempts } => {
                                let events_to_publish = Self::build_queued_events(
                                    result.emitted_events,
                                    event_id,
                                    &handler_id,
                                    execution_clone.correlation_id,
                                    execution_clone.hops,
                                    "-dlq",
                                );
                                self.store
                                    .resolve_handler(HandlerResolution::DeadLetter(HandlerDlq {
                                        event_id,
                                        handler_id,
                                        error,
                                        reason: "failed".to_string(),
                                        attempts,
                                        events_to_publish,
                                        log_entries: result.log_entries,
                                    }))
                                    .await?;
                            }
                            HandlerStatus::Retry { error, attempts } => {
                                // Compute backoff schedule in Engine
                                let mut next_execute_at = chrono::Utc::now();
                                if let Some(h) =
                                    executor.handler_registry().find_by_id(&handler_id)
                                {
                                    if let Some(base) = h.backoff {
                                        let multiplier =
                                            2u64.saturating_pow(attempts.max(1) as u32 - 1);
                                        let delay = base.saturating_mul(multiplier as u32);
                                        next_execute_at = chrono::Utc::now()
                                            + chrono::Duration::from_std(delay)
                                                .unwrap_or(chrono::Duration::seconds(60));
                                    }
                                }
                                self.store
                                    .resolve_handler(HandlerResolution::Retry {
                                        event_id,
                                        handler_id,
                                        error,
                                        new_attempts: attempts + 1,
                                        next_execute_at,
                                    })
                                    .await?;
                            }
                            HandlerStatus::Timeout => {
                                let dlq_events = self
                                    .build_global_dlq_event(
                                        &execution_clone,
                                        "Handler execution timed out".to_string(),
                                        "timeout",
                                    )
                                    .unwrap_or_default();
                                let events_to_publish = Self::build_queued_events(
                                    dlq_events,
                                    event_id,
                                    &handler_id,
                                    execution_clone.correlation_id,
                                    execution_clone.hops,
                                    "-dlq",
                                );
                                self.store
                                    .resolve_handler(HandlerResolution::DeadLetter(HandlerDlq {
                                        event_id,
                                        handler_id,
                                        error: "timeout".to_string(),
                                        reason: "timeout".to_string(),
                                        attempts: 0,
                                        events_to_publish,
                                        log_entries: result.log_entries,
                                    }))
                                    .await?;
                            }
                        },
                        Err(e) => {
                            let error_str = e.to_string();
                            let dlq_events = self
                                .build_global_dlq_event(
                                    &execution_clone,
                                    error_str.clone(),
                                    "settle_handler_error",
                                )
                                .unwrap_or_default();
                            let events_to_publish = Self::build_queued_events(
                                dlq_events,
                                event_id,
                                &handler_id,
                                execution_clone.correlation_id,
                                execution_clone.hops,
                                "-dlq",
                            );
                            self.store
                                .resolve_handler(HandlerResolution::DeadLetter(HandlerDlq {
                                    event_id,
                                    handler_id,
                                    error: error_str,
                                    reason: "settle_handler_error".to_string(),
                                    attempts: 0,
                                    events_to_publish,
                                    log_entries: Vec::new(),
                                }))
                                .await?;
                        }
                    }
                }
            }

            if !processed_any {
                // Check if there are future-dated handlers (from backoff or delay).
                // If so, sleep until the earliest one becomes ready instead of exiting.
                if let Some(earliest) = self.store.earliest_pending_handler_at().await? {
                    let now = chrono::Utc::now();
                    if earliest > now {
                        let wait = (earliest - now).to_std().unwrap_or_default();
                        tokio::time::sleep(wait).await;
                        continue;
                    }
                }
                break; // Settled!
            }
        }

        Ok(())
    }

    // --- Internal ---

    /// Try to build a DLQ event using the global mapper.
    /// Returns None if no global mapper is configured or if the mapper errors.
    fn build_global_dlq_event(
        &self,
        execution: &QueuedHandler,
        error: String,
        reason: &str,
    ) -> Option<Vec<crate::types::EmittedEvent>> {
        let global = self.global_dlq_mapper.as_ref()?;
        let info = crate::handler::DlqTerminalInfo {
            handler_id: execution.handler_id.clone(),
            source_event_type: event_type_short_name(&execution.event_type).to_string(),
            source_event_id: execution.event_id,
            error,
            reason: reason.to_string(),
            attempts: execution.attempts,
            max_attempts: execution.max_attempts,
        };
        match global(info) {
            Ok(mut emitted) => {
                if emitted.handler_id.is_none() {
                    emitted.handler_id = Some(execution.handler_id.clone());
                }
                Some(vec![emitted])
            }
            Err(e) => {
                tracing::warn!("global on_dlq mapper failed: {}", e);
                None
            }
        }
    }

    /// Hydrate cold aggregates, then persist every event to the global log.
    ///
    /// 1. For each matching aggregator, hydrate cold aggregates from Store
    /// 2. Always append the event (with aggregate metadata if aggregators match)
    async fn persist_and_hydrate(&self, event: &QueuedEvent) -> Result<()> {
        let short_name = event_type_short_name(&event.event_type);

        // Find matching aggregators to set aggregate metadata
        let matching = self.aggregators.find_by_event_type(&event.event_type);

        // Determine aggregate metadata from the first matching aggregator
        let (aggregate_type, aggregate_id) = matching.iter().find_map(|agg| {
            agg.extract_id_from_json(&event.payload)
                .map(|id| (agg.aggregate_type.clone(), id))
        }).map(|(t, id)| (Some(t), Some(id))).unwrap_or((None, None));

        // Step 1: Hydrate cold aggregates BEFORE appending (so we don't load
        // the event we're about to append)
        for agg in &matching {
            let agg_id = match agg.extract_id_from_json(&event.payload) {
                Some(id) => id,
                None => continue,
            };

            let key = format!("{}:{}", agg.aggregate_type, agg_id);

            if !self.aggregators.has_state(&key) {
                self.hydrate_aggregate(
                    &agg.aggregate_type,
                    agg_id,
                    &key,
                )
                .await?;
            }
        }

        // Step 2: Always append to global log
        let mut metadata = self.event_metadata.clone();
        if let Some(ref hid) = event.handler_id {
            metadata.insert(
                "handler_id".to_string(),
                serde_json::Value::String(hid.clone()),
            );
        }

        self.store
            .append_event(NewEvent {
                event_id: event.event_id,
                parent_id: event.parent_id,
                correlation_id: event.correlation_id,
                event_type: short_name.to_string(),
                payload: event.payload.clone(),
                created_at: event.created_at,
                aggregate_type,
                aggregate_id,
                metadata,
            })
            .await?;

        Ok(())
    }

    /// Hydrate a single aggregate from Store (with optional snapshot acceleration).
    async fn hydrate_aggregate(
        &self,
        aggregate_type: &str,
        aggregate_id: Uuid,
        key: &str,
    ) -> Result<()> {
        // Try snapshot first
        if let Some(snapshot) = self
            .store
            .load_snapshot(aggregate_type, aggregate_id)
            .await?
        {
            let agg = self.aggregators.find_first_by_aggregate_type(aggregate_type);
            if let Some(agg) = agg {
                let mut state = agg.deserialize_state(snapshot.state)?;

                // Load remaining events after snapshot version
                let remaining = self
                    .store
                    .load_stream(aggregate_type, aggregate_id, Some(snapshot.version))
                    .await?;

                if !remaining.is_empty() {
                    let event_pairs: Vec<(&str, &serde_json::Value)> = remaining
                        .iter()
                        .map(|e| (e.event_type.as_str(), &e.payload))
                        .collect();

                    self.aggregators.replay_events_onto(
                        aggregate_type,
                        state.as_mut(),
                        &event_pairs,
                        &self.upcasters,
                    )?;
                }

                let final_version = snapshot.version + remaining.len() as u64;
                self.aggregators
                    .set_state(key, Arc::from(state), final_version, snapshot.version);
                return Ok(());
            }
        }

        // No snapshot — full replay
        let events = self
            .store
            .load_stream(aggregate_type, aggregate_id, None)
            .await?;
        if events.is_empty() {
            return Ok(());
        }

        let event_pairs: Vec<(&str, &serde_json::Value)> = events
            .iter()
            .map(|e| (e.event_type.as_str(), &e.payload))
            .collect();

        let last_version = events.last().unwrap().version.unwrap_or(events.len() as u64);

        if let Some(state) = self.aggregators.replay_events(
            aggregate_type,
            &event_pairs,
            &self.upcasters,
        )? {
            self.aggregators
                .set_state(key, Arc::from(state), last_version, 0);
        }

        Ok(())
    }

    async fn publish_event<E>(
        &self,
        event: E,
        correlation_id_override: Option<Uuid>,
    ) -> Result<ProcessHandle>
    where
        E: Clone + Send + Sync + serde::Serialize + serde::de::DeserializeOwned + 'static,
    {
        // Auto-register codec so the event can be decoded in the settle loop
        let codec = Arc::new(crate::event_codec::EventCodec {
            event_type: std::any::type_name::<E>().to_string(),
            type_id: std::any::TypeId::of::<E>(),
            decode: Arc::new(|payload| {
                let event: E = serde_json::from_value(payload.clone())?;
                Ok(Arc::new(event))
            }),
        });
        self.handlers.register_codec(codec);

        let event_id = Uuid::new_v4();
        let correlation_id = correlation_id_override.unwrap_or_else(Uuid::new_v4);
        let event_type = std::any::type_name::<E>().to_string();
        let payload = serde_json::to_value(&event).expect("Event must be serializable");
        let ephemeral: Arc<dyn std::any::Any + Send + Sync> = Arc::new(event);

        info!(
            "Publishing event: type={}, correlation_id={}",
            event_type, correlation_id
        );

        let queued = QueuedEvent {
            id: 0,
            event_id,
            parent_id: None,
            correlation_id,
            event_type: event_type.clone(),
            payload: payload.clone(),
            hops: 0,
            retry_count: 0,
            batch_id: None,
            batch_index: None,
            batch_size: None,
            handler_id: None,
            created_at: chrono::Utc::now(),
            ephemeral: Some(ephemeral),
        };

        self.store.publish(queued).await?;

        Ok(ProcessHandle {
            correlation_id,
            event_id,
            store: Some(self.store.clone()),
        })
    }

    async fn publish_output(
        &self,
        output: crate::handler::EventOutput,
        correlation_id_override: Option<Uuid>,
    ) -> Result<ProcessHandle> {
        if let Some(codec) = &output.codec {
            self.handlers.register_codec(codec.clone());
        }

        let payload = output.payload;

        let event_id = Uuid::new_v4();
        let correlation_id = correlation_id_override.unwrap_or_else(Uuid::new_v4);

        info!(
            "Publishing event output: type={}, correlation_id={}",
            output.event_type, correlation_id
        );

        let queued = QueuedEvent {
            id: 0,
            event_id,
            parent_id: None,
            correlation_id,
            event_type: output.event_type,
            payload,
            hops: 0,
            retry_count: 0,
            batch_id: None,
            batch_index: None,
            batch_size: None,
            handler_id: None,
            created_at: chrono::Utc::now(),
            ephemeral: output.ephemeral,
        };

        self.store.publish(queued).await?;

        Ok(ProcessHandle {
            correlation_id,
            event_id,
            store: Some(self.store.clone()),
        })
    }

    /// Auto-snapshot aggregates if the event count since last snapshot exceeds the threshold.
    async fn maybe_auto_snapshot(
        &self,
        event: &QueuedEvent,
        threshold: u64,
    ) -> Result<()> {
        let matching = self.aggregators.find_by_event_type(&event.event_type);

        for agg in matching {
            let aggregate_id = match agg.extract_id_from_json(&event.payload) {
                Some(id) => id,
                None => continue,
            };

            let key = format!("{}:{}", agg.aggregate_type, aggregate_id);
            let version = self.aggregators.get_version(&key);
            let snapshot_at = self.aggregators.get_snapshot_at_version(&key);

            if version - snapshot_at >= threshold {
                let state_ref = match self.aggregators.get_state(&key) {
                    Some(s) => s,
                    None => continue,
                };

                let state_json = agg.serialize_state(state_ref.as_ref())?;

                self.store
                    .save_snapshot(Snapshot {
                        aggregate_type: agg.aggregate_type.clone(),
                        aggregate_id,
                        version,
                        state: state_json,
                        created_at: chrono::Utc::now(),
                    })
                    .await?;

                self.aggregators.update_snapshot_at_version(&key, version);
            }
        }

        Ok(())
    }

    /// Apply event to aggregator state.
    fn apply_to_aggregators(&self, event: &QueuedEvent) {
        self.aggregators
            .apply_event(&event.event_type, &event.payload);
    }

    /// Build pre-assigned `QueuedEvent`s from `EmittedEvent`s (ID gen lifted into Engine).
    ///
    /// `id_infix` differentiates ID derivation namespaces: `""` for normal
    /// handler output, `"-dlq"` for DLQ terminal events, etc.
    fn build_queued_events(
        emitted: Vec<EmittedEvent>,
        event_id: Uuid,
        handler_id: &str,
        correlation_id: Uuid,
        parent_hops: i32,
        id_infix: &str,
    ) -> Vec<QueuedEvent> {
        emitted
            .into_iter()
            .enumerate()
            .map(|(idx, e)| {
                let new_event_id = Uuid::new_v5(
                    &NAMESPACE_SEESAW,
                    format!(
                        "{}-{}{}-{}-{}",
                        event_id, handler_id, id_infix, e.event_type, idx
                    )
                    .as_bytes(),
                );
                QueuedEvent {
                    id: 0,
                    event_id: new_event_id,
                    parent_id: Some(event_id),
                    correlation_id,
                    event_type: e.event_type,
                    payload: e.payload,
                    hops: parent_hops + 1,
                    retry_count: 0,
                    batch_id: e.batch_id,
                    batch_index: e.batch_index,
                    batch_size: e.batch_size,
                    handler_id: e.handler_id,
                    created_at: chrono::Utc::now(),
                    ephemeral: e.ephemeral,
                }
            })
            .collect()
    }

}

impl<D> Clone for Engine<D>
where
    D: Send + Sync + 'static,
{
    fn clone(&self) -> Self {
        Self {
            store: self.store.clone(),
            deps: self.deps.clone(),
            handlers: self.handlers.clone(),
            aggregators: self.aggregators.clone(),
            upcasters: self.upcasters.clone(),
            snapshot_every: self.snapshot_every,
            event_metadata: self.event_metadata.clone(),
            global_dlq_mapper: self.global_dlq_mapper.clone(),
        }
    }
}
