//! A bounded, task-backed callback lane for host policy work.
//!
//! Remote workbench control remains owned by the worker actor. Callback policy
//! is enqueued on this independent lane, so a remote cell never waits while a
//! transport or session lock is held.
use crate::protocol::{Generation, HookSlot, OperationId, SessionId, TargetStamp};
use anyhow::{ensure, Context, Result};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::{
    future::Future,
    pin::Pin,
    sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    },
    time::{Duration, Instant},
};
use tokio::sync::{mpsc, oneshot};

/// The placement selected for the originating workbench generation.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum CallbackTarget {
    Local,
    Remote { stamp: TargetStamp },
}

/// Immutable execution identity carried with every policy callback.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CallbackContext {
    pub session_id: SessionId,
    pub execution_id: OperationId,
    pub target: CallbackTarget,
    pub generation: Generation,
    pub policy_revision: String,
}

impl CallbackContext {
    pub fn local(
        session_id: SessionId,
        execution_id: OperationId,
        generation: Generation,
        policy_revision: impl Into<String>,
    ) -> Self {
        Self {
            session_id,
            execution_id,
            target: CallbackTarget::Local,
            generation,
            policy_revision: policy_revision.into(),
        }
    }

    pub fn remote(
        session_id: SessionId,
        execution_id: OperationId,
        stamp: TargetStamp,
        generation: Generation,
        policy_revision: impl Into<String>,
    ) -> Self {
        Self {
            session_id,
            execution_id,
            target: CallbackTarget::Remote { stamp },
            generation,
            policy_revision: policy_revision.into(),
        }
    }

    fn validate(&self) -> Result<()> {
        ensure!(
            !self.policy_revision.trim().is_empty(),
            "callback policy revision is empty"
        );
        if let CallbackTarget::Remote { stamp } = &self.target {
            ensure!(
                !stamp.target_id.as_str().trim().is_empty()
                    && !stamp.namespace_id.as_str().trim().is_empty()
                    && !stamp.namespace_generation.as_str().trim().is_empty(),
                "callback target binding is incomplete"
            );
        }
        Ok(())
    }
}

/// One host callback request, including the source worker's complete pin.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CallbackRequest {
    pub context: CallbackContext,
    pub slot: HookSlot,
    pub input: Value,
}

impl CallbackRequest {
    pub fn new(context: CallbackContext, slot: HookSlot, input: Value) -> Result<Self> {
        context.validate()?;
        Ok(Self {
            context,
            slot,
            input,
        })
    }
}

/// A callback handler runs outside the caller and broker queue task.
pub type CallbackFuture = Pin<Box<dyn Future<Output = Result<Value>> + Send>>;
pub type CallbackHandler = Arc<dyn Fn(CallbackRequest) -> CallbackFuture + Send + Sync>;

struct QueuedCallback {
    request: CallbackRequest,
    reply: oneshot::Sender<Result<Value>>,
}

#[derive(Default)]
struct BrokerState {
    submitted: AtomicUsize,
    serviced: AtomicUsize,
}

#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct CallbackStats {
    pub submitted: usize,
    pub serviced: usize,
}

/// A bounded callback broker. It has no session or database lock and can be
/// awaited from the remote worker's host-request path without re-entering it.
#[derive(Clone)]
pub struct CallbackBroker {
    tx: mpsc::Sender<QueuedCallback>,
    state: Arc<BrokerState>,
    last_context: Arc<tokio::sync::Mutex<Option<CallbackContext>>>,
}

impl CallbackBroker {
    pub fn new(capacity: usize, handler: CallbackHandler) -> Result<Self> {
        ensure!(capacity > 0, "callback broker capacity must be positive");
        let (tx, mut rx) = mpsc::channel::<QueuedCallback>(capacity);
        let state = Arc::new(BrokerState::default());
        let last_context = Arc::new(tokio::sync::Mutex::new(None));
        let worker_state = state.clone();
        let worker_context = last_context.clone();
        tokio::spawn(async move {
            while let Some(queued) = rx.recv().await {
                if queued.reply.is_closed() {
                    continue;
                }
                worker_state.serviced.fetch_add(1, Ordering::Relaxed);
                *worker_context.lock().await = Some(queued.request.context.clone());
                let handler = handler.clone();
                tokio::spawn(async move {
                    let result = handler(queued.request).await;
                    let _ = queued.reply.send(result);
                });
            }
        });
        Ok(Self {
            tx,
            state,
            last_context,
        })
    }

    /// Enqueue and await a callback without holding caller-owned state.
    pub async fn submit(&self, request: CallbackRequest, budget: Duration) -> Result<Value> {
        request.context.validate()?;
        ensure!(
            budget > Duration::ZERO,
            "callback deadline must be positive"
        );
        self.state.submitted.fetch_add(1, Ordering::Relaxed);
        let (reply, receive) = oneshot::channel();
        let deadline = Instant::now() + budget;
        let queued = QueuedCallback { request, reply };
        let remaining = deadline.saturating_duration_since(Instant::now());
        tokio::time::timeout(remaining, self.tx.send(queued))
            .await
            .context("callback broker enqueue timed out")?
            .context("callback broker is closed")?;
        let remaining = deadline.saturating_duration_since(Instant::now());
        tokio::time::timeout(remaining, receive)
            .await
            .context("callback handler timed out")?
            .context("callback handler dropped its reply")?
    }

    pub fn stats(&self) -> CallbackStats {
        CallbackStats {
            submitted: self.state.submitted.load(Ordering::Relaxed),
            serviced: self.state.serviced.load(Ordering::Relaxed),
        }
    }

    pub async fn last_context(&self) -> Option<CallbackContext> {
        self.last_context.lock().await.clone()
    }
}