feat(shared-topics): thread the shared flag through pubsub triggers (D2)

Add `shared` to BundleTrigger::Pubsub + PubsubTriggerSpec + the trigger
identity (so toggling re-diffs) + current_trigger_identity. validate_bundle_for
now allows a `shared` pubsub trigger on a group and requires the topic
pattern's ROOT segment (events.* -> events) be a declared kind='topic'
collection; a wildcard root is a runtime match. Apply-side plumbing only —
the shared publish path + dispatch boundary land next.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-07-02 21:37:31 +02:00
parent ae8f0be748
commit 1c7e8886d8
3 changed files with 54 additions and 35 deletions

View File

@@ -234,6 +234,11 @@ pub enum BundleTrigger {
retry_max_attempts: Option<u32>, retry_max_attempts: Option<u32>,
#[serde(default)] #[serde(default)]
sealed: bool, sealed: bool,
/// §11.6 D2: `true` for a shared-TOPIC group trigger — fires when a
/// subtree app publishes to the group's shared topic namespace, not on
/// per-app publishes.
#[serde(default)]
shared: bool,
}, },
Email { Email {
script: String, script: String,
@@ -298,18 +303,18 @@ impl BundleTrigger {
} }
} }
/// §11.6: whether this template watches a SHARED collection. Only the /// §11.6: whether this template watches a SHARED collection. The collection-
/// collection-scoped event kinds (kv/docs/files) can be shared; pubsub + /// scoped event kinds (kv/docs/files) and pubsub (D2: a shared TOPIC) can be
/// the app-only kinds are never shared. /// shared; cron/email/queue are never shared here (a shared queue consumer
/// is authored as a `[[triggers.queue]]` and handled via materialization).
#[must_use] #[must_use]
pub fn shared(&self) -> bool { pub fn shared(&self) -> bool {
match self { match self {
Self::Kv { shared, .. } | Self::Docs { shared, .. } | Self::Files { shared, .. } => { Self::Kv { shared, .. }
*shared | Self::Docs { shared, .. }
} | Self::Files { shared, .. }
Self::Cron { .. } | Self::Pubsub { .. } | Self::Email { .. } | Self::Queue { .. } => { | Self::Pubsub { shared, .. } => *shared,
false Self::Cron { .. } | Self::Email { .. } | Self::Queue { .. } => false,
}
} }
} }
@@ -381,8 +386,9 @@ impl BundleTrigger {
script, script,
topic_pattern, topic_pattern,
sealed, sealed,
shared,
.. ..
} => format!("pubsub|{script}|{topic_pattern}|{sealed}"), } => format!("pubsub|{script}|{topic_pattern}|{sealed}|{shared}"),
Self::Email { script, .. } => format!("email|{script}"), Self::Email { script, .. } => format!("email|{script}"),
Self::Queue { queue_name, .. } => format!("queue|{queue_name}"), Self::Queue { queue_name, .. } => format!("queue|{queue_name}"),
} }
@@ -764,37 +770,41 @@ impl ApplyService {
// store once at apply (shared-group-secret model); materialization // store once at apply (shared-group-secret model); materialization
// copies the sealed bytes verbatim. // copies the sealed bytes verbatim.
for t in &bundle.triggers { for t in &bundle.triggers {
// §11.6: a `shared` trigger watches the group's SHARED collection. // §11.6: a `shared` trigger watches a SHARED collection/namespace.
// Only the collection-scoped kinds can be shared (pubsub has no // kv/docs/files watch a shared collection (name = the glob); D2
// shared topic store yet), and the watched collection must be one // pubsub watches a shared TOPIC namespace (name = the topic
// the SAME group declares shared, of the matching kind. // pattern's ROOT segment, e.g. `events.*` → `events`). The watched
// name must be declared shared of the matching kind on the SAME
// group; a wildcard can't be statically resolved, so it's allowed
// (runtime match).
if t.shared() { if t.shared() {
let kind = t.kind_str(); let (watched, coll_kind) = match t {
if !matches!(
t,
BundleTrigger::Kv { .. } BundleTrigger::Kv { .. }
| BundleTrigger::Docs { .. } | BundleTrigger::Docs { .. }
| BundleTrigger::Files { .. } | BundleTrigger::Files { .. } => {
) { (t.collection_glob().unwrap_or_default(), t.kind_str())
return Err(ApplyError::Invalid(format!( }
"a `shared` trigger must be a kv/docs/files kind; `{kind}` \ BundleTrigger::Pubsub { topic_pattern, .. } => {
has no shared collection store" (topic_pattern.split('.').next().unwrap_or(""), "topic")
))); }
} _ => {
let glob = t.collection_glob().unwrap_or_default(); return Err(ApplyError::Invalid(format!(
// A concrete (non-glob) collection must be declared shared of "a `shared` trigger must be a kv/docs/files/pubsub kind; \
// the same kind on this group. A wildcard glob can't be `{}` has no shared collection store",
// statically resolved, so it's allowed (runtime match). t.kind_str()
let is_wildcard = glob.contains('*'); )));
}
};
let is_wildcard = watched.contains('*');
let declared = bundle let declared = bundle
.collections .collections
.iter() .iter()
.any(|c| c.kind == kind && c.name.eq_ignore_ascii_case(glob)); .any(|c| c.kind == coll_kind && c.name.eq_ignore_ascii_case(watched));
if !is_wildcard && !declared { if !is_wildcard && !declared {
return Err(ApplyError::Invalid(format!( return Err(ApplyError::Invalid(format!(
"a `shared` {kind} trigger watches collection `{glob}`, \ "a `shared` trigger watches `{watched}`, which this group \
which this group does not declare as a shared {kind} \ does not declare as a shared {coll_kind} collection (add \
collection (add it to `collections`)" it to `collections`)"
))); )));
} }
} }
@@ -3403,7 +3413,7 @@ fn current_trigger_identity(t: &Trigger, name_by_id: &HashMap<ScriptId, String>)
schedule, timezone, .. schedule, timezone, ..
} => Some(format!("cron|{script}|{schedule}|{timezone}")), } => Some(format!("cron|{script}|{schedule}|{timezone}")),
TriggerDetails::Pubsub { topic_pattern } => { TriggerDetails::Pubsub { topic_pattern } => {
Some(format!("pubsub|{script}|{topic_pattern}|{sealed}")) Some(format!("pubsub|{script}|{topic_pattern}|{sealed}|{shared}"))
} }
TriggerDetails::Email { .. } => Some(format!("email|{script}")), TriggerDetails::Email { .. } => Some(format!("email|{script}")),
TriggerDetails::Queue { queue_name, .. } => Some(format!("queue|{queue_name}")), TriggerDetails::Queue { queue_name, .. } => Some(format!("queue|{queue_name}")),
@@ -4224,6 +4234,7 @@ mod tests {
dispatch_mode: None, dispatch_mode: None,
retry_max_attempts: None, retry_max_attempts: None,
sealed: false, sealed: false,
shared: false,
}) })
.is_err()); .is_err());
assert!(validate_trigger_shape(&BundleTrigger::Pubsub { assert!(validate_trigger_shape(&BundleTrigger::Pubsub {
@@ -4232,6 +4243,7 @@ mod tests {
dispatch_mode: None, dispatch_mode: None,
retry_max_attempts: None, retry_max_attempts: None,
sealed: false, sealed: false,
shared: false,
}) })
.is_ok()); .is_ok());
// Queue visibility floor matches the interactive API (>= 30): a value // Queue visibility floor matches the interactive API (>= 30): a value

View File

@@ -207,7 +207,9 @@ pub async fn run(app_ident: &str, dir: &Path, force: bool, mode: OutputMode) ->
topic_pattern: d.topic_pattern, topic_pattern: d.topic_pattern,
dispatch_mode, dispatch_mode,
retry_max_attempts, retry_max_attempts,
// app-owned triggers are never sealed/shared (group-only).
sealed: false, sealed: false,
shared: false,
}); });
} }
"queue" => { "queue" => {

View File

@@ -487,6 +487,11 @@ pub struct PubsubTriggerSpec {
/// See [`KvTriggerSpec::sealed`]. /// See [`KvTriggerSpec::sealed`].
#[serde(default, skip_serializing_if = "is_false")] #[serde(default, skip_serializing_if = "is_false")]
pub sealed: bool, pub sealed: bool,
/// §11.6 D2: `true` for a shared-TOPIC group trigger — fires when a subtree
/// app publishes to the group's declared shared topic, not on per-app
/// publishes. Group-only; requires a declared `kind = "topic"` collection.
#[serde(default, skip_serializing_if = "is_false")]
pub shared: bool,
} }
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]