diff --git a/src/spaces/notifications.rs b/src/spaces/notifications.rs
new file mode 100644
--- /dev/null
+++ b/src/spaces/notifications.rs
@@ -0,0 +1,86 @@
+use serde::Deserialize;
+use uuid::Uuid;
+
+use crate::db::DatabaseBackend;
+use crate::error::AppError;
+use crate::spaces::db;
+use crate::spaces::types::*;
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WriteNotification {
+    pub space_uri: String,
+    pub author_did: String,
+    pub collection: String,
+    pub rkey: String,
+    pub action: WriteAction,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum WriteAction {
+    Create,
+    Update,
+    Delete,
+}
+
+/// Process a write notification by queuing a sync pull for the affected member.
+///
+/// This marks the member's sync state as pending so the next sync pass picks it up.
+pub async fn handle_write_notification(
+    pool: &sqlx::AnyPool,
+    backend: DatabaseBackend,
+    space_id: &str,
+    notification: &WriteNotification,
+) -> Result<(), AppError> {
+    let existing = db::get_sync_state(pool, backend, space_id, &notification.author_did).await?;
+
+    let state = SpaceSyncState {
+        id: existing
+            .map(|s| s.id)
+            .unwrap_or_else(|| Uuid::new_v4().to_string()),
+        space_id: space_id.to_string(),
+        member_did: notification.author_did.clone(),
+        cursor: None,
+        last_synced_at: None,
+        status: SyncStatus::Pending,
+        error: None,
+    };
+
+    db::upsert_sync_state(pool, backend, &state).await?;
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn write_action_deserializes() {
+        let action: WriteAction = serde_json::from_str("\"create\"").unwrap();
+        assert!(matches!(action, WriteAction::Create));
+
+        let action: WriteAction = serde_json::from_str("\"update\"").unwrap();
+        assert!(matches!(action, WriteAction::Update));
+
+        let action: WriteAction = serde_json::from_str("\"delete\"").unwrap();
+        assert!(matches!(action, WriteAction::Delete));
+    }
+
+    #[test]
+    fn write_notification_deserializes() {
+        let json = r#"{
+            "spaceUri": "ats://did:plc:owner/com.example.forum/main",
+            "authorDid": "did:plc:alice",
+            "collection": "com.example.forum.post",
+            "rkey": "3k2abc",
+            "action": "create"
+        }"#;
+
+        let notif: WriteNotification = serde_json::from_str(json).unwrap();
+        assert_eq!(notif.author_did, "did:plc:alice");
+        assert_eq!(notif.collection, "com.example.forum.post");
+        assert!(matches!(notif.action, WriteAction::Create));
+    }
+}
diff --git a/src/spaces/sync.rs b/src/spaces/sync.rs
new file mode 100644
--- /dev/null
+++ b/src/spaces/sync.rs
@@ -0,0 +1,293 @@
+use uuid::Uuid;
+
+use crate::db::DatabaseBackend;
+use crate::db::now_rfc3339;
+use crate::error::AppError;
+use crate::profile::resolve_pds_endpoint;
+use crate::spaces::types::*;
+use crate::spaces::{db, members};
+
+/// Sync all members of a space by pulling records from their PDSes.
+pub async fn sync_space(
+    http: &reqwest::Client,
+    pool: &sqlx::AnyPool,
+    backend: DatabaseBackend,
+    plc_url: &str,
+    space_id: &str,
+    collections: &[String],
+) -> Result<SyncSpaceResult, AppError> {
+    let resolved = members::resolve_members(pool, backend, space_id).await?;
+    let mut results = Vec::new();
+
+    for member in &resolved {
+        let result = sync_member(
+            http,
+            pool,
+            backend,
+            plc_url,
+            space_id,
+            &member.did,
+            collections,
+        )
+        .await;
+
+        results.push(MemberSyncResult {
+            did: member.did.clone(),
+            records_synced: result.as_ref().map(|r| r.records_synced).unwrap_or(0),
+            error: result.err().map(|e| e.to_string()),
+        });
+    }
+
+    let total = results.iter().map(|r| r.records_synced).sum();
+
+    Ok(SyncSpaceResult {
+        members_processed: results.len(),
+        total_records_synced: total,
+        member_results: results,
+    })
+}
+
+/// Sync records from a single member's PDS for a given space.
+pub async fn sync_member(
+    http: &reqwest::Client,
+    pool: &sqlx::AnyPool,
+    backend: DatabaseBackend,
+    plc_url: &str,
+    space_id: &str,
+    member_did: &str,
+    collections: &[String],
+) -> Result<MemberSyncSummary, AppError> {
+    let state_id = match db::get_sync_state(pool, backend, space_id, member_did).await? {
+        Some(s) => s.id,
+        None => {
+            let id = Uuid::new_v4().to_string();
+            let initial = SpaceSyncState {
+                id: id.clone(),
+                space_id: space_id.to_string(),
+                member_did: member_did.to_string(),
+                cursor: None,
+                last_synced_at: None,
+                status: SyncStatus::Pending,
+                error: None,
+            };
+            db::upsert_sync_state(pool, backend, &initial).await?;
+            id
+        }
+    };
+
+    // Mark as syncing
+    let syncing_state = SpaceSyncState {
+        id: state_id.clone(),
+        space_id: space_id.to_string(),
+        member_did: member_did.to_string(),
+        cursor: None,
+        last_synced_at: None,
+        status: SyncStatus::Syncing,
+        error: None,
+    };
+    db::upsert_sync_state(pool, backend, &syncing_state).await?;
+
+    let result = pull_member_records(
+        http,
+        pool,
+        backend,
+        plc_url,
+        space_id,
+        member_did,
+        collections,
+    )
+    .await;
+
+    match result {
+        Ok(summary) => {
+            let done = SpaceSyncState {
+                id: state_id,
+                space_id: space_id.to_string(),
+                member_did: member_did.to_string(),
+                cursor: summary.cursor.clone(),
+                last_synced_at: Some(now_rfc3339()),
+                status: SyncStatus::Synced,
+                error: None,
+            };
+            db::upsert_sync_state(pool, backend, &done).await?;
+            Ok(summary)
+        }
+        Err(e) => {
+            let err_state = SpaceSyncState {
+                id: state_id,
+                space_id: space_id.to_string(),
+                member_did: member_did.to_string(),
+                cursor: None,
+                last_synced_at: Some(now_rfc3339()),
+                status: SyncStatus::Error,
+                error: Some(e.to_string()),
+            };
+            db::upsert_sync_state(pool, backend, &err_state).await?;
+            Err(e)
+        }
+    }
+}
+
+async fn pull_member_records(
+    http: &reqwest::Client,
+    pool: &sqlx::AnyPool,
+    backend: DatabaseBackend,
+    plc_url: &str,
+    space_id: &str,
+    member_did: &str,
+    collections: &[String],
+) -> Result<MemberSyncSummary, AppError> {
+    let pds_url = resolve_pds_endpoint(http, plc_url, member_did).await?;
+    let mut total_records = 0usize;
+    let mut last_cursor = None;
+
+    for collection in collections {
+        let mut cursor: Option<String> = None;
+        loop {
+            let (records, next_cursor) = fetch_records_page(
+                http,
+                &pds_url,
+                member_did,
+                collection,
+                cursor.as_deref(),
+                100,
+            )
+            .await?;
+
+            if records.is_empty() {
+                break;
+            }
+
+            for record in &records {
+                let uri = record["uri"].as_str().unwrap_or("");
+                let rkey = extract_rkey(uri);
+                let cid = record["cid"].as_str().unwrap_or("").to_string();
+                let value = record
+                    .get("value")
+                    .cloned()
+                    .unwrap_or(serde_json::Value::Null);
+
+                let space_record_uri = format!("ats://{space_id}/{member_did}/{collection}/{rkey}");
+
+                let space_record = SpaceRecord {
+                    uri: space_record_uri,
+                    space_id: space_id.to_string(),
+                    author_did: member_did.to_string(),
+                    collection: collection.clone(),
+                    rkey: rkey.to_string(),
+                    record: value,
+                    cid,
+                    indexed_at: now_rfc3339(),
+                };
+
+                db::upsert_space_record(pool, backend, &space_record).await?;
+                total_records += 1;
+            }
+
+            last_cursor = next_cursor.clone();
+            cursor = next_cursor;
+
+            if cursor.is_none() {
+                break;
+            }
+        }
+    }
+
+    Ok(MemberSyncSummary {
+        records_synced: total_records,
+        cursor: last_cursor,
+    })
+}
+
+async fn fetch_records_page(
+    http: &reqwest::Client,
+    pds_url: &str,
+    repo: &str,
+    collection: &str,
+    cursor: Option<&str>,
+    limit: u32,
+) -> Result<(Vec<serde_json::Value>, Option<String>), AppError> {
+    let mut url = format!(
+        "{}/xrpc/com.atproto.repo.listRecords?repo={}&collection={}&limit={}",
+        pds_url.trim_end_matches('/'),
+        repo,
+        collection,
+        limit,
+    );
+
+    if let Some(c) = cursor {
+        url.push_str(&format!("&cursor={c}"));
+    }
+
+    let resp = http
+        .get(&url)
+        .send()
+        .await
+        .map_err(|e| AppError::Internal(format!("PDS request failed: {e}")))?;
+
+    if !resp.status().is_success() {
+        let status = resp.status();
+        return Err(AppError::Internal(format!(
+            "PDS listRecords failed with {status} for {repo}/{collection}"
+        )));
+    }
+
+    let body: serde_json::Value = resp
+        .json()
+        .await
+        .map_err(|e| AppError::Internal(format!("invalid PDS response: {e}")))?;
+
+    let records = body["records"].as_array().cloned().unwrap_or_default();
+
+    let next_cursor = body["cursor"].as_str().map(|s| s.to_string());
+
+    Ok((records, next_cursor))
+}
+
+fn extract_rkey(uri: &str) -> &str {
+    uri.rsplit('/').next().unwrap_or("")
+}
+
+// ---------------------------------------------------------------------------
+// Result types
+// ---------------------------------------------------------------------------
+
+pub struct SyncSpaceResult {
+    pub members_processed: usize,
+    pub total_records_synced: usize,
+    pub member_results: Vec<MemberSyncResult>,
+}
+
+pub struct MemberSyncResult {
+    pub did: String,
+    pub records_synced: usize,
+    pub error: Option<String>,
+}
+
+pub struct MemberSyncSummary {
+    pub records_synced: usize,
+    pub cursor: Option<String>,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn extract_rkey_from_at_uri() {
+        assert_eq!(
+            extract_rkey("at://did:plc:abc/app.bsky.feed.post/3k2abc"),
+            "3k2abc"
+        );
+    }
+
+    #[test]
+    fn extract_rkey_from_empty() {
+        assert_eq!(extract_rkey(""), "");
+    }
+
+    #[test]
+    fn extract_rkey_no_slash() {
+        assert_eq!(extract_rkey("singlevalue"), "singlevalue");
+    }
+}