frame-host 0.4.0

//! The document-service transport binding (IRIDIUM-A3 R11, C1 as ruled):
//! frame-host is the thin embedding host — the service logic lives in
//! `frame-doc-service`; THIS module owns only the concrete bus wiring:
//!
//! - a [`FeedSink`] over the bus SDK's TCP push client, publishing every
//!   feed frame to the declared feed channel;
//! - the authoring pump: a TCP channel subscription on the declared
//!   authoring channel whose delivered bytes are pushed into
//!   [`DocumentService::receive_command`], every refusal logged loudly.
//!
//! Disconnect-release (C8, verified at liminal 0.3.0): the channel
//! delivery surface exposes NO participant-lifecycle observability for a
//! channel publisher, so v1 releases via the DECLARED policy expiry
//! (`[document].lease_expiry_ms`, observed lazily by the authority) plus
//! the page's best-effort `release-lease` on unload. The gap is recorded
//! as an upstream liminal ask in the leg's report; epoch fencing keeps
//! every failure mode safe (a dead holder's lease expires; its ghost
//! proposals fence).

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

use frame_authority::{AuthorityConfig, DocumentId};
use frame_core::component::ComponentId;
use frame_doc_service::{DocumentBinding, DocumentService, FeedSink, SinkError};
use frame_editor_wire::snapshot::PresentationInputs;
use frame_state::{ComponentSchema, ComponentStateStore, ComponentStoreHandle};
use liminal_sdk::remote::{PushClient, SubscriptionStream};

use crate::config::{DocumentSection, FrameConfig};
use crate::embedded::EmbeddedLiminal;
use crate::error::HostError;
use crate::syntax_theme::default_syntax_theme;

/// The authoring pump's receive quantum. Delivery is server push over the
/// blocking SDK stream; this bound only caps how long one idle block lasts
/// (and therefore the shutdown latency) — it is NOT a polling interval
/// (the frame-host lifecycle-logger precedent, `runtime.rs`).
const PUMP_QUANTUM: Duration = Duration::from_millis(500);

/// The live document binding: the service plus its bus wiring. Shutdown
/// order: stop the pump, join it, then tear the service down.
pub struct DocBinding {
    service: Arc<DocumentService>,
    pump: JoinHandle<()>,
    stop: Arc<AtomicBool>,
}

/// The feed publisher: one TCP push client on the declared feed channel.
/// The client is `!Sync` (it owns its ack receiver), and two threads
/// legitimately publish — the service's feed pump and its resync path —
/// so the mutex serializes them.
struct LiminalFeedSink {
    client: std::sync::Mutex<PushClient>,
    channel: String,
}

impl FeedSink for LiminalFeedSink {
    fn publish(&self, frame: &str) -> Result<(), SinkError> {
        self.client
            .lock()
            .map_err(|_| SinkError::new("feed publisher mutex poisoned"))?
            .publish(&self.channel, frame.as_bytes().to_vec())
            .map_err(|error| SinkError::new(format!("feed publish failed: {error}")))
    }
}

impl DocBinding {
    /// Boots the binding when `[document]` is configured: the frame-state
    /// store, the document service, the feed publisher, and the authoring
    /// pump — each failure typed and loud, nothing half-up.
    ///
    /// # Errors
    ///
    /// Returns the first typed failure from content reading, state-store
    /// opening, service boot, bus client connection, or thread start.
    pub fn boot(
        liminal: &EmbeddedLiminal,
        config: &FrameConfig,
    ) -> Result<Option<Self>, HostError> {
        let Some(section) = &config.document else {
            return Ok(None);
        };
        let initial_content = std::fs::read_to_string(&section.content_path).map_err(|source| {
            HostError::DocumentContent {
                path: section.content_path.clone(),
                source,
            }
        })?;
        let handle = open_store(section)?;
        let authority_config = AuthorityConfig::declare(
            Duration::from_millis(section.lease_expiry_ms),
            section.journal_length_bound,
            Duration::from_millis(section.quiesce_window_ms),
        )
        .map_err(|source| HostError::DocumentAuthorityConfig { source })?;
        let document =
            DocumentId::new(section.id.clone()).map_err(|error| HostError::ConfigContract {
                detail: format!("[document].id rejected by the model: {error}"),
            })?;
        let binding = DocumentBinding {
            document,
            language: section.language.clone(),
            initial_content,
            component_id: section.component_id.clone(),
            presentation: PresentationInputs {
                dark_theme: section.dark_theme,
                // C5, amended 2026-07-21: a highlighter must visibly
                // highlight out of the box. [document].syntax_theme, when
                // declared, REPLACES the built-in default wholesale (no
                // per-key merging); when absent, the built-in default
                // palette (copied from the proven example theme, never
                // invented) is used instead. The snapshot therefore always
                // carries Some(theme) — never the perpetually-absent member
                // C5 originally read as monochrome-by-default.
                syntax_theme: Some(
                    section
                        .syntax_theme
                        .clone()
                        .unwrap_or_else(default_syntax_theme),
                ),
            },
        };
        let address = liminal.tcp_addr().to_string();
        let auth_token = config
            .bus
            .auth
            .as_ref()
            .map(|auth| auth.token.clone().into_bytes());
        let sink = LiminalFeedSink {
            client: std::sync::Mutex::new(connect_publisher(&address, auth_token.as_deref())?),
            channel: section.feed_channel.clone(),
        };
        let service = Arc::new(
            DocumentService::start(handle, authority_config, binding, Arc::new(sink))
                .map_err(|source| HostError::DocumentService { source })?,
        );
        // The pinned SDK's SubscriptionStream sends an empty auth token by
        // construction (0.3.0); a token-gated bus + [document] is refused
        // at config load, so this open is always against an open server.
        let stream = SubscriptionStream::open(&address, &section.authoring_channel, Vec::new())
            .map_err(|error| HostError::DocumentBusClient {
                component: "authoring subscriber",
                detail: error.to_string(),
            })?;
        let stop = Arc::new(AtomicBool::new(false));
        let pump = spawn_authoring_pump(stream, Arc::clone(&service), Arc::clone(&stop))?;
        tracing::info!(
            document = %section.id,
            feed_channel = %section.feed_channel,
            authoring_channel = %section.authoring_channel,
            "document service live behind the embedded bus"
        );
        Ok(Some(Self {
            service,
            pump,
            stop,
        }))
    }

    /// Ordered teardown: stop the pump, join it, tear the service down.
    ///
    /// # Errors
    ///
    /// Returns typed pump-join and service-shutdown failures.
    pub fn shutdown(self) -> Result<(), HostError> {
        self.stop.store(true, Ordering::SeqCst);
        self.pump.join().map_err(|_| HostError::AuthoringPump {
            detail: "the authoring pump thread panicked".to_owned(),
        })?;
        let service = Arc::try_unwrap(self.service).map_err(|_| HostError::AuthoringPump {
            detail: "the authoring pump still holds the service after join".to_owned(),
        })?;
        service
            .shutdown()
            .map_err(|source| HostError::DocumentService { source })
    }
}

/// Connects the feed publisher, with the bus auth token when one gates the
/// server.
fn connect_publisher(address: &str, auth_token: Option<&[u8]>) -> Result<PushClient, HostError> {
    let connected = match auth_token {
        Some(token) => PushClient::connect_with_auth(address, token),
        None => PushClient::connect(address),
    };
    connected.map_err(|error| HostError::DocumentBusClient {
        component: "feed publisher",
        detail: error.to_string(),
    })
}

/// Opens (or reopens) the document service's frame-state namespace.
fn open_store(section: &DocumentSection) -> Result<ComponentStoreHandle, HostError> {
    let store = ComponentStateStore::open(&section.state_dir)
        .map_err(|source| HostError::DocumentState { source })?;
    let component = ComponentId::derive("frame-doc-service", &section.id);
    if let Ok(handle) = store.component_handle(component) {
        return Ok(handle);
    }
    // First boot for this namespace: declare and install.
    store
        .declare_schema(component, ComponentSchema::lww())
        .map_err(|source| HostError::DocumentState { source })?;
    store
        .install_storage(component)
        .map_err(|source| HostError::DocumentState { source })
}

/// The authoring pump: delivered command bytes go into the service; every
/// refusal is logged loudly; the thread exits on the stop flag or on
/// stream death (a dead wire is a loud death, never a silent one).
fn spawn_authoring_pump(
    stream: SubscriptionStream,
    service: Arc<DocumentService>,
    stop: Arc<AtomicBool>,
) -> Result<JoinHandle<()>, HostError> {
    std::thread::Builder::new()
        .name("frame-doc-service-authoring-pump".to_owned())
        .spawn(move || {
            loop {
                if stop.load(Ordering::SeqCst) {
                    tracing::info!("authoring pump exiting on shutdown");
                    return;
                }
                let waited = Instant::now();
                match stream.recv_timeout(PUMP_QUANTUM) {
                    Ok(delivered) => match String::from_utf8(delivered.payload().to_vec()) {
                        Ok(bytes) => match service.receive_command(&bytes) {
                            Ok(outcome) => {
                                tracing::info!(?outcome, "authoring command admitted");
                            }
                            Err(refusal) => {
                                tracing::warn!(%refusal, "authoring command REFUSED (loud, nothing applied)");
                            }
                        },
                        Err(error) => {
                            tracing::warn!(%error, "authoring frame is not UTF-8; refused loudly");
                        }
                    },
                    Err(error) => {
                        // The SDK folds the idle timeout and reader death into
                        // one Connection error; a sub-quantum return means the
                        // reader is gone (an mpsc disconnect returns
                        // immediately). Flagged as an upstream SDK seam in the
                        // leg's report.
                        if waited.elapsed() < PUMP_QUANTUM / 2 && !stop.load(Ordering::SeqCst) {
                            tracing::error!(
                                %error,
                                "authoring subscription DIED; the pump exits loudly — commands no longer reach the service"
                            );
                            return;
                        }
                    }
                }
            }
        })
        .map_err(|source| HostError::AuthoringPump {
            detail: format!("failed to start the authoring pump: {source}"),
        })
}