//! lifecycle truth for remote namespaces.
//!
//! m2 deliberately advertises **connection-scoped persistence**. a remote
//! namespace belongs to one live connection owner; losing that connection
//! loses the namespace. reconnecting creates a new owner, namespace id, and
//! generation. this module does not claim reattachment, and `ControlPersist`
//! is not evidence of one.

use crate::{
    protocol::{OperationId, SessionId},
    remote::{
        bootstrap::{ensure_target_readiness, BootstrapConfig},
        bundle::BundleSpec,
    },
    store::Store,
    target::{NamespaceGeneration, NamespaceId, RemoteTarget, TargetId},
};
use anyhow::{bail, ensure, Context, Result};
use serde::{Deserialize, Serialize};
use serde_json::json;
use std::{
    path::PathBuf,
    sync::{Arc, Mutex},
};

/// Persistence offered by a remote workbench.
///
/// The only M2 variant is [`PersistenceMode::ConnectionScoped`]. In
/// particular, there is no reattach variant until owner proof and deduplicated
/// operation receipts exist.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PersistenceMode {
    ConnectionScoped,
}

impl PersistenceMode {
    pub const ADVERTISED: Self = Self::ConnectionScoped;
}

/// Whether the owner of a connection-scoped namespace can still speak for it.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
pub enum ConnectionState {
    Connected,
    Lost { reason: String },
}

/// A public snapshot suitable for an SDK or diagnostic response.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LifecycleSnapshot {
    pub mode: PersistenceMode,
    pub namespace_id: NamespaceId,
    pub namespace_generation: NamespaceGeneration,
    pub owner_id: String,
    pub state: ConnectionState,
}

/// A narrow, typed diagnosis that permits the explicit bootstrap repair lane.
///
/// A command's nonzero exit is deliberately absent. Callers must first have a
/// bootstrap boundary identify one of these failures; strings and booleans are
/// not repair authorization.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum BootstrapFailure {
    MissingPrerequisite { detail: String },
    CandidateRejected { detail: String },
    VerificationFailed { detail: String },
    IncompatibleEnvironment { detail: String },
}

impl BootstrapFailure {
    fn detail(&self) -> &str {
        match self {
            Self::MissingPrerequisite { detail }
            | Self::CandidateRejected { detail }
            | Self::VerificationFailed { detail }
            | Self::IncompatibleEnvironment { detail } => detail,
        }
    }

    fn validate(&self) -> Result<()> {
        ensure_reason(self.detail())
    }
}

/// An attributable request for the exceptional repair lane.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RepairRequest {
    pub target: TargetId,
    pub operation: OperationId,
    pub reason: BootstrapFailure,
    /// If set, this repair is adjacent to an operation whose remote outcome is
    /// unknown. The repair is recorded, then refused; it can never replay it.
    pub uncertain_operation: Option<OperationId>,
}

impl RepairRequest {
    pub fn new(target: TargetId, operation: OperationId, reason: BootstrapFailure) -> Result<Self> {
        reason.validate()?;
        Ok(Self {
            target,
            operation,
            reason,
            uncertain_operation: None,
        })
    }

    pub fn with_uncertain_operation(mut self, operation: OperationId) -> Self {
        self.uncertain_operation = Some(operation);
        self
    }
}

/// A typed refusal for a repair that could otherwise replay an uncertain cell.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RepairError {
    UncertainMutation { operation: OperationId },
}

impl std::fmt::Display for RepairError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::UncertainMutation { operation } => write!(
                f,
                "repair refused: operation {operation} has unknown outcome; replay is forbidden"
            ),
        }
    }
}
impl std::error::Error for RepairError {}

/// Durable evidence that the explicit repair lane was admitted.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RepairReceipt {
    pub target: TargetId,
    pub operation: OperationId,
    pub event_id: i64,
    pub status: String,
    pub normal_kernel_required: bool,
}

fn ensure_reason(detail: &str) -> Result<()> {
    ensure!(
        !detail.trim().is_empty() && detail.len() <= 4096,
        "bootstrap diagnosis is empty or too large"
    );
    Ok(())
}

/// Production repair action for a diagnosed bootstrap failure.
///
/// This is deliberately a concrete bootstrap invocation, not a caller-supplied
/// closure: repair uses the same target readiness path as first admission.
#[derive(Clone, Debug)]
pub struct BootstrapRepair {
    target: RemoteTarget,
    spec: BundleSpec,
    bundles_root: PathBuf,
    config: BootstrapConfig,
}

impl BootstrapRepair {
    pub fn new(
        target: RemoteTarget,
        spec: BundleSpec,
        bundles_root: impl Into<PathBuf>,
        config: BootstrapConfig,
    ) -> Self {
        Self {
            target,
            spec,
            bundles_root: bundles_root.into(),
            config,
        }
    }

    /// Runs the real content-addressed bootstrap path after recording the
    /// diagnosis. An uncertain operation is recorded as refused and never
    /// reaches the bootstrap action.
    pub async fn repair(
        &self,
        store: &Store,
        session: &SessionId,
        request: RepairRequest,
    ) -> Result<RepairReceipt> {
        request.reason.validate()?;
        ensure!(
            request.target == self.target.id,
            "repair target does not match configured bootstrap target"
        );
        let uncertain = request.uncertain_operation.clone();
        let status = if uncertain.is_some() {
            "refused_uncertain"
        } else {
            "started"
        };
        let admission_event = store
            .append(
                session,
                "remote.repair",
                json!({
                    "target": request.target,
                    "operation": request.operation,
                    "reason": request.reason,
                    "uncertain_operation": uncertain,
                    "status": status,
                    "replay": "forbidden",
                    "mode": PersistenceMode::ConnectionScoped,
                }),
            )
            .await?;
        if let Some(operation) = uncertain {
            return Err(RepairError::UncertainMutation { operation }.into());
        }

        let outcome = match ensure_target_readiness(
            &self.target,
            &self.spec,
            &self.bundles_root,
            &self.config,
        )
        .await
        {
            Ok(outcome) => outcome,
            Err(error) => {
                store
                    .append(
                        session,
                        "remote.repair",
                        json!({
                            "target": self.target.id,
                            "operation": request.operation,
                            "status": "failed",
                            "admission_event": admission_event,
                            "reason": request.reason,
                            "error": format!("{error:#}"),
                            "replay": "forbidden",
                        }),
                    )
                    .await?;
                return Err(error).context("diagnosed bootstrap repair failed");
            }
        };
        let event_id = store
            .append(
                session,
                "remote.repair",
                json!({
                    "target": self.target.id,
                    "operation": request.operation,
                    "status": "succeeded",
                    "admission_event": admission_event,
                    "bundle_id": outcome.manifest().bundle_id,
                    "warm_reuse": outcome.is_warm(),
                    "replay": "forbidden",
                }),
            )
            .await?;
        Ok(RepairReceipt {
            target: self.target.id.clone(),
            operation: request.operation,
            event_id,
            status: "succeeded".into(),
            normal_kernel_required: true,
        })
    }
}

/// Host-owned lease for one connection-scoped remote namespace.
///
/// `reconnect` never mutates this lease back to `Connected`: callers must
/// replace it with the newly returned lease, making accidental old-handle
/// reuse visible in both the namespace id and generation.
#[derive(Clone, Debug)]
pub struct ConnectionScopedLifecycle {
    namespace_id: NamespaceId,
    namespace_generation: NamespaceGeneration,
    owner_id: String,
    state: Arc<Mutex<ConnectionState>>,
}

impl ConnectionScopedLifecycle {
    /// Creates the initial lease for an already-admitted remote connection.
    pub fn new(namespace_id: NamespaceId, namespace_generation: NamespaceGeneration) -> Self {
        Self {
            namespace_id,
            namespace_generation,
            owner_id: uuid::Uuid::new_v4().to_string(),
            state: Arc::new(Mutex::new(ConnectionState::Connected)),
        }
    }

    /// M2's advertised persistence contract.
    #[must_use]
    pub const fn mode(&self) -> PersistenceMode {
        PersistenceMode::ConnectionScoped
    }

    #[must_use]
    pub fn snapshot(&self) -> LifecycleSnapshot {
        let state = self
            .state
            .lock()
            .map(|state| state.clone())
            .unwrap_or_else(|_| ConnectionState::Lost {
                reason: "lifecycle state lock poisoned".into(),
            });
        LifecycleSnapshot {
            mode: self.mode(),
            namespace_id: self.namespace_id.clone(),
            namespace_generation: self.namespace_generation.clone(),
            owner_id: self.owner_id.clone(),
            state,
        }
    }

    #[must_use]
    pub fn namespace_id(&self) -> &NamespaceId {
        &self.namespace_id
    }

    #[must_use]
    pub fn namespace_generation(&self) -> &NamespaceGeneration {
        &self.namespace_generation
    }

    #[must_use]
    pub fn is_connected(&self) -> bool {
        self.state
            .lock()
            .map(|state| matches!(*state, ConnectionState::Connected))
            .unwrap_or(false)
    }

    /// Refuses host work after the connection owner has been lost.
    pub fn ensure_connected(&self) -> Result<()> {
        let state = self
            .state
            .lock()
            .map_err(|_| anyhow::anyhow!("remote lifecycle state lock poisoned"))?;
        if let ConnectionState::Lost { reason } = &*state {
            bail!(
                "remote namespace {}@{} is lost: {reason}; reconnect explicitly",
                self.namespace_id,
                self.namespace_generation
            );
        }
        Ok(())
    }

    /// Records an uncertain transport loss exactly once.
    ///
    /// The first loss reason wins because a later observation cannot turn an
    /// unknown remote outcome into a known one.
    pub fn mark_lost(&self, reason: impl Into<String>) -> Result<bool> {
        let mut state = self
            .state
            .lock()
            .map_err(|_| anyhow::anyhow!("remote lifecycle state lock poisoned"))?;
        if matches!(*state, ConnectionState::Lost { .. }) {
            return Ok(false);
        }
        *state = ConnectionState::Lost {
            reason: reason.into(),
        };
        Ok(true)
    }

    /// Starts an independent owner and namespace after an explicit reconnect.
    /// The old lease remains `Lost`; it is never revived or reused.
    pub fn reconnect(&self) -> Result<Self> {
        self.ensure_lost()?;
        Ok(Self::new(
            fresh_namespace_id()?,
            NamespaceGeneration::fresh(),
        ))
    }

    fn ensure_lost(&self) -> Result<()> {
        let state = self
            .state
            .lock()
            .map_err(|_| anyhow::anyhow!("remote lifecycle state lock poisoned"))?;
        if matches!(*state, ConnectionState::Connected) {
            bail!("remote namespace is still connected; disconnect before reconnect")
        }
        Ok(())
    }
}

/// Makes an opaque namespace id for an explicit new connection owner.
pub fn fresh_namespace_id() -> Result<NamespaceId> {
    NamespaceId::parse(format!("ns-{}", uuid::Uuid::new_v4().simple()))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn connection_scoped_loss_is_terminal_and_reconnect_is_new() -> Result<()> {
        let old = ConnectionScopedLifecycle::new(
            NamespaceId::parse("old")?,
            NamespaceGeneration::parse("generation-old")?,
        );
        assert_eq!(old.mode(), PersistenceMode::ConnectionScoped);
        assert!(old.is_connected());
        assert!(old.mark_lost("relay closed")?);
        assert!(!old.mark_lost("later observation")?);
        assert!(old.ensure_connected().is_err());

        let new = old.reconnect()?;
        assert!(new.is_connected());
        assert_ne!(old.namespace_id(), new.namespace_id());
        assert_ne!(old.namespace_generation(), new.namespace_generation());
        assert_ne!(old.snapshot().owner_id, new.snapshot().owner_id);
        assert!(old.ensure_connected().is_err());
        Ok(())
    }
}