windmill/backend/windmill-api-integration-tests/tests/datatable_roles.rs

//! Who may connect to a data table as which role, across the two shapes an entry can take: one
//! that owns its database, and a fork's pointer at it.

use serde_json::{json, Value};
use sqlx::{Pool, Postgres};

use windmill_test_utils::*;

fn client() -> reqwest::Client {
    reqwest::Client::new()
}

fn authed(builder: reqwest::RequestBuilder, token: &str) -> reqwest::RequestBuilder {
    builder.header("Authorization", format!("Bearer {token}"))
}

/// The `analytics` role's tenant list as stored, so a cascade can be observed directly.
async fn tenants(db: &Pool<Postgres>, w_id: &str) -> Vec<String> {
    let value: Option<Value> = sqlx::query_scalar(
        "SELECT datatable->'datatables'->'main'->'permissions'->'roles'->'role1'->'tenants'
         FROM workspace_settings WHERE workspace_id = $1",
    )
    .bind(w_id)
    .fetch_one(db)
    .await
    .unwrap();
    serde_json::from_value(value.unwrap_or(json!([]))).unwrap()
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn freeing_a_principal_takes_its_datatable_tenant(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let base = format!("http://localhost:{port}/api/w/test-workspace");

    assert_eq!(
        tenants(&db, "test-workspace").await,
        vec!["u/test-user-2", "g/analysts", "f/finance"]
    );

    let resp = authed(
        client().delete(format!("{base}/groups/delete/analysts")),
        "SECRET_TOKEN",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "delete group: {}", resp.text().await?);

    let resp = authed(
        client().delete(format!("{base}/folders/delete/finance")),
        "SECRET_TOKEN",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "delete folder: {}", resp.text().await?);

    let resp = authed(
        client().delete(format!("{base}/users/delete/test-user-2")),
        "SECRET_TOKEN",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "delete user: {}", resp.text().await?);

    // Leaving is the other way a membership ends, and there are two `/leave` routes — the one the
    // UI and the generated client call is this one. A tenant left behind here comes back with the
    // person on rejoin, or attaches to whoever takes the username next.
    sqlx::query(
        r#"UPDATE workspace_settings SET datatable = jsonb_set(datatable,
             '{datatables,main,permissions,roles,role1,tenants}', '["u/test-user-3"]'::jsonb)
           WHERE workspace_id = 'test-workspace'"#,
    )
    .execute(&db)
    .await?;
    let resp = authed(
        client().post(format!("{base}/workspaces/leave")),
        "SECRET_TOKEN_3",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "leave: {}", resp.text().await?);
    // Nothing left naming a principal that no longer exists: a later group or account reusing one
    // of those names must not inherit the access this one had.
    assert!(
        tenants(&db, "test-workspace").await.is_empty(),
        "leaving kept the tenant: {:?}",
        tenants(&db, "test-workspace").await
    );
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_fork_uses_the_data_table_it_points_at_but_never_administers_it(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let fork = format!("http://localhost:{port}/api/w/wm-fork-dt/workspaces");

    // `test-user-2` is an admin of the fork and a plain member of the parent. The roles they can
    // use are the ones the parent's tenants give them there, not what their fork admin bit says.
    let resp = authed(
        client().get(format!("{fork}/datatable_usable_roles/main")),
        "SECRET_TOKEN_2",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200);
    let body: Value = resp.json().await?;
    assert_eq!(body["roles"], json!(["analytics"]), "{body}");
    assert_eq!(body["default_role"], "analytics");

    // The drawer names the workspace that decides, and refuses to let the fork edit it.
    let resp = authed(
        client().get(format!("{fork}/datatable_permissions/main")),
        "SECRET_TOKEN_2",
    )
    .send()
    .await?;
    let body: Value = resp.json().await?;
    assert_eq!(body["governing_workspace_id"], "test-workspace");
    assert_eq!(body["editable"], false, "{body}");

    let resp = authed(
        client().post(format!("{fork}/datatable_permissions/main")),
        "SECRET_TOKEN_2",
    )
    .json(&json!({"permissioned": true, "default_role": "admin",
                  "roles": [{"id": "admin", "tenants": ["*"]}]}))
    .send()
    .await?;
    assert_eq!(
        resp.status(),
        401,
        "a fork admin widened the parent's access"
    );

    // Nor by saving the settings form: the pointer is server-owned, so a payload naming the
    // parent's database leaves the entry exactly as it was.
    let resp = authed(
        client().post(format!("{fork}/edit_datatable_config")),
        "SECRET_TOKEN_2",
    )
    .json(&json!({
        "settings": {"datatables": {"main": {
            "database": {"resource_type": "instance", "resource_path": "dt_main"}
        }}},
        "renames": [], "deleted_datatables": []
    }))
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);

    let entry: Option<Value> = sqlx::query_scalar(
        "SELECT datatable->'datatables'->'main' FROM workspace_settings WHERE workspace_id = $1",
    )
    .bind("wm-fork-dt")
    .fetch_one(&db)
    .await?;
    let entry = entry.unwrap();
    assert_eq!(
        entry["reference"]["workspace_id"], "test-workspace",
        "{entry}"
    );
    assert!(entry["database"].is_null(), "{entry}");
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_second_entry_on_the_same_database_is_reported_rather_than_governed(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    // A copy of the parent's entry, as a fork created before data table roles would hold. It keeps
    // its own access, so the owner is told about it instead of being told it is covered.
    sqlx::query(
        r#"UPDATE workspace_settings SET datatable = '{"datatables": {"copy": {
            "database": {"resource_type": "instance", "resource_path": "dt_main"}}}}'::jsonb
           WHERE workspace_id = 'wm-fork-dt'"#,
    )
    .execute(&db)
    .await?;

    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let resp = authed(
        client().get(format!(
            "http://localhost:{port}/api/w/test-workspace/workspaces/datatable_permissions/main"
        )),
        "SECRET_TOKEN",
    )
    .send()
    .await?;
    let body: Value = resp.json().await?;
    assert_eq!(
        body["ungoverned_reachers"],
        json!([{"workspace_id": "wm-fork-dt", "datatable": "copy"}]),
        "{body}"
    );
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_resource_backed_data_table_cannot_be_put_under_roles(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    // A role is a login on Windmill's own cluster. A resource-backed data table dials a host the
    // workspace admin chose, so accepting one here would hand that host a real cluster credential.
    sqlx::query(
        r#"UPDATE workspace_settings SET datatable = '{"datatables": {"byo": {
            "database": {"resource_type": "postgresql", "resource_path": "u/test-user/pg"}}}}'::jsonb
           WHERE workspace_id = 'test-workspace'"#,
    )
    .execute(&db)
    .await?;

    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let base = format!("http://localhost:{port}/api/w/test-workspace/workspaces");

    let resp = authed(
        client().get(format!("{base}/datatable_permissions/byo")),
        "SECRET_TOKEN",
    )
    .send()
    .await?;
    let body: Value = resp.json().await?;
    assert_eq!(body["supported"], false, "{body}");

    let resp = authed(
        client().post(format!("{base}/datatable_permissions/byo")),
        "SECRET_TOKEN",
    )
    .json(&json!({"permissioned": true, "default_role": "role1",
                  "roles": [{"id": "role1", "tenants": ["*"]}]}))
    .send()
    .await?;
    assert_eq!(resp.status(), 400, "{}", resp.text().await?);
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_fork_renaming_its_own_entry_leaves_the_governing_bookkeeping_alone(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    // The parent's migration definitions. A rename or delete through the fork's settings form
    // resolves through the pointer, so without a guard it would relabel or wipe these.
    sqlx::query(
        "INSERT INTO datatable_migrations (workspace_id, datatable, timestamp, name, code_up)
         VALUES ('test-workspace', 'main', 1, 'init', 'SELECT 1')",
    )
    .execute(&db)
    .await?;

    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let resp = authed(
        client().post(format!(
            "http://localhost:{port}/api/w/wm-fork-dt/workspaces/edit_datatable_config"
        )),
        "SECRET_TOKEN_2",
    )
    .json(&json!({
        "settings": {"datatables": {}},
        "renames": [],
        "deleted_datatables": ["main"]
    }))
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);

    let left: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM datatable_migrations WHERE workspace_id = 'test-workspace'",
    )
    .fetch_one(&db)
    .await?;
    assert_eq!(left, 1, "the fork's delete reached the parent's migrations");
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_caller_who_is_not_a_member_of_the_governing_workspace_reaches_nothing(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    // A fork member who was never added to the parent. Their fork membership says nothing there,
    // and the email lookup that would evaluate them as a member of it finds no row.
    sqlx::query(
        "INSERT INTO usr (workspace_id, email, username, is_admin, role)
         VALUES ('wm-fork-dt', 'test3@windmill.dev', 'test-user-3', false, 'User')",
    )
    .execute(&db)
    .await?;

    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let resp = authed(
        client().get(format!(
            "http://localhost:{port}/api/w/wm-fork-dt/workspaces/datatable_usable_roles/main"
        )),
        "SECRET_TOKEN_3",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200);
    let body: Value = resp.json().await?;
    assert_eq!(body["roles"], json!([]), "{body}");
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_caller_with_no_identity_reaches_a_permissioned_data_table_not_at_all(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    use windmill_common::workspaces::{get_datatable_resource_from_db, DatatableAccess};

    initialize_tracing().await;
    // The compatibility story for an agent worker that predates data table roles and sends no job
    // id: it keeps resolving an unpermissioned data table, and is refused on a permissioned one
    // rather than handed an unattributed admin connection.
    let refused = get_datatable_resource_from_db(
        &db,
        "test-workspace",
        "main",
        None,
        DatatableAccess::NoIdentity,
    )
    .await;
    assert!(refused.is_err(), "an unidentified caller was let in");

    sqlx::query(
        "UPDATE workspace_settings
         SET datatable = datatable #- '{datatables,main,permissions}'
         WHERE workspace_id = 'test-workspace'",
    )
    .execute(&db)
    .await?;
    let resolved = get_datatable_resource_from_db(
        &db,
        "test-workspace",
        "main",
        None,
        DatatableAccess::NoIdentity,
    )
    .await?;
    assert_eq!(resolved["dbname"], "dt_main", "{resolved}");
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn concurrent_role_creations_both_survive(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    // Postgres roles are cluster-wide and this cluster is shared with every other test database,
    // so the names have to be unique to this run.
    let suffix: String = uuid::Uuid::new_v4().simple().to_string()[..8].to_string();
    let names = [format!("wmtest_a_{suffix}"), format!("wmtest_b_{suffix}")];

    // The cluster DDL is not visible to another transaction until commit, so without the lock both
    // of these pass their `pg_roles` existence check and one loses — leaving a live cluster login
    // the catalog never recorded.
    let create = |name: String| async move {
        let resp = authed(
            client().post(format!(
                "http://localhost:{port}/api/settings/datatable_roles"
            )),
            "SECRET_TOKEN",
        )
        .json(&json!({ "name": name }))
        .send()
        .await?;
        let status = resp.status();
        let body = resp.text().await?;
        Ok::<_, anyhow::Error>((status, body))
    };
    let outcome = async {
        let (a, b) = tokio::join!(create(names[0].clone()), create(names[1].clone()));
        let (a, b) = (a?, b?);
        assert_eq!(a.0, 200, "{}", a.1);
        assert_eq!(b.0, 200, "{}", b.1);

        let catalog = windmill_common::datatable_roles::read_role_catalog(&db).await?;
        let recorded: Vec<&str> = catalog.values().map(|r| r.name.as_str()).collect();
        for name in &names {
            assert!(
                recorded.contains(&name.as_str()),
                "{name} is a live cluster login the catalog forgot: {recorded:?}"
            );
        }
        Ok::<_, anyhow::Error>(())
    }
    .await;

    // Roles are cluster-wide, so they outlive this test's throwaway database. Dropped whatever
    // happened above — a failing run is exactly the one that created them and did not record them.
    for name in &names {
        let _ = sqlx::query(&format!("DROP ROLE IF EXISTS \"{name}\""))
            .execute(&db)
            .await;
    }
    outcome
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn renaming_a_governing_data_table_carries_its_forks(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    // A pointer names the governing data table by name, so a rename that does not follow leaves
    // every fork resolving to nothing — the data table vanishes from their pickers and their jobs
    // stop, with nothing in the renaming workspace to suggest why.
    let resp = authed(
        client().post(format!(
            "http://localhost:{port}/api/w/test-workspace/workspaces/edit_datatable_config"
        )),
        "SECRET_TOKEN",
    )
    .json(&json!({
        "settings": {"datatables": {"renamed": {
            "database": {"resource_type": "instance", "resource_path": "dt_main"}
        }}},
        "renames": [{"from": "main", "to": "renamed"}],
        "deleted_datatables": []
    }))
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);

    let entry: Option<Value> = sqlx::query_scalar(
        "SELECT datatable->'datatables'->'main' FROM workspace_settings WHERE workspace_id = $1",
    )
    .bind("wm-fork-dt")
    .fetch_one(&db)
    .await?;
    let entry = entry.unwrap();
    assert_eq!(entry["reference"]["datatable"], "renamed", "{entry}");

    // And it still resolves, which is the thing the fork actually cares about.
    let resp = authed(
        client().get(format!(
            "http://localhost:{port}/api/w/wm-fork-dt/workspaces/datatable_usable_roles/main"
        )),
        "SECRET_TOKEN_2",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_rename_has_to_match_the_save_it_claims_to_describe(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let url =
        format!("http://localhost:{port}/api/w/test-workspace/workspaces/edit_datatable_config");

    let instance =
        |path: &str| json!({"database": {"resource_type": "instance", "resource_path": path}});

    // Fork pointers are rewritten from the rename list, so a rename nobody performed moves every
    // fork of one data table onto another. `main` survives this save, so it was not renamed.
    let resp = authed(client().post(&url), "SECRET_TOKEN")
        .json(&json!({
            "settings": {"datatables": {"main": instance("dt_main"), "decoy": instance("dt_two")}},
            "renames": [{"from": "main", "to": "decoy"}],
            "deleted_datatables": []
        }))
        .send()
        .await?;
    assert_eq!(resp.status(), 400, "a forged rename was accepted");

    let entry: Option<Value> = sqlx::query_scalar(
        "SELECT datatable->'datatables'->'main'->'reference' FROM workspace_settings
         WHERE workspace_id = $1",
    )
    .bind("wm-fork-dt")
    .fetch_one(&db)
    .await?;
    assert_eq!(
        entry.unwrap()["datatable"],
        "main",
        "the fork was repointed anyway"
    );

    // A swap is two renames whose sources and targets cross. It cannot be done one at a time —
    // `datatables` is keyed by name — so refusing it would be a regression, and applying the two
    // in order without a temporary name would carry `main`'s pointers back to `main`.
    let resp = authed(client().post(&url), "SECRET_TOKEN")
        .json(&json!({
            "settings": {"datatables": {"main": instance("dt_two"), "other": instance("dt_main")}},
            "renames": [{"from": "main", "to": "other"}, {"from": "other", "to": "main"}],
            "deleted_datatables": []
        }))
        .send()
        .await?;
    // `other` does not exist yet, so this particular pair is still refused — the swap shape is
    // covered by the pair below, which starts from two real data tables.
    assert_eq!(resp.status(), 400, "{}", resp.text().await?);

    sqlx::query(
        r#"UPDATE workspace_settings SET datatable = jsonb_set(datatable, '{datatables,other}',
             '{"database": {"resource_type": "instance", "resource_path": "dt_two"}}'::jsonb)
           WHERE workspace_id = 'test-workspace'"#,
    )
    .execute(&db)
    .await?;

    let resp = authed(client().post(&url), "SECRET_TOKEN")
        .json(&json!({
            "settings": {"datatables": {"main": instance("dt_two"), "other": instance("dt_main")}},
            "renames": [{"from": "main", "to": "other"}, {"from": "other", "to": "main"}],
            "deleted_datatables": []
        }))
        .send()
        .await?;
    assert_eq!(
        resp.status(),
        200,
        "a swap was refused: {}",
        resp.text().await?
    );

    // The fork named `main`, which is now called `other`.
    let entry: Option<Value> = sqlx::query_scalar(
        "SELECT datatable->'datatables'->'main'->'reference' FROM workspace_settings
         WHERE workspace_id = $1",
    )
    .bind("wm-fork-dt")
    .fetch_one(&db)
    .await?;
    assert_eq!(
        entry.unwrap()["datatable"],
        "other",
        "the swap did not carry the pointer"
    );
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_data_table_under_roles_is_not_copied_into_a_fork(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    // `pg_dump` carries no roles and the restore drops ACLs, so a copy would arrive with the
    // parent's tenants and none of the grants behind them: every role but admin denied by
    // Postgres in a data table that reads as configured. Refuse the copy rather than ship that,
    // and refuse it before any data moves.
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    // Both halves of the clone: the database the copy would land in, then the copy itself. The
    // first has to refuse too, or a permissioned fork leaves an empty registered database that
    // no data table entry names and nothing collects.
    let resp = authed(
        client().post(format!(
            "http://localhost:{port}/api/w/test-workspace/workspaces/create_pg_database"
        )),
        "SECRET_TOKEN",
    )
    .json(&json!({"source": "datatable://main", "target_dbname": "wm_fork_dt_copy"}))
    .send()
    .await?;
    assert_eq!(resp.status(), 400);
    assert!(
        resp.text().await?.contains("under roles"),
        "the fork's database was created for a copy that cannot happen"
    );

    let resp = authed(
        client().post(format!(
            "http://localhost:{port}/api/w/test-workspace/workspaces/import_pg_database"
        )),
        "SECRET_TOKEN",
    )
    .json(
        &json!({"source": "datatable://main", "target": "datatable://main",
                  "fork_behavior": "schema_only"}),
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 400);
    assert!(
        resp.text().await?.contains("under roles"),
        "the copy was refused for some other reason"
    );
    Ok(())
}

/// The fork's `forked_from` for one of its entries; `None` whether it is absent or `null`.
async fn forked_from_of(db: &Pool<Postgres>, name: &str) -> Option<Value> {
    sqlx::query_scalar::<_, Option<Value>>(
        "SELECT datatable->'datatables'->$1::text->'forked_from'
         FROM workspace_settings WHERE workspace_id = 'wm-fork-dt'",
    )
    .bind(name)
    .fetch_one(db)
    .await
    .unwrap()
    .filter(|v| !v.is_null())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_clone_stamp_is_carried_but_its_schema_baseline_advances(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    // Whether an entry is a clone is what marks its database droppable, so a save can neither
    // stamp nor unstamp one. The schema baseline inside the stamp is what the fork's schema diff
    // advances after applying a change; dropping it would offer that same change again.
    sqlx::query(
        r#"UPDATE workspace_settings SET datatable = '{"datatables": {
             "clone": {"database": {"resource_type": "instance", "resource_path": "wm_fork_dt__clone"},
                       "forked_from": {"schema": {}}},
             "plain": {"database": {"resource_type": "instance", "resource_path": "dt_plain"}}}}'::jsonb
           WHERE workspace_id = 'wm-fork-dt'"#,
    )
    .execute(&db)
    .await?;
    let server = ApiServer::start(db.clone()).await?;
    let url = format!(
        "http://localhost:{}/api/w/wm-fork-dt/workspaces/edit_datatable_config",
        server.addr.port()
    );
    let clone_db = json!({"resource_type": "instance", "resource_path": "wm_fork_dt__clone"});
    let plain_db = json!({"resource_type": "instance", "resource_path": "dt_plain"});
    let baseline = json!({"schema": {"public": {"orders": {"id": "int4"}}}});

    let resp = authed(client().post(&url), "SECRET_TOKEN_2")
        .json(&json!({"settings": {"datatables": {
            "clone": {"database": clone_db, "forked_from": baseline},
            "plain": {"database": plain_db, "forked_from": {"schema": {}}}
        }}}))
        .send()
        .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
    assert_eq!(
        forked_from_of(&db, "clone").await,
        Some(baseline.clone()),
        "the schema diff's baseline did not advance"
    );
    assert_eq!(
        forked_from_of(&db, "plain").await,
        None,
        "a save stamped a clone"
    );

    let resp = authed(client().post(&url), "SECRET_TOKEN_2")
        .json(&json!({"settings": {"datatables": {
            "clone": {"database": clone_db}, "plain": {"database": plain_db}
        }}}))
        .send()
        .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
    assert_eq!(
        forked_from_of(&db, "clone").await,
        Some(baseline),
        "a save unstamped a clone"
    );
    Ok(())
}

/// Live replication listeners on `main`: a fork's through its pointer, the governing workspace's
/// own (in the `?role=` form), and a fork capture — plus one on another data table, which no
/// deletion of `main` may touch. Each is held by a server, as a running stream would be.
async fn plant_live_streams(db: &Pool<Postgres>) -> anyhow::Result<()> {
    sqlx::query(
        r#"INSERT INTO postgres_trigger (path, script_path, is_flow, workspace_id, edited_by,
             postgres_resource_path, replication_slot_name, publication_name, permissioned_as,
             server_id)
           VALUES
             ('u/test-user-2/fork_stream', 'u/test-user-2/s', false, 'wm-fork-dt', 'test-user-2',
              'datatable://main', 'slot_fork', 'pub_fork', 'u/test-user-2', 'srv'),
             ('u/test-user/own_stream', 'u/test-user/s', false, 'test-workspace', 'test-user',
              'datatable://main?role=admin', 'slot_own', 'pub_own', 'u/test-user', 'srv'),
             ('u/test-user-2/unrelated', 'u/test-user-2/s', false, 'wm-fork-dt', 'test-user-2',
              'datatable://other', 'slot_other', 'pub_other', 'u/test-user-2', 'srv')"#,
    )
    .execute(db)
    .await?;
    sqlx::query(
        r#"INSERT INTO capture_config (workspace_id, path, is_flow, trigger_kind, owner, email,
             trigger_config, server_id)
           VALUES ('wm-fork-dt', 'u/test-user-2/s', false, 'postgres', 'u/test-user-2',
                   'test2@windmill.dev', '{"postgres_resource_path": "datatable://main"}', 'srv')"#,
    )
    .execute(db)
    .await?;
    Ok(())
}

/// Which server holds each planted listener; `None` is a bounced one, free for a reconnect.
async fn stream_servers(
    db: &Pool<Postgres>,
) -> anyhow::Result<std::collections::HashMap<String, Option<String>>> {
    let rows: Vec<(String, Option<String>)> = sqlx::query_as(
        "SELECT path, server_id FROM postgres_trigger
         UNION ALL
         SELECT 'capture:' || path, server_id FROM capture_config",
    )
    .fetch_all(db)
    .await?;
    Ok(rows.into_iter().collect())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn deleting_a_governing_data_table_bounces_the_streams_reading_it(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    // A replication stream keeps the connection it opened while its entry resolved. Unbounced, a
    // fork's stream reads on through a pointer that no longer resolves to anything.
    plant_live_streams(&db).await?;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    let resp = authed(
        client().post(format!(
            "http://localhost:{port}/api/w/test-workspace/workspaces/edit_datatable_config"
        )),
        "SECRET_TOKEN",
    )
    // The way the settings-sync CLI saves: the new document alone, with no deletion hint.
    .json(&json!({"settings": {"datatables": {}}}))
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);

    let servers = stream_servers(&db).await?;
    for bounced in [
        "u/test-user-2/fork_stream",
        "u/test-user/own_stream",
        "capture:u/test-user-2/s",
    ] {
        assert_eq!(
            servers[bounced], None,
            "{bounced} kept streaming: {servers:?}"
        );
    }
    assert_eq!(
        servers["u/test-user-2/unrelated"].as_deref(),
        Some("srv"),
        "a stream on another data table was bounced: {servers:?}"
    );
    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn deleting_a_governing_workspace_bounces_its_forks_streams(
    db: Pool<Postgres>,
) -> anyhow::Result<()> {
    initialize_tracing().await;
    // The governing workspace's own listeners go with it; each fork's stays behind on the
    // connection it opened while this workspace still governed the pointer.
    plant_live_streams(&db).await?;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    let resp = authed(
        client().delete(format!(
            "http://localhost:{port}/api/workspaces/delete/test-workspace"
        )),
        "SECRET_TOKEN",
    )
    .send()
    .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);

    let servers = stream_servers(&db).await?;
    for bounced in ["u/test-user-2/fork_stream", "capture:u/test-user-2/s"] {
        assert_eq!(
            servers[bounced], None,
            "{bounced} kept streaming: {servers:?}"
        );
    }
    assert_eq!(
        servers["u/test-user-2/unrelated"].as_deref(),
        Some("srv"),
        "a stream on another data table was bounced: {servers:?}"
    );
    Ok(())
}