Publish collaborative project state

Signed-off-by: John Tennant <jtennant@squareup.com>

crates/buzz-db/src/store/project_state.rs

@@ -490,7 +490,7 @@ impl Db {
             let project_owner: Vec<u8> = coordinate.try_get("project_owner")?;
             let project_d_tag: String = coordinate.try_get("project_d_tag")?;
             if let Some(candidate) = self
-                .load_project_state_projection_candidate(
+                .load_pending_project_state_projection(
                     community_id,
                     &project_owner,
                     &project_d_tag,
@@ -504,7 +504,13 @@ impl Db {
         Ok(candidates)
     }
 
-    async fn load_project_state_projection_candidate(
+    /// Load one coherent Project state when it requires publication by
+    /// `projection_pubkey`.
+    ///
+    /// The returned candidate is safe to sign outside the transaction because
+    /// [`Self::commit_project_state_projection`] revalidates every observed
+    /// head field while holding the same Project coordinate lock.
+    pub async fn load_pending_project_state_projection(
         &self,
         community_id: CommunityId,
         project_owner: &[u8],

crates/buzz-relay/src/handlers/mod.rs

@@ -28,6 +28,8 @@ pub mod moderation_notices;
 pub mod product_feedback;
 /// NIP-PC collaborative Project change validation and execution.
 pub mod project_change;
+/// Relay signing and durable repair for NIP-PC Project State projections.
+pub mod project_state_projection;
 #[allow(dead_code, missing_docs)]
 pub mod push_lease;
 /// NIP-43 relay membership admin command handler (kinds 9030–9032).

crates/buzz-relay/src/handlers/project_change.rs

@@ -207,6 +207,21 @@ pub(crate) async fn handle(
         ProjectChangeApplyResult::InvalidMutation(message)
         | ProjectChangeApplyResult::InvalidBase(message) => return Err(invalid(message)),
     };
+    if let Err(error) = super::project_state_projection::publish_project_state_for_coordinate(
+        tenant,
+        state,
+        &parsed.owner,
+        &parsed.d_tag,
+    )
+    .await
+    {
+        tracing::warn!(
+            project_owner = %hex::encode(&parsed.owner),
+            project_d_tag = %parsed.d_tag,
+            %error,
+            "accepted Project change awaits projection repair"
+        );
+    }
     Ok(IngestResult {
         event_id: event.id.to_hex(),
         accepted: true,

crates/buzz-relay/src/handlers/project_state_projection.rs

@@ -0,0 +1,195 @@
+//! Relay signing and durable repair for NIP-PC Project State projections.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use anyhow::{anyhow, Context};
+use buzz_core::event::StoredEvent;
+use buzz_core::kind::KIND_PROJECT_STATE;
+use buzz_core::tenant::TenantContext;
+use buzz_db::project_state::{ProjectStateProjectionCandidate, ProjectStateProjectionCommitResult};
+use nostr::{Event, EventBuilder, Timestamp};
+use tokio_util::sync::CancellationToken;
+
+use crate::state::AppState;
+
+use super::event::dispatch_persistent_event;
+
+const RECONCILE_BATCH_SIZE: i64 = 100;
+const RECONCILE_INTERVAL: Duration = Duration::from_secs(60);
+
+fn projection_created_at(now: u64, previous: Option<u64>) -> anyhow::Result<u64> {
+    let after_previous = previous
+        .map(|value| {
+            value
+                .checked_add(1)
+                .ok_or_else(|| anyhow!("Project projection timestamp overflow"))
+        })
+        .transpose()?;
+    Ok(after_previous.map_or(now, |value| now.max(value)))
+}
+
+fn sign_candidate(
+    candidate: &ProjectStateProjectionCandidate,
+    relay_keypair: &nostr::Keys,
+) -> anyhow::Result<Event> {
+    let template = candidate.template();
+    let created_at =
+        projection_created_at(Timestamp::now().as_secs(), candidate.previous_created_at())?;
+    EventBuilder::new(template.kind, template.content.clone())
+        .tags(template.tags.clone())
+        .custom_created_at(Timestamp::from(created_at))
+        .sign_with_keys(relay_keypair)
+        .context("sign Project State projection")
+}
+
+async fn publish_candidate(
+    tenant: &TenantContext,
+    state: &Arc<AppState>,
+    candidate: ProjectStateProjectionCandidate,
+) -> anyhow::Result<bool> {
+    if candidate.community_id() != tenant.community() {
+        return Err(anyhow!("Project projection candidate crossed communities"));
+    }
+    let event = sign_candidate(&candidate, &state.relay_keypair)?;
+    let result = state
+        .db
+        .commit_project_state_projection(&candidate, &event)
+        .await
+        .context("commit Project State projection")?;
+    if result == ProjectStateProjectionCommitResult::Stale {
+        return Ok(false);
+    }
+
+    let stored = StoredEvent::new(event, None);
+    let relay_pubkey = state.relay_keypair.public_key().to_hex();
+    dispatch_persistent_event(
+        tenant,
+        state,
+        &stored,
+        KIND_PROJECT_STATE,
+        &relay_pubkey,
+        None,
+    )
+    .await;
+    Ok(true)
+}
+
+/// Publish the current projection for one accepted Project command, if pending.
+pub(crate) async fn publish_project_state_for_coordinate(
+    tenant: &TenantContext,
+    state: &Arc<AppState>,
+    project_owner: &[u8],
+    project_d_tag: &str,
+) -> anyhow::Result<bool> {
+    let relay_pubkey = state.relay_keypair.public_key().to_bytes();
+    let Some(candidate) = state
+        .db
+        .load_pending_project_state_projection(
+            tenant.community(),
+            project_owner,
+            project_d_tag,
+            &relay_pubkey,
+        )
+        .await
+        .context("load pending Project State projection")?
+    else {
+        return Ok(false);
+    };
+    publish_candidate(tenant, state, candidate).await
+}
+
+/// Repair a bounded batch of durable Project State publication markers.
+pub async fn reconcile_project_state_projections(state: &Arc<AppState>) -> anyhow::Result<usize> {
+    let relay_pubkey = state.relay_keypair.public_key().to_bytes();
+    let candidates = state
+        .db
+        .load_pending_project_state_projections(&relay_pubkey, RECONCILE_BATCH_SIZE)
+        .await
+        .context("load pending Project State projections")?;
+    let mut committed = 0;
+    for candidate in candidates {
+        let community_id = candidate.community_id();
+        let result = async {
+            let host = state
+                .db
+                .lookup_community_host(community_id)
+                .await?
+                .ok_or_else(|| anyhow!("Project projection community has no active host"))?;
+            let tenant = TenantContext::resolved(community_id, host);
+            publish_candidate(&tenant, state, candidate).await
+        }
+        .await;
+        match result {
+            Ok(true) => committed += 1,
+            Ok(false) => {}
+            Err(error) => {
+                tracing::warn!(%community_id, %error, "Project State projection repair failed")
+            }
+        }
+    }
+    Ok(committed)
+}
+
+/// Run periodic bounded repair until graceful shutdown is requested.
+pub fn spawn_project_state_projection_reconciler(
+    state: Arc<AppState>,
+    cancel: CancellationToken,
+) -> tokio::task::JoinHandle<()> {
+    tokio::spawn(async move {
+        match reconcile_project_state_projections(&state).await {
+            Ok(count) if count > 0 => {
+                tracing::info!(count, "Project State projections repaired on startup")
+            }
+            Ok(_) => {}
+            Err(error) => {
+                tracing::warn!(%error, "Project State startup reconciliation failed")
+            }
+        }
+
+        let mut interval = tokio::time::interval(RECONCILE_INTERVAL);
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+        interval.tick().await;
+        loop {
+            tokio::select! {
+                _ = cancel.cancelled() => break,
+                _ = interval.tick() => {
+                    match reconcile_project_state_projections(&state).await {
+                        Ok(count) if count > 0 => {
+                            tracing::info!(count, "Project State projections repaired")
+                        }
+                        Ok(_) => {}
+                        Err(error) => tracing::warn!(%error, "Project State projection reconciliation failed"),
+                    }
+                }
+            }
+        }
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn projection_timestamp_is_strictly_monotonic() {
+        assert_eq!(projection_created_at(100, None).expect("timestamp"), 100);
+        assert_eq!(
+            projection_created_at(100, Some(50)).expect("timestamp"),
+            100
+        );
+        assert_eq!(
+            projection_created_at(100, Some(100)).expect("timestamp"),
+            101
+        );
+        assert_eq!(
+            projection_created_at(100, Some(200)).expect("timestamp"),
+            201
+        );
+    }
+
+    #[test]
+    fn projection_timestamp_rejects_overflow() {
+        assert!(projection_created_at(100, Some(u64::MAX)).is_err());
+    }
+}

crates/buzz-relay/src/main.rs

@@ -467,6 +467,16 @@ async fn main() -> anyhow::Result<()> {
     );
     let state = Arc::new(app_state);
 
+    // Project changes commit independently of their relay-signed read model.
+    // Start best-effort repair immediately without making listener availability
+    // depend on it, then retry a bounded batch periodically.
+    let project_state_projection_cancel = CancellationToken::new();
+    let mut project_state_projection_task =
+        buzz_relay::handlers::project_state_projection::spawn_project_state_projection_reconciler(
+            Arc::clone(&state),
+            project_state_projection_cancel.clone(),
+        );
+
     // Inter-relay mesh (BUZZ_MESH seam). `boot_mesh` returns None when the
     // kill switch is off — nothing is bound, published, or spawned, so the
     // relay behaves byte-identically to a build without the mesh. When
@@ -1151,6 +1161,18 @@ async fn main() -> anyhow::Result<()> {
 
     serve(router, health_router, Arc::clone(&state)).await?;
     state.community_revalidator_cancel.cancel();
+    project_state_projection_cancel.cancel();
+    if tokio::time::timeout(
+        std::time::Duration::from_secs(5),
+        &mut project_state_projection_task,
+    )
+    .await
+    .is_err()
+    {
+        project_state_projection_task.abort();
+        let _ = project_state_projection_task.await;
+        tracing::warn!("Project State projection reconciler did not stop within the drain window");
+    }
 
     // Signal the audit worker to stop accepting, flush buffered entries, and
     // exit. Uses a CancellationToken so it works regardless of how many