//! Reactive processor support: system-table mutation builders, the
//! checkpoint-ack watermark cache, and the option/status/runtime records
//! shared between `AedbInstance` and the reactive processor API.

use crate::catalog::types::{Row, Value};
use crate::catalog::{
    REACTIVE_PROCESSOR_CHECKPOINTS_TABLE, REACTIVE_PROCESSOR_DLQ_TABLE,
    REACTIVE_PROCESSOR_REGISTRY_TABLE, SYSTEM_SCOPE_ID,
};
use crate::commit::validation::Mutation;
use crate::error::AedbError;
use crate::system_now_micros;
use crate::{AedbInstance, EventOutboxRecord};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::time::Instant;
use tokio::sync::Mutex as AsyncMutex;
use tokio::task::JoinHandle;

pub(crate) fn reactive_processor_checkpoint_mutation(
    processor_name: &str,
    checkpoint_seq: u64,
) -> Mutation {
    Mutation::Upsert {
        project_id: crate::catalog::SYSTEM_PROJECT_ID.to_string(),
        scope_id: SYSTEM_SCOPE_ID.to_string(),
        table_name: REACTIVE_PROCESSOR_CHECKPOINTS_TABLE.to_string(),
        primary_key: vec![Value::Text(processor_name.to_string().into())],
        row: Row::from_values(vec![
            Value::Text(processor_name.to_string().into()),
            Value::Integer(checkpoint_seq as i64),
            Value::Timestamp(system_now_micros() as i64),
        ]),
    }
}

pub(crate) fn reactive_processor_registry_mutation(
    processor_name: &str,
    options: &ReactiveProcessorOptions,
    enabled: bool,
) -> Result<Mutation, AedbError> {
    let options_json =
        serde_json::to_string(options).map_err(|e| AedbError::Encode(e.to_string()))?;
    Ok(Mutation::Upsert {
        project_id: crate::catalog::SYSTEM_PROJECT_ID.to_string(),
        scope_id: SYSTEM_SCOPE_ID.to_string(),
        table_name: REACTIVE_PROCESSOR_REGISTRY_TABLE.to_string(),
        primary_key: vec![Value::Text(processor_name.to_string().into())],
        row: Row::from_values(vec![
            Value::Text(processor_name.to_string().into()),
            Value::Json(options_json.into()),
            Value::Boolean(enabled),
            Value::Timestamp(system_now_micros() as i64),
        ]),
    })
}

pub(crate) fn reactive_processor_dlq_mutation(
    processor_name: &str,
    events: &[EventOutboxRecord],
    error: &str,
    attempts: u32,
) -> Mutation {
    let failed_at = system_now_micros() as i64;
    let rows = events
        .iter()
        .map(|event| {
            Row::from_values(vec![
                Value::Text(processor_name.to_string().into()),
                Value::Integer(event.commit_seq as i64),
                Value::Text(event.topic.clone().into()),
                Value::Text(event.event_key.clone().into()),
                Value::Json(event.payload_json.clone().into()),
                Value::Text(error.to_string().into()),
                Value::Integer(attempts as i64),
                Value::Timestamp(failed_at),
            ])
        })
        .collect();
    Mutation::UpsertBatch {
        project_id: crate::catalog::SYSTEM_PROJECT_ID.to_string(),
        scope_id: SYSTEM_SCOPE_ID.to_string(),
        table_name: REACTIVE_PROCESSOR_DLQ_TABLE.to_string(),
        rows,
    }
}

#[derive(Debug, Clone, Default)]
pub(crate) struct ReactiveCheckpointAckState {
    pub(crate) last_persisted_seq: u64,
    pub(crate) last_touch_micros: u64,
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub(crate) struct ReactiveCheckpointAckCacheKey {
    pub(crate) processor_name: String,
    pub(crate) caller_id: Option<String>,
}

pub(crate) const REACTIVE_ACK_CACHE_MAX_ENTRIES: usize = 1_024;
const REACTIVE_ACK_CACHE_EVICT_BATCH: usize = 128;

pub(crate) fn prune_reactive_ack_cache(
    cache: &mut HashMap<ReactiveCheckpointAckCacheKey, ReactiveCheckpointAckState>,
) {
    if cache.len() <= REACTIVE_ACK_CACHE_MAX_ENTRIES {
        return;
    }
    let excess = cache.len() - REACTIVE_ACK_CACHE_MAX_ENTRIES;
    let prune_count = excess
        .saturating_add(REACTIVE_ACK_CACHE_EVICT_BATCH)
        .min(cache.len());
    let mut oldest: Vec<(ReactiveCheckpointAckCacheKey, u64)> = cache
        .iter()
        .map(|(k, v)| (k.clone(), v.last_touch_micros))
        .collect();
    oldest.sort_by_key(|(_, ts)| *ts);
    for (key, _) in oldest.into_iter().take(prune_count) {
        cache.remove(&key);
    }
}

pub type ReactiveProcessorHandlerFuture =
    Pin<Box<dyn Future<Output = Result<(), AedbError>> + Send>>;
pub type ReactiveProcessorHandler = Arc<
    dyn Fn(Arc<AedbInstance>, Vec<EventOutboxRecord>) -> ReactiveProcessorHandlerFuture
        + Send
        + Sync,
>;

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct ReactiveProcessorOptions {
    pub caller_id: Option<String>,
    pub topic_filter: Option<String>,
    pub run_on_interval: bool,
    pub max_allowed_lag_commits: Option<u64>,
    pub max_allowed_stall_ms: Option<u64>,
    pub max_events_per_run: usize,
    pub max_bytes_per_run: usize,
    pub max_run_duration_ms: u64,
    pub run_interval_ms: u64,
    pub idle_backoff_ms: u64,
    pub checkpoint_watermark_commits: u64,
    pub max_retries: u32,
    pub retry_backoff_ms: u64,
}

impl Default for ReactiveProcessorOptions {
    fn default() -> Self {
        Self {
            caller_id: None,
            topic_filter: None,
            run_on_interval: false,
            max_allowed_lag_commits: None,
            max_allowed_stall_ms: None,
            max_events_per_run: 256,
            max_bytes_per_run: 2 * 1024 * 1024,
            max_run_duration_ms: 250,
            run_interval_ms: 20,
            idle_backoff_ms: 50,
            checkpoint_watermark_commits: 64,
            max_retries: 3,
            retry_backoff_ms: 100,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ReactiveProcessorRuntimeStatus {
    pub processor_name: String,
    pub running: bool,
    pub runs_total: u64,
    pub processed_events_total: u64,
    pub failures_total: u64,
    pub retries_total: u64,
    pub dead_lettered_total: u64,
    pub last_processed_seq: u64,
    pub last_error: Option<String>,
    pub last_run_started_micros: Option<u64>,
    pub last_run_completed_micros: Option<u64>,
    pub last_success_micros: Option<u64>,
    pub last_failure_micros: Option<u64>,
    pub last_retry_micros: Option<u64>,
    pub last_sleep_ms: u64,
    pub last_batch_events: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReactiveProcessorInfo {
    pub processor_name: String,
    pub options: ReactiveProcessorOptions,
    pub enabled: bool,
    pub running: bool,
    pub updated_at_micros: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReactiveProcessorHealth {
    pub processor_name: String,
    pub enabled: bool,
    pub running: bool,
    pub checkpoint_seq: u64,
    pub head_seq: u64,
    pub lag_commits: u64,
    pub runs_total: u64,
    pub processed_events_total: u64,
    pub failures_total: u64,
    pub retries_total: u64,
    pub dead_lettered_total: u64,
    pub last_processed_seq: u64,
    pub last_error: Option<String>,
    pub last_run_started_micros: Option<u64>,
    pub last_run_completed_micros: Option<u64>,
    pub last_success_micros: Option<u64>,
    pub last_failure_micros: Option<u64>,
    pub last_retry_micros: Option<u64>,
    pub last_sleep_ms: u64,
    pub last_batch_events: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReactiveProcessorSloStatus {
    pub processor_name: String,
    pub breached: bool,
    pub enabled: bool,
    pub running: bool,
    pub lag_commits: u64,
    pub max_allowed_lag_commits: Option<u64>,
    pub stall_ms: Option<u64>,
    pub max_allowed_stall_ms: Option<u64>,
    pub reasons: Vec<String>,
}

#[derive(Debug)]
pub(crate) struct ReactiveProcessorRuntime {
    pub(crate) stop: Arc<AtomicBool>,
    pub(crate) join: JoinHandle<()>,
    pub(crate) stats: Arc<AsyncMutex<ReactiveProcessorRuntimeStatus>>,
}

#[derive(Debug, Clone)]
pub(crate) struct ReactiveProcessorRetryState {
    pub(crate) events: Vec<EventOutboxRecord>,
    pub(crate) last_seq: u64,
    pub(crate) attempts: u32,
    pub(crate) next_retry_at: Instant,
    pub(crate) last_error: String,
}

#[derive(Debug, Clone)]
pub(crate) struct ReactiveProcessorRegistration {
    pub(crate) processor_name: String,
    pub(crate) options: ReactiveProcessorOptions,
    pub(crate) enabled: bool,
    pub(crate) updated_at_micros: u64,
}
