PiCloud/crates/picloud-cli/tests/shared_queues.rs

//! §11.6 D3 — shared durable QUEUES, declarative authoring end to end via `pic`.
//! A group declares a shared `queue` collection + a `[[triggers.queue]]
//! shared = true` consumer over it; the template applies, `pic triggers ls
//! --group` shows `shared = true`; a shared queue consumer over an UNDECLARED
//! collection is rejected, and `shared = true` on an app queue is rejected.
//!
//! The live behaviour — competing per-descendant consumers claim one group
//! store exactly once (SKIP LOCKED), the dispatcher drains the group store — is
//! pinned deterministically by `manager-core/tests/group_queue.rs` +
//! `stateful_templates.rs`; driving the async dispatcher from a journey is
//! deliberately avoided (codebase norm, mirroring `shared_topics`).

use std::fs;

use tempfile::TempDir;

use crate::common;
use crate::common::cleanup::{AppGuard, GroupGuard, ScriptGuard};

fn manifest_dir() -> TempDir {
    let dir = TempDir::new().expect("tempdir");
    fs::create_dir_all(dir.path().join("scripts")).expect("scripts dir");
    dir
}

fn group_script_id(env: &common::TestEnv, group: &str, name: &str) -> String {
    let ls = common::pic_as(env)
        .args(["scripts", "ls", "--group", group])
        .output()
        .expect("scripts ls");
    let table = String::from_utf8(ls.stdout).unwrap();
    table
        .lines()
        .map(common::cells)
        .find(|c| c.get(2) == Some(&name))
        .and_then(|c| c.first().map(|s| (*s).to_string()))
        .unwrap_or_else(|| panic!("group script `{name}` not found:\n{table}"))
}

#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
#[test]
fn shared_queue_consumer_applies_lists_and_validates() {
    let Some(fx) = common::fixture_or_skip() else {
        return;
    };
    let env = common::admin_env(fx);
    let group = common::unique_slug("shq-grp");

    let _g = GroupGuard::new(&env.url, &env.token, &group);
    common::pic_as(&env)
        .args(["groups", "create", &group])
        .assert()
        .success();

    // Group handler for the shared queue.
    let dir = manifest_dir();
    fs::write(
        dir.path().join("scripts/on-task.rhai"),
        r#"log::info("shared task"); "ok""#,
    )
    .unwrap();
    common::pic_as(&env)
        .args(["scripts", "deploy"])
        .arg(dir.path().join("scripts/on-task.rhai"))
        .args(["--group", &group, "--name", "on-task"])
        .assert()
        .success();
    let _gs = ScriptGuard::new(
        &env.url,
        &env.token,
        &group_script_id(&env, &group, "on-task"),
    );

    // Group declares a shared `queue` collection `tasks` + a shared queue
    // consumer over it.
    let gmanifest = format!(
        "[group]\nslug = \"{group}\"\nname = \"ShQG\"\n\
         collections = [{{ name = \"tasks\", kind = \"queue\" }}]\n\n\
         [[triggers.queue]]\nscript = \"on-task\"\nqueue_name = \"tasks\"\nshared = true\n"
    );
    let gpath = dir.path().join("group.toml");
    fs::write(&gpath, &gmanifest).unwrap();
    common::pic_as(&env)
        .args(["apply", "--file"])
        .arg(&gpath)
        .assert()
        .success();

    // `pic triggers ls --group` shows shared = true for the queue consumer.
    let ls = String::from_utf8(
        common::pic_as(&env)
            .args(["triggers", "ls", "--group", &group])
            .output()
            .unwrap()
            .stdout,
    )
    .unwrap();
    let row = ls
        .lines()
        .map(common::cells)
        .find(|c| c.contains(&"on-task"))
        .unwrap_or_else(|| panic!("no on-task trigger row:\n{ls}"));
    assert!(
        row.contains(&"true"),
        "the shared column must read true for the shared queue consumer:\n{ls}"
    );

    // A shared queue consumer over a queue the group does NOT declare shared is
    // rejected at apply.
    let bad_group = format!(
        "[group]\nslug = \"{group}\"\nname = \"ShQG\"\n\
         collections = [{{ name = \"tasks\", kind = \"queue\" }}]\n\n\
         [[triggers.queue]]\nscript = \"on-task\"\nqueue_name = \"undeclared\"\nshared = true\n"
    );
    fs::write(&gpath, &bad_group).unwrap();
    let out = common::pic_as(&env)
        .args(["apply", "--file"])
        .arg(&gpath)
        .output()
        .expect("apply bad");
    assert!(
        !out.status.success(),
        "a shared queue consumer over an undeclared queue must be rejected"
    );

    // `shared = true` on an APP queue trigger is rejected.
    let app = common::unique_slug("shq-app");
    let _a = AppGuard::new(&env.url, &env.token, &app);
    common::pic_as(&env)
        .args(["apps", "create", &app])
        .assert()
        .success();
    fs::write(
        dir.path().join("scripts/app-h.rhai"),
        r#"log::info("app"); "ok""#,
    )
    .unwrap();
    let amanifest = format!(
        "[app]\nslug = \"{app}\"\nname = \"App\"\n\n\
         [[scripts]]\nname = \"app-h\"\nfile = \"scripts/app-h.rhai\"\n\n\
         [[triggers.queue]]\nscript = \"app-h\"\nqueue_name = \"tasks\"\nshared = true\n"
    );
    let apath = dir.path().join("app.toml");
    fs::write(&apath, &amanifest).unwrap();
    let out = common::pic_as(&env)
        .args(["apply", "--file"])
        .arg(&apath)
        .output()
        .expect("apply app");
    assert!(
        !out.status.success(),
        "a shared queue consumer on an app is rejected (apps don't own shared queues)"
    );
    let err = String::from_utf8_lossy(&out.stderr);
    assert!(
        err.to_lowercase().contains("shared"),
        "the rejection must mention shared:\n{err}"
    );
}