frame-conv 0.2.0

//! Embedded-engine harness for the R2/R3 workflow acceptance: compiles
//! AWL sources to `.aion` archives at test time (the AWL-SANCTION seam —
//! the only external road to an executing workflow), opens a persistent
//! libsql disk store, and builds the embedded engine inside a
//! harness-owned runtime. The engine and its runtime live for the length
//! of the test; frame-conv reaches it ONLY through the one named
//! `from_substrate_client` wall seam.

use std::error::Error;
use std::num::NonZeroUsize;
use std::path::PathBuf;
use std::sync::Arc;

use aion::engine::builder::WorkflowPackageSource;
use aion::signal::ConcreteSignalRouter;
use aion::{Engine, EngineBuilder};
use aion_awl_package::compile_and_assemble_awl;
use aion_client::Client;
use aion_store_libsql::LibSqlStore;

use super::store_dir;

/// A live embedded aion engine over a persistent disk store.
pub struct EmbeddedAion {
    /// Harness-owned runtime the engine's background tasks live on.
    pub runtime: tokio::runtime::Runtime,
    /// The engine, shareable into any number of embedded clients.
    pub engine: Arc<Engine>,
    /// Scratch directory holding the store and compiled archives.
    pub dir: PathBuf,
}

impl EmbeddedAion {
    /// Compiles each AWL source, loads the archives, and builds the
    /// engine on a fresh disk store under `label`'s scratch directory.
    pub fn start(label: &str, awl_sources: &[&str]) -> Result<Self, Box<dyn Error>> {
        Self::start_at(store_dir(label)?, awl_sources, |builder| builder)
    }

    /// Builds (or REBUILDS — same `dir` = the restart half of a crash
    /// window, recovery driven by the store's committed truth alone) an
    /// engine over the named directory. `customize` finishes the builder
    /// (activity dispatchers, recovery seams) before `build()`.
    pub fn start_at(
        dir: PathBuf,
        awl_sources: &[&str],
        customize: impl FnOnce(EngineBuilder) -> EngineBuilder,
    ) -> Result<Self, Box<dyn Error>> {
        let mut archives = Vec::new();
        for (index, source) in awl_sources.iter().enumerate() {
            let prepared = compile_and_assemble_awl(source, &dir)
                .map_err(|error| format!("AWL compile refused source {index}: {error}"))?;
            let path = dir.join(format!("workflow-{index}.aion"));
            std::fs::write(&path, &prepared.archive)?;
            archives.push(path);
        }
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(2)
            .enable_all()
            .build()?;
        // Two build()-time opt-ins the deferred seams otherwise refuse
        // typed: event_streaming installs the publisher behind
        // Engine::subscribe (without it every subscription is empty),
        // and the concrete signal-router factory installs
        // record-before-deliver signal routing.
        let engine = runtime.block_on(async {
            let store = LibSqlStore::open(dir.join("aion.db")).await?;
            let mut builder = EngineBuilder::new()
                .store(store)
                .in_memory_visibility()
                .event_streaming(NonZeroUsize::new(256).expect("nonzero"))
                .signal_router_factory(|runtime, handoff| {
                    Arc::new(ConcreteSignalRouter::new(runtime, handoff))
                });
            for archive in &archives {
                builder = builder.load_workflows(WorkflowPackageSource::Path(archive.clone()));
            }
            customize(builder)
                .build()
                .await
                .map_err(|error| Box::<dyn Error>::from(error.to_string()))
        })?;
        Ok(Self {
            runtime,
            engine: Arc::new(engine),
            dir,
        })
    }

    /// A fresh embedded substrate client over this engine.
    pub fn client(&self) -> Client {
        Client::embedded(Arc::clone(&self.engine))
    }
}