frame-host 0.4.1

//! The fragment-assembly authority (F-5b R1, ruled (i)-into-(ii); the
//! F5B-T1 ruling carries it on ONE CHANNEL).
//!
//! Boots `frame-fragments`' [`LiveAssembly`] — the server-assembled,
//! R4-ordered, incarnation-fenced complete-assembly surface — publishing
//! [`AssemblyUpdate`]s on ONE bus channel ([`ASSEMBLY_CHANNEL`]). This is
//! the authority stream the browser surface consumes: the assembly IS the
//! roster, absence from the complete set IS withdrawal truth, and
//! per-fragment `generation` fences staleness — the three convergence
//! gaps the per-component `frame.fragments.*` world cannot close from
//! inside a channel (its own design's stated limit).
//!
//! Why a channel and not a conversation (finding F5B-T1, ruled
//! 2026-07-24): the published JS SDK has no participant-protocol client —
//! channels are the landed D3 browser road (upstream ASK-7 records the
//! gap; the conversation-carried authority is the named future seam
//! behind this same payload codec). Every publication is a COMPLETE
//! assembly, so any single delivery fully resyncs a consumer; the newest
//! update is also served over HTTP ([`crate::server::ASSEMBLY_ROUTE`]) in
//! the same JSON shape as the channel payload — a browser bootstraps from
//! the snapshot and follows the channel with ONE codec.

use std::num::NonZeroUsize;
use std::sync::Arc;

use frame_core::registry::ComponentRegistry;
use frame_fragments::{AssemblyPublisher, AssemblyUpdate, LiveAssembly};
use liminal_sdk::remote::PushClient;
use liminal_server::config::{ChannelDef, ServerConfig};

use crate::error::HostError;

/// The one channel the assembly authority publishes on, advertised to the
/// page as `assemblyChannel` in `/frame/config.json`. Host-owned
/// infrastructure (like the health listener), not operator vocabulary.
pub const ASSEMBLY_CHANNEL: &str = "frame.assembly";

/// Bounded lifecycle-subscription buffer handed to the assembly worker —
/// same sizing rationale as the announcer's event buffer: drop-oldest with
/// a counted lag, and every lag increase forces a full authority resync.
const ASSEMBLY_EVENT_BUFFER: usize = 256;

/// The embedded bus config with the host-owned assembly channel declared.
/// The bus refuses subscriptions to unconfigured channels, and the
/// authority is always-on, so the host declares its own infrastructure
/// channel (schema-less, non-durable) rather than demanding a config line
/// for machinery the operator did not compose. An operator who DID declare
/// it keeps their declaration verbatim. The injected channel is
/// deliberately absent from the SERVED `channels` roster — that roster is
/// the application's; the assembly channel travels as `assemblyChannel`.
#[must_use]
pub fn bus_with_assembly_channel(bus: &ServerConfig) -> ServerConfig {
    let mut bus = bus.clone();
    if bus
        .channels
        .iter()
        .any(|channel| channel.name == ASSEMBLY_CHANNEL)
    {
        return bus;
    }
    tracing::info!(
        channel = ASSEMBLY_CHANNEL,
        "declaring the host-owned assembly channel on the embedded bus"
    );
    bus.channels.push(ChannelDef {
        name: ASSEMBLY_CHANNEL.to_owned(),
        schema_ref: None,
        durable: false,
        loaded_schema: None,
    });
    bus
}

/// The channel sink: one native bus connection publishing each complete
/// update as its JSON bytes — the exact bytes `/frame/assembly.json`
/// serves, so the browser decodes both with one codec.
struct ChannelSink {
    client: PushClient,
}

impl AssemblyPublisher for ChannelSink {
    type Error = std::io::Error;

    fn publish(&mut self, update: &AssemblyUpdate) -> Result<(), Self::Error> {
        let payload = serde_json::to_vec(update)
            .map_err(|error| std::io::Error::other(format!("encoding the update: {error}")))?;
        self.client
            .publish(ASSEMBLY_CHANNEL, payload)
            .map_err(|error| std::io::Error::other(error.to_string()))
    }
}

/// The booted authority: the live worker publishing on
/// [`ASSEMBLY_CHANNEL`]. Dropping the LAST clone of `live` closes and
/// joins the worker (event wake, no timer; queued truth still publishes —
/// the F5B-F1 fix); [`crate::application::Application`] holds one clone
/// for teardown and the shell server holds one for snapshot serving.
pub struct AssemblyAuthority {
    live: Arc<LiveAssembly>,
}

impl AssemblyAuthority {
    /// Connects a native publisher to the embedded bus's real bound TCP
    /// address and starts [`LiveAssembly`] over the host's component
    /// registry, publishing on [`ASSEMBLY_CHANNEL`]. The initial complete
    /// statement (revision 1, even when empty) is the first publication.
    ///
    /// # Errors
    ///
    /// Returns a typed failure when the bus connection cannot be made or
    /// the assembly worker cannot start.
    pub fn boot(
        bus_tcp: &str,
        auth_token: Option<&[u8]>,
        registry: Arc<ComponentRegistry>,
    ) -> Result<Self, HostError> {
        let client = match auth_token {
            Some(token) => PushClient::connect_with_auth(bus_tcp, token),
            None => PushClient::connect(bus_tcp),
        }
        .map_err(|error| HostError::AssemblyAuthority {
            detail: format!("connecting the assembly publisher: {error}"),
        })?;
        let capacity =
            NonZeroUsize::new(ASSEMBLY_EVENT_BUFFER).ok_or(HostError::SynchronizationPoisoned)?;
        let live =
            LiveAssembly::start(registry, capacity, ChannelSink { client }).map_err(|error| {
                HostError::AssemblyAuthority {
                    detail: format!("starting the live assembly worker: {error}"),
                }
            })?;
        Ok(Self {
            live: Arc::new(live),
        })
    }

    /// The channel the authority publishes on, advertised to the page as
    /// `assemblyChannel` in `/frame/config.json`.
    #[must_use]
    pub const fn channel(&self) -> &'static str {
        ASSEMBLY_CHANNEL
    }

    /// A snapshot handle for the shell server's assembly route.
    #[must_use]
    pub fn live(&self) -> Arc<LiveAssembly> {
        Arc::clone(&self.live)
    }

    /// The newest published update in the exact channel wire shape, when
    /// the worker has published one.
    ///
    /// # Errors
    ///
    /// Propagates the worker's typed failure (a stopped worker reports the
    /// publish failure that stopped it).
    pub fn current_update(
        live: &LiveAssembly,
    ) -> Result<Option<AssemblyUpdate>, frame_fragments::AssemblyError> {
        Ok(live.current()?.map(|published| AssemblyUpdate {
            revision: published.revision,
            assembly: (*published.assembly).clone(),
        }))
    }
}