frame-conv 0.2.0

//! One durable workflow conversation: identity, contribution,
//! observation, and explicit inspection.

use std::num::NonZeroU64;
use std::sync::Arc;

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

use super::client::phase_from_status;
use super::error::WorkflowCallError;
use super::id::WorkflowIdentity;
use super::observe::WorkflowObservation;
use super::outcome::WorkflowInspection;

/// A handle to one durable workflow conversation — a typed facade over
/// exactly one aion workflow/run lineage. Cheap to clone-reconstruct
/// via [`super::WorkflowClient::conversation`]; holds no state beyond
/// identity and the shared runtime.
pub struct WorkflowConversation {
    pub(super) runtime: Arc<Runtime>,
    pub(super) client: Client,
    pub(super) identity: WorkflowIdentity,
}

impl std::fmt::Debug for WorkflowConversation {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("WorkflowConversation")
            .field("identity", &self.identity)
            .finish_non_exhaustive()
    }
}

impl WorkflowConversation {
    /// The caller-persistable identity of this conversation and run.
    #[must_use]
    pub const fn identity(&self) -> WorkflowIdentity {
        self.identity
    }

    /// Sends a typed contribution (aion signal) to THIS concrete run.
    ///
    /// `Ok(())` means the substrate ACCEPTED the contribution for
    /// durable delivery — it is not a delivery or processing
    /// confirmation. Processing news arrives on observation as
    /// [`super::WorkflowProgressKind::ContributionReceived`]; a reply,
    /// where one is required, is S2 request-response territory, not this
    /// seam.
    ///
    /// # Errors
    ///
    /// Returns [`WorkflowCallError::Unknown`] for an absent conversation,
    /// [`WorkflowCallError::NotLive`] for a terminal run, or the rest of
    /// the closed grammar for transport fates.
    pub fn contribute<T>(&self, name: &str, value: &T) -> Result<(), WorkflowCallError>
    where
        T: Serialize + ?Sized,
    {
        let workflow_id = self.workflow_id();
        let run_id = self.run_id();
        self.runtime
            .block_on(
                self.client
                    .signal_typed(&workflow_id, Some(&run_id), name, value),
            )
            .map_err(|error| WorkflowCallError::from_client(&error))
    }

    /// Opens observation of this conversation from an explicit committed
    /// sequence cursor: `resume_from` is the first sequence wanted, so
    /// `1` replays the full recorded history before splicing into live
    /// delivery, gap-free and duplicate-free (the pinned substrate
    /// contract). Zero is unrepresentable by type.
    #[must_use]
    pub fn observe_from(&self, resume_from: NonZeroU64) -> WorkflowObservation {
        let stream = self
            .client
            .subscribe_workflow_from(&self.workflow_id(), resume_from);
        WorkflowObservation::new(Arc::clone(&self.runtime), stream)
    }

    /// Fetches a point-in-time description for EXPLICIT inspection.
    ///
    /// Never a polling authority: progress and completion are observed
    /// exclusively through [`Self::observe_from`], which parks until the
    /// substrate publishes. The phase returned here is a pure projection
    /// of committed history at the moment of the call.
    ///
    /// # Errors
    ///
    /// Returns [`WorkflowCallError::Unknown`] for an absent conversation
    /// or the rest of the closed grammar for transport fates.
    pub fn inspect(&self) -> Result<WorkflowInspection, WorkflowCallError> {
        let workflow_id = self.workflow_id();
        let run_id = self.run_id();
        let description = self
            .runtime
            .block_on(self.client.describe(&workflow_id, Some(&run_id)))
            .map_err(|error| WorkflowCallError::from_client(&error))?;
        Ok(WorkflowInspection {
            phase: phase_from_status(status_from_events(&description.history)),
            recorded_events: description.history.len() as u64,
        })
    }

    fn workflow_id(&self) -> WorkflowId {
        WorkflowId::new(self.identity.conversation.as_uuid())
    }

    fn run_id(&self) -> RunId {
        RunId::new(self.identity.run.as_uuid())
    }
}