diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index ea81bc354b..7c2a3f4415 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -36,9 +36,9 @@ pub(crate) use runtime::{ }; pub use store::{ admin_moderation, allowlist, api_token, archived_identities, channel, channel_members, - community, deletion, dm, event, feed, git_repo, moderation, partition, product_feedback, push, - reaction, relay_admin_actions, relay_invite, relay_members, relay_operators, reminder, - replaceable, thread, usage, user, workflow, + community, deletion, dm, event, feed, git_repo, moderation, partition, product_feedback, + project_state, push, reaction, relay_admin_actions, relay_invite, relay_members, + relay_operators, reminder, replaceable, thread, usage, user, workflow, }; pub use allowlist::AllowlistEntry; @@ -50,6 +50,9 @@ pub use community::{ }; pub use error::{DbError, Result}; pub use event::{EventQuery, DEFAULT_MAX_PAGE_LIMIT}; +pub use project_state::{ + ProjectChangeApplyResult, ProjectRelatedChannelChange, ProjectStateSnapshot, +}; pub use reaction::ReactionEventInsertOutcome; pub use reminder::DueReminder; pub use usage::UsageMetricsLeader; diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 66ac1d2f8f..487916fc5e 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -699,7 +699,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 42); + assert_eq!(migrations.len(), 43); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -1815,6 +1815,17 @@ mod tests { let mut expected_fences = migration.fence_attachments.clone(); expected_fences.remove("product_feedback"); expected_fences.remove("rate_limit_violations"); + expected_fences.extend( + surface( + MIGRATOR + .iter() + .find(|item| item.version == 43) + .expect("migration 0043") + .sql + .as_ref(), + ) + .fence_attachments, + ); assert_eq!( expected_fences, schema.fence_attachments, "write-fence attachment targets differ after recovery policy" diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index f0fd3332ac..8abd74b7bf 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -173,7 +173,7 @@ pub async fn verify_channel_roster_fence_behavior(pool: &sqlx::PgPool) -> Result /// Take the per-channel membership lock. MUST be the first statement in the /// transaction that then reads roles/owner counts and writes membership, so the /// whole check-then-write sequence is atomic against a concurrent one. -async fn acquire_channel_membership_lock( +pub(crate) async fn acquire_channel_membership_lock( tx: &mut Transaction<'_, Postgres>, community_id: CommunityId, channel_id: Uuid, diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index c7fcdc09f6..749d63fc23 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -69,6 +69,9 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[ "moderation_actions", "moderation_reports", "parameterized_event_watermarks", + "project_change_receipts", + "project_related_channels", + "project_state_heads", "pubkey_allowlist", "push_leases", "push_match_queue", @@ -87,6 +90,9 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[ /// Foreign-key-safe child-before-parent order for the PostgreSQL purge. pub const PURGE_SCOPED_TABLES: &[&str] = &[ + "project_change_receipts", + "project_related_channels", + "project_state_heads", "workflow_approvals", "scheduled_workflow_fires", "workflow_runs", diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index 1fa1273eb0..ac5b2817d5 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -30,6 +30,8 @@ pub mod moderation; pub mod partition; /// Buzz product-feedback sidecar persistence. pub mod product_feedback; +/// Collaborative Project state and command receipts (NIP-PC). +pub mod project_state; /// Community-scoped push lease and durable wake-outbox persistence. pub mod push; /// Reaction persistence. diff --git a/crates/buzz-db/src/store/project_state.rs b/crates/buzz-db/src/store/project_state.rs new file mode 100644 index 0000000000..ea4318f5cf --- /dev/null +++ b/crates/buzz-db/src/store/project_state.rs @@ -0,0 +1,613 @@ +//! Transactional persistence for collaborative Project state. + +use std::collections::BTreeSet; + +use buzz_core::kind::{event_kind_u32, KIND_PROJECT, KIND_PROJECT_CHANGE}; +use buzz_core::CommunityId; +use nostr::Event; +use sqlx::Row; +use uuid::Uuid; + +use crate::event::insert_event_in_transaction; +use crate::replaceable::event_replacement_lock_key; +use crate::{Db, DbError, Result}; + +const RELATED_CHANNEL_CAP: usize = 64; + +/// A validated Project related-channel mutation supplied by the relay parser. +#[derive(Clone, Copy, Debug)] +pub struct ProjectRelatedChannelChange<'a> { + /// Owner pubkey from the canonical kind:30621 coordinate. + pub project_owner: &'a [u8], + /// Verbatim `d` value from the canonical kind:30621 coordinate. + pub project_d_tag: &'a str, + /// Relational revision the actor observed. + pub expected_revision: i64, + /// Channels to add to the effective related-channel set. + pub add: &'a [Uuid], + /// Channels to remove from the effective related-channel set. + pub remove: &'a [Uuid], +} + +/// Canonical effective state returned for relay projection signing. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ProjectStateSnapshot { + /// Monotonic relational revision produced by the command. + pub revision: i64, + /// Current owner-signed kind:30621 identity event id. + pub identity_event_id: Vec, + /// Command event id that produced this revision. + pub change_event_id: Vec, + /// Sorted effective related-channel set. + pub related_channels: Vec, +} + +/// Outcome of applying one Project related-channel command. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum ProjectChangeApplyResult { + /// The command and its effective state committed atomically. + Applied(ProjectStateSnapshot), + /// This exact command already committed, possibly before later revisions. + Duplicate { + /// Revision originally produced by the command. + applied_revision: i64, + }, + /// No live owner-signed Project identity exists at the coordinate. + ProjectNotFound, + /// The materialized Project is deleted and awaits owner recreation. + ProjectDeleted, + /// The actor is neither the Project owner nor an active home-channel owner/admin. + Forbidden, + /// The expected revision is stale, or lifecycle state changed first. + Conflict { + /// Current materialized revision. + current_revision: i64, + }, + /// The patch is invalid against current effective state. + InvalidMutation(String), + /// The current owner identity cannot be materialized as canonical v1 state. + InvalidBase(String), +} + +fn tag_parts(tag: &serde_json::Value) -> Option> { + tag.as_array()? + .iter() + .map(serde_json::Value::as_str) + .collect() +} + +fn parse_base_state( + tags: &serde_json::Value, +) -> std::result::Result<(Option, BTreeSet), String> { + let mut home = None; + let mut related = BTreeSet::new(); + for tag in tags.as_array().ok_or("Project tags are not an array")? { + let Some(parts) = tag_parts(tag) else { + return Err("Project contains a non-string tag".into()); + }; + match parts.as_slice() { + ["buzz-channel", value] => { + let channel = Uuid::parse_str(value) + .ok() + .filter(|id| id.to_string() == *value); + home = channel; + } + ["buzz-related-channel", value] => { + let channel = Uuid::parse_str(value) + .ok() + .filter(|id| id.to_string() == *value) + .ok_or("Project contains a non-canonical related channel")?; + if !related.insert(channel) { + return Err("Project contains a duplicate related channel".into()); + } + } + ["buzz-related-channel", ..] => { + return Err("Project contains a malformed related-channel tag".into()); + } + _ => {} + } + } + if related.len() > RELATED_CHANNEL_CAP { + return Err("Project contains more than 64 related channels".into()); + } + if home.is_some_and(|channel| related.contains(&channel)) { + return Err("Project home channel cannot also be related".into()); + } + Ok((home, related)) +} + +fn validate_patch(change: ProjectRelatedChannelChange<'_>) -> Option { + if change.project_owner.len() != 32 { + return Some("Project owner must be 32 bytes".into()); + } + if change.project_d_tag.is_empty() || change.project_d_tag.len() > crate::event::D_TAG_MAX_LEN { + return Some("Project d tag is empty or too long".into()); + } + if change.expected_revision < 1 { + return Some("expected revision must be positive".into()); + } + if change.add.is_empty() && change.remove.is_empty() { + return Some("Project change must not be empty".into()); + } + if change.add.len() > RELATED_CHANNEL_CAP || change.remove.len() > RELATED_CHANNEL_CAP { + return Some("Project change exceeds the 64-channel patch bound".into()); + } + let add = change.add.iter().copied().collect::>(); + let remove = change.remove.iter().copied().collect::>(); + if add.len() != change.add.len() || remove.len() != change.remove.len() { + return Some("Project change contains a duplicate channel".into()); + } + if !add.is_disjoint(&remove) { + return Some("Project change adds and removes the same channel".into()); + } + None +} + +impl Db { + /// Authorize and atomically apply one v1 Project related-channel command. + /// + /// Lock order is community deletion fence, Project coordinate, then current + /// home-channel membership. The accepted command event, normalized state, + /// CAS advance, and durable replay receipt share one transaction. + pub async fn apply_project_related_channel_change( + &self, + community_id: CommunityId, + event: &Event, + change: ProjectRelatedChannelChange<'_>, + ) -> Result { + if event_kind_u32(event) != KIND_PROJECT_CHANGE { + return Err(DbError::InvalidData( + "Project change persistence requires kind 47010".into(), + )); + } + if let Some(message) = validate_patch(change) { + return Ok(ProjectChangeApplyResult::InvalidMutation(message)); + } + + let mut tx = self.begin_transaction().await?; + self.deletion_store() + .guard_transaction(&mut tx, community_id) + .await?; + let coordinate_lock = event_replacement_lock_key( + community_id, + KIND_PROJECT as i32, + change.project_owner, + Some(change.project_d_tag.as_bytes()), + ); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(coordinate_lock) + .execute(&mut *tx) + .await?; + + let receipt = sqlx::query( + "SELECT project_owner, project_d_tag, applied_revision \ + FROM project_change_receipts WHERE community_id=$1 AND command_event_id=$2", + ) + .bind(community_id.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .fetch_optional(&mut *tx) + .await?; + if let Some(receipt) = receipt { + let owner: Vec = receipt.try_get("project_owner")?; + let d_tag: String = receipt.try_get("project_d_tag")?; + if owner != change.project_owner || d_tag != change.project_d_tag { + return Err(DbError::InvalidData( + "Project command receipt coordinate mismatch".into(), + )); + } + return Ok(ProjectChangeApplyResult::Duplicate { + applied_revision: receipt.try_get("applied_revision")?, + }); + } + + let head = sqlx::query( + "SELECT revision, deleted, identity_event_id FROM project_state_heads \ + WHERE community_id=$1 AND project_owner=$2 AND project_d_tag=$3 FOR UPDATE", + ) + .bind(community_id.as_uuid()) + .bind(change.project_owner) + .bind(change.project_d_tag) + .fetch_optional(&mut *tx) + .await?; + if head.as_ref().is_some_and(|row| row.get("deleted")) { + return Ok(ProjectChangeApplyResult::ProjectDeleted); + } + + let base = sqlx::query( + "SELECT id, tags FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ + AND d_tag=$4 AND deleted_at IS NULL ORDER BY created_at DESC, id ASC LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(KIND_PROJECT as i32) + .bind(change.project_owner) + .bind(change.project_d_tag) + .fetch_optional(&mut *tx) + .await?; + let Some(base) = base else { + return Ok(ProjectChangeApplyResult::ProjectNotFound); + }; + let identity_event_id: Vec = base.try_get("id")?; + let base_tags: serde_json::Value = base.try_get("tags")?; + let (home_channel, base_related) = match parse_base_state(&base_tags) { + Ok(state) => state, + Err(message) => return Ok(ProjectChangeApplyResult::InvalidBase(message)), + }; + + let current_revision = if let Some(head) = head { + let revision: i64 = head.try_get("revision")?; + let materialized_identity: Vec = head.try_get("identity_event_id")?; + if materialized_identity != identity_event_id { + return Ok(ProjectChangeApplyResult::Conflict { + current_revision: revision, + }); + } + revision + } else { + sqlx::query( + "INSERT INTO project_state_heads \ + (community_id, project_owner, project_d_tag, revision, identity_event_id, last_event_id) \ + VALUES ($1,$2,$3,1,$4,$4)", + ) + .bind(community_id.as_uuid()) + .bind(change.project_owner) + .bind(change.project_d_tag) + .bind(&identity_event_id) + .execute(&mut *tx) + .await?; + for channel in &base_related { + sqlx::query( + "INSERT INTO project_related_channels \ + (community_id, project_owner, project_d_tag, channel_id) VALUES ($1,$2,$3,$4)", + ) + .bind(community_id.as_uuid()) + .bind(change.project_owner) + .bind(change.project_d_tag) + .bind(channel) + .execute(&mut *tx) + .await?; + } + 1 + }; + + let actor = event.pubkey.to_bytes(); + if actor.as_slice() != change.project_owner { + let Some(home_channel) = home_channel else { + return Ok(ProjectChangeApplyResult::Forbidden); + }; + crate::channel_members::acquire_channel_membership_lock( + &mut tx, + community_id, + home_channel, + ) + .await?; + let role: Option = sqlx::query_scalar( + "SELECT member.role::text FROM channel_members member \ + JOIN channels channel ON channel.community_id=member.community_id \ + AND channel.id=member.channel_id \ + WHERE member.community_id=$1 AND member.channel_id=$2 AND member.pubkey=$3 \ + AND member.removed_at IS NULL AND channel.archived_at IS NULL \ + AND channel.deleted_at IS NULL FOR SHARE OF channel, member", + ) + .bind(community_id.as_uuid()) + .bind(home_channel) + .bind(actor.as_slice()) + .fetch_optional(&mut *tx) + .await?; + if !matches!(role.as_deref(), Some("owner" | "admin")) { + return Ok(ProjectChangeApplyResult::Forbidden); + } + } + if change.expected_revision != current_revision { + return Ok(ProjectChangeApplyResult::Conflict { current_revision }); + } + + let rows = sqlx::query_scalar::<_, Uuid>( + "SELECT channel_id FROM project_related_channels \ + WHERE community_id=$1 AND project_owner=$2 AND project_d_tag=$3 ORDER BY channel_id", + ) + .bind(community_id.as_uuid()) + .bind(change.project_owner) + .bind(change.project_d_tag) + .fetch_all(&mut *tx) + .await?; + let mut related = rows.into_iter().collect::>(); + for channel in change.add { + if Some(*channel) == home_channel || !related.insert(*channel) { + return Ok(ProjectChangeApplyResult::InvalidMutation( + "cannot add the home channel or an already-related channel".into(), + )); + } + } + for channel in change.remove { + if !related.remove(channel) { + return Ok(ProjectChangeApplyResult::InvalidMutation( + "cannot remove a channel that is not related".into(), + )); + } + } + if related.len() > RELATED_CHANNEL_CAP { + return Ok(ProjectChangeApplyResult::InvalidMutation( + "effective Project state exceeds 64 related channels".into(), + )); + } + + let (_, inserted) = insert_event_in_transaction(&mut tx, community_id, event, None).await?; + if !inserted { + return Err(DbError::InvalidData( + "Project command event exists without a durable receipt".into(), + )); + } + for channel in change.remove { + sqlx::query( + "DELETE FROM project_related_channels WHERE community_id=$1 \ + AND project_owner=$2 AND project_d_tag=$3 AND channel_id=$4", + ) + .bind(community_id.as_uuid()) + .bind(change.project_owner) + .bind(change.project_d_tag) + .bind(channel) + .execute(&mut *tx) + .await?; + } + for channel in change.add { + sqlx::query( + "INSERT INTO project_related_channels \ + (community_id, project_owner, project_d_tag, channel_id) VALUES ($1,$2,$3,$4)", + ) + .bind(community_id.as_uuid()) + .bind(change.project_owner) + .bind(change.project_d_tag) + .bind(channel) + .execute(&mut *tx) + .await?; + } + let revision: i64 = sqlx::query_scalar( + "UPDATE project_state_heads SET revision=revision+1, last_event_id=$4, \ + updated_at=transaction_timestamp() WHERE community_id=$1 AND project_owner=$2 \ + AND project_d_tag=$3 AND revision=$5 AND revision < 9223372036854775807 \ + RETURNING revision", + ) + .bind(community_id.as_uuid()) + .bind(change.project_owner) + .bind(change.project_d_tag) + .bind(event.id.as_bytes().as_slice()) + .bind(current_revision) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| DbError::InvalidData("Project revision overflow or CAS failure".into()))?; + sqlx::query( + "INSERT INTO project_change_receipts \ + (community_id, command_event_id, project_owner, project_d_tag, applied_revision) \ + VALUES ($1,$2,$3,$4,$5)", + ) + .bind(community_id.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .bind(change.project_owner) + .bind(change.project_d_tag) + .bind(revision) + .execute(&mut *tx) + .await?; + + let snapshot = ProjectStateSnapshot { + revision, + identity_event_id, + change_event_id: event.id.as_bytes().to_vec(), + related_channels: related.into_iter().collect(), + }; + tx.commit().await?; + Ok(ProjectChangeApplyResult::Applied(snapshot)) + } +} + +#[cfg(test)] +mod tests { + use buzz_core::channel::{ChannelType, ChannelVisibility}; + use nostr::{EventBuilder, Keys, Kind, Tag}; + use sqlx::postgres::PgPoolOptions; + + use super::*; + + const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 -- local test-only credentials + + fn command( + actor: &Keys, + owner: &Keys, + slug: &str, + expected: i64, + add: &[Uuid], + remove: &[Uuid], + ) -> Event { + let content = serde_json::json!({ + "v": 1, + "patch": {"related_channels": {"add": add, "remove": remove}} + }) + .to_string(); + EventBuilder::new(Kind::Custom(KIND_PROJECT_CHANGE as u16), content) + .tags([ + Tag::parse([ + "a", + &format!("30621:{}:{slug}", owner.public_key().to_hex()), + ]) + .expect("a tag"), + Tag::parse(["expected-revision", &expected.to_string()]).expect("revision tag"), + ]) + .sign_with_keys(actor) + .expect("sign command") + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn transaction_enforces_authority_cas_replay_and_atomic_attribution() { + let database_url = + std::env::var("TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.into()); + let pool = PgPoolOptions::new() + .max_connections(5) + .connect(&database_url) + .await + .expect("connect test database"); + crate::migration::run_migrations(&pool) + .await + .expect("run migrations"); + let db = Db::from_pool(pool.clone()); + let community = CommunityId::from_uuid(Uuid::new_v4()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1,$2)") + .bind(community.as_uuid()) + .bind(format!("project-state-{}.example", community.as_uuid())) + .execute(&pool) + .await + .expect("insert community"); + + let owner = Keys::generate(); + let admin = Keys::generate(); + let member = Keys::generate(); + let home = Uuid::new_v4(); + crate::channel::create_channel_with_id( + &pool, + community, + home, + "project-home", + ChannelType::Stream, + ChannelVisibility::Private, + None, + owner.public_key().to_bytes().as_slice(), + None, + ) + .await + .expect("create home channel"); + for (keys, role) in [(&admin, "admin"), (&member, "member")] { + sqlx::query( + "INSERT INTO channel_members (community_id,channel_id,pubkey,role,invited_by) \ + VALUES ($1,$2,$3,$4::member_role,$5)", + ) + .bind(community.as_uuid()) + .bind(home) + .bind(keys.public_key().to_bytes().as_slice()) + .bind(role) + .bind(owner.public_key().to_bytes().as_slice()) + .execute(&pool) + .await + .expect("insert member"); + } + let seeded = Uuid::new_v4(); + let base = EventBuilder::new(Kind::Custom(KIND_PROJECT as u16), "") + .tags([ + Tag::parse(["d", "shared"]).expect("d tag"), + Tag::parse(["buzz-channel", &home.to_string()]).expect("home tag"), + Tag::parse(["buzz-related-channel", &seeded.to_string()]).expect("related tag"), + ]) + .sign_with_keys(&owner) + .expect("sign Project"); + db.insert_event(community, &base, None) + .await + .expect("insert Project"); + + let added = Uuid::new_v4(); + let first = command(&admin, &owner, "shared", 1, &[added], &[]); + let owner_bytes = owner.public_key().to_bytes(); + let applied = db + .apply_project_related_channel_change( + community, + &first, + ProjectRelatedChannelChange { + project_owner: owner_bytes.as_slice(), + project_d_tag: "shared", + expected_revision: 1, + add: &[added], + remove: &[], + }, + ) + .await + .expect("apply admin command"); + let ProjectChangeApplyResult::Applied(snapshot) = applied else { + panic!("expected applied command: {applied:?}"); + }; + assert_eq!(snapshot.revision, 2); + assert_eq!(snapshot.related_channels.len(), 2); + assert!(snapshot.related_channels.contains(&seeded)); + assert!(snapshot.related_channels.contains(&added)); + let later = command(&owner, &owner, "shared", 2, &[], &[seeded]); + assert!(matches!( + db.apply_project_related_channel_change( + community, + &later, + ProjectRelatedChannelChange { + project_owner: owner_bytes.as_slice(), + project_d_tag: "shared", + expected_revision: 2, + add: &[], + remove: &[seeded], + }, + ) + .await + .expect("apply owner command"), + ProjectChangeApplyResult::Applied(ProjectStateSnapshot { revision: 3, .. }) + )); + assert_eq!( + db.apply_project_related_channel_change( + community, + &first, + ProjectRelatedChannelChange { + project_owner: owner_bytes.as_slice(), + project_d_tag: "shared", + expected_revision: 1, + add: &[added], + remove: &[], + }, + ) + .await + .expect("replay accepted command"), + ProjectChangeApplyResult::Duplicate { + applied_revision: 2 + } + ); + + let denied_channel = Uuid::new_v4(); + let denied = command(&member, &owner, "shared", 3, &[denied_channel], &[]); + let denied_id = denied.id.as_bytes().to_vec(); + assert_eq!( + db.apply_project_related_channel_change( + community, + &denied, + ProjectRelatedChannelChange { + project_owner: owner_bytes.as_slice(), + project_d_tag: "shared", + expected_revision: 3, + add: &[denied_channel], + remove: &[], + }, + ) + .await + .expect("reject member"), + ProjectChangeApplyResult::Forbidden + ); + let rejected_rows: i64 = + sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(denied_id) + .fetch_one(&pool) + .await + .expect("count rejected event"); + assert_eq!(rejected_rows, 0); + + let stale_channel = Uuid::new_v4(); + let stale = command(&owner, &owner, "shared", 2, &[stale_channel], &[]); + assert_eq!( + db.apply_project_related_channel_change( + community, + &stale, + ProjectRelatedChannelChange { + project_owner: owner_bytes.as_slice(), + project_d_tag: "shared", + expected_revision: 2, + add: &[stale_channel], + remove: &[], + }, + ) + .await + .expect("reject stale CAS"), + ProjectChangeApplyResult::Conflict { + current_revision: 3 + } + ); + } +} diff --git a/migrations/0043_project_state.sql b/migrations/0043_project_state.sql new file mode 100644 index 0000000000..acbb344bed --- /dev/null +++ b/migrations/0043_project_state.sql @@ -0,0 +1,36 @@ +CREATE TABLE project_state_heads ( + community_id UUID NOT NULL REFERENCES communities(id), + project_owner BYTEA NOT NULL CHECK (octet_length(project_owner) = 32), + project_d_tag TEXT NOT NULL CHECK (project_d_tag <> ''), + revision BIGINT NOT NULL CHECK (revision > 0), + deleted BOOLEAN NOT NULL DEFAULT FALSE, + identity_event_id BYTEA NOT NULL CHECK (octet_length(identity_event_id) = 32), + last_event_id BYTEA NOT NULL CHECK (octet_length(last_event_id) = 32), + updated_at TIMESTAMPTZ NOT NULL DEFAULT transaction_timestamp(), + PRIMARY KEY (community_id, project_owner, project_d_tag) +); +CREATE TABLE project_related_channels ( + community_id UUID NOT NULL, + project_owner BYTEA NOT NULL CHECK (octet_length(project_owner) = 32), + project_d_tag TEXT NOT NULL CHECK (project_d_tag <> ''), + channel_id UUID NOT NULL, + PRIMARY KEY (community_id, project_owner, project_d_tag, channel_id), + FOREIGN KEY (community_id, project_owner, project_d_tag) + REFERENCES project_state_heads (community_id, project_owner, project_d_tag) + ON DELETE CASCADE +); +CREATE TABLE project_change_receipts ( + community_id UUID NOT NULL, + command_event_id BYTEA NOT NULL CHECK (octet_length(command_event_id) = 32), + project_owner BYTEA NOT NULL CHECK (octet_length(project_owner) = 32), + project_d_tag TEXT NOT NULL CHECK (project_d_tag <> ''), + applied_revision BIGINT NOT NULL CHECK (applied_revision > 1), + accepted_at TIMESTAMPTZ NOT NULL DEFAULT transaction_timestamp(), + PRIMARY KEY (community_id, command_event_id), + UNIQUE (community_id, project_owner, project_d_tag, applied_revision), + FOREIGN KEY (community_id, project_owner, project_d_tag) + REFERENCES project_state_heads (community_id, project_owner, project_d_tag) +); +SELECT attach_community_write_fence('project_state_heads'); +SELECT attach_community_write_fence('project_related_channels'); +SELECT attach_community_write_fence('project_change_receipts'); diff --git a/schema/schema.sql b/schema/schema.sql index 2335f8bf0b..83574002cf 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -3787,3 +3787,40 @@ CREATE CONSTRAINT TRIGGER authorization_event_receipt_cardinality AFTER INSERT ON authorization_events DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION authorization_operation_receipt_event_guard_v1(); + +CREATE TABLE project_state_heads ( + community_id UUID NOT NULL REFERENCES communities(id), + project_owner BYTEA NOT NULL CHECK (octet_length(project_owner) = 32), + project_d_tag TEXT NOT NULL CHECK (project_d_tag <> ''), + revision BIGINT NOT NULL CHECK (revision > 0), + deleted BOOLEAN NOT NULL DEFAULT FALSE, + identity_event_id BYTEA NOT NULL CHECK (octet_length(identity_event_id) = 32), + last_event_id BYTEA NOT NULL CHECK (octet_length(last_event_id) = 32), + updated_at TIMESTAMPTZ NOT NULL DEFAULT transaction_timestamp(), + PRIMARY KEY (community_id, project_owner, project_d_tag) +); +CREATE TABLE project_related_channels ( + community_id UUID NOT NULL, + project_owner BYTEA NOT NULL CHECK (octet_length(project_owner) = 32), + project_d_tag TEXT NOT NULL CHECK (project_d_tag <> ''), + channel_id UUID NOT NULL, + PRIMARY KEY (community_id, project_owner, project_d_tag, channel_id), + FOREIGN KEY (community_id, project_owner, project_d_tag) + REFERENCES project_state_heads (community_id, project_owner, project_d_tag) + ON DELETE CASCADE +); +CREATE TABLE project_change_receipts ( + community_id UUID NOT NULL, + command_event_id BYTEA NOT NULL CHECK (octet_length(command_event_id) = 32), + project_owner BYTEA NOT NULL CHECK (octet_length(project_owner) = 32), + project_d_tag TEXT NOT NULL CHECK (project_d_tag <> ''), + applied_revision BIGINT NOT NULL CHECK (applied_revision > 1), + accepted_at TIMESTAMPTZ NOT NULL DEFAULT transaction_timestamp(), + PRIMARY KEY (community_id, command_event_id), + UNIQUE (community_id, project_owner, project_d_tag, applied_revision), + FOREIGN KEY (community_id, project_owner, project_d_tag) + REFERENCES project_state_heads (community_id, project_owner, project_d_tag) +); +SELECT attach_community_write_fence('project_state_heads'); +SELECT attach_community_write_fence('project_related_channels'); +SELECT attach_community_write_fence('project_change_receipts');