frame-conv 0.2.0

//! Event-driven workflow observation: a parked blocking facade over the
//! substrate's per-workflow event stream, and the exhaustive typed
//! projection from recorded events to conversation vocabulary.

use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};

use aion_client::{ClientError, EventStream};
use aion_core::{Event, TimerCancelCause, WithTimeoutOutcome};
use tokio::runtime::Runtime;

use super::error::WorkflowObserveError;
use super::id::{WorkflowConversationId, WorkflowRunId, WorkflowStepId};
use super::outcome::{
    ScheduleNote, WorkflowFailure, WorkflowItem, WorkflowPayload, WorkflowProgress,
    WorkflowProgressKind, WorkflowStepFailure, WorkflowTerminal,
};

/// An open observation of one workflow conversation. Purely
/// event-driven (Ruling B): [`Self::next_item`] parks the calling thread in
/// the runtime until the substrate stream yields — no timer, no clock
/// read, no re-armed wait. The pinned stream resumes per workflow from
/// `last_seq + 1` on reattach and deduplicates by sequence, so items
/// arrive gap-free and duplicate-free without any frame-conv cursor.
///
/// Consumer-side law (characterized, pinned): replay over a REOPENED
/// run re-surfaces its superseded terminal as history — a terminal item
/// ends the run only when it is the frontier (nothing follows it). See
/// [`WorkflowTerminal`]'s docs before ending any consumer at the first
/// terminal observed.
pub struct WorkflowObservation {
    runtime: Arc<Runtime>,
    stream: EventStream,
}

/// A std-only future resolving the stream's next item; written by hand
/// so observation needs no executor-utility dependency.
struct NextItem<'stream> {
    stream: &'stream mut EventStream,
}

impl Future for NextItem<'_> {
    type Output = Option<Result<Event, ClientError>>;

    fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
        self.stream.as_mut().poll_next(context)
    }
}

impl WorkflowObservation {
    pub(super) fn new(runtime: Arc<Runtime>, stream: EventStream) -> Self {
        Self { runtime, stream }
    }

    /// Blocks until the conversation publishes its next item, and
    /// projects it. Returns until: a typed progress item, a member of
    /// the closed terminal set, or a typed observation failure (S3
    /// connection fate, or the loud typed end-of-stream refusal).
    ///
    /// # Errors
    ///
    /// Returns [`WorkflowObserveError::ConnectionLost`] when observation
    /// itself becomes unavailable, [`WorkflowObserveError::StreamEnded`]
    /// when the stream ends without a terminal, or
    /// [`WorkflowObserveError::Substrate`] for any other typed stream
    /// failure.
    pub fn next_item(&mut self) -> Result<WorkflowItem, WorkflowObserveError> {
        let item = self.runtime.block_on(NextItem {
            stream: &mut self.stream,
        });
        match item {
            Some(Ok(event)) => Ok(project(event)),
            Some(Err(error)) => Err(WorkflowObserveError::from_client(&error)),
            None => Err(WorkflowObserveError::StreamEnded),
        }
    }
}

/// The R2 projection: EXHAUSTIVE over the pinned `aion-core` `=0.10.0`
/// event enum — every recorded event has exactly one typed projection
/// preserving its sequence, so nothing is dropped or flattened, and an
/// upstream pin bump that adds a variant fails HERE at compile time
/// (the loud typed fate of an unknown event). Arm bodies live in the
/// named constructors below so the match stays one screen of law.
fn project(event: Event) -> WorkflowItem {
    let seq = event.envelope().seq;
    let kind = match event {
        Event::WorkflowCompleted { result, .. } => return completed(seq, result),
        Event::WorkflowFailed { error, .. } => return failed(seq, error),
        Event::WorkflowCancelled { reason, .. } => return cancelled(seq, reason),
        Event::WorkflowTimedOut { timeout, .. } => return timed_out(seq, timeout),
        Event::WorkflowContinuedAsNew {
            input,
            workflow_type,
            parent_run_id,
            ..
        } => return continued_as_new(seq, input, workflow_type, &parent_run_id),
        Event::WorkflowStarted {
            workflow_type,
            run_id,
            parent_run_id,
            ..
        } => opened(workflow_type, &run_id, parent_run_id.as_ref()),
        Event::WorkflowReopened {
            run_id, reopened, ..
        } => reopened_kind(&run_id, &reopened),
        Event::WorkflowPaused { reason, .. } => WorkflowProgressKind::Paused { reason },
        Event::WorkflowResumed { .. } => WorkflowProgressKind::Resumed,
        Event::SearchAttributesUpdated { attributes, .. } => attributes_noted(attributes),
        Event::ActivityScheduled {
            activity_id,
            activity_type,
            input,
            ..
        } => step_scheduled(&activity_id, activity_type, input),
        Event::ActivityStarted {
            activity_id,
            attempt,
            ..
        } => step_started(&activity_id, attempt),
        Event::ActivityCompleted {
            activity_id,
            result,
            attempt,
            ..
        } => step_completed(&activity_id, attempt, result),
        Event::ActivityFailed {
            activity_id,
            error,
            attempt,
            ..
        } => step_failed(&activity_id, attempt, error),
        Event::ActivityCancelled {
            activity_id,
            attempt,
            ..
        } => step_cancelled(&activity_id, attempt),
        Event::TimerStarted { timer_id, .. } => timer_armed(&timer_id),
        Event::TimerFired { timer_id, .. } => timer_fired(&timer_id),
        Event::TimerCancelled {
            timer_id, cause, ..
        } => timer_retired(&timer_id, cause),
        Event::WithTimeoutCompleted {
            timer_id,
            outcome,
            result,
            ..
        } => timeout_settled(&timer_id, &outcome, result),
        Event::SignalReceived { name, payload, .. } => contribution_received(name, payload),
        Event::SignalSent {
            target_workflow_id,
            name,
            ..
        } => contribution_forwarded(&target_workflow_id, name),
        Event::ChildWorkflowStarted {
            child_workflow_id,
            workflow_type,
            ..
        } => child_opened(&child_workflow_id, workflow_type),
        Event::ChildWorkflowCompleted {
            child_workflow_id,
            result,
            ..
        } => child_completed(&child_workflow_id, result),
        Event::ChildWorkflowFailed {
            child_workflow_id,
            error,
            ..
        } => child_failed(&child_workflow_id, error),
        Event::ChildWorkflowCancelled {
            child_workflow_id, ..
        } => child_cancelled(&child_workflow_id),
        Event::ScheduleCreated { schedule_id, .. } => noted(&schedule_id, ScheduleNote::Created),
        Event::ScheduleUpdated { schedule_id, .. } => noted(&schedule_id, ScheduleNote::Updated),
        Event::SchedulePaused { schedule_id, .. } => noted(&schedule_id, ScheduleNote::Paused),
        Event::ScheduleResumed { schedule_id, .. } => noted(&schedule_id, ScheduleNote::Resumed),
        Event::ScheduleDeleted { schedule_id, .. } => noted(&schedule_id, ScheduleNote::Deleted),
        Event::ScheduleTriggered {
            schedule_id,
            workflow_id,
            run_id,
            ..
        } => schedule_triggered(&schedule_id, &workflow_id, &run_id),
    };
    WorkflowItem::Progress(WorkflowProgress { seq, kind })
}

fn step_of(activity_id: &aion_core::ActivityId) -> WorkflowStepId {
    WorkflowStepId::from_position(activity_id.sequence_position())
}

fn conversation_of(workflow_id: &aion_core::WorkflowId) -> WorkflowConversationId {
    WorkflowConversationId::from_uuid(workflow_id.as_uuid())
}

fn attributes_noted(
    attributes: std::collections::HashMap<String, aion_core::SearchAttributeValue>,
) -> WorkflowProgressKind {
    WorkflowProgressKind::AttributesNoted {
        attributes: attributes.into_keys().collect(),
    }
}

fn step_started(activity_id: &aion_core::ActivityId, attempt: u32) -> WorkflowProgressKind {
    WorkflowProgressKind::StepStarted {
        step: step_of(activity_id),
        attempt,
    }
}

fn step_cancelled(activity_id: &aion_core::ActivityId, attempt: u32) -> WorkflowProgressKind {
    WorkflowProgressKind::StepCancelled {
        step: step_of(activity_id),
        attempt,
    }
}

fn child_cancelled(child_workflow_id: &aion_core::WorkflowId) -> WorkflowProgressKind {
    WorkflowProgressKind::ChildCancelled {
        child: conversation_of(child_workflow_id),
    }
}

fn cancelled(seq: u64, reason: String) -> WorkflowItem {
    WorkflowItem::Terminal(WorkflowTerminal::Cancelled { seq, reason })
}

fn timed_out(seq: u64, timeout: String) -> WorkflowItem {
    WorkflowItem::Terminal(WorkflowTerminal::TimedOut { seq, timeout })
}

fn timer_armed(timer_id: &aion_core::TimerId) -> WorkflowProgressKind {
    WorkflowProgressKind::TimerArmed {
        timer: timer_id.to_string(),
    }
}

fn timer_fired(timer_id: &aion_core::TimerId) -> WorkflowProgressKind {
    WorkflowProgressKind::TimerFired {
        timer: timer_id.to_string(),
    }
}

fn timer_retired(timer_id: &aion_core::TimerId, cause: TimerCancelCause) -> WorkflowProgressKind {
    WorkflowProgressKind::TimerRetired {
        timer: timer_id.to_string(),
        permanent: matches!(cause, TimerCancelCause::WorkflowIntent),
    }
}

fn contribution_forwarded(
    target_workflow_id: &aion_core::WorkflowId,
    name: String,
) -> WorkflowProgressKind {
    WorkflowProgressKind::ContributionForwarded {
        target: conversation_of(target_workflow_id),
        name,
    }
}

fn child_opened(
    child_workflow_id: &aion_core::WorkflowId,
    workflow_type: String,
) -> WorkflowProgressKind {
    WorkflowProgressKind::ChildOpened {
        child: conversation_of(child_workflow_id),
        workflow_kind: workflow_type,
    }
}

fn completed(seq: u64, result: aion_core::Payload) -> WorkflowItem {
    WorkflowItem::Terminal(WorkflowTerminal::Completed {
        seq,
        result: WorkflowPayload::from_core(result),
    })
}

fn failed(seq: u64, error: aion_core::WorkflowError) -> WorkflowItem {
    WorkflowItem::Terminal(WorkflowTerminal::Failed {
        seq,
        failure: WorkflowFailure::from_core(error),
    })
}

fn continued_as_new(
    seq: u64,
    input: aion_core::Payload,
    workflow_type: Option<String>,
    parent_run_id: &aion_core::RunId,
) -> WorkflowItem {
    WorkflowItem::Terminal(WorkflowTerminal::ContinuedAsNew {
        seq,
        continued_run: WorkflowRunId::from_uuid(parent_run_id.as_uuid()),
        next_input: WorkflowPayload::from_core(input),
        next_kind: workflow_type,
    })
}

fn opened(
    workflow_type: String,
    run_id: &aion_core::RunId,
    parent_run_id: Option<&aion_core::RunId>,
) -> WorkflowProgressKind {
    WorkflowProgressKind::Opened {
        workflow_kind: workflow_type,
        run: WorkflowRunId::from_uuid(run_id.as_uuid()),
        continued_from: parent_run_id.map(|run| WorkflowRunId::from_uuid(run.as_uuid())),
    }
}

fn reopened_kind(
    run_id: &aion_core::RunId,
    reopened: &[aion_core::ActivityId],
) -> WorkflowProgressKind {
    WorkflowProgressKind::Reopened {
        run: WorkflowRunId::from_uuid(run_id.as_uuid()),
        redispatched_steps: reopened.iter().map(step_of).collect(),
    }
}

fn step_scheduled(
    activity_id: &aion_core::ActivityId,
    activity_type: String,
    input: aion_core::Payload,
) -> WorkflowProgressKind {
    WorkflowProgressKind::StepScheduled {
        step: step_of(activity_id),
        step_kind: activity_type,
        input: WorkflowPayload::from_core(input),
    }
}

fn step_completed(
    activity_id: &aion_core::ActivityId,
    attempt: u32,
    result: aion_core::Payload,
) -> WorkflowProgressKind {
    WorkflowProgressKind::StepCompleted {
        step: step_of(activity_id),
        attempt,
        result: WorkflowPayload::from_core(result),
    }
}

fn step_failed(
    activity_id: &aion_core::ActivityId,
    attempt: u32,
    error: aion_core::ActivityError,
) -> WorkflowProgressKind {
    WorkflowProgressKind::StepFailed {
        step: step_of(activity_id),
        attempt,
        failure: WorkflowStepFailure::from_core(error),
    }
}

fn timeout_settled(
    timer_id: &aion_core::TimerId,
    outcome: &WithTimeoutOutcome,
    result: Option<aion_core::Payload>,
) -> WorkflowProgressKind {
    WorkflowProgressKind::TimeoutSettled {
        timer: timer_id.to_string(),
        timed_out: matches!(outcome, WithTimeoutOutcome::TimedOut),
        result: result.map(WorkflowPayload::from_core),
    }
}

fn contribution_received(name: String, payload: aion_core::Payload) -> WorkflowProgressKind {
    WorkflowProgressKind::ContributionReceived {
        name,
        payload: WorkflowPayload::from_core(payload),
    }
}

fn child_completed(
    child_workflow_id: &aion_core::WorkflowId,
    result: aion_core::Payload,
) -> WorkflowProgressKind {
    WorkflowProgressKind::ChildCompleted {
        child: conversation_of(child_workflow_id),
        result: WorkflowPayload::from_core(result),
    }
}

fn child_failed(
    child_workflow_id: &aion_core::WorkflowId,
    error: aion_core::WorkflowError,
) -> WorkflowProgressKind {
    WorkflowProgressKind::ChildFailed {
        child: conversation_of(child_workflow_id),
        failure: WorkflowFailure::from_core(error),
    }
}

fn noted(schedule_id: &aion_core::ScheduleId, note: ScheduleNote) -> WorkflowProgressKind {
    WorkflowProgressKind::ScheduleNoted {
        schedule: schedule_id.to_string(),
        note,
    }
}

fn schedule_triggered(
    schedule_id: &aion_core::ScheduleId,
    workflow_id: &aion_core::WorkflowId,
    run_id: &aion_core::RunId,
) -> WorkflowProgressKind {
    WorkflowProgressKind::ScheduleNoted {
        schedule: schedule_id.to_string(),
        note: ScheduleNote::Triggered {
            conversation: conversation_of(workflow_id),
            run: WorkflowRunId::from_uuid(run_id.as_uuid()),
        },
    }
}