crates/agent: authorization API uses DB snapshots

Update the API to fetch on-demand snapshots from the DB of
authorization state, bounding both the frequency with which snapshots
may be taken and also for how long a stale snapshot may be used.

A request will either succeed against a current snapshot (happy
fast path), or will fail. If it fails, the caller is asked to retry
after a given delay, and meanwhile the API will have typically
started to refresh the current snapshot.

If the request continues to fail against a snapshot which as taken
_after_ the issued-at time of the request token, only then is the error
considered permanently failed.

This means that requests are cheap, evaluted only by the agent-api,
but we *also* don't have caching artifacts that could result in false
errors when tasks or collections are being created or changed. Instead, we
incur a minor delay while a sufficiently-recent snapshot is taken.

A final change is that we now prefix match on shard ID templates and
journal name templates, rather than munging both into a more-approximate
prefix match over catalog names.

crates/agent-sql/src/data_plane.rs

@@ -68,70 +68,3 @@ pub async fn fetch_data_planes(
 
     Ok(r.into_iter().collect())
 }
-
-pub async fn fetch_data_plane_by_task_and_fqdn(
-    pool: &sqlx::PgPool,
-    task_shard: &str,
-    task_data_plane_fqdn: &str,
-) -> sqlx::Result<Option<tables::DataPlane>> {
-    sqlx::query_as!(
-        tables::DataPlane,
-        r#"
-        select
-            d.id as "control_id: Id",
-            d.data_plane_name,
-            d.data_plane_fqdn,
-            false as "is_default!: bool",
-            d.hmac_keys,
-            d.broker_address,
-            d.reactor_address,
-            d.ops_logs_name as "ops_logs_name: models::Collection",
-            d.ops_stats_name as "ops_stats_name: models::Collection"
-        from data_planes d
-        join live_specs t on t.data_plane_id = d.id
-        where d.data_plane_fqdn = $2 and starts_with($1::text, t.catalog_name)
-        "#,
-        task_shard,
-        task_data_plane_fqdn,
-    )
-    .fetch_optional(pool)
-    .await
-}
-
-pub async fn verify_task_authorization(
-    pool: &sqlx::PgPool,
-    task_shard: &str,
-    journal_name_or_prefix: &str,
-    required_role: &str,
-) -> sqlx::Result<Option<(String, models::Collection, models::Id, bool)>> {
-    let r = sqlx::query!(
-        r#"
-        select
-            t.catalog_name as "task_name: String",
-            c.catalog_name as "collection_name: models::Collection",
-            c.data_plane_id as "collection_data_plane_id: models::Id",
-            exists(
-                select 1
-                from internal.task_roles($1, $3::text::grant_capability) r
-                where starts_with($2, r.role_prefix)
-            ) as "authorized!: bool"
-        from live_specs t, live_specs c
-        where starts_with($1, t.catalog_name)
-          and starts_with($2, c.catalog_name)
-        "#,
-        task_shard,
-        journal_name_or_prefix,
-        required_role,
-    )
-    .fetch_optional(pool)
-    .await?;
-
-    Ok(r.map(|r| {
-        (
-            r.task_name,
-            r.collection_name,
-            r.collection_data_plane_id,
-            r.authorized,
-        )
-    }))
-}