//! Effect builder functions and types.

use std::any::{Any, TypeId};
use std::future::Future;
use std::marker::PhantomData;
use std::sync::Arc;
use std::time::Duration;

use anyhow::Result;
#[cfg(test)]
use uuid::Uuid;

use super::context::EffectContext;
use super::types::{AnyEvent, BoxFuture, Effect, Emit, EventOutput, JoinMode};
use crate::event_codec::EventCodec;

#[track_caller]
fn default_effect_id(prefix: &str) -> String {
    let location = std::panic::Location::caller();
    format!(
        "{prefix}@{}:{}:{}",
        location.file(),
        location.line(),
        location.column()
    )
}

fn emit_to_outputs<E: Send + Sync + 'static>(emit: Emit<E>) -> Vec<EventOutput> {
    // `Ok(())` is supported as observer semantics for typed handlers.
    if TypeId::of::<E>() == TypeId::of::<()>() {
        return Vec::new();
    }

    emit.into_vec()
        .into_iter()
        .map(EventOutput::new)
        .collect::<Vec<_>>()
}

// =============================================================================
// Type-State Markers
// =============================================================================

/// Marker for typed event effects (on::<E>()).
pub struct Typed<E>(PhantomData<E>);

/// Marker for any event effects (on_any()).
pub struct Untyped;

/// Marker for no filter.
pub struct NoFilter;

/// Marker for having a filter predicate.
pub struct WithFilter<F>(F);

/// Marker for having a filter_map predicate.
pub struct WithFilterMap<F, T>(F, PhantomData<T>);

/// Marker for no transition predicate.
pub struct NoTransition;

/// Marker for having a transition predicate.
pub struct WithTransition<S, P>(P, PhantomData<S>);

/// Marker for no started handler.
pub struct NoStarted;

/// Marker for having a started handler.
pub struct WithStarted<S, D, St>(St, PhantomData<(S, D)>);

fn typed_event_codec<E>() -> Arc<EventCodec>
where
    E: Clone + Send + Sync + 'static + serde::Serialize + serde::de::DeserializeOwned,
{
    Arc::new(EventCodec {
        event_type: std::any::type_name::<E>().to_string(),
        type_id: TypeId::of::<E>(),
        decode: Arc::new(|payload| {
            let event: E = serde_json::from_value(payload.clone())?;
            Ok(Arc::new(event))
        }),
        encode: Arc::new(|event_any| {
            event_any
                .downcast_ref::<E>()
                .and_then(|event| serde_json::to_value(event).ok())
        }),
    })
}

trait QueueCodecProvider {
    fn queue_codec() -> Option<Arc<EventCodec>>;
}

impl QueueCodecProvider for Untyped {
    fn queue_codec() -> Option<Arc<EventCodec>> {
        None
    }
}

impl<E> QueueCodecProvider for Typed<E>
where
    E: Clone + Send + Sync + 'static + serde::Serialize + serde::de::DeserializeOwned,
{
    fn queue_codec() -> Option<Arc<EventCodec>> {
        Some(typed_event_codec::<E>())
    }
}

// =============================================================================
// Unified EffectBuilder
// =============================================================================

/// A unified builder for effects using the type-state pattern.
pub struct EffectBuilder<EventType, Filter, Trans, Started> {
    filter: Filter,
    transition: Trans,
    started: Started,
    id: Option<String>,
    queued: bool,
    delay: Option<Duration>,
    timeout: Option<Duration>,
    max_attempts: u32,
    priority: Option<i32>,
    codec: Option<Arc<EventCodec>>,
    _marker: PhantomData<EventType>,
}

// =============================================================================
// Entry Points
// =============================================================================

/// Create an effect that handles a specific event type.
///
/// # Example
///
/// ```ignore
/// effect::on::<MyEvent>().then(|event, ctx| async move {
///     Ok(NextEvent { data: event.data })
/// })
/// ```
pub fn on<E: Send + Sync + 'static>() -> EffectBuilder<Typed<E>, NoFilter, NoTransition, NoStarted>
{
    EffectBuilder {
        filter: NoFilter,
        transition: NoTransition,
        started: NoStarted,
        id: None,
        queued: false,
        delay: None,
        timeout: None,
        max_attempts: 1,
        priority: None,
        codec: None,
        _marker: PhantomData,
    }
}

/// Create an effect that handles all events (observer pattern).
///
/// # Example
///
/// ```ignore
/// effect::on_any().then(|event, ctx| async move {
///     ctx.deps().metrics.increment("events");
///     Ok(())
/// })
/// ```
pub fn on_any() -> EffectBuilder<Untyped, NoFilter, NoTransition, NoStarted> {
    EffectBuilder {
        filter: NoFilter,
        transition: NoTransition,
        started: NoStarted,
        id: None,
        queued: false,
        delay: None,
        timeout: None,
        max_attempts: 1,
        priority: None,
        codec: None,
        _marker: PhantomData,
    }
}

// =============================================================================
// Builder Methods - Filter/FilterMap (Typed only)
// =============================================================================

impl<E, Trans, Started> EffectBuilder<Typed<E>, NoFilter, Trans, Started>
where
    E: Send + Sync + 'static,
{
    /// Add a filter predicate that must pass for the handler to run.
    pub fn filter<F>(self, predicate: F) -> EffectBuilder<Typed<E>, WithFilter<F>, Trans, Started>
    where
        F: Fn(&E) -> bool + Send + Sync + 'static,
    {
        EffectBuilder {
            filter: WithFilter(predicate),
            transition: self.transition,
            started: self.started,
            id: self.id,
            queued: self.queued,
            delay: self.delay,
            timeout: self.timeout,
            max_attempts: self.max_attempts,
            priority: self.priority,
            codec: self.codec,
            _marker: PhantomData,
        }
    }

    /// Extract data from events. Handler receives the extracted value.
    ///
    /// # Example
    ///
    /// ```ignore
    /// on::<MyEvent>()
    ///     .extract(|e| match e {
    ///         MyEvent::Variant { id, data } => Some((*id, data.clone())),
    ///         _ => None
    ///     })
    ///     .then(|(id, data), ctx| async move {
    ///         Ok(MyEvent::Processed { id, data })
    ///     })
    /// ```
    pub fn extract<F, T>(
        self,
        extractor: F,
    ) -> EffectBuilder<Typed<E>, WithFilterMap<F, T>, Trans, Started>
    where
        F: Fn(&E) -> Option<T> + Send + Sync + 'static,
        T: Clone + Send + Sync + 'static,
    {
        EffectBuilder {
            filter: WithFilterMap(extractor, PhantomData),
            transition: self.transition,
            started: self.started,
            id: self.id,
            queued: self.queued,
            delay: self.delay,
            timeout: self.timeout,
            max_attempts: self.max_attempts,
            priority: self.priority,
            codec: self.codec,
            _marker: PhantomData,
        }
    }
}

// =============================================================================
// Builder Methods - Transition (Both Typed and Untyped)
// =============================================================================

impl<EventType, Filter, Started> EffectBuilder<EventType, Filter, NoTransition, Started> {
    /// Add a state transition predicate. Handler only runs when predicate returns true.
    pub fn transition<S, P>(
        self,
        predicate: P,
    ) -> EffectBuilder<EventType, Filter, WithTransition<S, P>, Started>
    where
        S: Clone + Send + Sync + 'static,
        P: Fn(&S, &S) -> bool + Send + Sync + 'static,
    {
        EffectBuilder {
            filter: self.filter,
            transition: WithTransition(predicate, PhantomData),
            started: self.started,
            id: self.id,
            queued: self.queued,
            delay: self.delay,
            timeout: self.timeout,
            max_attempts: self.max_attempts,
            priority: self.priority,
            codec: self.codec,
            _marker: PhantomData,
        }
    }
}

// =============================================================================
// Builder Methods - Started (Both Typed and Untyped)
// =============================================================================

impl<EventType, Filter, Trans> EffectBuilder<EventType, Filter, Trans, NoStarted> {
    /// Add a started handler that runs when the store is activated.
    pub fn started<S, D, St, StFut>(
        self,
        started: St,
    ) -> EffectBuilder<EventType, Filter, Trans, WithStarted<S, D, St>>
    where
        S: Clone + Send + Sync + 'static,
        D: Send + Sync + 'static,
        St: Fn(EffectContext<S, D>) -> StFut + Send + Sync + 'static,
        StFut: Future<Output = Result<()>> + Send + 'static,
    {
        EffectBuilder {
            filter: self.filter,
            transition: self.transition,
            started: WithStarted(started, PhantomData),
            id: self.id,
            queued: self.queued,
            delay: self.delay,
            timeout: self.timeout,
            max_attempts: self.max_attempts,
            priority: self.priority,
            codec: self.codec,
            _marker: PhantomData,
        }
    }
}

// =============================================================================
// Builder Methods - Execution Configuration (All Builders)
// =============================================================================

impl<EventType, Filter, Trans, Started> EffectBuilder<EventType, Filter, Trans, Started> {
    /// Set a custom ID for this effect (default: auto-generated).
    pub fn id(mut self, id: impl Into<String>) -> Self {
        self.id = Some(id.into());
        self
    }
}

#[allow(private_bounds)]
impl<EventType, Filter, Trans, Started> EffectBuilder<EventType, Filter, Trans, Started>
where
    EventType: QueueCodecProvider,
{
    /// Force queued execution (default: inline).
    pub fn queued(mut self) -> Self {
        self.queued = true;
        self.codec = EventType::queue_codec();
        self
    }

    /// Add a delay before execution (triggers queued execution).
    pub fn delayed(mut self, duration: Duration) -> Self {
        self.delay = Some(duration);
        self.codec = EventType::queue_codec();
        self
    }

    /// Set execution timeout (triggers queued execution).
    pub fn timeout(mut self, duration: Duration) -> Self {
        self.timeout = Some(duration);
        self.codec = EventType::queue_codec();
        self
    }

    /// Set maximum retry attempts (default: 1 = no retry).
    /// Values > 1 trigger queued execution.
    pub fn retry(mut self, attempts: u32) -> Self {
        self.max_attempts = attempts;
        if attempts > 1 {
            self.codec = EventType::queue_codec();
        }
        self
    }

    /// Set execution priority (lower = higher priority, triggers queued execution).
    pub fn priority(mut self, level: i32) -> Self {
        self.priority = Some(level);
        self.codec = EventType::queue_codec();
        self
    }
}

// =============================================================================
// Join Builder (typed, queued-only)
// =============================================================================

/// Builder for durable same-batch join effects.
pub struct JoinEffectBuilder<E, Started> {
    inner: EffectBuilder<Typed<E>, NoFilter, NoTransition, Started>,
    mode: JoinMode,
}

impl<E, Started> EffectBuilder<Typed<E>, NoFilter, NoTransition, Started>
where
    E: Clone + Send + Sync + 'static,
{
    /// Configure this effect as a durable join effect.
    ///
    /// Join effects are always queued and never execute inline.
    pub fn join(self) -> JoinEffectBuilder<E, Started> {
        JoinEffectBuilder {
            inner: self,
            mode: JoinMode::SameBatch,
        }
    }
}

impl<E, Started> JoinEffectBuilder<E, Started>
where
    E: Clone + Send + Sync + 'static + serde::Serialize + serde::de::DeserializeOwned,
{
    /// Join terminal events that share the same `(correlation_id, batch_id)`.
    pub fn same_batch(mut self) -> Self {
        self.mode = JoinMode::SameBatch;
        self
    }

    /// Set the handler for joined batch execution.
    #[track_caller]
    #[allow(private_bounds)]
    pub fn then<S, D, H, Fut, R, O>(mut self, handler: H) -> Effect<S, D>
    where
        S: Clone + Send + Sync + 'static,
        D: Send + Sync + 'static,
        Started: StartedHandler<S, D>,
        H: Fn(Vec<E>, EffectContext<S, D>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<R>> + Send + 'static,
        R: Into<Emit<O>> + Send + 'static,
        O: Send + Sync + 'static,
    {
        let target = TypeId::of::<E>();
        let id = self
            .inner
            .id
            .take()
            .unwrap_or_else(|| default_effect_id(std::any::type_name::<E>()));
        let join_mode = self.mode;
        let typed_codec = self.inner.codec.take().unwrap_or_else(typed_event_codec::<E>);

        Effect {
            id,
            codecs: vec![typed_codec],
            can_handle: Arc::new(move |t| t == target),
            started: self.inner.started.into_started(),
            // append phase is handled by runtime; this per-item handler is a no-op
            handler: Arc::new(move |_, _, _| Box::pin(async { Ok(Vec::new()) })),
            join_mode: Some(join_mode),
            join_batch_handler: Some(Arc::new(move |values, ctx| {
                let mut typed = Vec::with_capacity(values.len());
                for value in values {
                    let Ok(downcasted) = value.downcast::<E>() else {
                        return Box::pin(async {
                            Err(anyhow::anyhow!("join batch item downcast failed"))
                        });
                    };
                    typed.push((*downcasted).clone());
                }

                let fut = handler(typed, ctx);
                Box::pin(async move {
                    let output: R = fut.await?;
                    let emit: Emit<O> = output.into();
                    Ok(emit_to_outputs(emit))
                })
            })),
            queued: true,
            delay: self.inner.delay,
            timeout: self.inner.timeout,
            max_attempts: self.inner.max_attempts.max(1),
            priority: self.inner.priority,
        }
    }
}

// =============================================================================
// Extraction Trait - Abstracts filter/filter_map logic
// =============================================================================

/// Trait for extracting a value from an event (handles filter/filter_map).
trait Extractor<E, T>: Send + Sync + 'static {
    fn extract(&self, event: &E) -> Option<T>;
}

impl<E: Clone + Send + Sync + 'static> Extractor<E, Arc<E>> for NoFilter {
    fn extract(&self, event: &E) -> Option<Arc<E>> {
        Some(Arc::new(event.clone()))
    }
}

impl<E, F> Extractor<E, Arc<E>> for WithFilter<F>
where
    E: Clone + Send + Sync + 'static,
    F: Fn(&E) -> bool + Send + Sync + 'static,
{
    fn extract(&self, event: &E) -> Option<Arc<E>> {
        if (self.0)(event) {
            Some(Arc::new(event.clone()))
        } else {
            None
        }
    }
}

impl<E, F, T> Extractor<E, T> for WithFilterMap<F, T>
where
    E: Send + Sync + 'static,
    F: Fn(&E) -> Option<T> + Send + Sync + 'static,
    T: Clone + Send + Sync + 'static,
{
    fn extract(&self, event: &E) -> Option<T> {
        (self.0)(event)
    }
}

// =============================================================================
// Transition Trait - Abstracts transition checking
// =============================================================================

/// Trait for checking state transitions.
trait TransitionChecker<S>: Send + Sync + 'static {
    fn should_run(&self, prev: &S, next: &S) -> bool;
}

impl<S> TransitionChecker<S> for NoTransition {
    fn should_run(&self, _prev: &S, _next: &S) -> bool {
        true
    }
}

impl<S, P> TransitionChecker<S> for WithTransition<S, P>
where
    S: Clone + Send + Sync + 'static,
    P: Fn(&S, &S) -> bool + Send + Sync + 'static,
{
    fn should_run(&self, prev: &S, next: &S) -> bool {
        (self.0)(prev, next)
    }
}

// =============================================================================
// Started Trait - Abstracts started handler
// =============================================================================

/// Trait for optional started handlers.
trait StartedHandler<S, D>: Send + Sync + 'static
where
    S: Send + Sync + 'static,
    D: Send + Sync + 'static,
{
    fn into_started(
        self,
    ) -> Option<Arc<dyn Fn(EffectContext<S, D>) -> BoxFuture<Result<()>> + Send + Sync>>;
}

impl<S, D> StartedHandler<S, D> for NoStarted
where
    S: Send + Sync + 'static,
    D: Send + Sync + 'static,
{
    fn into_started(
        self,
    ) -> Option<Arc<dyn Fn(EffectContext<S, D>) -> BoxFuture<Result<()>> + Send + Sync>> {
        None
    }
}

impl<S, D, St, StFut> StartedHandler<S, D> for WithStarted<S, D, St>
where
    S: Clone + Send + Sync + 'static,
    D: Send + Sync + 'static,
    St: Fn(EffectContext<S, D>) -> StFut + Send + Sync + 'static,
    StFut: Future<Output = Result<()>> + Send + 'static,
{
    fn into_started(
        self,
    ) -> Option<Arc<dyn Fn(EffectContext<S, D>) -> BoxFuture<Result<()>> + Send + Sync>> {
        let started = self.0;
        Some(Arc::new(move |ctx| Box::pin(started(ctx))))
    }
}

// =============================================================================
// then() - Unified Implementation for Typed Events
// =============================================================================

impl<E, Filter, Trans, Started> EffectBuilder<Typed<E>, Filter, Trans, Started>
where
    E: Clone + Send + Sync + 'static,
{
    /// Set the handler that returns the next event (terminal operation).
    ///
    /// The output type is inferred from the closure return type.
    /// - Return `Ok(SomeEvent)` to dispatch that event
    /// - Return `Ok(())` to dispatch nothing (observer pattern for typed events)
    ///
    /// # Example
    ///
    /// ```ignore
    /// // Chain effect - returns next event
    /// effect::on::<OrderPlaced>().then(|event, ctx| async move {
    ///     ctx.deps().mailer.send(&event).await?;
    ///     Ok(EmailSent { order_id: event.id })
    /// })
    ///
    /// // Observer effect - returns () to dispatch nothing
    /// effect::on::<OrderPlaced>().then(|event, ctx| async move {
    ///     println!("Order placed: {:?}", event);
    ///     Ok(())
    /// })
    /// ```
    #[track_caller]
    #[allow(private_bounds)]
    pub fn then<S, D, T, H, Fut, R, O>(self, handler: H) -> Effect<S, D>
    where
        S: Clone + Send + Sync + 'static,
        D: Send + Sync + 'static,
        T: Clone + Send + 'static,
        Filter: Extractor<E, T>,
        Trans: TransitionChecker<S>,
        Started: StartedHandler<S, D>,
        H: Fn(T, EffectContext<S, D>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<R>> + Send + 'static,
        R: Into<Emit<O>> + Send + 'static,
        O: Send + Sync + 'static,
    {
        let target = TypeId::of::<E>();
        let filter = self.filter;
        let transition = self.transition;
        let codec = self.codec;

        let id = self
            .id
            .unwrap_or_else(|| default_effect_id(std::any::type_name::<E>()));

        Effect {
            id,
            codecs: codec.into_iter().collect(),
            can_handle: Arc::new(move |t| t == target),
            started: self.started.into_started(),
            handler: Arc::new(move |value, _, ctx| {
                let typed = value.downcast::<E>().expect("type checked by can_handle");

                // Check transition
                if !transition.should_run(ctx.prev_state(), ctx.next_state()) {
                    return Box::pin(async { Ok(Vec::new()) });
                }

                // Extract/filter
                match filter.extract(&typed) {
                    Some(extracted) => {
                        let fut = handler(extracted, ctx);
                        Box::pin(async move {
                            let output: R = fut.await?;
                            let emit: Emit<O> = output.into();
                            Ok(emit_to_outputs(emit))
                        })
                    }
                    None => Box::pin(async { Ok(Vec::new()) }),
                }
            }),
            join_mode: None,
            join_batch_handler: None,
            queued: self.queued,
            delay: self.delay,
            timeout: self.timeout,
            max_attempts: self.max_attempts,
            priority: self.priority,
        }
    }

    /// Queue-capable variant of [`then`].
    ///
    /// Adds serde-based codec metadata so queue-backed workers can decode and
    /// execute this typed effect from persisted JSON events.
    #[track_caller]
    #[allow(private_bounds)]
    pub fn then_queue<S, D, T, H, Fut, R, O>(self, handler: H) -> Effect<S, D>
    where
        S: Clone + Send + Sync + 'static,
        D: Send + Sync + 'static,
        E: Clone + serde::Serialize + serde::de::DeserializeOwned,
        T: Clone + Send + 'static,
        Filter: Extractor<E, T>,
        Trans: TransitionChecker<S>,
        Started: StartedHandler<S, D>,
        H: Fn(T, EffectContext<S, D>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<R>> + Send + 'static,
        R: Into<Emit<O>> + Send + 'static,
        O: Send + Sync + 'static,
    {
        let target = TypeId::of::<E>();
        let filter = self.filter;
        let transition = self.transition;
        let codec = self.codec.unwrap_or_else(typed_event_codec::<E>);

        let id = self
            .id
            .unwrap_or_else(|| default_effect_id(std::any::type_name::<E>()));

        Effect {
            id,
            codecs: vec![codec],
            can_handle: Arc::new(move |t| t == target),
            started: self.started.into_started(),
            handler: Arc::new(move |value, _, ctx| {
                let typed = value.downcast::<E>().expect("type checked by can_handle");

                // Check transition
                if !transition.should_run(ctx.prev_state(), ctx.next_state()) {
                    return Box::pin(async { Ok(Vec::new()) });
                }

                // Extract/filter
                match filter.extract(&typed) {
                    Some(extracted) => {
                        let fut = handler(extracted, ctx);
                        Box::pin(async move {
                            let output: R = fut.await?;
                            let emit: Emit<O> = output.into();
                            Ok(emit_to_outputs(emit))
                        })
                    }
                    None => Box::pin(async { Ok(Vec::new()) }),
                }
            }),
            join_mode: None,
            join_batch_handler: None,
            queued: self.queued,
            delay: self.delay,
            timeout: self.timeout,
            max_attempts: self.max_attempts,
            priority: self.priority,
        }
    }
}

// =============================================================================
// then() - Untyped Events (on_any) - Observer Pattern
// =============================================================================

impl EffectBuilder<Untyped, NoFilter, NoTransition, NoStarted> {
    /// Set the handler for observing all events (terminal operation).
    /// Returns `Result<()>` - observers don't produce events.
    #[track_caller]
    pub fn then<S, D, H, Fut>(self, handler: H) -> Effect<S, D>
    where
        S: Clone + Send + Sync + 'static,
        D: Send + Sync + 'static,
        H: Fn(AnyEvent, EffectContext<S, D>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<()>> + Send + 'static,
    {
        let id = self.id.unwrap_or_else(|| default_effect_id("effect_any"));

        Effect {
            id,
            codecs: Vec::new(),
            can_handle: Arc::new(|_| true),
            started: None,
            handler: Arc::new(move |value, type_id, ctx| {
                let event = AnyEvent { value, type_id };
                let fut = handler(event, ctx);
                Box::pin(async move {
                    fut.await?;
                    Ok(Vec::new())
                })
            }),
            join_mode: None,
            join_batch_handler: None,
            queued: self.queued,
            delay: self.delay,
            timeout: self.timeout,
            max_attempts: self.max_attempts,
            priority: self.priority,
        }
    }
}

impl<S, P> EffectBuilder<Untyped, NoFilter, WithTransition<S, P>, NoStarted>
where
    S: Clone + Send + Sync + 'static,
    P: Fn(&S, &S) -> bool + Send + Sync + 'static,
{
    /// Set the handler for state transitions (terminal operation).
    /// Handler receives only context since transitions are about state changes.
    #[track_caller]
    pub fn then<D, H, Fut>(self, handler: H) -> Effect<S, D>
    where
        D: Send + Sync + 'static,
        H: Fn(EffectContext<S, D>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<()>> + Send + 'static,
    {
        let id = self
            .id
            .unwrap_or_else(|| default_effect_id("effect_transition"));
        let transition = self.transition.0;

        Effect {
            id,
            codecs: Vec::new(),
            can_handle: Arc::new(|_| true),
            started: None,
            handler: Arc::new(move |_, _, ctx| {
                if !transition(ctx.prev_state(), ctx.next_state()) {
                    return Box::pin(async { Ok(Vec::new()) });
                }
                let fut = handler(ctx);
                Box::pin(async move {
                    fut.await?;
                    Ok(Vec::new())
                })
            }),
            join_mode: None,
            join_batch_handler: None,
            queued: self.queued,
            delay: self.delay,
            timeout: self.timeout,
            max_attempts: self.max_attempts,
            priority: self.priority,
        }
    }
}

impl<S, D, St, StFut> EffectBuilder<Untyped, NoFilter, NoTransition, WithStarted<S, D, St>>
where
    S: Clone + Send + Sync + 'static,
    D: Send + Sync + 'static,
    St: Fn(EffectContext<S, D>) -> StFut + Send + Sync + 'static,
    StFut: Future<Output = Result<()>> + Send + 'static,
{
    /// Set the handler for observing all events with started (terminal operation).
    #[track_caller]
    pub fn then<H, Fut>(self, handler: H) -> Effect<S, D>
    where
        H: Fn(AnyEvent, EffectContext<S, D>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<()>> + Send + 'static,
    {
        let id = self
            .id
            .unwrap_or_else(|| default_effect_id("effect_any_started"));
        let started = self.started.0;

        Effect {
            id,
            codecs: Vec::new(),
            can_handle: Arc::new(|_| true),
            started: Some(Arc::new(move |ctx| Box::pin(started(ctx)))),
            handler: Arc::new(move |value, type_id, ctx| {
                let event = AnyEvent { value, type_id };
                let fut = handler(event, ctx);
                Box::pin(async move {
                    fut.await?;
                    Ok(Vec::new())
                })
            }),
            join_mode: None,
            join_batch_handler: None,
            queued: self.queued,
            delay: self.delay,
            timeout: self.timeout,
            max_attempts: self.max_attempts,
            priority: self.priority,
        }
    }
}

impl<S, D, St, StFut, P>
    EffectBuilder<Untyped, NoFilter, WithTransition<S, P>, WithStarted<S, D, St>>
where
    S: Clone + Send + Sync + 'static,
    D: Send + Sync + 'static,
    St: Fn(EffectContext<S, D>) -> StFut + Send + Sync + 'static,
    StFut: Future<Output = Result<()>> + Send + 'static,
    P: Fn(&S, &S) -> bool + Send + Sync + 'static,
{
    /// Set the handler for state transitions with started (terminal operation).
    #[track_caller]
    pub fn then<H, Fut>(self, handler: H) -> Effect<S, D>
    where
        H: Fn(EffectContext<S, D>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<()>> + Send + 'static,
    {
        let id = self
            .id
            .unwrap_or_else(|| default_effect_id("effect_transition_started"));
        let started = self.started.0;
        let transition = self.transition.0;

        Effect {
            id,
            codecs: Vec::new(),
            can_handle: Arc::new(|_| true),
            started: Some(Arc::new(move |ctx| Box::pin(started(ctx)))),
            handler: Arc::new(move |_, _, ctx| {
                if !transition(ctx.prev_state(), ctx.next_state()) {
                    return Box::pin(async { Ok(Vec::new()) });
                }
                let fut = handler(ctx);
                Box::pin(async move {
                    fut.await?;
                    Ok(Vec::new())
                })
            }),
            join_mode: None,
            join_batch_handler: None,
            queued: self.queued,
            delay: self.delay,
            timeout: self.timeout,
            max_attempts: self.max_attempts,
            priority: self.priority,
        }
    }
}

// =============================================================================
// =============================================================================
// group() - Compose multiple effects
// =============================================================================

/// Compose multiple effects into a single effect.
///
/// # Example
///
/// ```ignore
/// effect::group([
///     effect::on::<EventA>().then(handle_a),
///     effect::on::<EventB>().then(handle_b),
/// ])
/// ```
#[track_caller]
pub fn group<S, D>(effects: impl IntoIterator<Item = Effect<S, D>>) -> Effect<S, D>
where
    S: Clone + Send + Sync + 'static,
    D: Send + Sync + 'static,
{
    let effects: Arc<Vec<Effect<S, D>>> = Arc::new(effects.into_iter().collect());
    let mut codecs = Vec::new();
    for effect in effects.iter() {
        codecs.extend(effect.codecs().iter().cloned());
    }

    Effect {
        id: default_effect_id("effect_group"),
        codecs,
        can_handle: {
            let effects = effects.clone();
            Arc::new(move |type_id| effects.iter().any(|e| (e.can_handle)(type_id)))
        },
        started: {
            let effects = effects.clone();
            Some(Arc::new(move |ctx: EffectContext<S, D>| {
                let effects = effects.clone();
                Box::pin(async move {
                    for effect in effects.iter() {
                        if let Some(ref started) = effect.started {
                            started(ctx.clone()).await?;
                        }
                    }
                    Ok(())
                })
            }))
        },
        handler: {
            let effects = effects.clone();
            Arc::new(
                move |value: Arc<dyn Any + Send + Sync>,
                      type_id: TypeId,
                      ctx: EffectContext<S, D>| {
                    let effects = effects.clone();
                    Box::pin(async move {
                        let mut outputs = Vec::new();
                        let mut first_error: Option<anyhow::Error> = None;
                        for effect in effects.iter() {
                            if (effect.can_handle)(type_id) {
                                match (effect.handler)(value.clone(), type_id, ctx.clone()).await {
                                    Ok(mut emitted) => outputs.append(&mut emitted),
                                    Err(e) => {
                                        // Capture first error but continue processing
                                        if first_error.is_none() {
                                            first_error = Some(e);
                                        }
                                    }
                                }
                            }
                        }
                        // Return first error after all effects have run
                        if let Some(err) = first_error {
                            return Err(err);
                        }
                        Ok(outputs)
                    })
                },
            )
        },
        join_mode: None,
        join_batch_handler: None,
        queued: false,
        delay: None,
        timeout: None,
        max_attempts: 1,
        priority: None,
    }
}

// =============================================================================
// task() - Long-running task without event handling
// =============================================================================

/// Create a long-running task that runs on store activation.
///
/// Use for background tasks that don't need to handle events.
///
/// # Example
///
/// ```ignore
/// effect::task(|ctx| async move {
///     while !ctx.is_cancelled() {
///         // background work
///     }
///     Ok(())
/// })
/// ```
#[track_caller]
pub fn task<S, D, F, Fut>(f: F) -> Effect<S, D>
where
    S: Clone + Send + Sync + 'static,
    D: Send + Sync + 'static,
    F: Fn(EffectContext<S, D>) -> Fut + Send + Sync + 'static,
    Fut: Future<Output = Result<()>> + Send + 'static,
{
    Effect {
        id: default_effect_id("effect_task"),
        codecs: Vec::new(),
        can_handle: Arc::new(|_| false),
        started: Some(Arc::new(move |ctx| Box::pin(f(ctx)))),
        handler: Arc::new(|_, _, _| Box::pin(async { Ok(Vec::new()) })),
        join_mode: None,
        join_batch_handler: None,
        queued: false,
        delay: None,
        timeout: None,
        max_attempts: 1,
        priority: None,
    }
}

// =============================================================================
// bridge() - Bidirectional relay connection
// =============================================================================

/// Create a bidirectional bridge to a relay channel.
///
/// Forwards store events to relay and relay events back to store.
/// Prevents feedback loops using unique origin tracking.
#[track_caller]
pub fn bridge<S, D>(tx: pipedream::WeakSender, rx: pipedream::RelayReceiver) -> Effect<S, D>
where
    S: Clone + Send + Sync + 'static,
    D: Send + Sync + 'static,
{
    use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};

    static BRIDGE_ID_COUNTER: AtomicU64 = AtomicU64::new(1 << 32);
    let bridge_origin_id = BRIDGE_ID_COUNTER.fetch_add(1, AtomicOrdering::Relaxed);

    let tx = Arc::new(tx);
    let rx = Arc::new(rx);

    Effect {
        id: default_effect_id("effect_bridge"),
        codecs: Vec::new(),
        can_handle: Arc::new(|_| true),
        started: {
            let rx = rx.clone();
            Some(Arc::new(move |ctx: EffectContext<S, D>| {
                let rx = rx.clone();
                let mut receiver = rx.subscribe_all();

                // Spawn background task to forward events from bridge
                tokio::spawn(async move {
                    while let Some(envelope) = receiver.recv().await {
                        if envelope.origin() == bridge_origin_id {
                            continue;
                        }
                        let type_id = envelope.type_id();
                        ctx.emit_any(envelope.value, type_id);
                    }
                });

                Box::pin(async { Ok(()) })
            }))
        },
        handler: {
            let tx = tx.clone();
            Arc::new(move |value, type_id, _ctx| {
                let tx = tx.clone();
                Box::pin(async move {
                    tx.send_any_with_origin(value, type_id, bridge_origin_id)
                        .await
                        .map_err(|e| anyhow::anyhow!("{:?}", e))?;
                    Ok(Vec::new())
                })
            })
        },
        join_mode: None,
        join_batch_handler: None,
        queued: false,
        delay: None,
        timeout: None,
        max_attempts: 1,
        priority: None,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    #[derive(Clone, Default)]
    struct TestState;

    #[derive(Clone, Default)]
    struct TestDeps;

    #[derive(Clone)]
    struct TestEvent {
        value: i32,
    }

    #[derive(Clone)]
    struct OutputEvent {
        result: i32,
    }

    #[derive(Clone)]
    struct OtherEvent;

    fn create_test_ctx() -> EffectContext<TestState, TestDeps> {
        use parking_lot::RwLock;

        struct NoopEmitter;
        impl super::super::context::EventEmitter<TestState, TestDeps> for NoopEmitter {
            fn emit(
                &self,
                _: TypeId,
                _: Arc<dyn Any + Send + Sync>,
                _: EffectContext<TestState, TestDeps>,
            ) {
            }
        }

        let state = Arc::new(TestState);
        let live_state = Arc::new(RwLock::new(TestState));
        EffectContext::new(
            "test_effect".to_string(),
            "test_idempotency_key".to_string(),
            Uuid::nil(),
            Uuid::nil(),
            state.clone(),
            state,
            live_state,
            Arc::new(TestDeps),
            Arc::new(NoopEmitter),
        )
    }

    #[test]
    fn test_on_can_handle() {
        let effect: Effect<TestState, TestDeps> =
            on::<TestEvent>().then(|_, _| async { Ok(OutputEvent { result: 1 }) });

        assert!(effect.can_handle(TypeId::of::<TestEvent>()));
        assert!(!effect.can_handle(TypeId::of::<OtherEvent>()));
    }

    #[test]
    fn test_on_any_can_handle() {
        let effect: Effect<TestState, TestDeps> = on_any().then(|_, _| async { Ok(()) });

        assert!(effect.can_handle(TypeId::of::<TestEvent>()));
        assert!(effect.can_handle(TypeId::of::<OtherEvent>()));
    }

    #[test]
    fn test_group_can_handle() {
        let effect: Effect<TestState, TestDeps> = group([
            on::<TestEvent>().then(|_, _| async { Ok(OutputEvent { result: 1 }) }),
            on::<OtherEvent>().then(|_, _| async { Ok(OutputEvent { result: 2 }) }),
        ]);

        assert!(effect.can_handle(TypeId::of::<TestEvent>()));
        assert!(effect.can_handle(TypeId::of::<OtherEvent>()));
        assert!(!effect.can_handle(TypeId::of::<String>()));
    }

    #[tokio::test]
    async fn test_then_returns_event() {
        let effect: Effect<TestState, TestDeps> = on::<TestEvent>().then(|event, _| async move {
            Ok(OutputEvent {
                result: event.value * 2,
            })
        });

        let ctx = create_test_ctx();
        let event: Arc<dyn Any + Send + Sync> = Arc::new(TestEvent { value: 21 });

        let result = effect
            .call_handler(event, TypeId::of::<TestEvent>(), ctx)
            .await
            .unwrap();

        assert_eq!(result.len(), 1);
        let output = &result[0];
        assert_eq!(output.type_id, TypeId::of::<OutputEvent>());
        let downcasted = output.value.clone().downcast::<OutputEvent>().unwrap();
        assert_eq!(downcasted.result, 42);
    }

    #[tokio::test]
    async fn test_filter_skips_non_matching() {
        let counter = Arc::new(AtomicUsize::new(0));
        let counter_clone = counter.clone();

        let effect: Effect<TestState, TestDeps> =
            on::<TestEvent>()
                .filter(|e| e.value > 10)
                .then(move |event, _| {
                    let counter = counter_clone.clone();
                    async move {
                        counter.fetch_add(1, Ordering::SeqCst);
                        Ok(OutputEvent {
                            result: event.value,
                        })
                    }
                });

        let ctx = create_test_ctx();

        // Should skip - value <= 10
        let event1: Arc<dyn Any + Send + Sync> = Arc::new(TestEvent { value: 5 });
        let result1 = effect
            .call_handler(event1, TypeId::of::<TestEvent>(), ctx.clone())
            .await
            .unwrap();
        assert!(result1.is_empty());
        assert_eq!(counter.load(Ordering::SeqCst), 0);

        // Should run - value > 10
        let event2: Arc<dyn Any + Send + Sync> = Arc::new(TestEvent { value: 15 });
        let result2 = effect
            .call_handler(event2, TypeId::of::<TestEvent>(), ctx)
            .await
            .unwrap();
        assert_eq!(result2.len(), 1);
        assert_eq!(counter.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn test_filter_map_transforms_and_filters() {
        let effect: Effect<TestState, TestDeps> = on::<TestEvent>()
            .extract(|e| {
                if e.value > 10 {
                    Some((e.value, e.value * 2))
                } else {
                    None
                }
            })
            .then(|(original, doubled), _| async move {
                Ok(OutputEvent {
                    result: original + doubled,
                })
            });

        let ctx = create_test_ctx();

        // Should skip - value <= 10
        let event1: Arc<dyn Any + Send + Sync> = Arc::new(TestEvent { value: 5 });
        let result1 = effect
            .call_handler(event1, TypeId::of::<TestEvent>(), ctx.clone())
            .await
            .unwrap();
        assert!(result1.is_empty());

        // Should run with transformed value - value > 10
        let event2: Arc<dyn Any + Send + Sync> = Arc::new(TestEvent { value: 15 });
        let result2 = effect
            .call_handler(event2, TypeId::of::<TestEvent>(), ctx)
            .await
            .unwrap();

        assert_eq!(result2.len(), 1);
        let output = result2[0]
            .value
            .clone()
            .downcast::<OutputEvent>()
            .unwrap();
        assert_eq!(output.result, 15 + 30); // original + doubled
    }

    #[test]
    fn test_task_never_handles_events() {
        let effect: Effect<TestState, TestDeps> = task(|_ctx| async { Ok(()) });

        assert!(!effect.can_handle(TypeId::of::<TestEvent>()));
        assert!(!effect.can_handle(TypeId::of::<OtherEvent>()));
    }

    #[tokio::test]
    async fn test_task_started_called() {
        let counter = Arc::new(AtomicUsize::new(0));
        let counter_clone = counter.clone();

        let effect: Effect<TestState, TestDeps> = task(move |_ctx| {
            let counter = counter_clone.clone();
            async move {
                counter.fetch_add(1, Ordering::SeqCst);
                Ok(())
            }
        });

        let ctx = create_test_ctx();
        effect.call_started(ctx).await.unwrap();

        assert_eq!(counter.load(Ordering::SeqCst), 1);
    }
}
