feat: multi-authority space host with dynamic discovery and projection syncer

Cargo.toml

@@ -17,6 +17,7 @@ members = [
   "rsky-satnav",
   "rsky-space",
   "rsky-space-host",
+  "rsky-spaces-parity",
   "rsky-syntax",
   "rsky-video",
   "rsky-wintermute",

rsky-daemon/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rsky-daemon"
-version = "0.4.0"
+version = "0.6.3"
 authors = ["Rudy Fraser <him@rudyfraser.com>"]
 description = "atproto permissioned-data syncer daemon: pulls, verifies, and indexes permissioned repos from members' PDSes"
 edition = "2021"
@@ -36,6 +36,7 @@ chrono = { version = "0.4", features = ["serde"] }
 axum = "0.7"
 base64 = "0.22"
 sha2 = { workspace = true }
+secp256k1 = { workspace = true }
 
 [dev-dependencies]
 secp256k1 = { workspace = true }

rsky-daemon/Dockerfile

@@ -0,0 +1,82 @@
+# Use the official Rust image.
+# https://hub.docker.com/_/rust
+FROM rust AS builder
+
+WORKDIR /usr/src/rsky
+
+# Copy workspace and all crate manifests for dependency resolution
+COPY Cargo.toml Cargo.lock rust-toolchain ./
+COPY palomar-sync/Cargo.toml palomar-sync/Cargo.toml
+COPY rsky-common/Cargo.toml rsky-common/Cargo.toml
+COPY rsky-crypto/Cargo.toml rsky-crypto/Cargo.toml
+COPY rsky-daemon/Cargo.toml rsky-daemon/Cargo.toml
+COPY rsky-feedgen/Cargo.toml rsky-feedgen/Cargo.toml
+COPY rsky-firehose/Cargo.toml rsky-firehose/Cargo.toml
+COPY rsky-identity/Cargo.toml rsky-identity/Cargo.toml
+COPY rsky-jetstream-subscriber/Cargo.toml rsky-jetstream-subscriber/Cargo.toml
+COPY rsky-labeler/Cargo.toml rsky-labeler/Cargo.toml
+COPY rsky-lexicon/Cargo.toml rsky-lexicon/Cargo.toml
+COPY rsky-oauth/Cargo.toml rsky-oauth/Cargo.toml
+COPY rsky-pds/Cargo.toml rsky-pds/Cargo.toml
+COPY rsky-relay/Cargo.toml rsky-relay/Cargo.toml
+COPY rsky-repo/Cargo.toml rsky-repo/Cargo.toml
+COPY rsky-satnav/Cargo.toml rsky-satnav/Cargo.toml
+COPY rsky-space/Cargo.toml rsky-space/Cargo.toml
+COPY rsky-space-host/Cargo.toml rsky-space-host/Cargo.toml
+COPY rsky-spaces-parity/Cargo.toml rsky-spaces-parity/Cargo.toml
+COPY rsky-syntax/Cargo.toml rsky-syntax/Cargo.toml
+COPY rsky-video/Cargo.toml rsky-video/Cargo.toml
+COPY rsky-wintermute/Cargo.toml rsky-wintermute/Cargo.toml
+
+# Copy real source for library crates that rsky-daemon depends on
+COPY rsky-common/src rsky-common/src
+COPY rsky-crypto/src rsky-crypto/src
+COPY rsky-identity/src rsky-identity/src
+COPY rsky-lexicon/src rsky-lexicon/src
+COPY rsky-oauth/src rsky-oauth/src
+COPY rsky-space/src rsky-space/src
+COPY rsky-syntax/src rsky-syntax/src
+
+# Stub out the remaining workspace members so cargo can resolve the workspace
+RUN mkdir -p \
+    palomar-sync/src rsky-feedgen/src rsky-firehose/src \
+    rsky-jetstream-subscriber/src rsky-labeler/src rsky-pds/src \
+    rsky-relay/src rsky-repo/src rsky-satnav/src rsky-space-host/src \
+    rsky-video/src rsky-wintermute/src && \
+    for crate in palomar-sync rsky-feedgen rsky-firehose \
+        rsky-jetstream-subscriber rsky-labeler rsky-pds rsky-relay \
+        rsky-satnav rsky-space-host rsky-video rsky-wintermute; do \
+        echo 'fn main() {}' > $crate/src/main.rs; \
+    done && \
+    touch rsky-pds/src/lib.rs rsky-repo/src/lib.rs && \
+    mkdir -p rsky-spaces-parity/src && touch rsky-spaces-parity/src/lib.rs && \
+    mkdir -p rsky-wintermute/src/bin && \
+    for bin in queue_backfill fix_blob_refs plc_import label_sync car_loader \
+        cleanup_stale_deactivated; do \
+        echo 'fn main() {}' > rsky-wintermute/src/bin/$bin.rs; \
+    done
+
+# Create an empty src directory to trick Cargo into thinking it's a valid Rust project
+RUN mkdir -p rsky-daemon/src && echo "fn main() {}" > rsky-daemon/src/main.rs
+
+# Install production dependencies and build a release artifact.
+RUN cargo build --release --package rsky-daemon
+
+# Now copy the real source code and build the final binary
+COPY rsky-daemon/src rsky-daemon/src
+
+RUN cargo build --release --package rsky-daemon
+
+FROM debian:bookworm-slim
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends ca-certificates && \
+    rm -rf /var/lib/apt/lists/*
+WORKDIR /usr/src/rsky
+COPY --from=builder /usr/src/rsky/target/release/rsky-daemon rsky-daemon
+LABEL org.opencontainers.image.source=https://github.com/blacksky-algorithms/rsky
+# State is sqlite + a persisted key: mount a volume covering DAEMON_INDEX_DB_PATH
+# and DAEMON_DPOP_KEY_PATH. Required env: DAEMON_SPACE_HOST_URL,
+# DAEMON_SERVICE_IDENTITY, DAEMON_SERVICE_SIGNING_KEY_HEX,
+# DAEMON_SPACE_HOST_MINT_TOKEN, and DAEMON_SPACE_URI and/or DAEMON_SPACES_URL
+# (+ DAEMON_SPACES_API_KEY). See rsky-daemon/src/config.rs for the full reference.
+CMD ["sh", "-c", "DAEMON_NOTIFY_BIND=${DAEMON_NOTIFY_BIND:-0.0.0.0:8055} ./rsky-daemon"]

rsky-daemon/src/appview.rs

@@ -0,0 +1,271 @@
+//! Projection to the appview's record ingress.
+
+use async_trait::async_trait;
+use serde_json::Value;
+use std::collections::BTreeMap;
+
+use crate::error::Result;
+use crate::feeds::{ProjectRecord, ProjectRecordsRequest, ProjectionIngress, ProjectionOperation};
+use crate::projection::Projector;
+use crate::router::{SyncEvent, LIKE_COLLECTION, MODERATION_ACTION_COLLECTION, POST_COLLECTION};
+
+/// Projects a space's posts, likes and moderation state to the appview.
+pub struct AppviewProjector<I: ProjectionIngress> {
+    ingress: I,
+    space: String,
+}
+
+impl<I: ProjectionIngress> AppviewProjector<I> {
+    pub fn new(ingress: I, space: impl Into<String>) -> Self {
+        Self {
+            ingress,
+            space: space.into(),
+        }
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    fn op(
+        &self,
+        author: &str,
+        uri: &str,
+        cid: Option<&str>,
+        revision: &str,
+        operation: ProjectionOperation,
+        collection: &str,
+        record: Option<Value>,
+    ) -> ProjectRecord {
+        ProjectRecord {
+            space: self.space.clone(),
+            author: author.to_string(),
+            uri: uri.to_string(),
+            cid: cid.map(str::to_string),
+            revision: revision.to_string(),
+            operation,
+            collection: collection.to_string(),
+            record,
+            action_uri: None,
+            neg: None,
+        }
+    }
+
+    fn moderation_op(
+        &self,
+        author: &str,
+        revision: &str,
+        uri: &str,
+        record: &Value,
+    ) -> Option<ProjectRecord> {
+        if record["val"].as_str() != Some("remove") {
+            return None;
+        }
+        // A lift is its own signed record carrying the action it reverses,
+        // so it names that action's URI rather than its own.
+        let neg = record["neg"].as_bool() == Some(true);
+        let (action_uri, cid) = if neg {
+            (record["action"]["uri"].as_str(), None)
+        } else {
+            (Some(uri), record["subject"]["cid"].as_str())
+        };
+        Some(ProjectRecord {
+            action_uri: Some(action_uri?.to_string()),
+            neg: Some(neg),
+            ..self.op(
+                author,
+                uri,
+                cid,
+                revision,
+                if neg {
+                    ProjectionOperation::Unflag
+                } else {
+                    ProjectionOperation::Flag
+                },
+                MODERATION_ACTION_COLLECTION,
+                None,
+            )
+        })
+    }
+}
+
+#[async_trait]
+impl<I: ProjectionIngress> Projector for AppviewProjector<I> {
+    fn name(&self) -> &'static str {
+        "appview"
+    }
+
+    async fn project(&self, author: &str, revision: &str, events: &[SyncEvent]) -> Result<()> {
+        // One URI may change more than once in a batch; only its final state
+        // is worth sending.
+        let mut final_records: BTreeMap<&str, &SyncEvent> = BTreeMap::new();
+        for event in events {
+            final_records.insert(event.uri(), event);
+        }
+        let ops: Vec<ProjectRecord> = final_records
+            .into_values()
+            .filter_map(|event| match event {
+                SyncEvent::PostCreated {
+                    uri, cid, record, ..
+                } => Some(self.op(
+                    author,
+                    uri,
+                    Some(cid),
+                    revision,
+                    ProjectionOperation::Create,
+                    POST_COLLECTION,
+                    Some(record.clone()),
+                )),
+                SyncEvent::PostDeleted { uri, .. } => Some(self.op(
+                    author,
+                    uri,
+                    None,
+                    revision,
+                    ProjectionOperation::Delete,
+                    POST_COLLECTION,
+                    None,
+                )),
+                SyncEvent::LikeCreated {
+                    uri, cid, record, ..
+                } => Some(self.op(
+                    author,
+                    uri,
+                    Some(cid),
+                    revision,
+                    ProjectionOperation::Create,
+                    LIKE_COLLECTION,
+                    Some(record.clone()),
+                )),
+                SyncEvent::LikeDeleted { uri, .. } => Some(self.op(
+                    author,
+                    uri,
+                    None,
+                    revision,
+                    ProjectionOperation::Delete,
+                    LIKE_COLLECTION,
+                    None,
+                )),
+                SyncEvent::ModerationAction {
+                    uri,
+                    record: Some(record),
+                    ..
+                } => self.moderation_op(author, revision, uri, record),
+                SyncEvent::ModerationAction { record: None, .. } => None,
+            })
+            .collect();
+        if ops.is_empty() {
+            return Ok(());
+        }
+        self.ingress
+            .project_records(&ProjectRecordsRequest { ops })
+            .await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::feeds::tests::{post_created, CapturingIngress, AUTHOR, SPACE};
+    use serde_json::json;
+
+    fn like_created(rkey: &str) -> SyncEvent {
+        SyncEvent::LikeCreated {
+            uri: format!("{SPACE}/{AUTHOR}/{LIKE_COLLECTION}/{rkey}"),
+            author: AUTHOR.to_string(),
+            cid: "bafylike".to_string(),
+            rev: "3krev".to_string(),
+            record: json!({
+                "$type": LIKE_COLLECTION,
+                "subject": {"uri": format!("{SPACE}/{AUTHOR}/{POST_COLLECTION}/3ka"), "cid": "bafypost"},
+                "createdAt": "2026-08-19T00:00:00Z",
+            }),
+        }
+    }
+
+    #[tokio::test]
+    async fn posts_likes_and_removals_all_project() {
+        let projector = AppviewProjector::new(CapturingIngress::default(), SPACE);
+        let action_uri = format!("{SPACE}/did:plc:community/{MODERATION_ACTION_COLLECTION}/3m");
+        projector
+            .project(
+                AUTHOR,
+                "3krev",
+                &[
+                    post_created("3ka", "hello"),
+                    like_created("3kl"),
+                    SyncEvent::ModerationAction {
+                        uri: action_uri.clone(),
+                        rev: "3krev".to_string(),
+                        record: Some(json!({
+                            "val": "remove",
+                            "subject": {"uri": format!("{SPACE}/{AUTHOR}/{POST_COLLECTION}/3ka"), "cid": "bafypost"},
+                        })),
+                    },
+                ],
+            )
+            .await
+            .unwrap();
+
+        let sent = projector.ingress.sent.lock().unwrap();
+        assert_eq!(sent.len(), 1);
+        let ops = &sent[0].ops;
+        assert_eq!(ops.len(), 3);
+        let collections: Vec<&str> = ops.iter().map(|op| op.collection.as_str()).collect();
+        assert!(collections.contains(&POST_COLLECTION));
+        assert!(collections.contains(&LIKE_COLLECTION));
+        assert!(collections.contains(&MODERATION_ACTION_COLLECTION));
+        let flagged = ops
+            .iter()
+            .find(|op| op.collection == MODERATION_ACTION_COLLECTION)
+            .unwrap();
+        assert_eq!(flagged.operation, ProjectionOperation::Flag);
+        assert_eq!(flagged.action_uri.as_deref(), Some(action_uri.as_str()));
+        let like = ops
+            .iter()
+            .find(|op| op.collection == LIKE_COLLECTION)
+            .unwrap();
+        assert!(like.record.as_ref().unwrap()["subject"]["uri"]
+            .as_str()
+            .unwrap()
+            .contains("/space/"));
+    }
+
+    #[tokio::test]
+    async fn only_the_final_state_of_a_uri_is_sent() {
+        let projector = AppviewProjector::new(CapturingIngress::default(), SPACE);
+        let uri = format!("{SPACE}/{AUTHOR}/{POST_COLLECTION}/3ka");
+        projector
+            .project(
+                AUTHOR,
+                "3krev",
+                &[
+                    post_created("3ka", "first"),
+                    SyncEvent::PostDeleted {
+                        uri: uri.clone(),
+                        author: AUTHOR.to_string(),
+                    },
+                ],
+            )
+            .await
+            .unwrap();
+
+        let sent = projector.ingress.sent.lock().unwrap();
+        assert_eq!(sent[0].ops.len(), 1);
+        assert_eq!(sent[0].ops[0].operation, ProjectionOperation::Delete);
+    }
+
+    #[tokio::test]
+    async fn a_batch_with_nothing_to_send_makes_no_request() {
+        let projector = AppviewProjector::new(CapturingIngress::default(), SPACE);
+        projector
+            .project(
+                AUTHOR,
+                "3krev",
+                &[SyncEvent::ModerationAction {
+                    uri: format!("{SPACE}/did:plc:community/{MODERATION_ACTION_COLLECTION}/3m"),
+                    rev: "3krev".to_string(),
+                    record: None,
+                }],
+            )
+            .await
+            .unwrap();
+        assert!(projector.ingress.sent.lock().unwrap().is_empty());
+    }
+}