use crate::catalog::SYSTEM_PROJECT_ID;
use crate::catalog::types::{Row, Value};
use crate::commit::executor::CommitResult;
use crate::commit::tx::{
    ReadKey, ReadSet, ReadSetEntry, TransactionEnvelope, WriteClass, WriteIntent,
};
use crate::commit::validation::Mutation;
use crate::error::{AedbError, ResourceType as ErrorResourceType};
use crate::permission::{CallerContext, Permission};
use crate::query::plan::ConsistencyMode;
use crate::query_authorization::ensure_external_caller_allowed;
use crate::reactive_support::{
    ReactiveCheckpointAckCacheKey, ReactiveProcessorHandler, ReactiveProcessorHealth,
    ReactiveProcessorInfo, ReactiveProcessorOptions, ReactiveProcessorRegistration,
    ReactiveProcessorRetryState, ReactiveProcessorRuntime, ReactiveProcessorRuntimeStatus,
    ReactiveProcessorSloStatus, prune_reactive_ack_cache, reactive_processor_checkpoint_mutation,
    reactive_processor_dlq_mutation, reactive_processor_registry_mutation,
};
use crate::storage::encoded_key::EncodedKey;
use crate::{
    AedbInstance, EVENT_OUTBOX_TABLE, EventOutboxRecord, EventStreamPage,
    REACTIVE_PROCESSOR_CHECKPOINTS_TABLE, REACTIVE_PROCESSOR_REGISTRY_TABLE, ReactiveProcessorLag,
    SYSTEM_SCOPE_ID, system_now_micros,
};
use std::collections::HashSet;
use std::future::Future;
use std::ops::Bound;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};
use tokio::sync::Mutex as AsyncMutex;

impl AedbInstance {
    pub async fn emit_event(
        &self,
        project_id: &str,
        scope_id: &str,
        topic: &str,
        event_key: String,
        payload_json: String,
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::EmitEvent {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            topic: topic.to_string(),
            event_key,
            payload_json,
        })
        .await
    }

    pub async fn emit_event_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        topic: &str,
        event_key: String,
        payload_json: String,
    ) -> Result<CommitResult, AedbError> {
        self.commit_as(
            caller,
            Mutation::EmitEvent {
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                topic: topic.to_string(),
                event_key,
                payload_json,
            },
        )
        .await
    }

    pub async fn read_event_stream(
        &self,
        topic_filter: Option<&str>,
        from_commit_seq_exclusive: u64,
        limit: usize,
        consistency: ConsistencyMode,
    ) -> Result<EventStreamPage, AedbError> {
        if self.require_authenticated_calls {
            return Err(AedbError::PermissionDenied(
                "authenticated caller required in secure mode".into(),
            ));
        }
        self.read_event_stream_internal(
            topic_filter,
            from_commit_seq_exclusive,
            limit,
            consistency,
            None,
        )
        .await
    }

    pub async fn read_event_stream_as(
        &self,
        caller: &CallerContext,
        topic_filter: Option<&str>,
        from_commit_seq_exclusive: u64,
        limit: usize,
        consistency: ConsistencyMode,
    ) -> Result<EventStreamPage, AedbError> {
        ensure_external_caller_allowed(caller)?;
        self.read_event_stream_internal(
            topic_filter,
            from_commit_seq_exclusive,
            limit,
            consistency,
            Some(caller),
        )
        .await
    }

    async fn read_event_stream_internal(
        &self,
        topic_filter: Option<&str>,
        from_commit_seq_exclusive: u64,
        limit: usize,
        consistency: ConsistencyMode,
        caller: Option<&CallerContext>,
    ) -> Result<EventStreamPage, AedbError> {
        let lease = self.acquire_snapshot(consistency).await?;
        let snapshot_seq = lease.view.seq;
        if limit == 0 {
            return Ok(EventStreamPage {
                events: Vec::new(),
                next_commit_seq: None,
                snapshot_seq,
            });
        }
        if let Some(caller) = caller {
            let required = Permission::TableRead {
                project_id: SYSTEM_PROJECT_ID.to_string(),
                scope_id: SYSTEM_SCOPE_ID.to_string(),
                table_name: EVENT_OUTBOX_TABLE.to_string(),
            };
            if !lease
                .view
                .catalog
                .has_permission(&caller.caller_id, &required)
            {
                return Err(AedbError::PermissionDenied("permission denied".into()));
            }
        }
        let Some(table) =
            lease
                .view
                .keyspace
                .table(SYSTEM_PROJECT_ID, SYSTEM_SCOPE_ID, EVENT_OUTBOX_TABLE)
        else {
            return Ok(EventStreamPage {
                events: Vec::new(),
                next_commit_seq: None,
                snapshot_seq,
            });
        };
        let mut events = Vec::new();
        let start_seq = from_commit_seq_exclusive.saturating_add(1);
        let Ok(start_seq_i64) = i64::try_from(start_seq) else {
            return Ok(EventStreamPage {
                events: Vec::new(),
                next_commit_seq: None,
                snapshot_seq,
            });
        };
        let start_key = EncodedKey::from_values(&[
            Value::Integer(start_seq_i64),
            Value::Text("".into()),
            Value::Text("".into()),
        ]);
        for (_, stored) in table
            .rows
            .range((Bound::Included(start_key), Bound::Unbounded))
        {
            if events.len() >= limit {
                break;
            }
            let row = lease.view.keyspace.materialize_row(stored)?;
            let (
                Some(Value::Integer(commit_seq_i64)),
                Some(Value::Timestamp(ts_i64)),
                Some(Value::Text(project_id)),
                Some(Value::Text(scope_id)),
                Some(Value::Text(topic)),
                Some(Value::Text(event_key)),
                Some(Value::Json(payload)),
            ) = (
                row.values.first(),
                row.values.get(1),
                row.values.get(2),
                row.values.get(3),
                row.values.get(4),
                row.values.get(5),
                row.values.get(6),
            )
            else {
                continue;
            };
            let Ok(commit_seq) = u64::try_from(*commit_seq_i64) else {
                continue;
            };
            if let Some(filter_topic) = topic_filter
                && topic.as_str() != filter_topic
            {
                continue;
            }
            let Ok(ts_micros) = u64::try_from(*ts_i64) else {
                continue;
            };
            events.push(EventOutboxRecord {
                commit_seq,
                ts_micros,
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                topic: topic.to_string(),
                event_key: event_key.to_string(),
                payload_json: payload.to_string(),
            });
        }
        let next_commit_seq = events.last().map(|e| e.commit_seq);
        Ok(EventStreamPage {
            events,
            next_commit_seq,
            snapshot_seq,
        })
    }

    pub async fn ack_reactive_processor_checkpoint(
        &self,
        processor_name: &str,
        checkpoint_seq: u64,
    ) -> Result<CommitResult, AedbError> {
        if self.require_authenticated_calls {
            return Err(AedbError::PermissionDenied(
                "authenticated caller required in secure mode".into(),
            ));
        }
        if processor_name.trim().is_empty() {
            return Err(AedbError::Validation(
                "processor_name cannot be empty".into(),
            ));
        }
        let envelope = self
            .build_reactive_processor_checkpoint_envelope(processor_name, checkpoint_seq)
            .await?;
        self.commit_envelope_prevalidated_internal("ack_reactive_processor_checkpoint", envelope)
            .await
    }

    pub async fn ack_reactive_processor_checkpoint_as(
        &self,
        caller: CallerContext,
        processor_name: &str,
        checkpoint_seq: u64,
    ) -> Result<CommitResult, AedbError> {
        ensure_external_caller_allowed(&caller)?;
        if processor_name.trim().is_empty() {
            return Err(AedbError::Validation(
                "processor_name cannot be empty".into(),
            ));
        }
        self.ensure_reactive_processor_checkpoint_write_allowed(&caller)
            .await?;
        let mut envelope = self
            .build_reactive_processor_checkpoint_envelope(processor_name, checkpoint_seq)
            .await?;
        envelope.caller = Some(caller);
        self.commit_envelope_prevalidated_system_internal(
            "ack_reactive_processor_checkpoint_as",
            envelope,
        )
        .await
    }

    pub async fn ack_reactive_processor_checkpoint_batched(
        &self,
        processor_name: &str,
        checkpoint_seq: u64,
        watermark_commits: u64,
    ) -> Result<Option<CommitResult>, AedbError> {
        if self.require_authenticated_calls {
            return Err(AedbError::PermissionDenied(
                "authenticated caller required in secure mode".into(),
            ));
        }
        if processor_name.trim().is_empty() {
            return Err(AedbError::Validation(
                "processor_name cannot be empty".into(),
            ));
        }
        if watermark_commits == 0 {
            return Err(AedbError::Validation(
                "watermark_commits must be > 0".into(),
            ));
        }
        let cache_key = ReactiveCheckpointAckCacheKey {
            processor_name: processor_name.to_string(),
            caller_id: None,
        };
        let should_persist = {
            let cache = self.reactive_processor_ack_watermarks.lock();
            let last_persisted = cache
                .get(&cache_key)
                .map(|s| s.last_persisted_seq)
                .unwrap_or(0);
            checkpoint_seq > last_persisted
                && (last_persisted == 0
                    || checkpoint_seq.saturating_sub(last_persisted) >= watermark_commits)
        };
        if !should_persist {
            return Ok(None);
        }
        let envelope = self
            .build_reactive_processor_checkpoint_envelope(processor_name, checkpoint_seq)
            .await?;
        let committed = self
            .commit_envelope_prevalidated_internal(
                "ack_reactive_processor_checkpoint_batched",
                envelope,
            )
            .await?;
        {
            let mut cache = self.reactive_processor_ack_watermarks.lock();
            let state = cache.entry(cache_key).or_default();
            state.last_persisted_seq = state.last_persisted_seq.max(checkpoint_seq);
            state.last_touch_micros = system_now_micros();
            prune_reactive_ack_cache(&mut cache);
        }
        Ok(Some(committed))
    }

    pub async fn ack_reactive_processor_checkpoint_batched_as(
        &self,
        caller: CallerContext,
        processor_name: &str,
        checkpoint_seq: u64,
        watermark_commits: u64,
    ) -> Result<Option<CommitResult>, AedbError> {
        ensure_external_caller_allowed(&caller)?;
        if processor_name.trim().is_empty() {
            return Err(AedbError::Validation(
                "processor_name cannot be empty".into(),
            ));
        }
        if watermark_commits == 0 {
            return Err(AedbError::Validation(
                "watermark_commits must be > 0".into(),
            ));
        }
        let cache_key = ReactiveCheckpointAckCacheKey {
            processor_name: processor_name.to_string(),
            caller_id: Some(caller.caller_id.clone()),
        };
        let should_persist = {
            let cache = self.reactive_processor_ack_watermarks.lock();
            let last_persisted = cache
                .get(&cache_key)
                .map(|s| s.last_persisted_seq)
                .unwrap_or(0);
            checkpoint_seq > last_persisted
                && (last_persisted == 0
                    || checkpoint_seq.saturating_sub(last_persisted) >= watermark_commits)
        };
        if !should_persist {
            return Ok(None);
        }
        self.ensure_reactive_processor_checkpoint_write_allowed(&caller)
            .await?;
        let mut envelope = self
            .build_reactive_processor_checkpoint_envelope(processor_name, checkpoint_seq)
            .await?;
        envelope.caller = Some(caller);
        let committed = self
            .commit_envelope_prevalidated_system_internal(
                "ack_reactive_processor_checkpoint_batched_as",
                envelope,
            )
            .await?;
        {
            let mut cache = self.reactive_processor_ack_watermarks.lock();
            let state = cache.entry(cache_key).or_default();
            state.last_persisted_seq = state.last_persisted_seq.max(checkpoint_seq);
            state.last_touch_micros = system_now_micros();
            prune_reactive_ack_cache(&mut cache);
        }
        Ok(Some(committed))
    }

    async fn ensure_reactive_processor_checkpoint_write_allowed(
        &self,
        caller: &CallerContext,
    ) -> Result<(), AedbError> {
        let lease = self.acquire_snapshot(ConsistencyMode::AtLatest).await?;
        let required = Permission::TableWrite {
            project_id: SYSTEM_PROJECT_ID.to_string(),
            scope_id: SYSTEM_SCOPE_ID.to_string(),
            table_name: REACTIVE_PROCESSOR_CHECKPOINTS_TABLE.to_string(),
        };
        if !lease
            .view
            .catalog
            .has_permission(&caller.caller_id, &required)
        {
            return Err(AedbError::PermissionDenied("permission denied".into()));
        }
        Ok(())
    }

    pub(crate) async fn build_reactive_processor_checkpoint_envelope(
        &self,
        processor_name: &str,
        checkpoint_seq: u64,
    ) -> Result<TransactionEnvelope, AedbError> {
        let lease = self.acquire_snapshot(ConsistencyMode::AtLatest).await?;
        let checkpoint_key = Value::Text(processor_name.to_string().into());
        let checkpoint_pk = EncodedKey::from_values(std::slice::from_ref(&checkpoint_key));
        let checkpoint_table = lease.view.keyspace.table(
            SYSTEM_PROJECT_ID,
            SYSTEM_SCOPE_ID,
            REACTIVE_PROCESSOR_CHECKPOINTS_TABLE,
        );
        let current_checkpoint =
            match checkpoint_table.and_then(|table| table.rows.get(&checkpoint_pk)) {
                Some(stored) => lease
                    .view
                    .keyspace
                    .materialize_row(stored)?
                    .values
                    .get(1)
                    .and_then(|v| match v {
                        Value::Integer(i) => u64::try_from(*i).ok(),
                        _ => None,
                    })
                    .unwrap_or(0),
                None => 0,
            };
        if checkpoint_seq < current_checkpoint {
            return Err(AedbError::Validation(format!(
                "checkpoint_seq {checkpoint_seq} regresses current checkpoint {current_checkpoint}"
            )));
        }
        let checkpoint_version = checkpoint_table
            .and_then(|table| table.version_of(&checkpoint_pk))
            .unwrap_or(0);
        let primary_key = vec![checkpoint_key.clone()];
        let read_set = ReadSet {
            points: vec![ReadSetEntry {
                key: ReadKey::TableRow {
                    project_id: SYSTEM_PROJECT_ID.to_string(),
                    scope_id: SYSTEM_SCOPE_ID.to_string(),
                    table_name: REACTIVE_PROCESSOR_CHECKPOINTS_TABLE.to_string(),
                    primary_key,
                },
                version_at_read: checkpoint_version,
            }],
            ranges: Vec::new(),
        };

        Ok(TransactionEnvelope {
            caller: None,
            idempotency_key: None,
            write_class: WriteClass::Standard,
            assertions: Vec::new(),
            read_set,
            write_intent: WriteIntent {
                mutations: vec![reactive_processor_checkpoint_mutation(
                    processor_name,
                    checkpoint_seq,
                )],
            },
            base_seq: lease.view.seq,
        })
    }

    pub async fn reactive_processor_lag(
        &self,
        processor_name: &str,
        consistency: ConsistencyMode,
    ) -> Result<ReactiveProcessorLag, AedbError> {
        if self.require_authenticated_calls {
            return Err(AedbError::PermissionDenied(
                "authenticated caller required in secure mode".into(),
            ));
        }
        if processor_name.trim().is_empty() {
            return Err(AedbError::Validation(
                "processor_name cannot be empty".into(),
            ));
        }
        self.reactive_processor_lag_internal(processor_name, consistency, None)
            .await
    }

    async fn reactive_processor_lag_internal(
        &self,
        processor_name: &str,
        consistency: ConsistencyMode,
        caller: Option<&CallerContext>,
    ) -> Result<ReactiveProcessorLag, AedbError> {
        let lease = self.acquire_snapshot(consistency).await?;
        if let Some(caller) = caller {
            let required = Permission::TableRead {
                project_id: SYSTEM_PROJECT_ID.to_string(),
                scope_id: SYSTEM_SCOPE_ID.to_string(),
                table_name: REACTIVE_PROCESSOR_CHECKPOINTS_TABLE.to_string(),
            };
            if !lease
                .view
                .catalog
                .has_permission(&caller.caller_id, &required)
            {
                return Err(AedbError::PermissionDenied("permission denied".into()));
            }
        }
        let mut checkpoint_seq = 0u64;
        if let Some(table) = lease.view.keyspace.table(
            SYSTEM_PROJECT_ID,
            SYSTEM_SCOPE_ID,
            REACTIVE_PROCESSOR_CHECKPOINTS_TABLE,
        ) {
            let pk = EncodedKey::from_values(&[Value::Text(processor_name.to_string().into())]);
            if let Some(stored) = table.rows.get(&pk) {
                let row = lease.view.keyspace.materialize_row(stored)?;
                if let Some(Value::Integer(v)) = row.values.get(1) {
                    checkpoint_seq = u64::try_from(*v).unwrap_or(0);
                }
            }
        }
        let head_seq = lease.view.seq;
        Ok(ReactiveProcessorLag {
            processor_name: processor_name.to_string(),
            checkpoint_seq,
            head_seq,
            lag_commits: head_seq.saturating_sub(checkpoint_seq),
        })
    }

    pub async fn reactive_processor_lag_as(
        &self,
        caller: &CallerContext,
        processor_name: &str,
        consistency: ConsistencyMode,
    ) -> Result<ReactiveProcessorLag, AedbError> {
        ensure_external_caller_allowed(caller)?;
        if processor_name.trim().is_empty() {
            return Err(AedbError::Validation(
                "processor_name cannot be empty".into(),
            ));
        }
        self.reactive_processor_lag_internal(processor_name, consistency, Some(caller))
            .await
    }

    pub async fn reactive_processor_runtime_status(
        &self,
        processor_name: &str,
    ) -> Option<ReactiveProcessorRuntimeStatus> {
        let (running, stats) = {
            let runtimes = self.reactive_processor_runtimes.lock().await;
            let runtime = runtimes.get(processor_name)?;
            (true, Arc::clone(&runtime.stats))
        };
        let mut status = stats.lock().await.clone();
        status.running = running;
        Some(status)
    }

    pub async fn stop_reactive_processor(&self, processor_name: &str) -> Result<(), AedbError> {
        self.pause_reactive_processor(processor_name).await
    }

    /// Stop all running reactive-processor loops WITHOUT mutating their
    /// persisted enabled/disabled registry state, so they auto-resume on the
    /// next start.
    ///
    /// Used during `shutdown()`: the loops hold a `Weak<AedbInstance>` and
    /// upgrade it for the duration of each iteration. If they keep running while
    /// the caller drops its `Arc`, a loop mid-iteration keeps the instance alive
    /// (and the exclusive data-directory lock held), so a fresh open of the same
    /// directory in the same process would spuriously fail. Aborting the loops
    /// here lets the instance fully drop and release the lock.
    pub(crate) async fn stop_all_reactive_runtimes_for_shutdown(&self) {
        let runtimes: Vec<ReactiveProcessorRuntime> = {
            let mut map = self.reactive_processor_runtimes.lock().await;
            map.drain().map(|(_, runtime)| runtime).collect()
        };
        for runtime in runtimes {
            runtime.stop.store(true, Ordering::Release);
            runtime.join.abort();
            let _ = runtime.join.await;
        }
    }

    pub async fn pause_reactive_processor(&self, processor_name: &str) -> Result<(), AedbError> {
        if processor_name.trim().is_empty() {
            return Err(AedbError::Validation(
                "processor_name cannot be empty".into(),
            ));
        }
        if let Some((options, _)) = self
            .load_reactive_processor_registration(processor_name)
            .await?
        {
            self.persist_reactive_processor_registration(processor_name, &options, false)
                .await?;
        }
        let runtime = {
            let mut runtimes = self.reactive_processor_runtimes.lock().await;
            runtimes.remove(processor_name)
        };
        if let Some(runtime) = runtime {
            runtime.stop.store(true, Ordering::Release);
            runtime.join.abort();
            let _ = runtime.join.await;
        }
        Ok(())
    }

    pub async fn resume_reactive_processor(
        self: &Arc<Self>,
        processor_name: &str,
    ) -> Result<(), AedbError> {
        if processor_name.trim().is_empty() {
            return Err(AedbError::Validation(
                "processor_name cannot be empty".into(),
            ));
        }
        {
            let runtimes = self.reactive_processor_runtimes.lock().await;
            if runtimes.contains_key(processor_name) {
                return Ok(());
            }
        }
        let registration = self
            .load_reactive_processor_registration(processor_name)
            .await?
            .ok_or_else(|| AedbError::NotFound {
                resource_type: ErrorResourceType::Table,
                resource_id: format!(
                    "{}.{}.{}:{}",
                    SYSTEM_PROJECT_ID,
                    SYSTEM_SCOPE_ID,
                    REACTIVE_PROCESSOR_REGISTRY_TABLE,
                    processor_name
                ),
            })?;
        let handler = {
            let handlers = self.reactive_processor_handlers.lock();
            handlers.get(processor_name).cloned().ok_or_else(|| {
                AedbError::Validation(format!(
                    "reactive processor handler not registered: {processor_name}"
                ))
            })?
        };
        let (options, _) = registration;
        self.start_reactive_processor_with_handler(processor_name, options, handler, true)
            .await
    }

    pub async fn list_reactive_processors(
        &self,
        consistency: ConsistencyMode,
    ) -> Result<Vec<ReactiveProcessorInfo>, AedbError> {
        let mut registrations = self
            .load_reactive_processor_registrations(consistency)
            .await?;
        registrations.sort_by(|a, b| a.processor_name.cmp(&b.processor_name));
        let running_names: HashSet<String> = {
            let runtimes = self.reactive_processor_runtimes.lock().await;
            runtimes.keys().cloned().collect()
        };
        Ok(registrations
            .into_iter()
            .map(|r| ReactiveProcessorInfo {
                running: running_names.contains(&r.processor_name),
                processor_name: r.processor_name,
                options: r.options,
                enabled: r.enabled,
                updated_at_micros: r.updated_at_micros,
            })
            .collect())
    }

    pub async fn reactive_processor_health(
        &self,
        processor_name: &str,
        consistency: ConsistencyMode,
    ) -> Result<ReactiveProcessorHealth, AedbError> {
        if processor_name.trim().is_empty() {
            return Err(AedbError::Validation(
                "processor_name cannot be empty".into(),
            ));
        }
        let registration = self
            .load_reactive_processor_registration_record(processor_name, consistency)
            .await?;
        let runtime_status = self.reactive_processor_runtime_status(processor_name).await;
        let lag = self
            .reactive_processor_lag_internal(processor_name, consistency, None)
            .await?;
        let running = runtime_status.as_ref().map(|s| s.running).unwrap_or(false);
        let enabled = registration.as_ref().map(|r| r.enabled).unwrap_or(running);
        let status = runtime_status.unwrap_or_else(|| ReactiveProcessorRuntimeStatus {
            processor_name: processor_name.to_string(),
            running,
            ..ReactiveProcessorRuntimeStatus::default()
        });
        Ok(ReactiveProcessorHealth {
            processor_name: processor_name.to_string(),
            enabled,
            running,
            checkpoint_seq: lag.checkpoint_seq,
            head_seq: lag.head_seq,
            lag_commits: lag.lag_commits,
            runs_total: status.runs_total,
            processed_events_total: status.processed_events_total,
            failures_total: status.failures_total,
            retries_total: status.retries_total,
            dead_lettered_total: status.dead_lettered_total,
            last_processed_seq: status.last_processed_seq,
            last_error: status.last_error,
            last_run_started_micros: status.last_run_started_micros,
            last_run_completed_micros: status.last_run_completed_micros,
            last_success_micros: status.last_success_micros,
            last_failure_micros: status.last_failure_micros,
            last_retry_micros: status.last_retry_micros,
            last_sleep_ms: status.last_sleep_ms,
            last_batch_events: status.last_batch_events,
        })
    }

    pub async fn reactive_processor_slo_status(
        &self,
        processor_name: &str,
        consistency: ConsistencyMode,
    ) -> Result<ReactiveProcessorSloStatus, AedbError> {
        if processor_name.trim().is_empty() {
            return Err(AedbError::Validation(
                "processor_name cannot be empty".into(),
            ));
        }
        let registration = self
            .load_reactive_processor_registration_record(processor_name, consistency)
            .await?
            .ok_or_else(|| AedbError::NotFound {
                resource_type: ErrorResourceType::Table,
                resource_id: format!(
                    "{}.{}.{}:{}",
                    SYSTEM_PROJECT_ID,
                    SYSTEM_SCOPE_ID,
                    REACTIVE_PROCESSOR_REGISTRY_TABLE,
                    processor_name
                ),
            })?;
        let health = self
            .reactive_processor_health(processor_name, consistency)
            .await?;
        Ok(Self::build_reactive_processor_slo_status(
            &registration.options,
            health,
        ))
    }

    pub async fn list_reactive_processor_slo_statuses(
        &self,
        consistency: ConsistencyMode,
    ) -> Result<Vec<ReactiveProcessorSloStatus>, AedbError> {
        let infos = self.list_reactive_processors(consistency).await?;
        let mut out = Vec::with_capacity(infos.len());
        for info in infos {
            let health = self
                .reactive_processor_health(&info.processor_name, consistency)
                .await?;
            out.push(Self::build_reactive_processor_slo_status(
                &info.options,
                health,
            ));
        }
        out.sort_by(|a, b| a.processor_name.cmp(&b.processor_name));
        Ok(out)
    }

    pub async fn enforce_reactive_processor_slos(
        &self,
        consistency: ConsistencyMode,
    ) -> Result<(), AedbError> {
        let statuses = self
            .list_reactive_processor_slo_statuses(consistency)
            .await?;
        let breaches: Vec<ReactiveProcessorSloStatus> =
            statuses.into_iter().filter(|s| s.breached).collect();
        if breaches.is_empty() {
            return Ok(());
        }
        let details = breaches
            .iter()
            .map(|s| {
                format!(
                    "{}: {}",
                    s.processor_name,
                    if s.reasons.is_empty() {
                        "unknown breach".to_string()
                    } else {
                        s.reasons.join("; ")
                    }
                )
            })
            .collect::<Vec<_>>()
            .join(" | ");
        Err(AedbError::Unavailable {
            message: format!("reactive processor SLO breach: {details}"),
        })
    }

    pub async fn start_reactive_processor<F, Fut>(
        self: &Arc<Self>,
        processor_name: &str,
        options: ReactiveProcessorOptions,
        handler: F,
    ) -> Result<(), AedbError>
    where
        F: Fn(Arc<AedbInstance>, Vec<EventOutboxRecord>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<(), AedbError>> + Send + 'static,
    {
        let handler: ReactiveProcessorHandler =
            Arc::new(move |db, events| Box::pin(handler(db, events)));
        self.start_reactive_processor_with_handler(processor_name, options, handler, true)
            .await
    }

    pub async fn register_reactive_processor_handler<F, Fut>(
        self: &Arc<Self>,
        processor_name: &str,
        handler: F,
    ) -> Result<bool, AedbError>
    where
        F: Fn(Arc<AedbInstance>, Vec<EventOutboxRecord>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<(), AedbError>> + Send + 'static,
    {
        if processor_name.trim().is_empty() {
            return Err(AedbError::Validation(
                "processor_name cannot be empty".into(),
            ));
        }
        let handler: ReactiveProcessorHandler =
            Arc::new(move |db, events| Box::pin(handler(db, events)));
        self.reactive_processor_handlers
            .lock()
            .insert(processor_name.to_string(), Arc::clone(&handler));
        let Some((options, enabled)) = self
            .load_reactive_processor_registration(processor_name)
            .await?
        else {
            return Ok(false);
        };
        if !enabled {
            return Ok(false);
        }
        {
            let runtimes = self.reactive_processor_runtimes.lock().await;
            if runtimes.contains_key(processor_name) {
                return Ok(false);
            }
        }
        self.start_reactive_processor_with_handler(processor_name, options, handler, false)
            .await?;
        Ok(true)
    }

    async fn start_reactive_processor_with_handler(
        self: &Arc<Self>,
        processor_name: &str,
        options: ReactiveProcessorOptions,
        handler: ReactiveProcessorHandler,
        persist_registry: bool,
    ) -> Result<(), AedbError> {
        if processor_name.trim().is_empty() {
            return Err(AedbError::Validation(
                "processor_name cannot be empty".into(),
            ));
        }
        if options.max_events_per_run == 0
            || options.max_bytes_per_run == 0
            || options.max_run_duration_ms == 0
            || options.run_interval_ms == 0
            || options.idle_backoff_ms == 0
            || options.checkpoint_watermark_commits == 0
            || (options.max_retries > 0 && options.retry_backoff_ms == 0)
        {
            return Err(AedbError::Validation(
                "reactive processor options must be > 0".into(),
            ));
        }
        let caller = if let Some(raw) = options.caller_id.as_ref() {
            let caller = CallerContext::new(raw.clone());
            ensure_external_caller_allowed(&caller)?;
            Some(caller)
        } else if self.require_authenticated_calls {
            return Err(AedbError::PermissionDenied(
                "reactive processor caller_id required in secure mode".into(),
            ));
        } else {
            None
        };
        if persist_registry {
            self.persist_reactive_processor_registration(processor_name, &options, true)
                .await?;
        }
        // Seed checkpoint state so caller-scoped ACKs never race a missing internal table.
        let _ = self
            .commit_prevalidated_system_internal(
                "init_reactive_processor_checkpoint",
                reactive_processor_checkpoint_mutation(processor_name, 0),
            )
            .await?;
        let stats = Arc::new(AsyncMutex::new(ReactiveProcessorRuntimeStatus {
            processor_name: processor_name.to_string(),
            running: true,
            ..ReactiveProcessorRuntimeStatus::default()
        }));
        {
            let runtimes = self.reactive_processor_runtimes.lock().await;
            if runtimes.contains_key(processor_name) {
                return Err(AedbError::Validation(format!(
                    "reactive processor already running: {processor_name}"
                )));
            }
        }
        let weak = Arc::downgrade(self);
        let stop = Arc::new(AtomicBool::new(false));
        let stop_loop = Arc::clone(&stop);
        let processor_name_owned = processor_name.to_string();
        let options_owned = options.clone();
        let caller_owned = caller.clone();
        let stats_loop = Arc::clone(&stats);
        let handler_loop = Arc::clone(&handler);
        let join = tokio::spawn(async move {
            let mut retry: Option<ReactiveProcessorRetryState> = None;
            loop {
                if stop_loop.load(Ordering::Acquire) {
                    break;
                }
                let loop_started_micros = system_now_micros();
                let Some(db) = weak.upgrade() else {
                    break;
                };
                let mut sleep_ms = options_owned.idle_backoff_ms;
                let mut add_processed = 0u64;
                let mut add_failures = 0u64;
                let mut add_retries = 0u64;
                let mut add_dead_lettered = 0u64;
                let mut last_processed_seq = 0u64;
                let mut last_error: Option<String> = None;

                if let Some(pending) = retry.as_mut() {
                    if Instant::now() < pending.next_retry_at {
                        let wait_ms = (pending.next_retry_at - Instant::now()).as_millis() as u64;
                        sleep_ms = wait_ms.max(1).min(options_owned.idle_backoff_ms.max(1));
                    } else {
                        match AedbInstance::process_reactive_processor_batch(
                            Arc::clone(&db),
                            &processor_name_owned,
                            caller_owned.as_ref(),
                            &options_owned,
                            &handler_loop,
                            pending.events.clone(),
                            pending.last_seq,
                        )
                        .await
                        {
                            Ok((processed, last_seq)) => {
                                add_processed = processed;
                                last_processed_seq = last_seq;
                                last_error = None;
                                retry = None;
                                sleep_ms = options_owned.run_interval_ms;
                            }
                            Err(err) => {
                                add_failures = 1;
                                last_error = Some(err.to_string());
                                if pending.attempts >= options_owned.max_retries {
                                    let exhausted = pending.clone();
                                    let _ = db
                                        .dead_letter_reactive_processor_batch(
                                            &processor_name_owned,
                                            caller_owned.as_ref(),
                                            &exhausted.events,
                                            &err.to_string(),
                                            exhausted.attempts.saturating_add(1),
                                        )
                                        .await;
                                    let _ = if let Some(caller) = caller_owned.as_ref() {
                                        db.ack_reactive_processor_checkpoint_batched_as(
                                            caller.clone(),
                                            &processor_name_owned,
                                            exhausted.last_seq,
                                            options_owned.checkpoint_watermark_commits,
                                        )
                                        .await
                                    } else {
                                        db.ack_reactive_processor_checkpoint_batched(
                                            &processor_name_owned,
                                            exhausted.last_seq,
                                            options_owned.checkpoint_watermark_commits,
                                        )
                                        .await
                                    };
                                    add_dead_lettered = exhausted.events.len() as u64;
                                    last_processed_seq = exhausted.last_seq;
                                    retry = None;
                                    sleep_ms = options_owned.run_interval_ms;
                                } else {
                                    pending.attempts = pending.attempts.saturating_add(1);
                                    let exp = pending.attempts.saturating_sub(1).min(8);
                                    let backoff =
                                        options_owned.retry_backoff_ms.saturating_mul(1u64 << exp);
                                    pending.last_error = err.to_string();
                                    pending.next_retry_at =
                                        Instant::now() + Duration::from_millis(backoff.max(1));
                                    add_retries = 1;
                                    sleep_ms = backoff.max(1);
                                }
                            }
                        }
                    }
                } else {
                    match AedbInstance::fetch_reactive_processor_batch(
                        Arc::clone(&db),
                        &processor_name_owned,
                        caller_owned.as_ref(),
                        &options_owned,
                    )
                    .await
                    {
                        Ok(events) if events.is_empty() => {
                            if options_owned.run_on_interval {
                                match AedbInstance::process_reactive_processor_batch(
                                    Arc::clone(&db),
                                    &processor_name_owned,
                                    caller_owned.as_ref(),
                                    &options_owned,
                                    &handler_loop,
                                    Vec::new(),
                                    0,
                                )
                                .await
                                {
                                    Ok((_, _)) => {
                                        last_error = None;
                                        sleep_ms = options_owned.run_interval_ms;
                                    }
                                    Err(err) => {
                                        add_failures = 1;
                                        last_error = Some(err.to_string());
                                        if options_owned.max_retries > 0 {
                                            add_retries = 1;
                                            let backoff = options_owned.retry_backoff_ms.max(1);
                                            retry = Some(ReactiveProcessorRetryState {
                                                events: Vec::new(),
                                                last_seq: 0,
                                                attempts: 1,
                                                next_retry_at: Instant::now()
                                                    + Duration::from_millis(backoff),
                                                last_error: err.to_string(),
                                            });
                                            sleep_ms = backoff;
                                        }
                                    }
                                }
                            }
                        }
                        Ok(events) => {
                            let last_seq = events.last().map(|e| e.commit_seq).unwrap_or(0);
                            match AedbInstance::process_reactive_processor_batch(
                                Arc::clone(&db),
                                &processor_name_owned,
                                caller_owned.as_ref(),
                                &options_owned,
                                &handler_loop,
                                events.clone(),
                                last_seq,
                            )
                            .await
                            {
                                Ok((processed, seq)) => {
                                    add_processed = processed;
                                    last_processed_seq = seq;
                                    sleep_ms = options_owned.run_interval_ms;
                                }
                                Err(err) => {
                                    add_failures = 1;
                                    last_error = Some(err.to_string());
                                    if options_owned.max_retries == 0 {
                                        let _ = db
                                            .dead_letter_reactive_processor_batch(
                                                &processor_name_owned,
                                                caller_owned.as_ref(),
                                                &events,
                                                &err.to_string(),
                                                1,
                                            )
                                            .await;
                                        let _ = if let Some(caller) = caller_owned.as_ref() {
                                            db.ack_reactive_processor_checkpoint_batched_as(
                                                caller.clone(),
                                                &processor_name_owned,
                                                last_seq,
                                                options_owned.checkpoint_watermark_commits,
                                            )
                                            .await
                                        } else {
                                            db.ack_reactive_processor_checkpoint_batched(
                                                &processor_name_owned,
                                                last_seq,
                                                options_owned.checkpoint_watermark_commits,
                                            )
                                            .await
                                        };
                                        add_dead_lettered = events.len() as u64;
                                        last_processed_seq = last_seq;
                                        sleep_ms = options_owned.run_interval_ms;
                                    } else {
                                        add_retries = 1;
                                        let backoff = options_owned.retry_backoff_ms.max(1);
                                        retry = Some(ReactiveProcessorRetryState {
                                            events,
                                            last_seq,
                                            attempts: 1,
                                            next_retry_at: Instant::now()
                                                + Duration::from_millis(backoff),
                                            last_error: err.to_string(),
                                        });
                                        sleep_ms = backoff;
                                    }
                                }
                            }
                        }
                        Err(err) => {
                            add_failures = 1;
                            last_error = Some(err.to_string());
                        }
                    }
                }

                {
                    let mut s = stats_loop.lock().await;
                    s.last_run_started_micros = Some(loop_started_micros);
                    s.runs_total = s.runs_total.saturating_add(1);
                    s.processed_events_total =
                        s.processed_events_total.saturating_add(add_processed);
                    s.failures_total = s.failures_total.saturating_add(add_failures);
                    s.retries_total = s.retries_total.saturating_add(add_retries);
                    s.dead_lettered_total = s.dead_lettered_total.saturating_add(add_dead_lettered);
                    s.last_sleep_ms = sleep_ms;
                    s.last_batch_events = add_processed.saturating_add(add_dead_lettered);
                    if last_processed_seq > 0 {
                        s.last_processed_seq = s.last_processed_seq.max(last_processed_seq);
                    }
                    if let Some(err) = last_error {
                        s.last_error = Some(err);
                    } else if add_processed > 0 || add_dead_lettered > 0 {
                        s.last_error = None;
                    }
                    let now = system_now_micros();
                    s.last_run_completed_micros = Some(now);
                    if add_processed > 0 || add_dead_lettered > 0 {
                        s.last_success_micros = Some(now);
                    }
                    if add_failures > 0 {
                        s.last_failure_micros = Some(now);
                    }
                    if add_retries > 0 {
                        s.last_retry_micros = Some(now);
                    }
                }
                tokio::time::sleep(Duration::from_millis(sleep_ms)).await;
            }
            let Some(db) = weak.upgrade() else {
                return;
            };
            if let Some(runtime) =
                db.reactive_processor_runtimes
                    .try_lock()
                    .ok()
                    .and_then(|runtimes| {
                        runtimes
                            .get(&processor_name_owned)
                            .map(|r| Arc::clone(&r.stats))
                    })
            {
                let mut s = runtime.lock().await;
                s.running = false;
            }
        });

        let mut runtimes = self.reactive_processor_runtimes.lock().await;
        if runtimes.contains_key(processor_name) {
            stop.store(true, Ordering::Release);
            join.abort();
            return Err(AedbError::Validation(format!(
                "reactive processor already running: {processor_name}"
            )));
        }
        self.reactive_processor_handlers
            .lock()
            .insert(processor_name.to_string(), handler);
        runtimes.insert(
            processor_name.to_string(),
            ReactiveProcessorRuntime { stop, join, stats },
        );
        Ok(())
    }

    async fn persist_reactive_processor_registration(
        &self,
        processor_name: &str,
        options: &ReactiveProcessorOptions,
        enabled: bool,
    ) -> Result<CommitResult, AedbError> {
        let mutation = reactive_processor_registry_mutation(processor_name, options, enabled)?;
        self.commit_prevalidated_system_internal(
            "persist_reactive_processor_registration",
            mutation,
        )
        .await
    }

    async fn load_reactive_processor_registration(
        &self,
        processor_name: &str,
    ) -> Result<Option<(ReactiveProcessorOptions, bool)>, AedbError> {
        self.load_reactive_processor_registration_record(processor_name, ConsistencyMode::AtLatest)
            .await
            .map(|opt| opt.map(|r| (r.options, r.enabled)))
    }

    async fn load_reactive_processor_registration_record(
        &self,
        processor_name: &str,
        consistency: ConsistencyMode,
    ) -> Result<Option<ReactiveProcessorRegistration>, AedbError> {
        let lease = self.acquire_snapshot(consistency).await?;
        let Some(table) = lease.view.keyspace.table(
            SYSTEM_PROJECT_ID,
            SYSTEM_SCOPE_ID,
            REACTIVE_PROCESSOR_REGISTRY_TABLE,
        ) else {
            return Ok(None);
        };
        let pk = EncodedKey::from_values(&[Value::Text(processor_name.to_string().into())]);
        let Some(stored) = table.rows.get(&pk) else {
            return Ok(None);
        };
        let row = lease.view.keyspace.materialize_row(stored)?;
        Self::decode_reactive_processor_registration_row(&row)
    }

    async fn load_reactive_processor_registrations(
        &self,
        consistency: ConsistencyMode,
    ) -> Result<Vec<ReactiveProcessorRegistration>, AedbError> {
        let lease = self.acquire_snapshot(consistency).await?;
        let Some(table) = lease.view.keyspace.table(
            SYSTEM_PROJECT_ID,
            SYSTEM_SCOPE_ID,
            REACTIVE_PROCESSOR_REGISTRY_TABLE,
        ) else {
            return Ok(Vec::new());
        };
        let mut out = Vec::with_capacity(table.rows.len());
        for stored in table.rows.values() {
            let row = lease.view.keyspace.materialize_row(stored)?;
            out.push(
                Self::decode_reactive_processor_registration_row(&row)?.ok_or_else(|| {
                    AedbError::Validation(
                        "reactive processor registry row missing processor_name".into(),
                    )
                })?,
            );
        }
        Ok(out)
    }

    fn decode_reactive_processor_registration_row(
        row: &Row,
    ) -> Result<Option<ReactiveProcessorRegistration>, AedbError> {
        let Some(Value::Text(processor_name)) = row.values.first() else {
            return Ok(None);
        };
        let Some(Value::Json(options_json)) = row.values.get(1) else {
            return Err(AedbError::Validation(
                "reactive processor registry options missing".into(),
            ));
        };
        let Some(Value::Boolean(enabled)) = row.values.get(2) else {
            return Err(AedbError::Validation(
                "reactive processor registry enabled flag missing".into(),
            ));
        };
        let updated_at_micros = match row.values.get(3) {
            Some(Value::Timestamp(v)) => u64::try_from(*v).unwrap_or(0),
            _ => 0,
        };
        let options: ReactiveProcessorOptions =
            serde_json::from_str(options_json).map_err(|e| AedbError::Validation(e.to_string()))?;
        Ok(Some(ReactiveProcessorRegistration {
            processor_name: processor_name.to_string(),
            options,
            enabled: *enabled,
            updated_at_micros,
        }))
    }

    fn build_reactive_processor_slo_status(
        options: &ReactiveProcessorOptions,
        health: ReactiveProcessorHealth,
    ) -> ReactiveProcessorSloStatus {
        let now = system_now_micros();
        let reference_ts = health
            .last_success_micros
            .or(health.last_run_completed_micros);
        let stall_ms = reference_ts.map(|ts| now.saturating_sub(ts) / 1_000);
        let mut reasons = Vec::new();
        let mut breached = false;
        if health.enabled {
            if let Some(max_lag) = options.max_allowed_lag_commits
                && health.lag_commits > max_lag
            {
                breached = true;
                reasons.push(format!(
                    "lag_commits={} exceeds max_allowed_lag_commits={}",
                    health.lag_commits, max_lag
                ));
            }
            if let Some(max_stall) = options.max_allowed_stall_ms {
                match stall_ms {
                    Some(stall) if stall > max_stall => {
                        breached = true;
                        reasons.push(format!(
                            "stall_ms={} exceeds max_allowed_stall_ms={}",
                            stall, max_stall
                        ));
                    }
                    None => {
                        breached = true;
                        reasons.push(
                            "stall_ms unavailable; processor has not produced a completed run"
                                .to_string(),
                        );
                    }
                    _ => {}
                }
            }
        }
        ReactiveProcessorSloStatus {
            processor_name: health.processor_name.clone(),
            breached,
            enabled: health.enabled,
            running: health.running,
            lag_commits: health.lag_commits,
            max_allowed_lag_commits: options.max_allowed_lag_commits,
            stall_ms,
            max_allowed_stall_ms: options.max_allowed_stall_ms,
            reasons,
        }
    }

    async fn fetch_reactive_processor_batch(
        db: Arc<Self>,
        processor_name: &str,
        caller: Option<&CallerContext>,
        options: &ReactiveProcessorOptions,
    ) -> Result<Vec<EventOutboxRecord>, AedbError> {
        let lag = if let Some(caller) = caller {
            db.reactive_processor_lag_as(caller, processor_name, ConsistencyMode::AtLatest)
                .await?
        } else {
            db.reactive_processor_lag(processor_name, ConsistencyMode::AtLatest)
                .await?
        };
        let mut from_seq = lag.checkpoint_seq;
        let deadline = Instant::now() + Duration::from_millis(options.max_run_duration_ms);
        let mut events = Vec::new();
        let mut bytes = 0usize;
        'read: while events.len() < options.max_events_per_run
            && bytes < options.max_bytes_per_run
            && Instant::now() < deadline
        {
            let limit = (options.max_events_per_run - events.len()).clamp(1, 128);
            let page = if let Some(caller) = caller {
                db.read_event_stream_as(
                    caller,
                    options.topic_filter.as_deref(),
                    from_seq,
                    limit,
                    ConsistencyMode::AtLatest,
                )
                .await?
            } else {
                db.read_event_stream(
                    options.topic_filter.as_deref(),
                    from_seq,
                    limit,
                    ConsistencyMode::AtLatest,
                )
                .await?
            };
            if page.events.is_empty() {
                break;
            }
            for event in page.events {
                let approx_bytes =
                    event.payload_json.len() + event.topic.len() + event.event_key.len() + 64;
                if !events.is_empty()
                    && bytes.saturating_add(approx_bytes) > options.max_bytes_per_run
                {
                    break 'read;
                }
                bytes = bytes.saturating_add(approx_bytes);
                from_seq = event.commit_seq;
                events.push(event);
                if events.len() >= options.max_events_per_run
                    || bytes >= options.max_bytes_per_run
                    || Instant::now() >= deadline
                {
                    break 'read;
                }
            }
            if page.next_commit_seq.is_none() {
                break;
            }
        }
        Ok(events)
    }

    async fn process_reactive_processor_batch(
        db: Arc<Self>,
        processor_name: &str,
        caller: Option<&CallerContext>,
        options: &ReactiveProcessorOptions,
        handler: &ReactiveProcessorHandler,
        events: Vec<EventOutboxRecord>,
        last_seq: u64,
    ) -> Result<(u64, u64), AedbError> {
        if events.is_empty() {
            handler(Arc::clone(&db), events).await?;
            return Ok((0, last_seq));
        }
        let processed = events.len() as u64;
        handler(Arc::clone(&db), events).await?;
        let _ = if let Some(caller) = caller {
            db.ack_reactive_processor_checkpoint_batched_as(
                caller.clone(),
                processor_name,
                last_seq,
                options.checkpoint_watermark_commits,
            )
            .await?
        } else {
            db.ack_reactive_processor_checkpoint_batched(
                processor_name,
                last_seq,
                options.checkpoint_watermark_commits,
            )
            .await?
        };
        Ok((processed, last_seq))
    }

    async fn dead_letter_reactive_processor_batch(
        &self,
        processor_name: &str,
        caller: Option<&CallerContext>,
        events: &[EventOutboxRecord],
        error: &str,
        attempts: u32,
    ) -> Result<CommitResult, AedbError> {
        let mutation = reactive_processor_dlq_mutation(processor_name, events, error, attempts);
        if let Some(caller) = caller {
            self.commit_as(caller.clone(), mutation).await
        } else {
            self.commit_prevalidated_system_internal("reactive_processor_dead_letter", mutation)
                .await
        }
    }
}
