durare 0.3.1

//! Durable stream tests on the in-memory provider: a workflow writes an
//! append-only stream that an external reader drains in order, observing the
//! close (or the producer going inactive).

use durare::{DurableContext, DurableEngine, Error, InMemoryProvider, Result, WorkflowOptions};
use std::sync::Arc;
use std::time::Duration;

/// A producer writes three values and closes the stream; the reader drains them
/// in order and sees `closed`. Each write/close is recorded as its own step.
#[tokio::test]
async fn write_close_then_read_in_order() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("producer", |ctx: DurableContext, _: ()| async move {
        for i in 0..3_i64 {
            ctx.write_stream("nums", i).await?;
        }
        ctx.close_stream("nums").await?;
        Ok::<_, Error>(())
    });

    engine
        .start::<_, ()>("producer", (), WorkflowOptions::with_id("p"))
        .await?
        .result()
        .await?;

    let (values, closed): (Vec<i64>, bool) = engine.read_stream("p", "nums").await?;
    assert_eq!(values, vec![0, 1, 2]);
    assert!(closed);

    // The three writes and the close each landed as their own step.
    let steps = engine.get_workflow_steps("p").await?;
    let names: Vec<&str> = steps.iter().map(|s| s.name.as_str()).collect();
    assert_eq!(
        names,
        vec![
            "DBOS.writeStream",
            "DBOS.writeStream",
            "DBOS.writeStream",
            "DBOS.closeStream",
        ]
    );
    Ok(())
}

/// A reader blocked on `read_stream` returns once the producer finishes, even if
/// the stream was never explicitly closed: an inactive producer can write no
/// more.
#[tokio::test]
async fn read_stops_when_producer_finishes_without_close() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("producer", |ctx: DurableContext, _: ()| async move {
        ctx.write_stream("s", "a".to_string()).await?;
        ctx.write_stream("s", "b".to_string()).await?;
        Ok::<_, Error>(())
    });

    engine
        .start::<_, ()>("producer", (), WorkflowOptions::with_id("p2"))
        .await?
        .result()
        .await?;

    let (values, closed): (Vec<String>, bool) = engine.read_stream("p2", "s").await?;
    assert_eq!(values, vec!["a".to_string(), "b".to_string()]);
    assert!(closed, "an inactive producer ends the stream");
    Ok(())
}

/// A snapshot read returns immediately with whatever is available from an
/// offset, without waiting for the stream to close.
#[tokio::test]
async fn snapshot_reads_available_from_offset() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("producer", |ctx: DurableContext, _: ()| async move {
        for i in 0..3_i64 {
            ctx.write_stream("nums", i).await?;
        }
        ctx.close_stream("nums").await?;
        Ok::<_, Error>(())
    });
    engine
        .start::<_, ()>("producer", (), WorkflowOptions::with_id("p3"))
        .await?
        .result()
        .await?;

    let (head, closed): (Vec<i64>, bool) = engine.read_stream_snapshot("p3", "nums", 0).await?;
    assert_eq!(head, vec![0, 1, 2]);
    assert!(closed);

    let (tail, closed): (Vec<i64>, bool) = engine.read_stream_snapshot("p3", "nums", 2).await?;
    assert_eq!(tail, vec![2]);
    assert!(closed);
    Ok(())
}

/// Writing to a closed stream is an error, surfaced as a workflow failure.
#[tokio::test]
async fn writing_to_closed_stream_errors() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("bad", |ctx: DurableContext, _: ()| async move {
        ctx.close_stream("s").await?;
        ctx.write_stream("s", 1_i64).await?;
        Ok::<_, Error>(())
    });

    let res = engine
        .start::<_, ()>("bad", (), WorkflowOptions::with_id("p4"))
        .await?
        .result()
        .await;
    assert!(res.is_err(), "writing after close must fail");
    Ok(())
}

/// A workflow can read another workflow's stream from inside itself via
/// `ctx.read_stream`. Unlike the writes, the read is live and records no step,
/// so the consumer's step log stays empty.
#[tokio::test]
async fn workflow_reads_another_workflows_stream() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("producer", |ctx: DurableContext, _: ()| async move {
        for i in 1..=3_i64 {
            ctx.write_stream("nums", i).await?;
        }
        ctx.close_stream("nums").await?;
        Ok::<_, Error>(())
    });
    engine.register(
        "consumer",
        |ctx: DurableContext, producer_id: String| async move {
            let (values, _closed): (Vec<i64>, bool) = ctx.read_stream(&producer_id, "nums").await?;
            Ok::<_, Error>(values)
        },
    );

    engine
        .start::<_, ()>("producer", (), WorkflowOptions::with_id("prod"))
        .await?
        .result()
        .await?;

    let values: Vec<i64> = engine
        .start::<_, Vec<i64>>(
            "consumer",
            "prod".to_string(),
            WorkflowOptions::with_id("cons"),
        )
        .await?
        .result()
        .await?;
    assert_eq!(values, vec![1, 2, 3]);

    // The read is live, not a durable step: the consumer recorded none.
    let steps = engine.get_workflow_steps("cons").await?;
    assert!(steps.is_empty(), "ctx.read_stream records no step");
    Ok(())
}

/// `read_stream_values` yields each value as an asynchronous `Stream`, delivering
/// them in order while the producer is still running and ending once it closes.
#[tokio::test]
async fn async_stream_yields_values_incrementally() -> Result<()> {
    use durare::StreamExt;

    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("slow_producer", |ctx: DurableContext, _: ()| async move {
        ctx.write_stream("s", 1_i64).await?;
        ctx.sleep(Duration::from_millis(40)).await?;
        ctx.write_stream("s", 2_i64).await?;
        ctx.sleep(Duration::from_millis(40)).await?;
        ctx.write_stream("s", 3_i64).await?;
        ctx.close_stream("s").await?;
        Ok::<_, Error>(())
    });

    let producer = engine
        .start::<_, ()>("slow_producer", (), WorkflowOptions::with_id("p6"))
        .await?;

    // Pull from the async stream as values are committed; it ends at close.
    let mut got = Vec::new();
    let mut values = engine.read_stream_values::<i64>("p6", "s");
    while let Some(item) = values.next().await {
        got.push(item?);
    }
    assert_eq!(got, vec![1, 2, 3]);

    producer.result().await?;
    Ok(())
}

/// A reader on the async stream of a nonexistent workflow gets a single terminal
/// `Err`, then the stream ends.
#[tokio::test]
async fn async_stream_errors_on_missing_workflow() -> Result<()> {
    use durare::StreamExt;

    let engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    let mut values = engine.read_stream_values::<i64>("nope", "s");
    let first = values.next().await;
    assert!(
        matches!(first, Some(Err(_))),
        "missing workflow yields a terminal Err"
    );
    assert!(values.next().await.is_none(), "stream ends after the Err");
    Ok(())
}

/// The async stream drains every value even when the producer finishes *without*
/// closing the stream — the inactive-producer termination must still make a final
/// read pass, so a value committed just before the producer went terminal is not
/// dropped (the lost-value race the blocking `read_stream` also guards against).
#[tokio::test]
async fn async_stream_drains_when_producer_finishes_without_close() -> Result<()> {
    use durare::StreamExt;

    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("p_noclose", |ctx: DurableContext, _: ()| async move {
        ctx.write_stream("s", "a".to_string()).await?;
        ctx.write_stream("s", "b".to_string()).await?;
        Ok::<_, Error>(())
    });

    // The producer is already terminal before the reader starts, so the stream
    // ends via the inactive-producer path rather than a close sentinel.
    engine
        .start::<_, ()>("p_noclose", (), WorkflowOptions::with_id("pnc"))
        .await?
        .result()
        .await?;

    let mut got = Vec::new();
    let mut values = engine.read_stream_values::<String>("pnc", "s");
    while let Some(item) = values.next().await {
        got.push(item?);
    }
    assert_eq!(
        got,
        vec!["a".to_string(), "b".to_string()],
        "no value dropped when the producer goes inactive without closing"
    );
    Ok(())
}

/// `read_stream` drains values as the producer writes them: the reader starts
/// while the producer is still running and blocks until the close arrives.
#[tokio::test]
async fn read_drains_while_producer_runs() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("slow_producer", |ctx: DurableContext, _: ()| async move {
        ctx.write_stream("s", 1_i64).await?;
        ctx.sleep(Duration::from_millis(50)).await?;
        ctx.write_stream("s", 2_i64).await?;
        ctx.close_stream("s").await?;
        Ok::<_, Error>(())
    });

    // Start the producer in the background, then block draining its stream.
    let producer = engine
        .start::<_, ()>("slow_producer", (), WorkflowOptions::with_id("p5"))
        .await?;
    let (values, closed): (Vec<i64>, bool) = engine.read_stream("p5", "s").await?;
    assert_eq!(values, vec![1, 2]);
    assert!(closed);
    producer.result().await?;
    Ok(())
}