durare 0.3.1

//! Queue & dispatch tests, backend-free via the in-memory provider:
//! basic enqueue→dispatch, worker concurrency, priority ordering, delayed
//! enqueue, deduplication, and rate limiting.

use durare::{
    DurableContext, DurableEngine, Error, ErrorCode, InMemoryProvider, ListFilter, RateLimiter,
    Result, StateProvider, WorkflowOptions, WorkflowQueue, STATUS_DELAYED, STATUS_ENQUEUED,
};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

/// A queue that polls fast enough for tests.
fn test_queue(name: &str) -> WorkflowQueue {
    WorkflowQueue::new(name).base_polling_interval(Duration::from_millis(10))
}

#[tokio::test]
async fn enqueue_dispatches_and_completes() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("add_one", |_ctx: DurableContext, n: i64| async move {
        Ok::<_, Error>(n + 1)
    });
    engine.register_queue(test_queue("q"));
    engine.launch().await?;

    let handle = engine
        .start::<_, i64>("add_one", 41_i64, WorkflowOptions::default().queue("q"))
        .await?;
    assert_eq!(handle.result().await?, 42);

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

#[tokio::test]
async fn enqueue_to_unregistered_queue_errors() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("noop", |_ctx: DurableContext, _: ()| async move {
        Ok::<_, Error>(())
    });
    let res = engine
        .start::<_, ()>("noop", (), WorkflowOptions::default().queue("nope"))
        .await;
    assert!(matches!(res, Err(Error::UnknownQueue(ref q)) if q == "nope"));
    Ok(())
}

/// With worker_concurrency(1), at most one workflow from the queue runs at a
/// time on this executor.
#[tokio::test]
async fn worker_concurrency_is_enforced() -> Result<()> {
    static RUNNING: AtomicUsize = AtomicUsize::new(0);
    static MAX_SEEN: AtomicUsize = AtomicUsize::new(0);

    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("tracked", |_ctx: DurableContext, _: ()| async move {
        let now = RUNNING.fetch_add(1, Ordering::SeqCst) + 1;
        MAX_SEEN.fetch_max(now, Ordering::SeqCst);
        tokio::time::sleep(Duration::from_millis(40)).await;
        RUNNING.fetch_sub(1, Ordering::SeqCst);
        Ok::<_, Error>(())
    });
    engine.register_queue(test_queue("serial").worker_concurrency(1));
    engine.launch().await?;

    let mut handles = Vec::new();
    for i in 0..3 {
        handles.push(
            engine
                .start::<_, ()>(
                    "tracked",
                    (),
                    WorkflowOptions::with_id(format!("wf-conc-{i}")).queue("serial"),
                )
                .await?,
        );
    }
    for h in &mut handles {
        h.result().await?;
    }

    assert_eq!(
        MAX_SEEN.load(Ordering::SeqCst),
        1,
        "worker_concurrency(1) must serialize queue workflows"
    );
    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// On a priority-enabled serialized queue, lower priority values run first
/// regardless of enqueue order.
#[tokio::test]
async fn priority_orders_execution() -> Result<()> {
    let order: Arc<tokio::sync::Mutex<Vec<i64>>> = Arc::new(tokio::sync::Mutex::new(Vec::new()));

    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    let order_wf = order.clone();
    engine.register("record", move |_ctx: DurableContext, n: i64| {
        let order = order_wf.clone();
        async move {
            order.lock().await.push(n);
            Ok::<_, Error>(n)
        }
    });
    engine.register_queue(test_queue("prio").worker_concurrency(1).priority_enabled());

    // Enqueue before launch so all three are pending when dispatch begins.
    for (i, prio) in [(0, 3), (1, 1), (2, 2)] {
        let mut opts = WorkflowOptions::with_id(format!("wf-prio-{i}"));
        opts.priority = prio;
        let _ = engine
            .start::<_, i64>("record", prio as i64, opts.queue("prio"))
            .await?;
    }
    engine.launch().await?;

    // Wait until all three have run.
    let deadline = Instant::now() + Duration::from_secs(3);
    while order.lock().await.len() < 3 {
        assert!(Instant::now() < deadline, "queue did not drain in time");
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    assert_eq!(*order.lock().await, vec![1, 2, 3]);

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// A delayed enqueue parks in DELAYED, then runs once the delay expires.
#[tokio::test]
async fn delayed_enqueue_waits_then_runs() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("echo", |_ctx: DurableContext, n: i64| async move {
        Ok::<_, Error>(n)
    });
    engine.register_queue(test_queue("later"));
    engine.launch().await?;

    let started = Instant::now();
    let mut opts = WorkflowOptions::with_id("wf-delayed");
    opts.delay = Some(Duration::from_millis(150));
    let handle = engine
        .start::<_, i64>("echo", 7_i64, opts.queue("later"))
        .await?;

    assert_eq!(handle.get_status().await?.status, STATUS_DELAYED);
    assert_eq!(handle.result().await?, 7);
    assert!(
        started.elapsed() >= Duration::from_millis(120),
        "workflow must not run before its delay expires"
    );

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// set_workflow_delay reschedules a DELAYED workflow: a far-future delay is
/// shortened so the workflow runs almost immediately. Non-DELAYED workflows are
/// a no-op (returns false, no error).
#[tokio::test]
async fn set_workflow_delay_reschedules() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("echo", |_ctx: DurableContext, n: i64| async move {
        Ok::<_, Error>(n)
    });
    engine.register_queue(test_queue("resched"));
    engine.launch().await?;

    // Enqueue with a 60s delay so it would never run during the test...
    let mut opts = WorkflowOptions::with_id("wf-resched");
    opts.delay = Some(Duration::from_secs(60));
    let handle = engine
        .start::<_, i64>("echo", 9_i64, opts.queue("resched"))
        .await?;
    assert_eq!(handle.get_status().await?.status, STATUS_DELAYED);

    // ...then pull it forward to ~20ms from now.
    let started = Instant::now();
    assert!(
        engine
            .set_workflow_delay("wf-resched", Duration::from_millis(20))
            .await?,
        "rescheduling a DELAYED workflow must report a match"
    );
    assert_eq!(handle.result().await?, 9);
    assert!(
        started.elapsed() < Duration::from_secs(5),
        "the workflow must run on the shortened delay, not the original 60s"
    );

    // The now-completed workflow is no longer DELAYED: a further reschedule is a
    // silent no-op. A missing id likewise.
    assert!(
        !engine
            .set_workflow_delay("wf-resched", Duration::ZERO)
            .await?
    );
    assert!(!engine.set_workflow_delay("ghost", Duration::ZERO).await?);

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// Delay without a queue is rejected.
#[tokio::test]
async fn delay_requires_queue() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("noop", |_ctx: DurableContext, _: ()| async move {
        Ok::<_, Error>(())
    });
    let opts = WorkflowOptions {
        delay: Some(Duration::from_millis(10)),
        ..Default::default()
    };
    let res = engine.start::<_, ()>("noop", (), opts).await;
    assert!(res.is_err());
    Ok(())
}

/// Two different workflow ids with the same deduplication id on one queue:
/// the second enqueue is rejected.
#[tokio::test]
async fn dedup_id_rejects_duplicates() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("noop", |_ctx: DurableContext, _: ()| async move {
        Ok::<_, Error>(())
    });
    engine.register_queue(test_queue("dedup"));

    let mut opts = WorkflowOptions::with_id("wf-dedup-1");
    opts.dedup_id = Some("once".to_string());
    let _first = engine
        .start::<_, ()>("noop", (), opts.queue("dedup"))
        .await?;

    let mut opts = WorkflowOptions::with_id("wf-dedup-2");
    opts.dedup_id = Some("once".to_string());
    let err = match engine.start::<_, ()>("noop", (), opts.queue("dedup")).await {
        Ok(_) => panic!("same dedup id on the same queue must be rejected"),
        Err(e) => e,
    };
    assert_eq!(err.code(), ErrorCode::QueueDeduplicated);
    Ok(())
}

/// With `ReturnExisting`, a colliding deduplication id returns a handle to the
/// workflow already holding the slot instead of erroring; a non-default policy
/// without a dedup id is rejected.
#[tokio::test]
async fn dedup_return_existing_returns_the_holder() -> Result<()> {
    use durare::DeduplicationPolicy;
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("noop", |_ctx: DurableContext, _: ()| async move {
        Ok::<_, Error>(())
    });
    engine.register_queue(test_queue("dedup"));

    let first = engine
        .start::<_, ()>(
            "noop",
            (),
            WorkflowOptions::with_id("wf-1")
                .dedup_id("once")
                .queue("dedup"),
        )
        .await?;

    let again = engine
        .start::<_, ()>(
            "noop",
            (),
            WorkflowOptions::with_id("wf-2")
                .dedup_id("once")
                .dedup_policy(DeduplicationPolicy::ReturnExisting)
                .queue("dedup"),
        )
        .await?;
    assert_eq!(again.id(), first.id(), "returned the slot holder");
    assert_eq!(again.id(), "wf-1");

    assert!(
        engine
            .start::<_, ()>(
                "noop",
                (),
                WorkflowOptions::with_id("wf-3")
                    .dedup_policy(DeduplicationPolicy::ReturnExisting)
                    .queue("dedup")
            )
            .await
            .is_err(),
        "a non-default policy requires a dedup id"
    );
    Ok(())
}

/// Sending to a workflow id that does not exist is a typed, classifiable error.
#[tokio::test]
async fn send_to_nonexistent_workflow_is_typed() -> Result<()> {
    let engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    let err = engine
        .send("no-such-workflow", "hi", "topic")
        .await
        .expect_err("sending to a nonexistent workflow must fail");
    assert_eq!(err.code(), ErrorCode::NonExistentWorkflow);
    Ok(())
}

/// With a rate limit of 2 per long period, only 2 of 4 enqueued workflows
/// start; the rest stay ENQUEUED.
#[tokio::test]
async fn rate_limit_caps_starts() -> Result<()> {
    static STARTED: AtomicUsize = AtomicUsize::new(0);

    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("counted", |_ctx: DurableContext, _: ()| async move {
        STARTED.fetch_add(1, Ordering::SeqCst);
        Ok::<_, Error>(())
    });
    engine.register_queue(test_queue("limited").rate_limiter(RateLimiter {
        limit: 2,
        period: Duration::from_secs(60),
    }));

    let mut handles = Vec::new();
    for i in 0..4 {
        handles.push(
            engine
                .start::<_, ()>(
                    "counted",
                    (),
                    WorkflowOptions::with_id(format!("wf-rate-{i}")).queue("limited"),
                )
                .await?,
        );
    }
    engine.launch().await?;

    // Give the dispatcher several iterations to (not) over-dispatch.
    tokio::time::sleep(Duration::from_millis(200)).await;
    assert_eq!(
        STARTED.load(Ordering::SeqCst),
        2,
        "only `limit` workflows may start within the rate period"
    );
    let enqueued = futures_count_enqueued(&mut handles).await?;
    assert_eq!(enqueued, 2, "the overflow workflows must remain ENQUEUED");

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// The registry reports every registered queue, sorted by name, regardless of
/// which ones this process listens to.
#[tokio::test]
async fn list_registered_queues_is_sorted() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register_queue(WorkflowQueue::new("zebra").worker_concurrency(2));
    engine.register_queue(WorkflowQueue::new("alpha"));

    let names: Vec<String> = engine
        .list_registered_queues()
        .into_iter()
        .map(|q| q.name)
        .collect();
    assert_eq!(names, vec!["alpha".to_string(), "zebra".to_string()]);
    Ok(())
}

/// `listen_queues` dispatches only the named subset: an unlistened queue still
/// accepts enqueues, but nothing claims them in this process.
#[tokio::test]
async fn listen_queues_dispatches_only_listened() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("add_one", |_ctx: DurableContext, n: i64| async move {
        Ok::<_, Error>(n + 1)
    });
    engine.register_queue(test_queue("listened"));
    engine.register_queue(test_queue("ignored"));
    engine.listen_queues(["listened"]);
    engine.launch().await?;

    // The listened queue runs to completion.
    let run = engine
        .start::<_, i64>(
            "add_one",
            41_i64,
            WorkflowOptions::default().queue("listened"),
        )
        .await?;
    assert_eq!(run.result().await?, 42);

    // The ignored queue accepts the enqueue but never dispatches it here.
    let idle = engine
        .start::<_, i64>(
            "add_one",
            1_i64,
            WorkflowOptions::default().queue("ignored"),
        )
        .await?;
    tokio::time::sleep(Duration::from_millis(100)).await;
    assert_eq!(
        idle.get_status().await?.status,
        STATUS_ENQUEUED,
        "an unlistened queue is not dispatched by this process"
    );

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// `ListFilter::queues_only` returns only workflows that are on a queue,
/// excluding directly-run ones.
#[tokio::test]
async fn queues_only_filters_to_queued_workflows() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("noop", |_ctx: DurableContext, _: ()| async move {
        Ok::<_, Error>(())
    });
    engine.register_queue(test_queue("q"));
    engine.launch().await?;

    // One direct run, one enqueued.
    engine
        .start::<_, ()>("noop", (), WorkflowOptions::with_id("direct"))
        .await?
        .result()
        .await?;
    engine
        .start::<_, ()>("noop", (), WorkflowOptions::with_id("queued").queue("q"))
        .await?
        .result()
        .await?;

    let queued: Vec<String> = engine
        .list_workflows(&ListFilter {
            queues_only: true,
            ..Default::default()
        })
        .await?
        .into_iter()
        .map(|w| w.id)
        .collect();
    assert_eq!(queued, vec!["queued".to_string()]);

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// A partitioned queue applies worker concurrency *per partition*: with a limit
/// of 1 and two active partitions, one workflow from each runs at once, so two
/// run concurrently overall.
#[tokio::test]
async fn partitioned_queue_concurrency_is_per_partition() -> Result<()> {
    static CURRENT: AtomicUsize = AtomicUsize::new(0);
    static PEAK: AtomicUsize = AtomicUsize::new(0);

    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("work", |ctx: DurableContext, _: ()| async move {
        let now = CURRENT.fetch_add(1, Ordering::SeqCst) + 1;
        PEAK.fetch_max(now, Ordering::SeqCst);
        ctx.sleep(Duration::from_millis(80)).await?;
        CURRENT.fetch_sub(1, Ordering::SeqCst);
        Ok::<_, Error>(())
    });
    engine.register_queue(test_queue("pq").partitioned().worker_concurrency(1));
    engine.launch().await?;

    // Two workflows in each of two partitions.
    let mut handles = Vec::new();
    for part in ["a", "a", "b", "b"] {
        handles.push(
            engine
                .start::<_, ()>(
                    "work",
                    (),
                    WorkflowOptions::default().partition_key(part).queue("pq"),
                )
                .await?,
        );
    }
    for h in &mut handles {
        h.result().await?;
    }

    assert_eq!(
        PEAK.load(Ordering::SeqCst),
        2,
        "each partition runs one at a time, but the two partitions run in parallel"
    );

    engine.shutdown(Duration::from_secs(2)).await?;
    Ok(())
}

/// A workflow enqueued to a partitioned queue without a partition key is never
/// dispatched (partition discovery only sees keyed rows).
#[tokio::test]
async fn partitioned_queue_ignores_keyless_enqueue() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("noop", |_ctx: DurableContext, _: ()| async move {
        Ok::<_, Error>(())
    });
    engine.register_queue(test_queue("pq").partitioned());
    engine.launch().await?;

    let idle = engine
        .start::<_, ()>("noop", (), WorkflowOptions::default().queue("pq"))
        .await?;
    tokio::time::sleep(Duration::from_millis(80)).await;
    assert_eq!(
        idle.get_status().await?.status,
        STATUS_ENQUEUED,
        "no partition key means nothing dispatches it"
    );

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// launch() persists the registered queues into the database-backed registry
/// (the `queues` table), readable via list_queues() — the fleet-wide view the
/// conductor uses, distinct from the in-process list_registered_queues(). The
/// always-present internal queue stays unsurfaced.
#[tokio::test]
async fn launch_persists_the_queue_registry() -> Result<()> {
    let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
    engine.register("noop", |_ctx: DurableContext, _: ()| async move {
        Ok::<_, Error>(())
    });
    engine.register_queue(
        test_queue("emails")
            .worker_concurrency(4)
            .global_concurrency(10)
            .rate_limiter(RateLimiter {
                limit: 50,
                period: Duration::from_secs(60),
            }),
    );
    engine.register_queue(test_queue("billing").priority_enabled());
    engine.launch().await?;

    let queues = engine.list_queues().await?;
    // Sorted by name; the internal queue is not surfaced.
    let names: Vec<&str> = queues.iter().map(|q| q.name.as_str()).collect();
    assert_eq!(names, ["billing", "emails"]);

    let emails = queues.iter().find(|q| q.name == "emails").unwrap();
    assert_eq!(emails.worker_concurrency, Some(4));
    assert_eq!(emails.global_concurrency, Some(10));
    let rl = emails.rate_limit.as_ref().expect("rate limit persisted");
    assert_eq!(rl.limit, 50);
    assert_eq!(rl.period, Duration::from_secs(60));

    let billing = queues.iter().find(|q| q.name == "billing").unwrap();
    assert!(billing.priority_enabled);
    assert!(billing.rate_limit.is_none());

    engine.shutdown(Duration::from_secs(1)).await?;
    Ok(())
}

/// The launch-time conflict-policy gate resolved from the application version:
/// a process self-elects as latest when it first registers its version, so its
/// queue config lands on the first launch — but a straggler running an
/// already-registered older version (with a newer version present) must not
/// clobber the newer config. Pins the deliberate divergence from Go's ordering
/// (see PARITY.md); Go resolves the gate before self-registration.
#[tokio::test]
async fn launch_gate_self_elects_then_stragglers_do_not_clobber() -> Result<()> {
    let provider = Arc::new(InMemoryProvider::new());
    let noop = |_ctx: DurableContext, _: ()| async move { Ok::<_, Error>(()) };

    // v-old launches first: registers itself (latest so far) and writes q=1.
    {
        let mut e = DurableEngine::new_with_version(provider.clone(), "v-old").await?;
        e.register("noop", noop);
        e.register_queue(test_queue("q").worker_concurrency(1));
        e.launch().await?;
        e.shutdown(Duration::from_secs(1)).await?;
    }
    assert_eq!(provider.list_queues().await?[0].worker_concurrency, Some(1));

    // v-new launches: a newer version self-elects as latest and overwrites q=2.
    {
        let mut e = DurableEngine::new_with_version(provider.clone(), "v-new").await?;
        e.register("noop", noop);
        e.register_queue(test_queue("q").worker_concurrency(2));
        e.launch().await?;
        e.shutdown(Duration::from_secs(1)).await?;
    }
    assert_eq!(
        provider.list_queues().await?[0].worker_concurrency,
        Some(2),
        "the newer version self-elected and updated the config"
    );

    // v-old relaunches as a straggler: already registered (older timestamp) and
    // v-new is latest, so the gate is false — it must not revert the config.
    {
        let mut e = DurableEngine::new_with_version(provider.clone(), "v-old").await?;
        e.register("noop", noop);
        e.register_queue(test_queue("q").worker_concurrency(1));
        e.launch().await?;
        e.shutdown(Duration::from_secs(1)).await?;
    }
    assert_eq!(
        provider.list_queues().await?[0].worker_concurrency,
        Some(2),
        "an older-version straggler did not clobber the newer config"
    );

    Ok(())
}

/// upsert_queue's conflict policy: a first write inserts; a name collision does
/// nothing unless the caller may overwrite (the engine resolves that from the
/// application version).
#[tokio::test]
async fn queue_upsert_respects_conflict_policy() -> Result<()> {
    let provider = InMemoryProvider::new();
    provider
        .upsert_queue(&WorkflowQueue::new("q").worker_concurrency(1), true)
        .await?;

    // A writer that may NOT overwrite leaves the stored config untouched.
    provider
        .upsert_queue(&WorkflowQueue::new("q").worker_concurrency(9), false)
        .await?;
    let stored = provider.list_queues().await?;
    assert_eq!(stored.len(), 1);
    assert_eq!(
        stored[0].worker_concurrency,
        Some(1),
        "no-overwrite policy kept the original config"
    );

    // A writer that MAY overwrite replaces it.
    provider
        .upsert_queue(&WorkflowQueue::new("q").worker_concurrency(9), true)
        .await?;
    let stored = provider.list_queues().await?;
    assert_eq!(stored.len(), 1, "still one row (upsert, not insert)");
    assert_eq!(stored[0].worker_concurrency, Some(9), "update overwrote it");
    Ok(())
}

async fn futures_count_enqueued(handles: &mut [durare::WorkflowHandle<()>]) -> Result<usize> {
    let mut n = 0;
    for h in handles {
        if h.get_status().await?.status == STATUS_ENQUEUED {
            n += 1;
        }
    }
    Ok(n)
}