liminal-server 0.23.1

use std::collections::HashMap;
use std::sync::{Arc, Condvar, Mutex, MutexGuard};

use crate::server::participant::{ConnectionFateWorkItem, ParticipantSemanticError};
use liminal_protocol::wire::ConversationId;

/// The registered conversations this process has not restored yet, each with
/// the gate its first touch takes, and the server incarnation whose
/// unclean-restart repair a restore owes.
#[derive(Debug, Default)]
pub(in crate::server::participant::production) struct PendingRestores {
    gates: Mutex<GateMap>,
    repair_incarnation: Mutex<Option<u64>>,
    /// Conversations whose restore unit refused, each with the
    /// unclean-restart repair it still owes (`None` until startup owes one).
    /// A refused conversation loads later through the ordinary request path,
    /// which runs no repair, so its next touch runs the owed repair first.
    refused: Mutex<RefusedMap>,
    /// Connection fates met while their conversation was still pending
    /// (DUR-024, 11 Oct 2026): boot's connection-fate recovery no longer
    /// restores a conversation to apply a fate, which made the door's room
    /// replay for 30 minutes before the ports bound. The fate is applied
    /// inside the conversation's own restore unit, first, before the
    /// unclean-restart repair, with its class preserved.
    deferred_fates: Mutex<HashMap<ConversationId, Vec<ConnectionFateWorkItem>>>,
}

/// Each pending conversation's gate.
type GateMap = HashMap<ConversationId, Arc<Gate>>;

/// Where one pending conversation's restore unit stands. The lock is held
/// only to read or move the state, never across the unit, so a server stop
/// can wake every touch waiting on a unit that is still running (Gaia, 11
/// October 2026: the door's stop waited 60 s on the room's rebuild).
#[derive(Debug, Default)]
pub(super) struct Gate {
    pub(super) state: Mutex<GateState>,
    pub(super) changed: Condvar,
}

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(super) enum GateState {
    /// Nobody has started the unit.
    #[default]
    Pending,
    /// One thread is running the unit.
    Running,
    /// The unit has run, restored or refused.
    Done,
}

impl Gate {
    pub(super) fn state(&self) -> Result<MutexGuard<'_, GateState>, ParticipantSemanticError> {
        self.state
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "participant restore gate is poisoned".to_owned(),
            })
    }

    /// Records the unit as run and wakes every touch waiting on it.
    pub(super) fn done(&self) -> Result<(), ParticipantSemanticError> {
        *self.state()? = GateState::Done;
        self.changed.notify_all();
        Ok(())
    }
}

/// Each refused conversation's owed repair incarnation, behind its own lock.
type RefusedMap = HashMap<ConversationId, Arc<Mutex<Option<u64>>>>;

impl PendingRestores {
    pub(super) fn gates(&self) -> Result<MutexGuard<'_, GateMap>, ParticipantSemanticError> {
        self.gates
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "participant pending-restore lock is poisoned".to_owned(),
            })
    }

    /// Records the registered conversations as pending; answers how many.
    pub(in crate::server::participant::production) fn register(
        &self,
        conversation_ids: impl IntoIterator<Item = ConversationId>,
    ) -> Result<usize, ParticipantSemanticError> {
        let mut gates = self.gates()?;
        for conversation_id in conversation_ids {
            gates.entry(conversation_id).or_default();
        }
        Ok(gates.len())
    }

    pub(super) fn gate(
        &self,
        conversation_id: ConversationId,
    ) -> Result<Option<Arc<Gate>>, ParticipantSemanticError> {
        Ok(self.gates()?.get(&conversation_id).map(Arc::clone))
    }

    /// Wakes every touch waiting on a running unit, for a server stop: each
    /// rechecks the stop and leaves without waiting for the unit.
    pub(in crate::server::participant::production) fn wake_waiters(
        &self,
    ) -> Result<(), ParticipantSemanticError> {
        let gates: Vec<Arc<Gate>> = self.gates()?.values().map(Arc::clone).collect();
        for gate in gates {
            // Taken so a waiter between its stop check and its wait cannot
            // miss the wake.
            let state = gate.state()?;
            gate.changed.notify_all();
            drop(state);
        }
        Ok(())
    }

    pub(super) fn deferred(
        &self,
    ) -> Result<
        MutexGuard<'_, HashMap<ConversationId, Vec<ConnectionFateWorkItem>>>,
        ParticipantSemanticError,
    > {
        self.deferred_fates
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "participant deferred-fate lock is poisoned".to_owned(),
            })
    }

    /// Holds a connection fate for a conversation that is still pending and
    /// whose restore is not running: `true` when held, so the caller does
    /// not restore it now. The gate's state is held while the fate is
    /// recorded, so a restore that starts after this finds it, and a restore
    /// already running is not waited on here: the caller applies the fate
    /// through the ordinary path instead.
    pub(in crate::server::participant::production) fn defer_fate_if_pending(
        &self,
        conversation_id: ConversationId,
        work_item: &ConnectionFateWorkItem,
    ) -> Result<bool, ParticipantSemanticError> {
        let Some(gate) = self.gate(conversation_id)? else {
            return Ok(false);
        };
        let state = gate.state()?;
        if *state != GateState::Pending {
            return Ok(false);
        }
        self.deferred()?
            .entry(conversation_id)
            .or_default()
            .push(work_item.clone());
        drop(state);
        Ok(true)
    }

    /// The fates held for a conversation, taken by its restore unit.
    pub(super) fn take_deferred(
        &self,
        conversation_id: ConversationId,
    ) -> Result<Vec<ConnectionFateWorkItem>, ParticipantSemanticError> {
        Ok(self
            .deferred()?
            .remove(&conversation_id)
            .unwrap_or_default())
    }

    pub(super) fn finish(
        &self,
        conversation_id: ConversationId,
    ) -> Result<(), ParticipantSemanticError> {
        self.gates()?.remove(&conversation_id);
        Ok(())
    }

    /// The conversations still pending, in ascending id order.
    pub(in crate::server::participant::production) fn remaining(
        &self,
    ) -> Result<Vec<ConversationId>, ParticipantSemanticError> {
        let mut remaining: Vec<ConversationId> = self.gates()?.keys().copied().collect();
        remaining.sort_unstable();
        Ok(remaining)
    }

    /// Records the incarnation whose unclean-restart repair every restore
    /// from now on owes, and every conversation already refused owes too.
    pub(in crate::server::participant::production) fn owe_repair(
        &self,
        incarnation: u64,
    ) -> Result<(), ParticipantSemanticError> {
        *self.repair_incarnation()? = Some(incarnation);
        for owed in self.refused()?.values() {
            let mut owed = owed.lock().map_err(|_| poisoned_refusal())?;
            if owed.is_none() {
                *owed = Some(incarnation);
            }
        }
        Ok(())
    }

    pub(super) fn refused(&self) -> Result<MutexGuard<'_, RefusedMap>, ParticipantSemanticError> {
        self.refused.lock().map_err(|_| poisoned_refusal())
    }

    /// Records a refused restore with the repair it owes now, if any.
    pub(super) fn refuse(
        &self,
        conversation_id: ConversationId,
    ) -> Result<(), ParticipantSemanticError> {
        let owed = self.owed_repair()?;
        self.refused()?
            .entry(conversation_id)
            .or_insert_with(|| Arc::new(Mutex::new(owed)));
        Ok(())
    }

    pub(super) fn refused_gate(
        &self,
        conversation_id: ConversationId,
    ) -> Result<Option<Arc<Mutex<Option<u64>>>>, ParticipantSemanticError> {
        Ok(self.refused()?.get(&conversation_id).map(Arc::clone))
    }

    pub(super) fn clear_refused(
        &self,
        conversation_id: ConversationId,
    ) -> Result<(), ParticipantSemanticError> {
        self.refused()?.remove(&conversation_id);
        Ok(())
    }

    pub(super) fn repair_incarnation(
        &self,
    ) -> Result<MutexGuard<'_, Option<u64>>, ParticipantSemanticError> {
        self.repair_incarnation
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "participant repair-incarnation lock is poisoned".to_owned(),
            })
    }

    pub(in crate::server::participant::production) fn owed_repair(
        &self,
    ) -> Result<Option<u64>, ParticipantSemanticError> {
        Ok(*self.repair_incarnation()?)
    }
}

pub(super) fn poisoned_refusal() -> ParticipantSemanticError {
    ParticipantSemanticError::Internal {
        message: "participant refused-restore lock is poisoned".to_owned(),
    }
}