frame-conv 0.2.0

//! The workflow client: runtime ownership, connection, open, reopen, and
//! reconstruction from held identity.

use std::sync::Arc;

use aion_client::{Client, StartOptions};
use aion_core::{RunId, WorkflowId, WorkflowStatus};
use serde::Serialize;
use tokio::runtime::Runtime;

use super::conversation::WorkflowConversation;
use super::error::{WorkflowCallError, WorkflowConnectError};
use super::id::{WorkflowConversationId, WorkflowIdentity, WorkflowRunId};
use super::outcome::{ReopenedRun, WorkflowPhase};

/// Connection configuration for a remote workflow substrate endpoint.
#[derive(Clone, Debug)]
pub struct WorkflowAttachment {
    endpoint: String,
    namespace: Option<String>,
}

impl WorkflowAttachment {
    /// Names the substrate endpoint to attach to.
    #[must_use]
    pub fn new(endpoint: impl Into<String>) -> Self {
        Self {
            endpoint: endpoint.into(),
            namespace: None,
        }
    }

    /// Selects a namespace for every operation on this attachment.
    #[must_use]
    pub fn with_namespace(mut self, namespace: impl Into<String>) -> Self {
        self.namespace = Some(namespace.into());
        self
    }
}

/// A client for durable workflow conversations. Owns the module's
/// internal tokio runtime (A1): every substrate call crosses the aion
/// boundary through a parked `block_on` on this runtime — one runtime
/// per client, shared by every conversation and observation it opens
/// (the recorded granularity decision).
pub struct WorkflowClient {
    pub(super) runtime: Arc<Runtime>,
    pub(super) client: Client,
}

impl WorkflowClient {
    /// Connects to a remote workflow substrate endpoint.
    ///
    /// # Errors
    ///
    /// Returns [`WorkflowConnectError::Runtime`] when the internal
    /// runtime cannot be created, or [`WorkflowConnectError::Endpoint`]
    /// when the substrate refuses the configuration.
    pub fn connect(attachment: &WorkflowAttachment) -> Result<Self, WorkflowConnectError> {
        let runtime = build_runtime()?;
        let mut builder = Client::builder(attachment.endpoint.clone());
        if let Some(namespace) = &attachment.namespace {
            builder = builder.with_namespace(namespace.clone());
        }
        let client =
            runtime
                .block_on(builder.build())
                .map_err(|error| WorkflowConnectError::Endpoint {
                    message: error.to_string(),
                })?;
        Ok(Self {
            runtime: Arc::new(runtime),
            client,
        })
    }

    /// THE NAMED SUBSTRATE SEAM — the single sanctioned vocabulary-wall
    /// exemption, self-checked in `structural_witnesses.rs`: constructs a
    /// workflow client over a prebuilt, caller-configured substrate
    /// client. This is how an embedded-engine substrate (the R3 harness,
    /// and every R2 acceptance run) or a custom-authenticated remote
    /// client enters the conversation surface; no other public signature
    /// in this crate may name a substrate type.
    ///
    /// # Errors
    ///
    /// Returns [`WorkflowConnectError::Runtime`] when the internal
    /// runtime cannot be created.
    pub fn from_substrate_client(
        client: aion_client::Client,
    ) -> Result<Self, WorkflowConnectError> {
        Ok(Self {
            runtime: Arc::new(build_runtime()?),
            client,
        })
    }

    /// Opens a new durable workflow conversation: typed kind and input,
    /// EXPLICIT caller idempotency key.
    ///
    /// The honest weaker contract (gate-filled): the key is enforced per
    /// client instance only — open is NOT idempotent across process,
    /// client, or restart boundaries, and a double-open mints two
    /// distinct conversations, observable as two distinct
    /// [`WorkflowIdentity`] values. Reusing a key on this client for a
    /// DIFFERENT request is the typed
    /// [`WorkflowCallError::IdempotencyConflict`] refusal.
    ///
    /// # Errors
    ///
    /// Returns the closed [`WorkflowCallError`] grammar; an empty key is
    /// [`WorkflowCallError::InvalidArgument`].
    pub fn open<T>(
        &self,
        workflow_kind: &str,
        input: &T,
        idempotency_key: &str,
    ) -> Result<WorkflowConversation, WorkflowCallError>
    where
        T: Serialize + ?Sized,
    {
        let options = StartOptions {
            idempotency_key: Some(idempotency_key.to_owned()),
            ..StartOptions::default()
        };
        let handle = self
            .runtime
            .block_on(self.client.start_typed(workflow_kind, input, options))
            .map_err(|error| WorkflowCallError::from_client(&error))?;
        let identity = WorkflowIdentity {
            conversation: WorkflowConversationId::from_uuid(handle.workflow_id().as_uuid()),
            run: WorkflowRunId::from_uuid(handle.run_id().as_uuid()),
        };
        Ok(self.conversation(identity))
    }

    /// Reconstructs a conversation from caller-held identity. Pure
    /// reconstruction — no substrate round-trip; a dead or absent run
    /// surfaces as the typed refusal of whichever operation touches it.
    #[must_use]
    pub fn conversation(&self, identity: WorkflowIdentity) -> WorkflowConversation {
        WorkflowConversation {
            runtime: Arc::clone(&self.runtime),
            client: self.client.clone(),
            identity,
        }
    }

    /// Reopens a terminal reopenable run (Failed or Cancelled),
    /// re-driving it from committed history. Targets the conversation's
    /// latest run, or `run` when supplied. A live run is observed, never
    /// restarted: reopening it is the typed
    /// [`WorkflowCallError::InvalidState`] refusal.
    ///
    /// # Errors
    ///
    /// Returns [`WorkflowCallError::InvalidState`] for a non-reopenable
    /// run, [`WorkflowCallError::Unknown`] for an absent conversation,
    /// or the rest of the closed grammar for transport fates.
    pub fn reopen(
        &self,
        conversation: WorkflowConversationId,
        run: Option<WorkflowRunId>,
    ) -> Result<ReopenedRun, WorkflowCallError> {
        let workflow_id = WorkflowId::new(conversation.as_uuid());
        let run_id = run.map(|run| RunId::new(run.as_uuid()));
        let outcome = self
            .runtime
            .block_on(self.client.reopen(&workflow_id, run_id.as_ref()))
            .map_err(|error| WorkflowCallError::from_client(&error))?;
        Ok(ReopenedRun {
            run: WorkflowRunId::from_uuid(outcome.run_id.as_uuid()),
            phase: phase_from_status(outcome.status),
        })
    }
}

/// One multi-thread runtime with a single parked worker: the worker keeps
/// substrate transport tasks live between calls while caller threads
/// park in `block_on` at the boundary — event-driven wake mechanics,
/// recorded as the A1 report expectation.
fn build_runtime() -> Result<Runtime, WorkflowConnectError> {
    tokio::runtime::Builder::new_multi_thread()
        .worker_threads(1)
        .enable_all()
        .build()
        .map_err(|error| WorkflowConnectError::Runtime {
            message: error.to_string(),
        })
}

pub(super) fn phase_from_status(status: WorkflowStatus) -> WorkflowPhase {
    match status {
        WorkflowStatus::Running => WorkflowPhase::Running,
        WorkflowStatus::Completed => WorkflowPhase::Completed,
        WorkflowStatus::Failed => WorkflowPhase::Failed,
        WorkflowStatus::Cancelled => WorkflowPhase::Cancelled,
        WorkflowStatus::TimedOut => WorkflowPhase::TimedOut,
        WorkflowStatus::ContinuedAsNew => WorkflowPhase::ContinuedAsNew,
        WorkflowStatus::Paused => WorkflowPhase::Paused,
    }
}