diff --git a/backend/.sqlx/query-7300eb89029e0863241087fc10df7616db640054e804d4ca66958cad06008fdb.json b/backend/.sqlx/query-1acfeed9c7a5b1e3d2da262d338655dba6e43067a9912cc2b775830856390c5d.json similarity index 60% rename from backend/.sqlx/query-7300eb89029e0863241087fc10df7616db640054e804d4ca66958cad06008fdb.json rename to backend/.sqlx/query-1acfeed9c7a5b1e3d2da262d338655dba6e43067a9912cc2b775830856390c5d.json index cc564952cb..381a2b8ae4 100644 --- a/backend/.sqlx/query-7300eb89029e0863241087fc10df7616db640054e804d4ca66958cad06008fdb.json +++ b/backend/.sqlx/query-1acfeed9c7a5b1e3d2da262d338655dba6e43067a9912cc2b775830856390c5d.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "WITH del AS (\n DELETE FROM asset WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = $3\n )\n INSERT INTO notify_event (channel, payload)\n SELECT 'notify_asset_producer_change', $1 WHERE $3 = 'script'", + "query": "WITH del AS (\n DELETE FROM asset WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = $3\n RETURNING usage_access_type\n )\n INSERT INTO notify_event (channel, payload)\n SELECT 'notify_asset_producer_change', $1\n WHERE $3 = 'script'\n AND EXISTS (SELECT 1 FROM del WHERE usage_access_type IN ('w', 'rw'))", "describe": { "columns": [], "parameters": { @@ -23,5 +23,5 @@ }, "nullable": [] }, - "hash": "7300eb89029e0863241087fc10df7616db640054e804d4ca66958cad06008fdb" + "hash": "1acfeed9c7a5b1e3d2da262d338655dba6e43067a9912cc2b775830856390c5d" } diff --git a/backend/.sqlx/query-6a1998cb3a9a0898c0fd21c35cc41e309d7e066ee4869880c90e90dd66dc9d4d.json b/backend/.sqlx/query-1f375b37ff9f6f01972e284e84a7b2f9d2d323a3da55f20ff6671e8eba510043.json similarity index 51% rename from backend/.sqlx/query-6a1998cb3a9a0898c0fd21c35cc41e309d7e066ee4869880c90e90dd66dc9d4d.json rename to backend/.sqlx/query-1f375b37ff9f6f01972e284e84a7b2f9d2d323a3da55f20ff6671e8eba510043.json index 7f2b4a677a..152a88d082 100644 --- a/backend/.sqlx/query-6a1998cb3a9a0898c0fd21c35cc41e309d7e066ee4869880c90e90dd66dc9d4d.json +++ b/backend/.sqlx/query-1f375b37ff9f6f01972e284e84a7b2f9d2d323a3da55f20ff6671e8eba510043.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "WITH del AS (\n DELETE FROM asset WHERE workspace_id = $1 AND usage_kind = 'script'\n AND usage_path = (SELECT path FROM script WHERE hash = $2 AND workspace_id = $1)\n )\n INSERT INTO notify_event (channel, payload)\n VALUES ('notify_asset_producer_change', $1)", + "query": "WITH del AS (\n DELETE FROM asset WHERE workspace_id = $1 AND usage_kind = 'script'\n AND usage_path = (SELECT path FROM script WHERE hash = $2 AND workspace_id = $1)\n RETURNING usage_access_type\n )\n INSERT INTO notify_event (channel, payload)\n SELECT 'notify_asset_producer_change', $1\n WHERE EXISTS (SELECT 1 FROM del WHERE usage_access_type IN ('w', 'rw'))", "describe": { "columns": [], "parameters": { @@ -11,5 +11,5 @@ }, "nullable": [] }, - "hash": "6a1998cb3a9a0898c0fd21c35cc41e309d7e066ee4869880c90e90dd66dc9d4d" + "hash": "1f375b37ff9f6f01972e284e84a7b2f9d2d323a3da55f20ff6671e8eba510043" } diff --git a/backend/.sqlx/query-ab18d8765d6795eaa8035a8ac902790ff61549c5dd76e5b5cfb14b110a98abf2.json b/backend/.sqlx/query-ab18d8765d6795eaa8035a8ac902790ff61549c5dd76e5b5cfb14b110a98abf2.json new file mode 100644 index 0000000000..04131f8f88 --- /dev/null +++ b/backend/.sqlx/query-ab18d8765d6795eaa8035a8ac902790ff61549c5dd76e5b5cfb14b110a98abf2.json @@ -0,0 +1,60 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM asset\n WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = 'script'\n RETURNING kind AS \"kind!: AssetKind\", path,\n usage_access_type AS \"usage_access_type: AssetUsageAccessType\"", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "kind!: AssetKind", + "type_info": { + "Custom": { + "name": "asset_kind", + "kind": { + "Enum": [ + "s3object", + "resource", + "variable", + "ducklake", + "datatable", + "volume" + ] + } + } + } + }, + { + "ordinal": 1, + "name": "path", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "usage_access_type: AssetUsageAccessType", + "type_info": { + "Custom": { + "name": "asset_access_type", + "kind": { + "Enum": [ + "r", + "w", + "rw" + ] + } + } + } + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false, + false, + true + ] + }, + "hash": "ab18d8765d6795eaa8035a8ac902790ff61549c5dd76e5b5cfb14b110a98abf2" +} diff --git a/backend/.sqlx/query-bfb97d2f48157a1575b7b6f3e64e0d075701d541a44dd7a0b586d1f337ced5e1.json b/backend/.sqlx/query-bfb97d2f48157a1575b7b6f3e64e0d075701d541a44dd7a0b586d1f337ced5e1.json new file mode 100644 index 0000000000..a051b068a5 --- /dev/null +++ b/backend/.sqlx/query-bfb97d2f48157a1575b7b6f3e64e0d075701d541a44dd7a0b586d1f337ced5e1.json @@ -0,0 +1,63 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, columns)\n VALUES ($1, $2, $3, $4, $5, 'script', $6) ON CONFLICT DO NOTHING\n RETURNING usage_access_type AS \"usage_access_type: AssetUsageAccessType\"", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "usage_access_type: AssetUsageAccessType", + "type_info": { + "Custom": { + "name": "asset_access_type", + "kind": { + "Enum": [ + "r", + "w", + "rw" + ] + } + } + } + } + ], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + { + "Custom": { + "name": "asset_kind", + "kind": { + "Enum": [ + "s3object", + "resource", + "variable", + "ducklake", + "datatable", + "volume" + ] + } + } + }, + { + "Custom": { + "name": "asset_access_type", + "kind": { + "Enum": [ + "r", + "w", + "rw" + ] + } + } + }, + "Varchar", + "Jsonb" + ] + }, + "nullable": [ + true + ] + }, + "hash": "bfb97d2f48157a1575b7b6f3e64e0d075701d541a44dd7a0b586d1f337ced5e1" +} diff --git a/backend/.sqlx/query-a9e29764b5b9d94269e2b8aa755c71b61774c8ff8ae218d7a8d6ed0ac0169366.json b/backend/.sqlx/query-e041b20c1c4166b30ced8c6a0f50bcf0691ab2554ead6b489a8441a32fc0af82.json similarity index 64% rename from backend/.sqlx/query-a9e29764b5b9d94269e2b8aa755c71b61774c8ff8ae218d7a8d6ed0ac0169366.json rename to backend/.sqlx/query-e041b20c1c4166b30ced8c6a0f50bcf0691ab2554ead6b489a8441a32fc0af82.json index ef3968cdfa..037c796fc7 100644 --- a/backend/.sqlx/query-a9e29764b5b9d94269e2b8aa755c71b61774c8ff8ae218d7a8d6ed0ac0169366.json +++ b/backend/.sqlx/query-e041b20c1c4166b30ced8c6a0f50bcf0691ab2554ead6b489a8441a32fc0af82.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, columns)\n VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT DO NOTHING", + "query": "WITH ins AS (\n INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, columns)\n VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT DO NOTHING\n RETURNING usage_kind, usage_access_type\n )\n INSERT INTO notify_event (channel, payload)\n SELECT 'notify_asset_producer_change', $1\n FROM ins WHERE usage_kind = 'script' AND usage_access_type IN ('w', 'rw')", "describe": { "columns": [], "parameters": { @@ -52,5 +52,5 @@ }, "nullable": [] }, - "hash": "a9e29764b5b9d94269e2b8aa755c71b61774c8ff8ae218d7a8d6ed0ac0169366" + "hash": "e041b20c1c4166b30ced8c6a0f50bcf0691ab2554ead6b489a8441a32fc0af82" } diff --git a/backend/.sqlx/query-eeaad5c2284c1856cfc64ae0bc0dfc79b283c06987a6de399b6c1aaca94a2b7a.json b/backend/.sqlx/query-eeaad5c2284c1856cfc64ae0bc0dfc79b283c06987a6de399b6c1aaca94a2b7a.json new file mode 100644 index 0000000000..169fe55e62 --- /dev/null +++ b/backend/.sqlx/query-eeaad5c2284c1856cfc64ae0bc0dfc79b283c06987a6de399b6c1aaca94a2b7a.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO notify_event (channel, payload)\n VALUES ('notify_asset_producer_change', $1)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [] + }, + "hash": "eeaad5c2284c1856cfc64ae0bc0dfc79b283c06987a6de399b6c1aaca94a2b7a" +} diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index 135256f3d0..8115e2ef78 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -45,8 +45,8 @@ use windmill_dep_map::scoped_dependency_map::ScopedDependencyMap; use windmill_common::{ assets::{ clear_script_triggers, clear_static_asset_usage, clear_static_asset_usage_by_script_hash, - insert_script_trigger, insert_static_asset_usage, parse_duration_secs, - parse_pipeline_annotations, trigger_spec_to_row, AssetUsageKind, TriggerSpec, + insert_script_trigger, parse_duration_secs, parse_pipeline_annotations, + replace_static_asset_usage, trigger_spec_to_row, AssetUsageKind, TriggerSpec, }, error::{self, to_anyhow}, min_version::{MIN_VERSION_SUPPORTS_DEBOUNCING, MIN_VERSION_SUPPORTS_DEBOUNCING_V2}, @@ -1609,11 +1609,16 @@ async fn create_script_internal<'c>( ); } - clear_static_asset_usage(&mut *tx, &w_id, &script_path, AssetUsageKind::Script).await?; - for asset in effective_assets.as_ref().into_iter().flatten() { - insert_static_asset_usage(&mut *tx, &w_id, &asset, &ns.path, AssetUsageKind::Script) - .await?; - } + // Clear + reinsert this script's producer rows at script_path (== ns.path), + // invalidating the producer-writes cache once iff the write-producer set + // changed (see replace_static_asset_usage). + replace_static_asset_usage( + &mut tx, + &w_id, + &script_path, + effective_assets.as_deref().unwrap_or(&[]), + ) + .await?; // Pipeline trigger edges: wipe-and-reinsert per deploy so removing an // `// on ...` annotation drops the edge. Only Asset / Schedule produce diff --git a/backend/windmill-common/src/assets.rs b/backend/windmill-common/src/assets.rs index 4e241b4b99..2b70227454 100644 --- a/backend/windmill-common/src/assets.rs +++ b/backend/windmill-common/src/assets.rs @@ -1,4 +1,6 @@ -use sqlx::PgExecutor; +use std::collections::HashSet; + +use sqlx::{PgExecutor, Postgres, Transaction}; use crate::{error, scripts::ScriptHash}; @@ -33,9 +35,22 @@ pub async fn insert_static_asset_usage<'e>( .as_ref() .map(|cols| serde_json::to_value(cols).unwrap_or(serde_json::Value::Null)); + // Invalidate the per-workspace producer-writes cache only when this insert + // actually adds a write producer: the cache (asset_dispatch:: + // ASSET_PRODUCER_WRITES_CACHE) tracks script rows with 'w'/'rw' access, so + // a row that was a no-op (ON CONFLICT skipped), a flow usage, or read-only + // can't change it. Emitting in the same statement keeps the notify atomic + // with the insert and visible to pollers only on commit. See the matching + // delete-side guard in clear_static_asset_usage. sqlx::query!( - r#"INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, columns) - VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT DO NOTHING"#, + r#"WITH ins AS ( + INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, columns) + VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT DO NOTHING + RETURNING usage_kind, usage_access_type + ) + INSERT INTO notify_event (channel, payload) + SELECT 'notify_asset_producer_change', $1 + FROM ins WHERE usage_kind = 'script' AND usage_access_type IN ('w', 'rw')"#, workspace_id, asset.path, asset.kind as AssetKind, @@ -58,19 +73,23 @@ pub async fn clear_static_asset_usage<'e>( ) -> error::Result<()> { // Invalidate the per-workspace producer-writes cache that gates the // asset-trigger dispatch hook (windmill-queue - // asset_dispatch::ASSET_PRODUCER_WRITES_CACHE), but only for script - // usage. The notify is unconditional (not gated on rows deleted) so a - // deploy where the script *gains* its first asset still invalidates; - // emitting it in the same statement keeps it atomic with the change and - // visible to pollers only on commit. Flow usage doesn't affect the - // script producer set, so the INSERT is skipped via the `$3 = 'script'` - // guard. + // asset_dispatch::ASSET_PRODUCER_WRITES_CACHE). That cache only tracks + // script rows with 'w'/'rw' access, so emit the notify only when the + // delete actually removed such a write producer: flow usage, read-only + // usage, and deletes that matched no producer row leave the cache + // unchanged (the common case — most deploys touch no write asset). The + // matching add side lives in insert_static_asset_usage. Emitting in the + // same statement keeps the notify atomic with the delete and visible to + // pollers only on commit. sqlx::query!( r#"WITH del AS ( DELETE FROM asset WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = $3 + RETURNING usage_access_type ) INSERT INTO notify_event (channel, payload) - SELECT 'notify_asset_producer_change', $1 WHERE $3 = 'script'"#, + SELECT 'notify_asset_producer_change', $1 + WHERE $3 = 'script' + AND EXISTS (SELECT 1 FROM del WHERE usage_access_type IN ('w', 'rw'))"#, workspace_id, usage_path, usage_kind as AssetUsageKind @@ -85,16 +104,18 @@ pub async fn clear_static_asset_usage_by_script_hash<'e>( workspace_id: &str, script_hash: ScriptHash, ) -> error::Result<()> { - // Always script usage → unconditionally invalidate the producer-writes - // cache for this workspace (see clear_static_asset_usage), atomically - // with the delete. + // Always script usage → invalidate the producer-writes cache for this + // workspace, but only when the delete actually removed a write producer + // ('w'/'rw'); see clear_static_asset_usage. Atomic with the delete. sqlx::query!( r#"WITH del AS ( DELETE FROM asset WHERE workspace_id = $1 AND usage_kind = 'script' AND usage_path = (SELECT path FROM script WHERE hash = $2 AND workspace_id = $1) + RETURNING usage_access_type ) INSERT INTO notify_event (channel, payload) - VALUES ('notify_asset_producer_change', $1)"#, + SELECT 'notify_asset_producer_change', $1 + WHERE EXISTS (SELECT 1 FROM del WHERE usage_access_type IN ('w', 'rw'))"#, workspace_id, script_hash.0 ) @@ -103,6 +124,100 @@ pub async fn clear_static_asset_usage_by_script_hash<'e>( Ok(()) } +fn is_write_access(access: Option) -> bool { + matches!( + access, + Some(AssetUsageAccessType::W) | Some(AssetUsageAccessType::RW) + ) +} + +/// Clear and reinsert the full static-asset usage set of a script in one tx, +/// invalidating the producer-writes cache at most once and only on a real +/// change. The cache (asset_dispatch::ASSET_PRODUCER_WRITES_CACHE) keys a +/// workspace by the set of (kind, path) script rows with 'w'/'rw' access, so a +/// redeploy that keeps the same write producers must leave it untouched. We +/// diff the old write set (captured from the clearing DELETE's RETURNING) +/// against the new one and emit a single notify only when they differ — +/// emitting per statement, as clear/insert_static_asset_usage do, would fire +/// twice on every write-asset redeploy (the clear removes the row, the reinsert +/// adds it back). The notify rides the deploy tx, so it stays atomic with the +/// DML and visible to pollers only on commit. +/// +/// DELETE and INSERT of the same primary key cannot share a single statement +/// (both would read the pre-statement snapshot, so the reinsert's ON CONFLICT +/// would silently drop the row), which is why this clears and reinserts as +/// separate statements rather than one CTE. +/// +/// Performs no authorization itself: `tx` must already be scoped to a caller +/// authorized for `workspace_id`/`usage_path` (e.g. `user_db.begin(&authed)`, +/// which applies RLS), exactly like the sibling clear/insert helpers. +pub async fn replace_static_asset_usage( + tx: &mut Transaction<'_, Postgres>, + workspace_id: &str, + usage_path: &str, + assets: &[AssetWithAltAccessType], +) -> error::Result<()> { + let cleared = sqlx::query!( + r#"DELETE FROM asset + WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = 'script' + RETURNING kind AS "kind!: AssetKind", path, + usage_access_type AS "usage_access_type: AssetUsageAccessType""#, + workspace_id, + usage_path, + ) + .fetch_all(&mut **tx) + .await?; + let old_writes: HashSet<(AssetKind, String)> = cleared + .into_iter() + .filter(|r| is_write_access(r.usage_access_type)) + .map(|r| (r.kind, r.path)) + .collect(); + + // Build new_writes from rows actually inserted (RETURNING), not the + // requested slice: ON CONFLICT DO NOTHING is first-writer-wins, so a payload + // with duplicate (kind, path) entries at conflicting access types persists + // only the first. Since the DELETE above emptied this usage_path, every + // non-conflicting insert lands, so the inserted rows are exactly the new + // persisted set. + let mut new_writes: HashSet<(AssetKind, String)> = HashSet::new(); + for asset in assets { + let access = asset.access_type.or(asset.alt_access_type); + let columns_json = asset + .columns + .as_ref() + .map(|cols| serde_json::to_value(cols).unwrap_or(serde_json::Value::Null)); + let inserted = sqlx::query!( + r#"INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, columns) + VALUES ($1, $2, $3, $4, $5, 'script', $6) ON CONFLICT DO NOTHING + RETURNING usage_access_type AS "usage_access_type: AssetUsageAccessType""#, + workspace_id, + asset.path, + asset.kind as AssetKind, + access as Option, + usage_path, + columns_json as Option, + ) + .fetch_optional(&mut **tx) + .await?; + if let Some(row) = inserted { + if is_write_access(row.usage_access_type) { + new_writes.insert((asset.kind, asset.path.clone())); + } + } + } + + if old_writes != new_writes { + sqlx::query!( + r#"INSERT INTO notify_event (channel, payload) + VALUES ('notify_asset_producer_change', $1)"#, + workspace_id, + ) + .execute(&mut **tx) + .await?; + } + Ok(()) +} + // Wipe all pipeline trigger declarations held by the given runnable. Used at // deploy time: redeploying a script wipes its prior `// on` annotations so // removing them implicitly un-declares those edges. diff --git a/backend/windmill-common/tests/asset_producer_notify.rs b/backend/windmill-common/tests/asset_producer_notify.rs new file mode 100644 index 0000000000..d09b0114eb --- /dev/null +++ b/backend/windmill-common/tests/asset_producer_notify.rs @@ -0,0 +1,474 @@ +/*! + * Tests that the asset producer-writes cache invalidation + * (`notify_asset_producer_change`) is emitted only when a deploy actually + * changes the set of script write-producers ('w'/'rw' asset usage), not on + * every deploy. The cache (asset_dispatch::ASSET_PRODUCER_WRITES_CACHE) only + * tracks script rows with write access, so read-only usage, flow usage, and + * deploys touching no write asset must NOT emit an event. + */ + +use sqlx::{Pool, Postgres}; +use windmill_common::assets::{ + clear_static_asset_usage, clear_static_asset_usage_by_script_hash, insert_static_asset_usage, + replace_static_asset_usage, AssetKind, AssetUsageAccessType, AssetUsageKind, + AssetWithAltAccessType, +}; +use windmill_common::scripts::ScriptHash; + +const WS: &str = "test-workspace"; + +/// Run the deploy-time clear+reinsert against `usage_path` in its own tx, +/// mirroring how create_script_internal calls it. +async fn replace(db: &Pool, usage_path: &str, assets: &[AssetWithAltAccessType]) { + let mut tx = db.begin().await.expect("begin"); + replace_static_asset_usage(&mut tx, WS, usage_path, assets) + .await + .expect("replace static asset usage"); + tx.commit().await.expect("commit"); +} + +fn asset_at( + path: &str, + kind: AssetKind, + access: Option, +) -> AssetWithAltAccessType { + AssetWithAltAccessType { + path: path.to_string(), + kind, + access_type: access, + alt_access_type: None, + columns: None, + } +} + +async fn producer_notify_count(db: &Pool) -> i64 { + sqlx::query_scalar::<_, i64>( + "SELECT COUNT(*) FROM notify_event WHERE channel = 'notify_asset_producer_change'", + ) + .fetch_one(db) + .await + .expect("count notify events") +} + +async fn reset_notify(db: &Pool) { + sqlx::query("DELETE FROM notify_event WHERE channel = 'notify_asset_producer_change'") + .execute(db) + .await + .expect("reset notify events"); +} + +fn asset(access: Option) -> AssetWithAltAccessType { + AssetWithAltAccessType { + path: "u/test-user/res".to_string(), + kind: AssetKind::Resource, + access_type: access, + alt_access_type: None, + columns: None, + } +} + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn insert_write_script_asset_emits(db: Pool) { + insert_static_asset_usage( + &db, + WS, + &asset(Some(AssetUsageAccessType::W)), + "u/test-user/script", + AssetUsageKind::Script, + ) + .await + .unwrap(); + assert_eq!(producer_notify_count(&db).await, 1, "write asset must emit"); +} + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn insert_readonly_script_asset_does_not_emit(db: Pool) { + insert_static_asset_usage( + &db, + WS, + &asset(Some(AssetUsageAccessType::R)), + "u/test-user/script", + AssetUsageKind::Script, + ) + .await + .unwrap(); + assert_eq!( + producer_notify_count(&db).await, + 0, + "read-only asset is not a write producer" + ); +} + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn insert_write_flow_asset_does_not_emit(db: Pool) { + insert_static_asset_usage( + &db, + WS, + &asset(Some(AssetUsageAccessType::W)), + "u/test-user/flow", + AssetUsageKind::Flow, + ) + .await + .unwrap(); + assert_eq!( + producer_notify_count(&db).await, + 0, + "flow usage does not affect the script producer cache" + ); +} + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn clear_plain_script_does_not_emit(db: Pool) { + // No asset rows for this path: a plain script deploy must not emit. + clear_static_asset_usage(&db, WS, "u/test-user/plain", AssetUsageKind::Script) + .await + .unwrap(); + assert_eq!( + producer_notify_count(&db).await, + 0, + "plain script deploy must not emit" + ); +} + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn clear_removes_write_producer_emits(db: Pool) { + insert_static_asset_usage( + &db, + WS, + &asset(Some(AssetUsageAccessType::RW)), + "u/test-user/script", + AssetUsageKind::Script, + ) + .await + .unwrap(); + reset_notify(&db).await; // ignore the insert-side event; isolate the clear + + clear_static_asset_usage(&db, WS, "u/test-user/script", AssetUsageKind::Script) + .await + .unwrap(); + assert_eq!( + producer_notify_count(&db).await, + 1, + "removing a write producer must emit" + ); +} + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn clear_removes_only_readonly_does_not_emit(db: Pool) { + insert_static_asset_usage( + &db, + WS, + &asset(Some(AssetUsageAccessType::R)), + "u/test-user/script", + AssetUsageKind::Script, + ) + .await + .unwrap(); + reset_notify(&db).await; + + clear_static_asset_usage(&db, WS, "u/test-user/script", AssetUsageKind::Script) + .await + .unwrap(); + assert_eq!( + producer_notify_count(&db).await, + 0, + "removing only read-only usage must not emit" + ); +} + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn clear_flow_usage_does_not_emit(db: Pool) { + insert_static_asset_usage( + &db, + WS, + &asset(Some(AssetUsageAccessType::W)), + "u/test-user/flow", + AssetUsageKind::Flow, + ) + .await + .unwrap(); + reset_notify(&db).await; + + clear_static_asset_usage(&db, WS, "u/test-user/flow", AssetUsageKind::Flow) + .await + .unwrap(); + assert_eq!( + producer_notify_count(&db).await, + 0, + "clearing flow usage must not emit" + ); +} + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn clear_by_script_hash_gated_on_write(db: Pool) { + let hash = 123456789_i64; + sqlx::query( + r#"INSERT INTO script (workspace_id, hash, path, summary, description, content, + created_by, language, lock) + VALUES ($1, $2, 'u/test-user/byhash', '', '', '', 'test-user', 'deno', '')"#, + ) + .bind(WS) + .bind(hash) + .execute(&db) + .await + .unwrap(); + + // Read-only usage on that script path → clear must not emit. + insert_static_asset_usage( + &db, + WS, + &asset(Some(AssetUsageAccessType::R)), + "u/test-user/byhash", + AssetUsageKind::Script, + ) + .await + .unwrap(); + reset_notify(&db).await; + + clear_static_asset_usage_by_script_hash(&db, WS, ScriptHash(hash)) + .await + .unwrap(); + assert_eq!( + producer_notify_count(&db).await, + 0, + "by-hash clear of read-only usage must not emit" + ); + + // Now a write usage → by-hash clear must emit. + insert_static_asset_usage( + &db, + WS, + &asset(Some(AssetUsageAccessType::W)), + "u/test-user/byhash", + AssetUsageKind::Script, + ) + .await + .unwrap(); + reset_notify(&db).await; + + clear_static_asset_usage_by_script_hash(&db, WS, ScriptHash(hash)) + .await + .unwrap(); + assert_eq!( + producer_notify_count(&db).await, + 1, + "by-hash clear of write usage must emit" + ); +} + +const SP: &str = "u/test-user/script"; + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn replace_plain_deploy_does_not_emit(db: Pool) { + replace(&db, SP, &[]).await; + assert_eq!( + producer_notify_count(&db).await, + 0, + "deploy with no assets must not emit" + ); +} + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn replace_gaining_write_emits_once(db: Pool) { + replace( + &db, + SP, + &[asset_at( + "u/test-user/res", + AssetKind::Resource, + Some(AssetUsageAccessType::W), + )], + ) + .await; + assert_eq!( + producer_notify_count(&db).await, + 1, + "gaining a write producer must emit exactly once" + ); +} + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn replace_unchanged_write_set_does_not_emit(db: Pool) { + let assets = [asset_at( + "u/test-user/res", + AssetKind::Resource, + Some(AssetUsageAccessType::W), + )]; + replace(&db, SP, &assets).await; + reset_notify(&db).await; // isolate the redeploy + + // Redeploy with the identical write-producer set: clear removes the row and + // the reinsert adds it back, but the cache value is unchanged → no emit. + replace(&db, SP, &assets).await; + assert_eq!( + producer_notify_count(&db).await, + 0, + "redeploy keeping the same write producers must not emit" + ); +} + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn replace_dropping_write_emits(db: Pool) { + let assets = [asset_at( + "u/test-user/res", + AssetKind::Resource, + Some(AssetUsageAccessType::RW), + )]; + replace(&db, SP, &assets).await; + reset_notify(&db).await; + + // Redeploy with no assets: the write producer disappears → emit. + replace(&db, SP, &[]).await; + assert_eq!( + producer_notify_count(&db).await, + 1, + "dropping the last write producer must emit" + ); +} + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn replace_readonly_to_write_emits(db: Pool) { + replace( + &db, + SP, + &[asset_at( + "u/test-user/res", + AssetKind::Resource, + Some(AssetUsageAccessType::R), + )], + ) + .await; + reset_notify(&db).await; + + // Same asset path, access flips read-only → write: cache gains a row → emit. + replace( + &db, + SP, + &[asset_at( + "u/test-user/res", + AssetKind::Resource, + Some(AssetUsageAccessType::W), + )], + ) + .await; + assert_eq!( + producer_notify_count(&db).await, + 1, + "read-only → write transition must emit" + ); +} + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn replace_duplicate_entries_use_persisted_state(db: Pool) { + replace( + &db, + SP, + &[asset_at( + "u/test-user/res", + AssetKind::Resource, + Some(AssetUsageAccessType::W), + )], + ) + .await; + reset_notify(&db).await; + + // Redeploy with conflicting duplicates for the same (kind, path), read-only + // first: ON CONFLICT DO NOTHING persists the read-only row and drops the + // write one, so the write producer is really gone → must emit (the diff must + // follow the persisted state, not the requested slice). + replace( + &db, + SP, + &[ + asset_at( + "u/test-user/res", + AssetKind::Resource, + Some(AssetUsageAccessType::R), + ), + asset_at( + "u/test-user/res", + AssetKind::Resource, + Some(AssetUsageAccessType::W), + ), + ], + ) + .await; + assert_eq!( + producer_notify_count(&db).await, + 1, + "duplicate entries losing the persisted write producer must emit" + ); +} + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn replace_duplicate_entries_keeping_write_does_not_emit(db: Pool) { + replace( + &db, + SP, + &[asset_at( + "u/test-user/res", + AssetKind::Resource, + Some(AssetUsageAccessType::W), + )], + ) + .await; + reset_notify(&db).await; + + // Same duplicates but write first: the write row persists, so the write set + // is unchanged → no emit. + replace( + &db, + SP, + &[ + asset_at( + "u/test-user/res", + AssetKind::Resource, + Some(AssetUsageAccessType::W), + ), + asset_at( + "u/test-user/res", + AssetKind::Resource, + Some(AssetUsageAccessType::R), + ), + ], + ) + .await; + assert_eq!( + producer_notify_count(&db).await, + 0, + "duplicate entries keeping the persisted write producer must not emit" + ); +} + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn replace_changing_only_readonly_does_not_emit(db: Pool) { + replace( + &db, + SP, + &[asset_at( + "u/test-user/res", + AssetKind::Resource, + Some(AssetUsageAccessType::R), + )], + ) + .await; + reset_notify(&db).await; + + // Swap one read-only producer for another: no write producer either side, so + // the write-set cache is untouched → no emit. + replace( + &db, + SP, + &[asset_at( + "u/test-user/other", + AssetKind::Resource, + Some(AssetUsageAccessType::R), + )], + ) + .await; + assert_eq!( + producer_notify_count(&db).await, + 0, + "changes confined to read-only producers must not emit" + ); +}