fraiseql-server 2.15.0

//! Observer runtime for executing observers in response to database changes.
//!
//! This module integrates the fraiseql-observers crate with the server:
//! 1. Loads observer definitions from `tb_observer`
//! 2. Starts the `ChangeLogListener` to poll `tb_entity_change_log`
//! 3. Routes events through the `ObserverExecutor`
//! 4. Manages lifecycle (startup/shutdown)

use std::{
    collections::HashMap,
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};

use arc_swap::ArcSwap;
use fraiseql_core::runtime::subscription::{ChangeSpineEnvelope, SubscriptionOperation};
use fraiseql_observers::{
    ActionConfig as ObserverActionConfig, ActionExecutionDetail, ChangeLogListener,
    ChangeLogListenerConfig, EntityEvent as ObserverEntityEvent, EventMatcher, FailurePolicy,
    InMemoryTransport, ObserverDefinition, ObserverExecutor, RetryConfig as ObserverRetryConfig,
    checkpoint::{
        CheckpointState as ObserverCheckpointState, CheckpointStore, PostgresCheckpointStore,
    },
    config::{EmailSmtpConfig, TransportConfig, TransportKind},
    transport::{EventFilter, EventTransport},
};
use futures::StreamExt;
use sqlx::PgPool;
use tokio::{
    sync::{RwLock, mpsc, oneshot},
    task::JoinHandle,
};
use tracing::{debug, error, info, warn};

use crate::{
    ServerError,
    observers::{Observer, ObserverRepository},
    subscriptions::event_bridge::EntityEvent as BridgeEntityEvent,
};

#[cfg(test)]
mod tests;

/// Which event-source path [`ObserverRuntime::start`] takes for a given transport.
///
/// This is the listener-selection *seam* for #350: the PostgreSQL transport
/// drives the existing `ChangeLogListener` (LISTEN/NOTIFY) loop, while NATS and
/// the in-memory transport are driven generically through an
/// [`EventTransport`] stream. Keeping the decision in a pure function makes the
/// selection unit-testable without a database or broker, and guarantees a
/// non-Postgres selection never silently falls through to the PG listener.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ListenerSelection {
    /// PostgreSQL LISTEN/NOTIFY via `ChangeLogListener` (the default path).
    PostgresChangeLog,
    /// An `EventTransport` stream (NATS `JetStream`, or in-memory).
    TransportStream,
}

/// Map a transport kind to the event-source path `start()` will use.
pub(crate) const fn listener_selection(kind: TransportKind) -> ListenerSelection {
    match kind {
        TransportKind::Postgres => ListenerSelection::PostgresChangeLog,
        // NATS, in-memory, and any future fraiseql-observers transport variant
        // run through the generic stream path rather than silently using the PG
        // listener. `init_observer_runtime` has already refused/downgraded any
        // transport this binary cannot run, so reaching `start()` with an
        // unknown kind is benign (and still never lands on the PG listener).
        _ => ListenerSelection::TransportStream,
    }
}

/// Build the transport-layer [`NatsConfig`](fraiseql_observers::transport::NatsConfig)
/// from the operator-facing `[observers.runtime.transport.nats]` settings.
///
/// Note: NATS rejects stream names containing `.` or `_`; an offending
/// `stream_name` surfaces as a loud connection/stream-creation error from
/// `NatsTransport::new` rather than a silent fallback.
#[cfg(feature = "observers-nats")]
fn nats_config_from(
    cfg: &fraiseql_observers::config::NatsTransportConfig,
) -> fraiseql_observers::transport::NatsConfig {
    fraiseql_observers::transport::NatsConfig {
        url: cfg.url.clone(),
        stream_name: cfg.stream_name.clone(),
        consumer_name: cfg.consumer_name.clone(),
        subject_prefix: cfg.subject_prefix.clone(),
        ack_wait_secs: cfg.jetstream.ack_wait_secs,
        retention_max_messages: cfg.jetstream.max_msgs,
        retention_max_bytes: cfg.jetstream.max_bytes,
        ..Default::default()
    }
}

/// Configuration for the observer runtime
#[derive(Debug, Clone)]
pub struct ObserverRuntimeConfig {
    /// PostgreSQL connection pool
    pub pool: PgPool,

    /// How often to poll for new change log entries (milliseconds)
    pub poll_interval_ms: u64,

    /// Maximum events to fetch per batch
    pub batch_size: usize,

    /// Channel capacity for event backpressure
    pub channel_capacity: usize,

    /// Whether to automatically reload observers on changes
    pub auto_reload: bool,

    /// Interval to check for observer changes (seconds)
    pub reload_interval_secs: u64,

    /// Maximum entries the in-memory DLQ may hold (`None` = unbounded).
    ///
    /// Mirrors the `fraiseql-observers` library cap: when reached, the newest
    /// failed entry is dropped (drop-newest) with a warning and an overflow
    /// counter bump.
    pub max_dlq_size: Option<usize>,

    /// Event transport selection (PostgreSQL LISTEN/NOTIFY, NATS, in-memory).
    ///
    /// Resolved at boot from `[observers.runtime.transport]` plus
    /// `FRAISEQL_OBSERVER_TRANSPORT`/`FRAISEQL_NATS_*` env overrides, validated,
    /// and gated by `observer_transport_check` (in `server::initialization`).
    /// Consumed by [`ObserverRuntime::start`] to pick the event source.
    pub transport: TransportConfig,

    /// SMTP configuration for the email observer action (`[observers.runtime.email]`).
    ///
    /// `None` leaves the email action without a backend (it fails loud, #349);
    /// `Some(_)` builds a real `lettre` SMTP sender when the executor is created.
    pub email: Option<EmailSmtpConfig>,

    /// Whether to persist the triggering event payload into the
    /// `tb_observer_log.request_payload` column (#468).
    ///
    /// Default `false`: the non-sensitive audit columns (`action_type`,
    /// `action_index`, `response_status_code`, `response_payload` summary) are
    /// always written, but the request payload — which can carry PII from the
    /// entity row — is only stored when an operator opts in via
    /// `[observers.runtime].log_payloads`. Large payloads are truncated to a
    /// marker regardless.
    pub log_payloads: bool,

    /// Stable identity for the change-log cursor in `observer_checkpoints`
    /// (#805).
    ///
    /// The persisted checkpoint is keyed on this — one logical consumer of
    /// `core.tb_entity_change_log` — so a restarted runtime resumes from the
    /// last processed row instead of replaying the entire change log. All
    /// replicas of one deployment share this identity (they are the same
    /// logical consumer). Default: `"change_log"`.
    pub listener_id: String,

    /// Redis backend for the `cache`/`invalidate` observer transport (#985).
    ///
    /// `Some` mounts a [`fraiseql_observers::RedisCacheInvalidator`] onto the
    /// executor at boot, so a `cache` action declared in `fraiseql.toml` really
    /// invalidates keys.
    /// `None` with a declared `cache` action is a **boot error** — the operator
    /// learns at startup rather than one failed dispatch at a time.
    pub redis: Option<fraiseql_observers::config::RedisConfig>,
}

impl ObserverRuntimeConfig {
    /// Create config with defaults (PostgreSQL transport).
    #[must_use]
    pub fn new(pool: PgPool) -> Self {
        Self {
            pool,
            poll_interval_ms: 100,
            batch_size: 100,
            channel_capacity: 1000,
            auto_reload: true,
            reload_interval_secs: 60,
            max_dlq_size: None,
            transport: TransportConfig::default(),
            email: None,
            log_payloads: false,
            listener_id: "change_log".to_string(),
            redis: None,
        }
    }

    /// Set the Redis backend backing `cache`/`invalidate` actions (#985).
    #[must_use]
    pub fn with_redis(mut self, redis: Option<fraiseql_observers::config::RedisConfig>) -> Self {
        self.redis = redis;
        self
    }

    /// Set poll interval
    #[must_use]
    pub const fn with_poll_interval(mut self, ms: u64) -> Self {
        self.poll_interval_ms = ms;
        self
    }

    /// Set batch size
    #[must_use]
    pub const fn with_batch_size(mut self, size: usize) -> Self {
        self.batch_size = size;
        self
    }

    /// Set channel capacity
    #[must_use]
    pub const fn with_channel_capacity(mut self, capacity: usize) -> Self {
        self.channel_capacity = capacity;
        self
    }

    /// Set the in-memory DLQ size cap (`None` = unbounded).
    #[must_use]
    pub const fn with_max_dlq_size(mut self, max: Option<usize>) -> Self {
        self.max_dlq_size = max;
        self
    }

    /// Set the event transport configuration (already env-overridden + validated).
    #[must_use]
    pub fn with_transport(mut self, transport: TransportConfig) -> Self {
        self.transport = transport;
        self
    }

    /// Set the SMTP configuration for the email observer action.
    #[must_use]
    pub fn with_email(mut self, email: Option<EmailSmtpConfig>) -> Self {
        self.email = email;
        self
    }

    /// Enable persisting the triggering event payload into
    /// `tb_observer_log.request_payload` (#468). Off by default.
    #[must_use]
    pub const fn with_log_payloads(mut self, log_payloads: bool) -> Self {
        self.log_payloads = log_payloads;
        self
    }

    /// Set the stable checkpoint identity for the change-log cursor (#805).
    #[must_use]
    pub fn with_listener_id(mut self, listener_id: impl Into<String>) -> Self {
        self.listener_id = listener_id.into();
        self
    }
}

/// Runtime health status
#[derive(Debug, Clone)]
pub struct RuntimeHealth {
    /// Whether the runtime is running
    pub running: bool,

    /// Number of loaded observers
    pub observer_count: usize,

    /// Last checkpoint ID processed
    pub last_checkpoint: Option<i64>,

    /// Total events processed
    pub events_processed: u64,

    /// Total errors encountered
    pub errors: u64,
}

/// Observer runtime that manages the execution loop
pub struct ObserverRuntime {
    config:              ObserverRuntimeConfig,
    repository:          ObserverRepository,
    running:             Arc<AtomicBool>,
    /// Handle to the background processing task
    task_handle:         Option<JoinHandle<()>>,
    /// Channel to send shutdown signal
    shutdown_tx:         Option<mpsc::Sender<()>>,
    /// Statistics
    events_processed:    Arc<std::sync::atomic::AtomicU64>,
    errors:              Arc<std::sync::atomic::AtomicU64>,
    observer_count:      Arc<std::sync::atomic::AtomicUsize>,
    last_checkpoint:     Arc<std::sync::atomic::AtomicI64>,
    /// Hot-swappable components for reload
    matcher:             Arc<RwLock<Option<EventMatcher>>>,
    executor:            Arc<RwLock<Option<Arc<ObserverExecutor>>>>,
    /// `(entity_type, event_type)` → list of observer ids that should be logged
    /// for this combination.  Built locally in [`Self::load_observers`] and
    /// republished by an atomic `ArcSwap` store in `start` / `reload_observers`.
    ///
    /// The whole map is rebuilt off-line and then swapped in a single atomic
    /// pointer write, so concurrent CDC-event lookups always observe either
    /// the fully-populated pre-reload generation or the fully-populated
    /// post-reload generation — never a partial or empty index.  Readers
    /// remain lock-free: `load()` returns a cheap snapshot.
    entity_type_index:   Arc<ArcSwap<HashMap<(String, String), Vec<i64>>>>,
    /// In-memory DLQ shared across reloads and exposed to HTTP handlers.
    dlq:                 Arc<InMemoryDlq>,
    /// Optional sender to forward CDC events to `EventBridge` for GraphQL subscriptions
    event_bridge_sender: Option<mpsc::Sender<BridgeEntityEvent>>,
    /// Optional after:capture dispatch hook (#366). When set (the functions
    /// subsystem is configured), each change-log event is passed to it; the hook
    /// dispatches `after:capture` functions for genuinely-captured writes
    /// (`cdc_source == "fallback_trigger"`). `None` compiles the capture path out.
    capture_dispatch:    Option<CaptureDispatchFn>,
    /// The Redis cache-invalidation transport connected at boot (#985), kept so
    /// a hot reload re-mounts it. A reload rebuilds the executor from scratch;
    /// without this the `cache` actions of a reloaded observer set would silently
    /// stop working while every other action kept going.
    #[cfg(feature = "observers-cache")]
    cache_invalidator:   Option<Arc<fraiseql_observers::RedisCacheInvalidator>>,
}

/// A hook the observer runtime calls for each change-log event (#366).
///
/// Dispatches `after:capture` functions; it owns the function hooks + the `run_as`
/// query-executor factory, and is a no-op for non-captured (executor-written) rows.
pub type CaptureDispatchFn = Arc<dyn Fn(&ObserverEntityEvent) + Send + Sync>;

impl ObserverRuntime {
    /// Create a new observer runtime
    #[must_use]
    pub fn new(config: ObserverRuntimeConfig) -> Self {
        let repository = ObserverRepository::new(config.pool.clone());
        let max_dlq_size = config.max_dlq_size;

        Self {
            config,
            repository,
            running: Arc::new(AtomicBool::new(false)),
            task_handle: None,
            shutdown_tx: None,
            events_processed: Arc::new(std::sync::atomic::AtomicU64::new(0)),
            errors: Arc::new(std::sync::atomic::AtomicU64::new(0)),
            observer_count: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
            last_checkpoint: Arc::new(std::sync::atomic::AtomicI64::new(0)),
            matcher: Arc::new(RwLock::new(None)),
            executor: Arc::new(RwLock::new(None)),
            entity_type_index: Arc::new(ArcSwap::from_pointee(HashMap::new())),
            dlq: Arc::new(InMemoryDlq::new_with_max(max_dlq_size)),
            event_bridge_sender: None,
            capture_dispatch: None,
            #[cfg(feature = "observers-cache")]
            cache_invalidator: None,
        }
    }

    /// Set the `EventBridge` sender so CDC events are forwarded to GraphQL subscriptions.
    pub fn set_event_bridge_sender(&mut self, sender: mpsc::Sender<BridgeEntityEvent>) {
        self.event_bridge_sender = Some(sender);
    }

    /// Set the after:capture dispatch hook (#366) — called for each change-log event
    /// so `after:capture` functions fire on externally-captured writes.
    pub fn set_capture_dispatch(&mut self, hook: CaptureDispatchFn) {
        self.capture_dispatch = Some(hook);
    }

    /// Load observers from the database and convert to `ObserverDefinitions`.
    /// Returns (definitions, `entity_type_index`) tuple.
    /// `entity_type_index` maps (`entity_type`, `event_type`) -> `observer_id` for logging.
    async fn load_observers(
        &self,
    ) -> Result<
        (HashMap<String, ObserverDefinition>, HashMap<(String, String), Vec<i64>>),
        ServerError,
    > {
        // Load all enabled observers
        let query = crate::observers::ListObserversQuery {
            page:            1,
            page_size:       10000, // Load all
            entity_type:     None,
            event_type:      None,
            enabled:         Some(true),
            include_deleted: false,
        };

        let (observers, _total) = self.repository.list(&query, None).await?;

        let mut definitions = HashMap::new();
        let mut entity_type_index: HashMap<(String, String), Vec<i64>> = HashMap::new();

        for observer in observers {
            match Self::convert_observer(&observer) {
                Ok(definition) => {
                    // Index by (entity_type, event_type) for reverse lookup during logging
                    let entity_type =
                        observer.entity_type.clone().unwrap_or_else(|| "*".to_string());
                    let event_type =
                        observer.event_type.clone().unwrap_or_else(|| "INSERT".to_string());
                    entity_type_index
                        .entry((entity_type, event_type.to_uppercase()))
                        .or_default()
                        .push(observer.pk_observer);

                    definitions.insert(observer.name.clone(), definition);
                },
                Err(e) => {
                    warn!("Failed to convert observer {}: {}", observer.name, e);
                },
            }
        }

        info!("Loaded {} observers from database", definitions.len());
        Ok((definitions, entity_type_index))
    }

    /// Convert database Observer to `ObserverDefinition`.
    fn convert_observer(observer: &Observer) -> Result<ObserverDefinition, ServerError> {
        // Parse actions from JSONB
        let actions: Vec<ObserverActionConfig> = serde_json::from_value(observer.actions.clone())
            .map_err(|e| {
            ServerError::Validation(format!(
                "Failed to parse actions for observer {}: {}",
                observer.name, e
            ))
        })?;

        // Parse retry config — fall back to default if deserialization fails, but warn so
        // operators know the stored value is invalid and the observer may behave unexpectedly.
        let retry_config: ObserverRetryConfig =
            match serde_json::from_value(observer.retry_config.clone()) {
                Ok(cfg) => cfg,
                Err(e) => {
                    warn!(
                        observer = %observer.name,
                        error = %e,
                        "Observer retry_config could not be deserialized; using defaults. \
                         Check the stored JSON in tb_observer."
                    );
                    ObserverRetryConfig::default()
                },
            };

        Ok(ObserverDefinition {
            event_type: observer.event_type.clone().unwrap_or_else(|| "INSERT".to_string()),
            entity: observer.entity_type.clone().unwrap_or_else(|| "*".to_string()),
            condition: observer.condition_expression.clone(),
            actions,
            retry: retry_config,
            on_failure: FailurePolicy::default(),
        })
    }

    /// Start the observer runtime
    ///
    /// Selects the event source by the configured transport (#350): the default
    /// PostgreSQL transport drives the `ChangeLogListener` (LISTEN/NOTIFY) loop
    /// below; NATS `JetStream` and the in-memory transport are driven by the
    /// `start_transport_stream` path. A non-Postgres transport never silently
    /// falls through to the PG listener.
    ///
    /// # Errors
    ///
    /// Returns `ServerError` if the runtime is already running, initialization
    /// fails, or the configured transport cannot connect (e.g. an unreachable
    /// NATS broker) — the latter is the #350 dead-broker boot-failure contract.
    pub async fn start(&mut self) -> Result<(), ServerError> {
        if self.running.load(Ordering::SeqCst) {
            return Err(ServerError::ConfigError("Observer runtime already running".to_string()));
        }

        // Listener-selection seam (#350): a non-Postgres transport runs through
        // the generic EventTransport stream path, never the PG listener. Box the
        // delegated future so the (larger, broker-connecting) stream path does not
        // bloat `start()`'s future — and through it `Server::serve` — past the
        // `clippy::large_futures` budget.
        if listener_selection(self.config.transport.transport) == ListenerSelection::TransportStream
        {
            return Box::pin(self.start_transport_stream()).await;
        }

        info!("Starting observer runtime (PostgreSQL LISTEN/NOTIFY transport)...");

        // Load initial observers with entity_type index for logging
        let (observers, entity_type_index) = self.load_observers().await?;
        self.observer_count.store(observers.len(), Ordering::SeqCst);

        // #985: the mount decision reads the loaded action set, and
        // `EventMatcher::build` consumes it.
        let observers_for_mount = observers.clone();

        // Build event matcher
        let matcher = EventMatcher::build(observers).map_err(|e| {
            ServerError::ConfigError(format!("Failed to build event matcher: {}", e))
        })?;

        // Clone matcher for logging (we need it to find matching observers)
        let matcher_for_logging = matcher.clone();

        // Create executor with the shared in-memory DLQ
        let built = ObserverExecutor::new_with_email(
            matcher.clone(),
            self.dlq.clone(),
            self.config.email.as_ref(),
        )
        .map_err(|e| ServerError::ConfigError(format!("invalid observer email config: {e}")))?
        // #632: wire the pool so `database` actions call their PostgreSQL
        // function for real instead of failing "no pool wired".
        .with_database_pool(self.config.pool.clone());
        // #985: same idea for `cache` actions — mount the Redis transport, or
        // refuse to boot if one is declared with no backend (or no build support).
        #[cfg(feature = "observers-cache")]
        let built = {
            self.cache_invalidator =
                connect_cache_invalidator(&observers_for_mount, self.config.redis.as_ref()).await?;
            match self.cache_invalidator.clone() {
                Some(invalidator) => built.with_cache_invalidation(invalidator),
                None => built,
            }
        };
        #[cfg(not(feature = "observers-cache"))]
        reject_cache_actions_not_compiled_in(&observers_for_mount)?;
        let executor = Arc::new(built);

        // Store in shared references for hot reload
        {
            let mut m = self.matcher.write().await;
            *m = Some(matcher_for_logging.clone());
        }
        {
            let mut ex = self.executor.write().await;
            *ex = Some(executor.clone());
        }
        // Publish the freshly-built index with a single atomic pointer swap.
        // Concurrent lookup paths see either the empty initial map or this
        // fully-populated generation — never a half-built state.
        self.entity_type_index.store(Arc::new(entity_type_index));

        // #805: restore the durable cursor before constructing the listener.
        // Without this every process start seeds the cursor at 0 and re-fires
        // every historical observer action in the change log. A fresh database
        // gets the idempotent checkpoint DDL applied on the first load failure;
        // a failure after that is real and fails start() loudly.
        let checkpoint_store = PostgresCheckpointStore::new(self.config.pool.clone());
        let restored = if let Ok(state) = checkpoint_store.load(&self.config.listener_id).await {
            state
        } else {
            {
                crate::migration_lock::run_migration(
                    &self.config.pool,
                    fraiseql_observers::checkpoint::migration_sql(),
                )
                .await
                    .map_err(|e| {
                        ServerError::ConfigError(format!(
                            "observer checkpoint table is missing and could not be created \
                             (apply fraiseql-observers migration 02_create_observer_checkpoints.sql): {e}"
                        ))
                    })?;
                checkpoint_store.load(&self.config.listener_id).await.map_err(|e| {
                    ServerError::ConfigError(format!(
                        "failed to load observer checkpoint for listener '{}': {e}",
                        self.config.listener_id
                    ))
                })?
            }
        };
        if let Some(state) = &restored {
            info!(
                listener_id = %self.config.listener_id,
                resume_from = state.last_processed_id,
                "Restored observer change-log checkpoint; resuming (no replay)"
            );
            self.last_checkpoint.store(state.last_processed_id, Ordering::Relaxed);
        }

        // Create change log listener
        // The listener's dispatch ledger (#935) is keyed by the SAME identity as
        // the checkpoint above, so cursor and ledger describe one listener.
        let mut listener_config = ChangeLogListenerConfig::new(self.config.pool.clone())
            .with_poll_interval(self.config.poll_interval_ms)
            .with_batch_size(self.config.batch_size)
            .with_listener_id(self.config.listener_id.clone());
        if let Some(state) = &restored {
            listener_config = listener_config.with_resume_from(state.last_processed_id);
        }

        // Create shutdown channel
        let (shutdown_tx, mut shutdown_rx) = mpsc::channel::<()>(1);
        self.shutdown_tx = Some(shutdown_tx);

        // Clone state for the background task
        let running = self.running.clone();
        let events_processed = self.events_processed.clone();
        let errors = self.errors.clone();
        let last_checkpoint = self.last_checkpoint.clone();
        let poll_interval = Duration::from_millis(self.config.poll_interval_ms);
        let pool = self.config.pool.clone();
        let log_payloads = self.config.log_payloads;
        let listener_id = self.config.listener_id.clone();
        // Cumulative dispatched-event count carried across restarts via the
        // checkpoint row (#805).
        let mut checkpoint_event_count = restored.as_ref().map_or(0, |s| s.event_count);
        // #366: the after:capture dispatch hook, if the functions subsystem wired one.
        let capture_dispatch = self.capture_dispatch.clone();

        // Clone Arc references for hot reload
        let matcher_ref = Arc::clone(&self.matcher);
        let executor_ref = Arc::clone(&self.executor);
        let entity_type_index_ref = Arc::clone(&self.entity_type_index);

        // Clone optional EventBridge sender for forwarding CDC events to subscriptions
        let bridge_sender = self.event_bridge_sender.clone();

        // Extract non-optional initial values for the background task.
        // SAFETY: These were populated immediately above in this function before we
        // reach this point, so the Option is always Some here.  We surface a proper
        // error rather than panicking so callers get a useful diagnostic.
        let initial_matcher = {
            let m = self.matcher.read().await;
            m.clone().ok_or_else(|| {
                ServerError::ConfigError(
                    "matcher not initialised before spawning background task".to_string(),
                )
            })?
        };
        let initial_executor = {
            let ex = self.executor.read().await;
            ex.clone().ok_or_else(|| {
                ServerError::ConfigError(
                    "executor not initialised before spawning background task".to_string(),
                )
            })?
        };

        debug!("About to spawn background task");
        running.store(true, Ordering::SeqCst);

        // Create a oneshot channel so callers can await readiness before
        // inserting events — eliminates the race between `start()` returning
        // and the background task entering its poll loop.
        let (ready_tx, ready_rx) = oneshot::channel::<()>();

        // Spawn background processing task
        debug!("Calling tokio::spawn()");
        let handle = tokio::spawn(async move {
            let mut listener = ChangeLogListener::new(listener_config);
            // Start with the values captured at launch time; hot-reload replaces them.
            let mut current_matcher = initial_matcher;
            let mut current_executor = initial_executor;

            debug!("Observer runtime background task spawned");
            debug!("Poll interval: {:?}", poll_interval);
            info!("Observer runtime started, beginning event processing loop");

            // Signal that the background task is ready to process events.
            let _ = ready_tx.send(());

            loop {
                tokio::select! {
                    _ = shutdown_rx.recv() => {
                        info!("Observer runtime received shutdown signal");
                        break;
                    }
                    result = listener.next_batch() => {
                        // Refresh matcher/executor from shared slot in case a hot-reload occurred.
                        {
                            let m = matcher_ref.read().await;
                            if let Some(updated) = m.clone() {
                                current_matcher = updated;
                            }
                        }
                        {
                            let ex = executor_ref.read().await;
                            if let Some(updated) = ex.clone() {
                                current_executor = updated;
                            }
                        }

                        match result {
                            Ok(entries) => {
                                if entries.is_empty() {
                                    // No events, wait before polling again
                                    tokio::time::sleep(poll_interval).await;
                                    continue;
                                }

                                debug!("Processing batch of {} change log entries", entries.len());

                                for entry in &entries {
                                    // Convert ChangeLogEntry to EntityEvent
                                    let event = match entry.to_entity_event() {
                                        Ok(e) => e,
                                        Err(e) => {
                                            errors.fetch_add(1, Ordering::Relaxed);
                                            warn!("Failed to convert change log entry to event: {}", e);
                                            continue;
                                        }
                                    };

                                    process_entity_event(
                                        &event,
                                        &current_matcher,
                                        &current_executor,
                                        &entity_type_index_ref,
                                        &pool,
                                        bridge_sender.as_ref(),
                                        &events_processed,
                                        &errors,
                                        log_payloads,
                                    )
                                    .await;

                                    // #366: after:capture — drive functions on
                                    // externally-captured writes. The hook is a
                                    // no-op for executor-written (non-captured)
                                    // rows, so mediated writes never loop back.
                                    if let Some(ref dispatch) = capture_dispatch {
                                        dispatch(&event);
                                    }
                                }

                                // Record the batch as dispatched (#935) BEFORE the
                                // cursor advances: the ledger is what lets the
                                // poller reach back over the commit-lag window for
                                // late-committing rows without re-delivering these
                                // ones. Written after the actions ran, so a crash
                                // in between replays the batch — at-least-once,
                                // unchanged. A failure here is loud but not fatal:
                                // the cost is duplicate delivery, never a skip.
                                if let Err(e) = listener.record_dispatched(&entries).await {
                                    errors.fetch_add(1, Ordering::Relaxed);
                                    error!(
                                        "Failed to record dispatched change-log rows; this batch \
                                         may be re-delivered: {e}"
                                    );
                                }

                                // Advance the durable cursor (#805): persisted
                                // AFTER the batch's actions were dispatched,
                                // under the stable listener identity — never the
                                // entity type of whatever row came last. This is
                                // at-least-once delivery with a replay window of
                                // one batch; payloads carry the change-log row
                                // UUID as the dedup key.
                                if let Some(last_entry) = entries.last() {
                                    last_checkpoint.store(last_entry.id, Ordering::Relaxed);
                                    checkpoint_event_count += entries.len();

                                    let state = ObserverCheckpointState {
                                        listener_id: listener_id.clone(),
                                        last_processed_id: last_entry.id,
                                        last_processed_at: chrono::Utc::now(),
                                        batch_size: entries.len(),
                                        event_count: checkpoint_event_count,
                                    };
                                    match checkpoint_store.save(&listener_id, &state).await {
                                        Ok(()) => {
                                            debug!(
                                                "Checkpoint saved: listener_id={}, last_id={}",
                                                listener_id, last_entry.id
                                            );
                                        }
                                        Err(e) => {
                                            error!("Failed to save checkpoint: {}", e);
                                        }
                                    }
                                }
                            }
                            Err(e) => {
                                errors.fetch_add(1, Ordering::Relaxed);
                                error!("Failed to fetch entries from change log: {}", e);
                                // Back off on error
                                tokio::time::sleep(Duration::from_secs(1)).await;
                            }
                        }
                    }
                }

                if !running.load(Ordering::SeqCst) {
                    break;
                }
            }

            info!("Observer runtime stopped");
        });

        debug!("tokio::spawn() returned, storing task handle");
        self.task_handle = Some(handle);

        // Wait for the background task to signal readiness before returning.
        // This ensures callers can safely insert events immediately after start().
        ready_rx.await.map_err(|_| {
            ServerError::ConfigError(
                "observer background task exited before signalling readiness".to_string(),
            )
        })?;

        info!("Runtime started successfully");
        Ok(())
    }

    /// Start the runtime on a generic [`EventTransport`] stream (NATS, in-memory).
    ///
    /// Mirrors the PostgreSQL [`start`](Self::start) setup — load observers, build
    /// the matcher + executor, publish the entity-type index, share state for hot
    /// reload — but sources events from [`EventTransport::subscribe`] instead of
    /// the `ChangeLogListener`. Per-event dispatch, logging, and subscription
    /// forwarding go through the shared [`process_entity_event`] seam so behaviour
    /// matches the PG path. Transport-stream consumers do not persist change-log
    /// checkpoints — delivery state is owned by the transport (e.g. a NATS durable
    /// consumer).
    ///
    /// # Errors
    ///
    /// Returns `ServerError` if observers cannot be loaded, the matcher cannot be
    /// built, or the transport cannot connect/subscribe. An unreachable NATS
    /// broker fails here, before the runtime reports itself started — the #350
    /// dead-broker contract (no silent fallback to PostgreSQL).
    async fn start_transport_stream(&mut self) -> Result<(), ServerError> {
        let kind = self.config.transport.transport;
        info!("Starting observer runtime ({kind:?} transport)...");

        // Build the transport. Connecting happens here, so an unreachable broker
        // fails loudly before the runtime reports itself started (#350).
        let transport: Arc<dyn EventTransport> = match kind {
            TransportKind::InMemory => Arc::new(InMemoryTransport::new()),
            #[cfg(feature = "observers-nats")]
            TransportKind::Nats => {
                let nats_config = nats_config_from(&self.config.transport.nats);
                let connected = fraiseql_observers::transport::NatsTransport::new(nats_config)
                    .await
                    .map_err(|e| {
                        ServerError::ConfigError(format!(
                            "observer NATS transport failed to connect to {}: {e}",
                            self.config.transport.nats.url
                        ))
                    })?;
                Arc::new(connected)
            },
            #[cfg(not(feature = "observers-nats"))]
            TransportKind::Nats => {
                return Err(ServerError::ConfigError(
                    "observer transport = \"nats\" but this binary lacks the observers-nats \
                     feature"
                        .to_string(),
                ));
            },
            other => {
                // Postgres is handled by `start()` and never reaches here; an
                // unknown future kind was already refused/downgraded at init.
                return Err(ServerError::ConfigError(format!(
                    "unsupported observer transport for the stream path: {other:?}"
                )));
            },
        };

        // Load initial observers + build matcher/executor (same as the PG path).
        let (observers, entity_type_index) = self.load_observers().await?;
        self.observer_count.store(observers.len(), Ordering::SeqCst);
        let matcher = EventMatcher::build(observers).map_err(|e| {
            ServerError::ConfigError(format!("Failed to build event matcher: {}", e))
        })?;
        let executor = Arc::new(
            ObserverExecutor::new_with_email(
                matcher.clone(),
                self.dlq.clone(),
                self.config.email.as_ref(),
            )
            .map_err(|e| ServerError::ConfigError(format!("invalid observer email config: {e}")))?
            // #632: wire the pool so `database` actions call their PostgreSQL
            // function for real instead of failing "no pool wired".
            .with_database_pool(self.config.pool.clone()),
        );
        {
            let mut m = self.matcher.write().await;
            *m = Some(matcher.clone());
        }
        {
            let mut ex = self.executor.write().await;
            *ex = Some(executor.clone());
        }
        self.entity_type_index.store(Arc::new(entity_type_index));

        // Subscribe to the event stream (a second connection-time failure point).
        let mut stream = transport.subscribe(EventFilter::all_tenants()).await.map_err(|e| {
            ServerError::ConfigError(format!("failed to subscribe to observer transport: {e}"))
        })?;

        // Shutdown + readiness channels (same protocol as the PG path).
        let (shutdown_tx, mut shutdown_rx) = mpsc::channel::<()>(1);
        self.shutdown_tx = Some(shutdown_tx);
        let (ready_tx, ready_rx) = oneshot::channel::<()>();

        // Clone task-local state for the background loop.
        let running = self.running.clone();
        let events_processed = self.events_processed.clone();
        let errors = self.errors.clone();
        let pool = self.config.pool.clone();
        let log_payloads = self.config.log_payloads;
        let matcher_ref = Arc::clone(&self.matcher);
        let executor_ref = Arc::clone(&self.executor);
        let entity_type_index_ref = Arc::clone(&self.entity_type_index);
        let bridge_sender = self.event_bridge_sender.clone();

        let mut current_matcher = matcher;
        let mut current_executor = executor;

        running.store(true, Ordering::SeqCst);

        let handle = tokio::spawn(async move {
            info!("Observer runtime stream loop started, beginning event processing");
            // Signal readiness so callers can publish immediately after start().
            let _ = ready_tx.send(());

            loop {
                tokio::select! {
                    _ = shutdown_rx.recv() => {
                        info!("Observer runtime received shutdown signal");
                        break;
                    }
                    maybe_event = stream.next() => {
                        // Refresh matcher/executor in case a hot-reload occurred.
                        {
                            let m = matcher_ref.read().await;
                            if let Some(updated) = m.clone() {
                                current_matcher = updated;
                            }
                        }
                        {
                            let ex = executor_ref.read().await;
                            if let Some(updated) = ex.clone() {
                                current_executor = updated;
                            }
                        }

                        match maybe_event {
                            Some(Ok(event)) => {
                                process_entity_event(
                                    &event,
                                    &current_matcher,
                                    &current_executor,
                                    &entity_type_index_ref,
                                    &pool,
                                    bridge_sender.as_ref(),
                                    &events_processed,
                                    &errors,
                                    log_payloads,
                                )
                                .await;
                            },
                            Some(Err(e)) => {
                                errors.fetch_add(1, Ordering::Relaxed);
                                error!("Observer transport stream error: {}", e);
                            },
                            None => {
                                info!("Observer transport stream ended; stopping loop");
                                break;
                            },
                        }
                    }
                }

                if !running.load(Ordering::SeqCst) {
                    break;
                }
            }

            info!("Observer runtime stopped");
        });

        self.task_handle = Some(handle);

        ready_rx.await.map_err(|_| {
            ServerError::ConfigError(
                "observer background task exited before signalling readiness".to_string(),
            )
        })?;

        info!("Observer runtime started ({kind:?} transport)");
        Ok(())
    }

    /// Stop the observer runtime gracefully
    ///
    /// # Errors
    ///
    /// Returns `ServerError` if the shutdown signal fails to send.
    pub async fn stop(&mut self) -> Result<(), ServerError> {
        if !self.running.load(Ordering::SeqCst) {
            return Ok(());
        }

        info!("Stopping observer runtime...");
        self.running.store(false, Ordering::SeqCst);

        // Send shutdown signal
        if let Some(tx) = self.shutdown_tx.take() {
            let _ = tx.send(()).await;
        }

        // Wait for task to complete
        if let Some(handle) = self.task_handle.take() {
            let _ = tokio::time::timeout(Duration::from_secs(10), handle).await;
        }

        info!("Observer runtime stopped");
        Ok(())
    }

    /// Check if the runtime is running
    #[must_use]
    pub fn is_running(&self) -> bool {
        self.running.load(Ordering::SeqCst)
    }

    /// Whether a [`start`](Self::start) failure on this transport should be fatal
    /// at boot.
    ///
    /// The default PostgreSQL transport keeps the resilient "log and continue
    /// without observers" behaviour on a start failure. A non-Postgres transport
    /// (NATS) is an explicit operator choice for a broker-backed event source; if
    /// it cannot start — e.g. an unreachable broker — the server must not
    /// silently come up *without* it in production (#350). The boot path turns a
    /// start error into a boot failure when this is true and we are in production.
    #[must_use]
    pub(crate) fn transport_requires_broker(&self) -> bool {
        self.config.transport.transport != TransportKind::Postgres
    }

    /// The reader a resumed REST stream reads its catch-up from, when this runtime is
    /// the thing that recorded what was delivered (#1310).
    ///
    /// `Some` only on the PostgreSQL change-log path: that loop is what calls
    /// `record_dispatched`, so its dispatch ledger is the record of what
    /// `/{resource}/stream` delivered and in what order. A broker-backed runtime
    /// (NATS, in-memory) forwards events that never passed through that ledger — the
    /// local change log may not even be their source — so there is nothing here to
    /// resume from and the endpoint says so with `501 RESUMPTION_UNSUPPORTED`.
    ///
    /// Keyed on the same `listener_id` the poller writes under, because a ledger is
    /// per-listener and the wrong one would answer for a delivery order this deployment
    /// never had.
    #[must_use]
    pub fn stream_replay_reader(
        &self,
    ) -> Option<Arc<fraiseql_observers::listener::ChangeLogReplayReader>> {
        match listener_selection(self.config.transport.transport) {
            ListenerSelection::PostgresChangeLog => {
                Some(Arc::new(fraiseql_observers::listener::ChangeLogReplayReader::new(
                    self.config.pool.clone(),
                    self.config.listener_id.clone(),
                )))
            },
            ListenerSelection::TransportStream => None,
        }
    }

    /// Get a reference to the in-memory DLQ for use by HTTP handlers.
    #[must_use]
    pub(crate) const fn dlq(&self) -> &Arc<InMemoryDlq> {
        &self.dlq
    }

    /// Get a reference to the executor `RwLock` for use by DLQ retry handlers.
    #[must_use]
    pub(crate) const fn executor_ref(&self) -> &Arc<RwLock<Option<Arc<ObserverExecutor>>>> {
        &self.executor
    }

    /// Get runtime health status
    #[must_use]
    pub fn health(&self) -> RuntimeHealth {
        RuntimeHealth {
            running:          self.running.load(Ordering::SeqCst),
            observer_count:   self.observer_count.load(Ordering::SeqCst),
            last_checkpoint:  Some(self.last_checkpoint.load(Ordering::SeqCst)),
            events_processed: self.events_processed.load(Ordering::SeqCst),
            errors:           self.errors.load(Ordering::SeqCst),
        }
    }

    /// Reload observers from the database
    ///
    /// # Errors
    ///
    /// Returns `ServerError::Database` if loading observers fails.
    pub async fn reload_observers(&self) -> Result<usize, ServerError> {
        debug!("Reloading observers from database");

        // Load observers from database
        let (observers, new_entity_type_index) = self.load_observers().await?;
        let count = observers.len();

        // Build new matcher
        let new_matcher = EventMatcher::build(observers)
            .map_err(|e| ServerError::ConfigError(format!("Failed to build matcher: {}", e)))?;

        // Build new executor sharing the existing DLQ
        let rebuilt = ObserverExecutor::new_with_email(
            new_matcher.clone(),
            self.dlq.clone(),
            self.config.email.as_ref(),
        )
        .map_err(|e| ServerError::ConfigError(format!("invalid observer email config: {e}")))?
        // #632: keep the database pool wired across hot reloads.
        .with_database_pool(self.config.pool.clone());
        // #985: and keep the cache transport wired too — a reload rebuilds the
        // executor, so without this a reloaded `cache` action silently stops
        // invalidating while every other action keeps working. Mounted before
        // the `Arc`, so there is no shared owner to unwrap back out of.
        #[cfg(feature = "observers-cache")]
        let rebuilt = match self.cache_invalidator.clone() {
            Some(invalidator) => rebuilt.with_cache_invalidation(invalidator),
            None => rebuilt,
        };
        let new_executor = Arc::new(rebuilt);

        // Atomic swap - write locks block readers briefly
        debug!("Swapping matcher, executor, and entity_type_index atomically");

        {
            let mut m = self.matcher.write().await;
            *m = Some(new_matcher);
        }

        {
            let mut ex = self.executor.write().await;
            *ex = Some(new_executor);
        }

        // Republish the new index with a single atomic pointer swap.  See
        // the field doc on `entity_type_index`: readers always observe a
        // fully-populated generation (pre-reload or post-reload), never a
        // partial index.
        self.entity_type_index.store(Arc::new(new_entity_type_index));

        // Update count
        self.observer_count.store(count, Ordering::SeqCst);

        info!("Reloaded {} observers successfully", count);
        Ok(count)
    }
}

/// Process a single ready [`EntityEvent`](ObserverEntityEvent): match observers,
/// execute actions, write the per-observer execution log, and forward the event
/// to the GraphQL subscription bridge.
///
/// This is the shared seam between the PostgreSQL `ChangeLogListener` loop and
/// the generic [`EventTransport`] stream loop (#350): both paths converge here so
/// observer dispatch, logging, and subscription forwarding behave identically
/// regardless of how the event was sourced. The PostgreSQL loop additionally
/// persists a change-log checkpoint per batch (a PG-specific concern that stays
/// in that loop); transport-stream consumers manage their own delivery state.
#[allow(clippy::too_many_arguments)]
// Reason: a shared event-processing seam threading the runtime's task-local state
// (counters, shared index, pool, bridge sender) into both event-source loops.
async fn process_entity_event(
    event: &ObserverEntityEvent,
    matcher: &EventMatcher,
    executor: &Arc<ObserverExecutor>,
    entity_type_index: &ArcSwap<HashMap<(String, String), Vec<i64>>>,
    pool: &PgPool,
    bridge_sender: Option<&mpsc::Sender<BridgeEntityEvent>>,
    events_processed: &std::sync::atomic::AtomicU64,
    errors: &std::sync::atomic::AtomicU64,
    log_payloads: bool,
) {
    // Find matching observers using the current (possibly reloaded) matcher.
    // Used only as a fallback duration source for the audit log (#468) — the
    // executor performs its own matching internally for dispatch.
    let matching_observers = matcher.find_matches(event);

    // Process event using the current (possibly reloaded) executor.
    let process_result = executor.process_event(event).await;

    match process_result {
        Ok(summary) => {
            events_processed.fetch_add(1, Ordering::Relaxed);
            debug!(
                "Event {} processed: {} actions succeeded, {} skipped",
                event.id, summary.successful_actions, summary.conditions_skipped
            );

            // Write execution logs for each matched observer.
            // Look up observer IDs by (entity_type, event_type) from shared reference.
            let event_type_str = event.event_type.as_str().to_uppercase();
            let observer_ids = entity_type_index
                .load()
                .get(&(event.entity_type.clone(), event_type_str.clone()))
                .cloned();
            if let Some(observer_ids) = observer_ids {
                // #932: statuses must come from `tb_observer_log`'s own
                // vocabulary (migration 06's `ck_observer_log_status`). The
                // writer used to emit "error", which the CHECK rejects — so
                // every failure row was refused by the database and dropped with
                // only a warn, exactly when the audit record matters most.
                let status = if summary.successful_actions > 0 {
                    OBSERVER_LOG_STATUS_SUCCESS
                } else {
                    OBSERVER_LOG_STATUS_FAILED
                };
                // Pick the per-action detail that best represents this row's
                // outcome (#468): for a success row the first action that
                // succeeded, for an error row the first that failed. In the
                // common single-action observer case this is exact; the
                // aggregate-per-observer model (one row per matched observer) is
                // unchanged. `None` only when the executor produced no detail.
                let representative = if status == "success" {
                    summary.action_details.iter().find(|d| d.success)
                } else {
                    summary.action_details.iter().find(|d| !d.success)
                }
                .or_else(|| summary.action_details.first());

                // Fallback duration (the pre-#468 average over matched observers)
                // used only when no per-action detail is available.
                let fallback_duration_ms = if matching_observers.is_empty() {
                    None
                } else {
                    #[allow(clippy::cast_precision_loss, clippy::cast_possible_truncation)]
                    // Reason: observer count is small; average duration fits in i32.
                    Some((summary.total_duration_ms / matching_observers.len() as f64) as i32)
                };

                // Write a log entry for each matched observer. A failed audit-log
                // write must not be silently swallowed (M-observer-log-swallow): the
                // event itself processed, so this is non-fatal, but it is surfaced via
                // `warn!` (mirroring the EventBridge forward-failure handling below)
                // so the dropped audit row is observable.
                for observer_id in observer_ids {
                    write_observer_log(
                        pool,
                        observer_id,
                        event,
                        status,
                        representative,
                        fallback_duration_ms,
                        None,
                        log_payloads,
                    )
                    .await;
                }
            }

            // Forward processed event to EventBridge for GraphQL subscription delivery.
            // Only real entity changes are subscriber-visible (#773): `Custom` — the
            // Debezium `'r'` snapshot/read/no-op row — maps to `None` and is never
            // forwarded, so it cannot be fabricated into a phantom `Create`.
            if let (Some(sender), Some(bridge_event)) = (bridge_sender, bridge_event_for(event)) {
                forward_to_bridge(sender, bridge_event, &event.id.to_string()).await;
            }
        },
        Err(e) => {
            errors.fetch_add(1, Ordering::Relaxed);
            error!("Failed to process event {}: {}", event.id, e);

            // Write error logs for matched observers. This is an event-level
            // failure (e.g. condition-parser error), so there is no per-action
            // detail to attribute — only the error message is recorded.
            let event_type_str = event.event_type.as_str().to_uppercase();
            let observer_ids_err = entity_type_index
                .load()
                .get(&(event.entity_type.clone(), event_type_str))
                .cloned();
            if let Some(observer_ids) = observer_ids_err {
                let error_message = e.to_string();
                for observer_id in observer_ids {
                    write_observer_log(
                        pool,
                        observer_id,
                        event,
                        OBSERVER_LOG_STATUS_FAILED,
                        None,
                        None,
                        Some(&error_message),
                        log_payloads,
                    )
                    .await;
                }
            }
        },
    }
}

/// The subscriber-visible operation for an observer event kind (#773).
///
/// An **exhaustive** match over the closed [`EventKind`]: `Custom` (the Debezium
/// `'r'` snapshot/read/no-op row) is `None` — filtered before the subscription
/// bridge, never defaulted to `Create`. Adding an `EventKind` variant is a
/// compile error here instead of a silent fall-through.
pub(crate) const fn subscription_operation_for(
    kind: fraiseql_observers::EventKind,
) -> Option<SubscriptionOperation> {
    use fraiseql_observers::EventKind;
    match kind {
        EventKind::Created => Some(SubscriptionOperation::Create),
        EventKind::Updated => Some(SubscriptionOperation::Update),
        EventKind::Deleted => Some(SubscriptionOperation::Delete),
        EventKind::Custom => None,
    }
}

/// The bridge event one observer event becomes, or `None` when it is not
/// subscriber-visible.
///
/// The **one** projection from an observer event to the shape a subscriber receives.
/// It was inline in `process_entity_event` until a REST stream could be resumed
/// (#1310): a replayed event is read back from the change log and has to become the
/// same value the live path produces, and the only way to keep that true is for one
/// function to produce both. Two copies of this conversion would drift a field at a
/// time, and the drift would be invisible — a client would simply stop seeing the
/// tenant, or the envelope, on exactly the events it had to reconnect for (#1271, one
/// layer up).
pub(crate) fn bridge_event_for(event: &ObserverEntityEvent) -> Option<BridgeEntityEvent> {
    let operation = subscription_operation_for(event.event_type)?;

    let mut bridge_event = BridgeEntityEvent::new(
        &event.entity_type,
        event.entity_id.to_string(),
        operation,
        event.data.clone(),
    );

    // Propagate tenant_id for multi-tenant filtering.
    if let Some(ref tid) = event.tenant_id {
        bridge_event = bridge_event.with_tenant_id(tid);
    }

    // Propagate the Change-Spine envelope (actor / provenance) for client delivery
    // (#425). Built from the observer event's stamped fields; omitted entirely when the
    // producer stamped none. `seq` rides here, and is what a stream emits as its SSE
    // `id:` — so this is also where a resumable id comes from.
    let envelope = ChangeSpineEnvelope {
        actor_type:     event.actor_type.clone(),
        acting_for:     event.acting_for.clone(),
        schema_version: event.schema_version.clone(),
        tenant_id:      event.tenant_id.clone(),
        duration_ms:    event.duration_ms,
        seq:            event.seq,
    };
    if !envelope.is_empty() {
        bridge_event = bridge_event.with_change_spine(envelope);
    }

    Some(bridge_event)
}

/// Forward one subscriber-visible event into the `EventBridge` channel.
///
/// **Bounded, awaited send (#772):** when the bridge channel is full this waits until
/// the bridge loop drains a slot — upstream backpressure — instead of dropping the
/// event for every subscriber with only a `warn!` (the old `try_send`). Stalling here
/// stalls the change-log loop, which is the correct failure mode: the change log is
/// durable and its checkpoint only advances after the batch completes, so a slow
/// bridge delays delivery rather than losing events. `Err` means the bridge task is
/// gone (receiver dropped) — nothing can be delivered to subscriptions at all; logged
/// loudly as an error, not a debug-level shrug.
async fn forward_to_bridge(
    sender: &mpsc::Sender<BridgeEntityEvent>,
    bridge_event: BridgeEntityEvent,
    event_id: &str,
) {
    if let Err(e) = sender.send(bridge_event).await {
        error!("EventBridge task is gone; event {event_id} not forwarded to subscriptions: {e}");
    }
}

/// Maximum serialized size (bytes) of an event payload persisted into a JSONB
/// audit column. A larger payload is replaced with a small marker so a single
/// oversized event cannot bloat `tb_observer_log` (#468).
const MAX_LOG_PAYLOAD_BYTES: usize = 64 * 1024;

/// Clamp an event payload for persistence into `tb_observer_log.request_payload`.
///
/// Returns the value unchanged when its serialized form is within
/// [`MAX_LOG_PAYLOAD_BYTES`]; otherwise returns a marker object recording the
/// original size so the stored audit row stays bounded.
fn truncate_log_payload(data: &serde_json::Value) -> serde_json::Value {
    let size = serde_json::to_vec(data).map_or(0, |v| v.len());
    if size > MAX_LOG_PAYLOAD_BYTES {
        serde_json::json!({
            "_truncated": true,
            "_original_size_bytes": size,
        })
    } else {
        data.clone()
    }
}

/// Whether any loaded observer declares a `cache`/`invalidate` action.
///
/// Drives the #985 mount decision: the transport is only connected when
/// something actually uses it, so a deployment with no cache observers never
/// requires Redis.
#[cfg_attr(not(feature = "observers-cache"), allow(dead_code))]
// Reason: without `observers-cache` the transport is not compiled in; the check
// still runs, to refuse the declaration at boot.
fn declares_cache_action(observers: &HashMap<String, ObserverDefinition>) -> bool {
    observers
        .values()
        .flat_map(|o| o.actions.iter())
        .any(|a| matches!(a, ObserverActionConfig::Cache { .. }))
}

/// Connect the Redis cache-invalidation transport when the loaded observers need
/// it, and refuse to boot when they need it and it is not configured (#985).
///
/// Before this, `RedisCacheInvalidator` was reachable only by embedding the
/// library: the server builds its executor with `ObserverExecutor::new_with_email`,
/// and the only mount was an alternative constructor that discards the email
/// config. A `type = "cache"` action therefore loaded fine and then failed at
/// every single dispatch with "no backend wired" — honest, but the transport was
/// unreachable from any `fraiseql.toml`.
///
/// The no-backend case is a boot error rather than a per-event failure because
/// it is a configuration mistake, and an operator should find it out once, at
/// startup, not once per event forever.
///
/// # Errors
///
/// Returns `ServerError::ConfigError` when a `cache` action is declared with no
/// `[observers.runtime.redis]` backend, or when connecting to Redis fails.
#[cfg(feature = "observers-cache")]
async fn connect_cache_invalidator(
    observers: &HashMap<String, ObserverDefinition>,
    redis: Option<&fraiseql_observers::config::RedisConfig>,
) -> crate::Result<Option<Arc<fraiseql_observers::RedisCacheInvalidator>>> {
    if !declares_cache_action(observers) {
        return Ok(None);
    }

    let Some(redis_config) = redis else {
        return Err(ServerError::ConfigError(
            "an observer declares a `cache` action but no Redis backend is configured; \
             add [observers.runtime.redis] (url = \"redis://…\") or remove the action"
                .to_string(),
        ));
    };

    let invalidator = fraiseql_observers::RedisCacheInvalidator::connect(redis_config)
        .await
        .map_err(|e| {
            ServerError::ConfigError(format!(
                "an observer declares a `cache` action but the configured Redis backend \
                 ({}) could not be reached: {e}",
                redis_config.url
            ))
        })?;

    info!(
        redis_url = %redis_config.url,
        "Mounted the Redis cache-invalidation transport for `cache` observer actions"
    );
    Ok(Some(Arc::new(invalidator)))
}

/// Refuse to boot when a `cache` action is declared by a binary built without the
/// `observers-cache` feature (#985).
///
/// The transport is genuinely absent from this build, so there is nothing to
/// mount and every dispatch would fail identically, forever. Saying so once, at
/// boot, names the actual fix — rebuild with the feature — instead of leaving an
/// operator to infer it from a repeating per-event error.
///
/// # Errors
///
/// Returns `ServerError::ConfigError` when any enabled observer declares a
/// `cache` action.
#[cfg(not(feature = "observers-cache"))]
fn reject_cache_actions_not_compiled_in(
    observers: &HashMap<String, ObserverDefinition>,
) -> crate::Result<()> {
    if declares_cache_action(observers) {
        return Err(ServerError::ConfigError(
            "an observer declares a `cache` action but this binary was built without the              `observers-cache` feature, so the Redis cache-invalidation transport is not              compiled in; rebuild with --features observers-cache or remove the action"
                .to_string(),
        ));
    }
    Ok(())
}

/// The `tb_observer_log.status` this runtime writes for a dispatched observer
/// whose actions succeeded.
///
/// Both status constants are asserted to be members of
/// [`fraiseql_observers::migrations::OBSERVER_LOG_STATUSES`] — the vocabulary
/// migration 06's `ck_observer_log_status` CHECK accepts — by
/// `observer_log_statuses_are_accepted_by_the_shipped_check`. A status outside
/// it is refused by the database and the audit row vanishes with only a `warn!`
/// (#932).
const OBSERVER_LOG_STATUS_SUCCESS: &str = "success";

/// The `tb_observer_log.status` this runtime writes when an observer's actions
/// failed, or when the event could not be processed at all.
///
/// Was `"error"` — a value the CHECK constraint has never accepted, so every
/// failure row was silently rejected (#932).
const OBSERVER_LOG_STATUS_FAILED: &str = "failed";

/// Write a single `tb_observer_log` row, populating the per-action audit columns
/// from `detail` when available (#468).
///
/// `action_type`, `action_index`, `response_status_code`, and the
/// `response_payload` outcome summary are non-sensitive and always written when
/// a per-action detail is present. `request_payload` carries the triggering
/// event data — potentially PII — so it is only persisted when `log_payloads`
/// is set, and is truncated to a marker past [`MAX_LOG_PAYLOAD_BYTES`].
///
/// A failed write is surfaced via `warn!` rather than swallowed
/// (M-observer-log-swallow): the event itself already processed, so a dropped
/// audit row is non-fatal but must remain observable.
#[allow(clippy::too_many_arguments)]
// Reason: a single INSERT site threading the event, the per-action detail, and
// the privacy toggle so both success and error paths share one column contract.
async fn write_observer_log(
    pool: &PgPool,
    observer_id: i64,
    event: &ObserverEntityEvent,
    status: &str,
    detail: Option<&ActionExecutionDetail>,
    fallback_duration_ms: Option<i32>,
    fallback_error: Option<&str>,
    log_payloads: bool,
) {
    let action_index = detail.map(|d| i32::try_from(d.action_index).unwrap_or(i32::MAX));
    let action_type = detail.map(|d| d.action_type.clone());
    let status_code = detail.and_then(|d| d.status_code).map(i32::from);
    let duration_ms = detail
        .map(|d| {
            #[allow(clippy::cast_possible_truncation)]
            // Reason: action durations are small; milliseconds fit in i32.
            {
                d.duration_ms as i32
            }
        })
        .or(fallback_duration_ms);
    // Non-sensitive outcome summary (message + success flag); always recorded.
    let response_payload =
        detail.map(|d| serde_json::json!({ "message": d.message, "success": d.success }));
    // Request payload may carry PII → opt-in only, and truncated when large.
    let request_payload = log_payloads.then(|| truncate_log_payload(&event.data));
    // Prefer an explicit event-level error, else the action's own error.
    let error_message = fallback_error
        .map(ToString::to_string)
        .or_else(|| detail.and_then(|d| d.error_message.clone()));

    if let Err(e) = sqlx::query(
        // entity_id is bound as text (Uuid::to_string) and cast to the column's
        // uuid type — sqlx will not implicitly coerce text to uuid on INSERT.
        "INSERT INTO tb_observer_log
         (fk_observer, event_id, entity_type, entity_id, event_type, status,
          action_index, action_type, response_status_code, response_payload,
          request_payload, duration_ms, error_message, attempt_number, max_attempts)
         VALUES ($1, $2, $3, $4::uuid, $5, $6, $7, $8, $9, $10, $11, $12, $13, 1, 3)",
    )
    .bind(observer_id)
    .bind(event.id)
    .bind(&event.entity_type)
    .bind(event.entity_id.to_string())
    .bind(event.event_type.as_str())
    .bind(status)
    .bind(action_index)
    .bind(action_type)
    .bind(status_code)
    .bind(response_payload)
    .bind(request_payload)
    .bind(duration_ms)
    .bind(error_message)
    .execute(pool)
    .await
    {
        warn!(
            "Failed to write observer {status} log (observer {observer_id}, event {}): {e}",
            event.id
        );
    }
}

/// Simple in-memory Dead Letter Queue for development.
///
/// Honors an optional size cap (`max_size`) mirroring the `fraiseql-observers`
/// library policy so `max_dlq_size` means the same thing in the binary and the
/// embedder: when the cap is reached, [`push`](InMemoryDlq::push) **drops the
/// newest** failed entry, emits a `warn!` with the same fields as the library,
/// and bumps an overflow counter — the durable, loud signal (a lone warn line
/// scrolls past). The length check and the insert happen under the same items
/// mutex, so there is no separate-atomic TOCTOU.
pub(crate) struct InMemoryDlq {
    items:          std::sync::Mutex<Vec<fraiseql_observers::DlqItem>>,
    /// Function-trigger dispatch failures (after:mutation and future
    /// after:ingest), kept in the same store as observer-action failures but in a
    /// separate typed collection because they carry a
    /// [`FunctionDispatchRecord`](fraiseql_observers::FunctionDispatchRecord)
    /// rather than an `EntityEvent` + `ActionConfig`. Same size-cap policy.
    function_items: std::sync::Mutex<Vec<fraiseql_observers::FunctionDispatchRecord>>,
    /// Maximum retained entries; `None` = unbounded (back-compat default).
    max_size:       Option<usize>,
    /// Count of entries dropped because the DLQ was at capacity (drop-newest).
    /// Shared across observer-action and function-dispatch entries.
    overflow_count: std::sync::atomic::AtomicUsize,
}

impl InMemoryDlq {
    /// Create a DLQ with an optional retention cap (`None` = unbounded).
    pub(crate) const fn new_with_max(max_size: Option<usize>) -> Self {
        Self {
            items: std::sync::Mutex::new(Vec::new()),
            function_items: std::sync::Mutex::new(Vec::new()),
            max_size,
            overflow_count: std::sync::atomic::AtomicUsize::new(0),
        }
    }

    /// Returns the number of function-dispatch entries currently in the DLQ.
    pub(crate) fn function_count(&self) -> usize {
        self.function_items.lock().expect("function_items mutex poisoned").len()
    }

    /// Returns the number of items currently in the DLQ.
    pub(crate) fn count(&self) -> usize {
        self.items.lock().expect("items mutex poisoned").len()
    }

    /// Number of entries dropped because the DLQ was at capacity (drop-newest).
    ///
    /// The durable counterpart to the per-drop `warn!` — the loud overflow
    /// signal that survives log rotation.
    pub(crate) fn overflow_count(&self) -> usize {
        self.overflow_count.load(Ordering::Relaxed)
    }

    /// Returns all DLQ items as a cloned snapshot.
    pub(crate) fn list_all(&self) -> Vec<fraiseql_observers::DlqItem> {
        self.items.lock().expect("items mutex poisoned").clone()
    }

    /// Returns a single DLQ item by ID, if it exists.
    pub(crate) fn get(&self, id: uuid::Uuid) -> Option<fraiseql_observers::DlqItem> {
        self.items
            .lock()
            .expect("items mutex poisoned")
            .iter()
            .find(|item| item.id == id)
            .cloned()
    }

    /// Atomically remove and return a DLQ item by ID.
    ///
    /// The find and the remove happen under a single lock, so two concurrent
    /// retries on the same id cannot both observe it: exactly one caller gets
    /// `Some`, the rest get `None`. This makes DLQ retry **at-most-once per
    /// claim** (#344), closing the previous `get` → `process` → `remove` race
    /// that re-dispatched the action twice (turning at-least-once delivery into
    /// at-least-twice).
    pub(crate) fn try_claim(&self, id: uuid::Uuid) -> Option<fraiseql_observers::DlqItem> {
        let mut items = self.items.lock().expect("items mutex poisoned");
        let pos = items.iter().position(|item| item.id == id)?;
        Some(items.remove(pos))
    }

    /// Re-insert a previously-claimed item whose retry failed, **bypassing the
    /// size cap**.
    ///
    /// A claimed item was already counted against the cap before [`try_claim`]
    /// removed it; routing it back through [`push`](InMemoryDlq::push) would let
    /// drop-newest silently discard it if the DLQ refilled to `max_size` during
    /// the claim — re-introducing the exact silent loss the cap exists to
    /// prevent (#343/#344). The retry handlers use this to restore an item after
    /// a failed redispatch.
    pub(crate) fn reinsert(&self, item: fraiseql_observers::DlqItem) {
        self.items.lock().expect("items mutex poisoned").push(item);
    }
}

#[async_trait::async_trait]
impl fraiseql_observers::DeadLetterQueue for InMemoryDlq {
    async fn push(
        &self,
        event: fraiseql_observers::EntityEvent,
        action: fraiseql_observers::ActionConfig,
        error: String,
    ) -> fraiseql_observers::Result<uuid::Uuid> {
        let id = uuid::Uuid::new_v4();
        let mut items = self.items.lock().expect("items mutex poisoned");

        // Drop-newest when at capacity, mirroring the library policy
        // (`fraiseql-observers` executor): same `warn!` fields + an overflow
        // counter so the cap means the same thing in the binary and embedder.
        if let Some(max) = self.max_size {
            if items.len() >= max {
                warn!(
                    max_dlq_size = max,
                    action_type = action.action_type(),
                    event_id = %event.id,
                    "DLQ full; dropping failed action entry"
                );
                self.overflow_count.fetch_add(1, Ordering::Relaxed);
                // Drop the newest entry: it is intentionally not stored.
                return Ok(id);
            }
        }

        items.push(fraiseql_observers::DlqItem {
            id,
            event,
            action,
            error_message: error,
            attempts: 0,
        });
        Ok(id)
    }

    async fn get_pending(
        &self,
        limit: i64,
    ) -> fraiseql_observers::Result<Vec<fraiseql_observers::DlqItem>> {
        let items = self.items.lock().expect("items mutex poisoned");
        #[allow(clippy::cast_sign_loss, clippy::cast_possible_truncation)]
        // Reason: limit is a user-supplied i64 clamped to a small positive range; negative values
        // wrap to 0 safely
        let limit_usize = limit as usize;
        Ok(items.iter().take(limit_usize).cloned().collect())
    }

    async fn mark_success(&self, id: uuid::Uuid) -> fraiseql_observers::Result<()> {
        let mut items = self.items.lock().expect("items mutex poisoned");
        items.retain(|i| i.id != id);
        Ok(())
    }

    async fn mark_retry_failed(
        &self,
        id: uuid::Uuid,
        error: &str,
    ) -> fraiseql_observers::Result<()> {
        // Keep the item for a future retry / operator inspection: record the
        // failure on the existing fields instead of silently destroying it.
        // Items leave the DLQ only via `mark_success` or an explicit `remove`
        // (operator DELETE). This preserves the audit trail (#343) and is the
        // lifecycle the atomic claim model (#344) relies on.
        let mut items = self.items.lock().expect("items mutex poisoned");
        if let Some(item) = items.iter_mut().find(|i| i.id == id) {
            item.attempts = item.attempts.saturating_add(1);
            item.error_message = error.to_string();
        }
        Ok(())
    }

    async fn push_function(
        &self,
        record: fraiseql_observers::FunctionDispatchRecord,
    ) -> fraiseql_observers::Result<uuid::Uuid> {
        let id = record.id;
        let mut items = self.function_items.lock().expect("function_items mutex poisoned");

        // Drop-newest when at capacity, mirroring `push` and the library policy
        // so `max_dlq_size` means the same thing for function dispatches.
        if let Some(max) = self.max_size {
            if items.len() >= max {
                warn!(
                    max_dlq_size = max,
                    function = %record.function_name,
                    trigger = %record.trigger_type,
                    "DLQ full; dropping failed function dispatch entry"
                );
                self.overflow_count.fetch_add(1, Ordering::Relaxed);
                // The internal overflow counter now also reaches `/metrics` — an
                // eviction must never be Prometheus-invisible (#598).
                crate::function_metrics::record_dlq_eviction();
                crate::function_metrics::set_dlq_size(items.len());
                // Drop the newest entry: it is intentionally not stored.
                return Ok(id);
            }
        }

        items.push(record);
        crate::function_metrics::set_dlq_size(items.len());
        Ok(id)
    }

    async fn get_pending_functions(
        &self,
        limit: i64,
    ) -> fraiseql_observers::Result<Vec<fraiseql_observers::FunctionDispatchRecord>> {
        let items = self.function_items.lock().expect("function_items mutex poisoned");
        #[allow(clippy::cast_sign_loss, clippy::cast_possible_truncation)]
        // Reason: limit is a user-supplied i64 clamped to a small positive range; negative values
        // wrap to 0 safely
        let limit_usize = limit as usize;
        Ok(items.iter().take(limit_usize).cloned().collect())
    }
}