aion_client/
transcript.rs

1//! Typed client subscription for one activity-attempt transcript stream.
2//!
3//! This is the sole caller-side implementation of the transcript WebSocket
4//! contract. Consumers receive canonical activity events, plus an explicit lag
5//! item that tells them to reattach from their last durable `store_seq`.
6
7use std::pin::Pin;
8
9use aion_core::{ActivityEvent, ActivityId, RunId, WorkflowId};
10use aion_proto::{ProtoActivityId, ProtoRunId, ProtoWorkflowId, TranscriptSubscription};
11use futures::Stream;
12
13use crate::{Client, ClientError};
14
15/// Boxed stream returned by [`Client::subscribe_transcript`].
16pub type TranscriptStream =
17    Pin<Box<dyn Stream<Item = Result<TranscriptStreamItem, ClientError>> + Send>>;
18
19/// One decoded server frame on a transcript subscription.
20#[derive(Clone, Debug, PartialEq)]
21pub enum TranscriptStreamItem {
22    /// One canonical activity transcript event.
23    Event(Box<ActivityEvent>),
24    /// The server's per-subscription broadcast receiver lagged. The caller can
25    /// recover this leg by reattaching with its last applied durable sequence.
26    Lagged {
27        /// Number of live broadcast records skipped before the socket closed.
28        skipped: u64,
29    },
30}
31
32/// Full identity and optional durable resume cursor for one transcript stream.
33#[derive(Clone, Debug, PartialEq, Eq)]
34pub struct TranscriptTarget {
35    /// Workflow containing the activity.
36    pub workflow_id: WorkflowId,
37    /// Concrete workflow generation.
38    pub run_id: RunId,
39    /// Activity within the run.
40    pub activity_id: ActivityId,
41    /// Attempt within the activity.
42    pub attempt: u32,
43    /// Highest durable `store_seq` already applied. `None` requests the full
44    /// retained splice; ephemeral events do not advance this cursor.
45    pub after_seq: Option<u64>,
46}
47
48impl TranscriptTarget {
49    fn subscription(self, namespace: &str) -> TranscriptSubscription {
50        TranscriptSubscription {
51            namespace: namespace.to_owned(),
52            workflow_id: Some(ProtoWorkflowId::from(self.workflow_id)),
53            run_id: Some(ProtoRunId::from(self.run_id)),
54            activity_id: Some(ProtoActivityId::from(self.activity_id)),
55            attempt: self.attempt,
56            after_seq: self.after_seq,
57        }
58    }
59}
60
61impl Client {
62    /// Opens one typed transcript WebSocket subscription.
63    ///
64    /// Event frames and lag frames are decoded strictly. An unknown kind,
65    /// unknown field, malformed body, terminal wire error, or abnormal socket
66    /// close is returned as a named [`ClientError`]; no frame is discarded.
67    /// A [`TranscriptStreamItem::Lagged`] item is recoverable per leg: callers
68    /// should announce it and open a fresh subscription with their last applied
69    /// `store_seq` as [`TranscriptTarget::after_seq`].
70    ///
71    /// # Errors
72    ///
73    /// Returns [`ClientError::InvalidArgument`] for a missing/invalid stream
74    /// endpoint, [`ClientError::Unauthenticated`] for a rejected upgrade, or
75    /// [`ClientError::Unavailable`] when the socket cannot be established.
76    pub async fn subscribe_transcript(
77        &self,
78        target: TranscriptTarget,
79    ) -> Result<TranscriptStream, ClientError> {
80        crate::transport::transcript_ws::open(&self.config, target.subscription(self.namespace()))
81            .await
82    }
83}
aion_client/transcript.rs

aion_client/
transcript.rs
