feat(knot): repository crud

Signed-off-by: tjh <did:plc:65gha4t3avpfpzmvpbwovss7>

.sqlx/query-172ef5110ff75386a02a51265130c0e6ca58abe13a5d25c7787fa8f5c3d2c330.json

···
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT member_did FROM knot_member WHERE instance_name = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "member_did",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "172ef5110ff75386a02a51265130c0e6ca58abe13a5d25c7787fa8f5c3d2c330"
+}

.sqlx/query-17a82709db35100b19d26fb327c0f21f0fab6809185d0a3cd817092e34de75d2.json

···
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM repository WHERE did = $1 AND rkey = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "17a82709db35100b19d26fb327c0f21f0fab6809185d0a3cd817092e34de75d2"
+}

.sqlx/query-32f4262d9e9611fd25095a1f369c17a3645627fecd4c8e77bcf5a9e071d55988.json

···
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT rkey FROM repository WHERE did = $1 AND (rkey = $2 OR name = $2)",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "rkey",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "32f4262d9e9611fd25095a1f369c17a3645627fecd4c8e77bcf5a9e071d55988"
+}

.sqlx/query-41661dd412f5ae03fdf1e754a6011dd755ffeed6573029a726accb441329d1ad.json

···
-{
-  "db_name": "PostgreSQL",
-  "query": "INSERT INTO repository\n    (did, rkey, cid, name, knot, spindle, description, website, topics, source, labels, created_at, xrpc_create_at, jetstream_at)\nVALUES\n    ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)\nON CONFLICT\n    ON CONSTRAINT repository_pkey\n    DO UPDATE\n        SET jetstream_at = $14 WHERE repository.jetstream_at IS NULL\n    RETURNING\n        name, xrpc_create_at, OLD.jetstream_at as old_jetstream_at, NEW.jetstream_at AS new_jetstream_at\n",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "name",
-        "type_info": "Text"
-      },
-      {
-        "ordinal": 1,
-        "name": "xrpc_create_at",
-        "type_info": "Timestamptz"
-      },
-      {
-        "ordinal": 2,
-        "name": "old_jetstream_at",
-        "type_info": "Timestamptz"
-      },
-      {
-        "ordinal": 3,
-        "name": "new_jetstream_at",
-        "type_info": "Timestamptz"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Text",
-        "Text",
-        "Text",
-        "Text",
-        "Text",
-        "Text",
-        "Text",
-        "TextArray",
-        "Text",
-        "TextArray",
-        "Timestamptz",
-        "Timestamptz",
-        "Timestamptz"
-      ]
-    },
-    "nullable": [
-      false,
-      true,
-      true,
-      true
-    ]
-  },
-  "hash": "41661dd412f5ae03fdf1e754a6011dd755ffeed6573029a726accb441329d1ad"
-}

.sqlx/query-503462e658e2d8ebe9bdb3f22f486ad7744200182214262f9892c0e772a72450.json

···
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE repository\nSET\n    cid = $3,\n    name = $4,\n    knot = $5,\n    spindle = $6,\n    description = $7,\n    website = $8,\n    topics = $9,\n    source = $10,\n    labels = $11,\n    created_at = $12\nWHERE did = $1 AND rkey = $2\n",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "TextArray",
+        "Text",
+        "TextArray",
+        "Timestamptz"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "503462e658e2d8ebe9bdb3f22f486ad7744200182214262f9892c0e772a72450"
+}

.sqlx/query-6cf7d307f8c56a10893f384c52e5c1ecfd826dcfa5c3dcf3b07e05599e85a5e0.json

···
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO repository\n    (did, rkey, cid, name, knot, spindle, description, website, topics, source, labels, created_at, xrpc_create_at, jetstream_at)\nVALUES\n    ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)\nON CONFLICT\n    ON CONSTRAINT repository_pkey\n    DO UPDATE\n        SET jetstream_at = $14 WHERE repository.jetstream_at IS NULL\n    RETURNING\n        name,\n        OLD.xrpc_create_at AS old_xrpc_create_at,\n        NEW.xrpc_create_at AS new_xrpc_create_at,\n        OLD.jetstream_at as old_jetstream_at,\n        NEW.jetstream_at AS new_jetstream_at\n",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "name",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 1,
+        "name": "old_xrpc_create_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 2,
+        "name": "new_xrpc_create_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 3,
+        "name": "old_jetstream_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 4,
+        "name": "new_jetstream_at",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "TextArray",
+        "Text",
+        "TextArray",
+        "Timestamptz",
+        "Timestamptz",
+        "Timestamptz"
+      ]
+    },
+    "nullable": [
+      false,
+      true,
+      true,
+      true,
+      true
+    ]
+  },
+  "hash": "6cf7d307f8c56a10893f384c52e5c1ecfd826dcfa5c3dcf3b07e05599e85a5e0"
+}

.sqlx/query-789b78b11b38f55943a17f364c2000d9fa39b6ec75520805ff1e6e7aacb04297.json

···
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT member_did FROM repository_member WHERE repo_did = $1 AND repo_rkey = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "member_did",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "789b78b11b38f55943a17f364c2000d9fa39b6ec75520805ff1e6e7aacb04297"
+}

.sqlx/query-78db62d7d801f1b7a69ccde8738bfaa45b4c0899e2488692144e342686930147.json

···
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO repository\n    (did, rkey, cid, name, knot, spindle, description, website, topics, source, labels, created_at, xrpc_create_at, jetstream_at)\nVALUES\n    ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)\nON CONFLICT\n    ON CONSTRAINT repository_pkey\n    DO UPDATE\n        SET xrpc_create_at = $14 WHERE repository.xrpc_create_at IS NULL\n    RETURNING\n        name,\n        OLD.xrpc_create_at AS old_xrpc_create_at,\n        NEW.xrpc_create_at AS new_xrpc_create_at,\n        OLD.jetstream_at as old_jetstream_at,\n        NEW.jetstream_at AS new_jetstream_at\n",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "name",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 1,
+        "name": "old_xrpc_create_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 2,
+        "name": "new_xrpc_create_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 3,
+        "name": "old_jetstream_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 4,
+        "name": "new_jetstream_at",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "TextArray",
+        "Text",
+        "TextArray",
+        "Timestamptz",
+        "Timestamptz",
+        "Timestamptz"
+      ]
+    },
+    "nullable": [
+      false,
+      true,
+      true,
+      true,
+      true
+    ]
+  },
+  "hash": "78db62d7d801f1b7a69ccde8738bfaa45b4c0899e2488692144e342686930147"
+}

crates/knot/Cargo.toml

···
 data-encoding.workspace = true
 futures-util = "0.3.31"
 hyper-util = { version = "0.1.17", features = ["client"] }
+rayon = "1.11.0"
 rustc-hash = "2.1.1"
+sqlx = { version = "0.8.6", features = ["runtime-tokio", "tls-native-tls", "postgres", "time", "json", "macros", "derive"] }
 time.workspace = true
 tokio = { version = "1.47.1", features = ["io-util", "macros", "net", "process", "signal", "rt-multi-thread"] }
+tokio-rayon = "2.1.0"
 tokio-stream = { version = "0.1.17", features = ["time"] }
 tokio-tungstenite = "0.28.0"
 tower = { version = "0.5.2", features = ["buffer", "filter", "limit"] }
 tower-http = { version = "0.6.6", features = ["decompression-gzip", "request-id", "trace", "tracing"] }
 tracing-subscriber = { version = "0.3.20", features = ["env-filter"] }
-rayon = "1.11.0"
-tokio-rayon = "2.1.0"
-sqlx = { version = "0.8.6", features = ["runtime-tokio", "tls-native-tls", "postgres", "time", "json", "macros", "derive"] }
 
 [dev-dependencies]
 http-body-util = "0.1.3"

crates/knot/migrations/20251109121913_members.down.sql

···
+DROP TABLE repository_member;
+DROP TABLE knot_member;

crates/knot/migrations/20251109121913_members.up.sql

···
+CREATE TABLE repository_member (
+    repo_did   text NOT NULL,
+    repo_rkey  text NOT NULL,
+    member_did text NOT NULL,    
+
+    PRIMARY KEY (repo_did, repo_rkey, member_did),
+    FOREIGN KEY (repo_did, repo_rkey) REFERENCES repository (did, rkey) ON DELETE CASCADE
+);
+
+CREATE TABLE knot_member (
+    instance_name text NOT NULL,
+    member_did    text NOT NULL,
+
+    PRIMARY KEY (instance_name, member_did)
+);

crates/knot/sql/insert_repository.sql crates/knot/sql/insert_repository_jetstream.sql

···
     DO UPDATE
         SET jetstream_at = $14 WHERE repository.jetstream_at IS NULL
     RETURNING
-        name, xrpc_create_at, OLD.jetstream_at as old_jetstream_at, NEW.jetstream_at AS new_jetstream_at
+        name,
+        OLD.xrpc_create_at AS old_xrpc_create_at,
+        NEW.xrpc_create_at AS new_xrpc_create_at,
+        OLD.jetstream_at as old_jetstream_at,
+        NEW.jetstream_at AS new_jetstream_at

crates/knot/sql/insert_repository_xrpc.sql

···
+INSERT INTO repository
+    (did, rkey, cid, name, knot, spindle, description, website, topics, source, labels, created_at, xrpc_create_at, jetstream_at)
+VALUES
+    ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
+ON CONFLICT
+    ON CONSTRAINT repository_pkey
+    DO UPDATE
+        SET xrpc_create_at = $14 WHERE repository.xrpc_create_at IS NULL
+    RETURNING
+        name,
+        OLD.xrpc_create_at AS old_xrpc_create_at,
+        NEW.xrpc_create_at AS new_xrpc_create_at,
+        OLD.jetstream_at as old_jetstream_at,
+        NEW.jetstream_at AS new_jetstream_at

crates/knot/sql/update_repository.sql

···
+UPDATE repository
+SET
+    cid = $3,
+    name = $4,
+    knot = $5,
+    spindle = $6,
+    description = $7,
+    website = $8,
+    topics = $9,
+    source = $10,
+    labels = $11,
+    created_at = $12
+WHERE did = $1 AND rkey = $2

crates/knot/src/model.rs

···
 pub mod repository;
 
 use core::ops;
-use std::{path::PathBuf, sync::Arc};
+use std::sync::Arc;
 
 use atproto::did::Did;
 use axum::{
···
 
 use crate::{
     public::git::{Error, GitAuthorization, NotFound},
-    services::authorization::AuthorizationClaimsStore,
+    services::{
+        authorization::AuthorizationClaimsStore,
+        database::DataStoreError,
+        rbac::{Policy, RepositoryPushPolicy, RepositoryRef},
+    },
 };
 
 pub use knot_state::KnotState;
···
     }
 }
 
-pub trait RepositoryManager {
+#[derive(Debug, Clone)]
+pub struct ResolvedRepoPath {
+    pub owner: Box<Did>,
+    pub rkey: Box<str>,
+}
+
+pub trait RepositoryManager: Send + Sync {
     type Error;
 
-    fn resolve_repo_path<'a>(&'a self, repo: &'a RepoPath) -> BoxFuture<'a, Option<PathBuf>>;
+    fn resolve_repo_path<'a>(
+        &'a self,
+        repo: &'a RepoPath,
+    ) -> BoxFuture<'a, Result<ResolvedRepoPath, Self::Error>>;
 
     fn open<'a>(
         &'a self,
···
 
     /// Determine whether the specified account has permission to push to the
     /// specified repository.
-    fn can_push(&self, _repo: &RepoPath, _did: &Did) -> BoxFuture<'_, bool> {
-        async { false }.boxed()
-    }
+    fn can_push<'s: 'a, 'a>(&'s self, repo: &'a RepoPath, did: &'a Did) -> BoxFuture<'a, bool>;
 }
 
-impl RepositoryManager for KnotState {
-    type Error = Box<gix::open::Error>;
+#[derive(Debug, thiserror::Error)]
+pub enum RepositoryMangerError {
+    #[error("Failed to resolve identity: {0}")]
+    Resolve(#[from] identity::ResolveError),
+    #[error("Failed to lookup respository: {0}")]
+    Lookup(#[from] DataStoreError),
+    #[error("Failed to open repository: {0}")]
+    Open(#[from] gix::open::Error),
+    #[error("Repository not found")]
+    NotFound,
+}
 
-    fn resolve_repo_path<'a>(&'a self, repo: &'a RepoPath) -> BoxFuture<'a, Option<PathBuf>> {
+impl RepositoryManager for Knot {
+    type Error = RepositoryMangerError;
+
+    fn resolve_repo_path<'a>(
+        &'a self,
+        repo: &'a RepoPath,
+    ) -> BoxFuture<'a, Result<ResolvedRepoPath, Self::Error>> {
         use std::borrow::Cow;
 
+        let cache_key = (repo.owner.clone(), repo.name.clone());
+        if let Some(resolved) = self.repo_cache.read().unwrap().get(&cache_key).cloned() {
+            return async move { Ok(resolved) }.boxed();
+        }
+
         async move {
-            let repo_owner_did = match Did::parse(&repo.owner) {
+            let owner = match Did::parse(&repo.owner) {
                 Ok(did) => Cow::Borrowed(did),
                 Err(_) => {
                     // Assume the repo owner is a handle.
-                    let (did, _) = self.resolver().resolve(&repo.owner).await.ok()?;
+                    let (did, _) = self.resolver().resolve(&repo.owner).await?;
                     Cow::Owned(did)
                 }
             };
 
-            Some(
-                self.repository_path()
-                    .join(repo_owner_did.as_str())
-                    .join(repo.name()),
-            )
+            let rkey = self
+                .store()
+                .resolve_repository(&owner, repo.name())
+                .await?
+                .ok_or(RepositoryMangerError::NotFound)?;
+
+            let resolved = ResolvedRepoPath {
+                owner: owner.into_owned(),
+                rkey,
+            };
+
+            self.repo_cache
+                .write()
+                .unwrap()
+                .insert(cache_key, resolved.clone());
+
+            Ok(resolved)
         }
         .boxed()
     }
···
     ) -> BoxFuture<'a, Result<gix::Repository, Self::Error>> {
         use gix::open::Options;
 
+        let path = self.repository_path().join(repo.owner()).join(repo.name());
+        if let Ok(true) = std::fs::exists(&path) {
+            return async move {
+                let repository = Options::default()
+                    .strict_config(true)
+                    .open_path_as_is(true)
+                    .open(path)?;
+
+                let local = repository.to_thread_local();
+                assert!(local.is_bare());
+
+                Ok(local)
+            }
+            .boxed();
+        }
+
         async move {
+            let resolved = self.resolve_repo_path(repo).await?;
+
             let path = self
-                .resolve_repo_path(repo)
-                .await
-                .ok_or(Box::new(gix::open::Error::Io(std::io::Error::new(
-                    std::io::ErrorKind::NotFound,
-                    format!("failed to resolve repository owner '{}'", repo.owner),
-                ))))?;
+                .repository_path()
+                .join(resolved.owner.as_str())
+                .join(resolved.rkey.as_ref());
 
             let repository = Options::default()
                 .strict_config(true)
···
             assert!(local.is_bare());
 
             Ok(local)
+        }
+        .boxed()
+    }
+
+    fn can_push<'s: 'a, 'a>(&'s self, repo: &'a RepoPath, did: &'a Did) -> BoxFuture<'a, bool> {
+        async move {
+            if let Ok(resolved) = self.resolve_repo_path(repo).await {
+                let policy = RepositoryPushPolicy;
+                let repository = RepositoryRef::new(&resolved.owner, &resolved.rkey);
+                let result = policy
+                    .evaluate_access(
+                        &did,
+                        &crate::services::rbac::Action::RepositoryPush,
+                        &repository,
+                        self,
+                    )
+                    .await;
+
+                return matches!(result, crate::services::rbac::PolicyResult::Granted);
+            }
+
+            false
         }
         .boxed()
     }

crates/knot/src/model/knot_state.rs

···
 use std::{
     collections::HashMap,
     ops,
-    sync::{Arc, Mutex, MutexGuard},
+    sync::{Arc, Mutex, MutexGuard, RwLock},
     time::Duration,
 };
 
 use atproto::did::Did;
+use bytes::Bytes;
 use identity::{HttpClient, Resolver};
 use jetstream::JetstreamClient;
 use lexicon::com::atproto::repo::list_records;
···
 
 use crate::services::{authorization::AuthorizationClaimsStore, database::DataStore};
 
-use super::config::KnotConfiguration;
+use super::{ResolvedRepoPath, config::KnotConfiguration};
 
 #[derive(Debug)]
 pub struct KnotState {
     config: KnotConfiguration,
     jwt_claims: Mutex<HashMap<Box<str>, oauth::jwt::Claims>>,
+    pub(crate) repo_cache: RwLock<HashMap<(Box<str>, Box<str>), ResolvedRepoPath>>,
     public_http: reqwest::Client,
     resolver: Resolver,
     _jetstream: JetstreamClient,
···
 
         let inner = Arc::new(Self {
             config,
+            jwt_claims: Default::default(),
+            repo_cache: Default::default(),
             public_http,
             resolver,
             _jetstream: jetstream,
             store: database,
-            jwt_claims: Default::default(),
             pool,
         });
 
···
                     .ok()
             })
             .collect())
+    }
+
+    pub async fn fetch_pds_record(
+        &self,
+        did: &Did,
+        collection: &str,
+        rkey: &str,
+    ) -> anyhow::Result<Bytes> {
+        use url::Url;
+
+        fn get_record_url(mut pds: Url, repo: &Did, collection: &str, rkey: &str) -> Url {
+            pds.set_path("/xrpc/com.atproto.repo.getRecord");
+            let mut query = pds.query_pairs_mut();
+            query.append_pair("repo", repo.as_str());
+            query.append_pair("collection", collection);
+            query.append_pair("rkey", rkey);
+            drop(query);
+            pds
+        }
+
+        let (_, doc) = self.resolver.resolve(did.as_str()).await?;
+        let pds = &doc
+            .atproto_pds()
+            .ok_or(anyhow::anyhow!("DID document does not declare a pds"))?
+            .service_endpoint;
+
+        let response = self
+            .public_http
+            .get(get_record_url(pds.clone(), did, collection, rkey))
+            .send()
+            .await?
+            .error_for_status()?
+            .bytes()
+            .await?;
+
+        Ok(response)
+    }
+
+    pub fn create_repo(&self, did: &Did, rkey: &str, name: &str) -> anyhow::Result<()> {
+        let path = self.repository_path().join(did.as_str()).join(rkey);
+        let repo = gix::init_bare(&path)?;
+        tracing::info!(?repo, "created repository");
+
+        // Create a symlink to map the repository name -> rkey.
+        let symlink_path = self.repository_path().join(did.as_str()).join(name);
+        let _ = std::fs::remove_file(&symlink_path);
+        std::os::unix::fs::symlink(rkey, &symlink_path)?;
+
+        Ok(())
+    }
+
+    pub fn delete_repo(&self, did: &Did, rkey: &str) -> anyhow::Result<()> {
+        let path = self.repository_path().join(did.as_str()).join(rkey);
+
+        let mut delete_path = self.repository_path().join("deleted");
+        let _ = std::fs::create_dir_all(&delete_path);
+
+        delete_path.push(format!("{}-{}", did, rkey));
+        std::fs::rename(&path, &delete_path)?;
+        Ok(())
     }
 }
 

crates/knot/src/public/xrpc.rs

···
         .route("/sh.tangled.repo.blob", get(repo::handle_blob))
         .route("/sh.tangled.repo.branches", get(repo::handle_branches))
         .route("/sh.tangled.repo.create", post(repo::handle_create))
+        .route("/sh.tangled.repo.delete", post(repo::handle_delete))
         .route("/sh.tangled.repo.diff", get(repo::handle_diff))
         .route("/sh.tangled.repo.getDefaultBranch", get(repo::handle_get_default_branch))
         .route("/sh.tangled.repo.languages", get(repo::handle_languages))

crates/knot/src/public/xrpc/sh/tangled/repo.rs

···
 use axum::{Json, extract::State};
-use lexicon::sh::tangled::repo::{create, get_default_branch, languages, tree};
+use lexicon::sh::tangled::repo::{create, delete, get_default_branch, languages, tree};
 use tokio_rayon::AsyncThreadPool as _;
 
 use crate::{
-    model::{Knot, repository::GixRepository},
+    model::{Knot, errors, repository::GixRepository},
     public::xrpc::{XrpcQuery, XrpcResult},
     services::authorization::{Authorization, Verification},
     types::sh::tangled::repo::{blob, branches, diff, log, tags},
···
 
 /// Handler for XRPC procedure `sh.tangled.repo.create`.
 pub async fn handle_create(
-    State(_knot): State<Knot>,
+    State(knot): State<Knot>,
     authorization: Authorization<CreateVerification>,
     Json(params): Json<create::Input<'static>>,
 ) -> XrpcResult<()> {
-    let claims = authorization.claims();
+    use crate::services::rbac::{Action, Policy, PolicyResult::*, RepositoryCreatePolicy};
 
-    tracing::info!(?params, ?claims);
+    let claims = authorization.claims();
+    let policy = RepositoryCreatePolicy;
+    let can_create = policy
+        .evaluate_access(
+            &claims.iss.as_ref(),
+            &Action::RepositoryCreate,
+            &knot,
+            &knot,
+        )
+        .await;
+
+    if !matches!(can_create, Granted) {
+        return Err(errors::Forbidden(format!(
+            "'{}' does not have permission to create repositories on this knot",
+            claims.iss
+        )))?;
+    }
+
+    // Fetch repository record from pds.
+    let response = knot
+        .fetch_pds_record(&claims.iss, "sh.tangled.repo", &params.rkey)
+        .await
+        .map_err(errors::RepoError)?;
+
+    let record = serde_json::from_slice(&response).map_err(errors::RepoError)?;
+
+    let is_new = knot
+        .store()
+        .insert_repository_from_record(&record)
+        .await
+        .map_err(errors::RepoError)?;
+
+    if is_new && record.value.knot == knot.instance_name() {
+        knot.create_repo(&claims.iss, &params.rkey, &record.value.name)
+            .map_err(errors::RepoError)?;
+    }
+
+    Ok(().into())
+}
+
+#[derive(Debug, Default)]
+pub struct DeleteVerification;
+
+impl Verification for DeleteVerification {
+    const LEXICON_METHOD: &'static str = "sh.tangled.repo.delete";
+}
+
+/// Handler for XRPC procedure `sh.tangled.repo.delete`.
+pub async fn handle_delete(
+    State(knot): State<Knot>,
+    authorization: Authorization<DeleteVerification>,
+    Json(params): Json<delete::Input<'static>>,
+) -> XrpcResult<()> {
+    use crate::services::rbac::{
+        Action, Policy, PolicyResult::*, RepositoryDeletePolicy, RepositoryRef,
+    };
+
+    let claims = authorization.claims();
+    let policy = RepositoryDeletePolicy;
+    let repository = RepositoryRef::new(&params.did, &params.rkey);
+    let can_delete = policy
+        .evaluate_access(
+            &claims.iss.as_ref(),
+            &Action::RepositoryDelete,
+            &repository,
+            &knot,
+        )
+        .await;
+
+    if !matches!(can_delete, Granted) {
+        return Err(errors::Forbidden(format!(
+            "'{}' does not have permission to delete repository '{}/{}'",
+            claims.iss, params.did, params.rkey
+        )))?;
+    }
+
+    knot.store()
+        .delete_repository(&params.did, &params.rkey)
+        .await
+        .map_err(errors::RepoError)?;
+
+    knot.delete_repo(&params.did, &params.rkey)
+        .map_err(errors::RepoError)?;
+
     Ok(().into())
 }
 

crates/knot/src/services.rs

···
 pub mod authorization;
 pub mod database;
 pub mod jetstream;
+pub mod rbac;

crates/knot/src/services/database.rs

···
         xrpc_create_at: Option<&'a OffsetDateTime>,
         jetstream_at: Option<&'a OffsetDateTime>,
     ) -> BoxFuture<'a, Result<Option<InsertRepositoryResult>, Self::Error>>;
+
+    fn update_repository<'d: 'a, 'a>(
+        &'d self,
+        did: &'a Did,
+        rkey: &'a str,
+        cid: &'a str,
+        repo: &'a Repo<'a>,
+    ) -> BoxFuture<'a, Result<(), Self::Error>>;
+
+    fn delete_repository<'d: 'a, 'a>(
+        &'d self,
+        did: &'a Did,
+        rkey: &'a str,
+    ) -> BoxFuture<'a, Result<(), Self::Error>>;
+
+    fn resolve_repository<'d: 'a, 'a>(
+        &'d self,
+        did: &'a Did,
+        name_or_rkey: &'a str,
+    ) -> BoxFuture<'a, Result<Option<Box<str>>, Self::Error>>;
+
+    fn repository_members<'d: 'a, 'a>(
+        &'d self,
+        did: &'a Did,
+        rkey: &'a str,
+    ) -> BoxStream<'a, Result<Box<Did>, Self::Error>>;
+
+    fn knot_members<'d: 'a, 'a>(
+        &'d self,
+        instance_name: &'a str,
+    ) -> BoxStream<'a, Result<Box<Did>, Self::Error>>;
 }
 
 #[derive(Debug, thiserror::Error)]
 pub enum DataStoreError {
     #[error("Database error: {0}")]
     Sqlx(#[from] sqlx::Error),
+    #[error("Failed to parse DID from db: {0}")]
+    Did(#[from] atproto::did::Error),
     #[error("Failed to extract AT-URI: {0}")]
     AtUri(#[from] atproto::aturi::Error),
     #[error("{0}")]
···
         let is_new = match result {
             None => false,
             Some(result) => {
-                result.xrpc_create_at.is_none()
+                result.old_xrpc_create_at.is_none()
+                    && result.new_xrpc_create_at.is_none()
                     && result.old_jetstream_at.is_none()
                     && result.new_jetstream_at.is_some()
             }
         };
 
         Ok(is_new)
+    }
+
+    pub async fn insert_repository_from_record(
+        &self,
+        record: &Record<'_, Repo<'_>>,
+    ) -> Result<bool, DataStoreError> {
+        let uri = record.aturi()?;
+        let result = self
+            .inner
+            .insert_repository(
+                uri.did()
+                    .ok_or(anyhow::anyhow!("AT-URI with DID authority required"))?,
+                uri.rkey.ok_or(anyhow::anyhow!("Missing rkey"))?,
+                &record.cid,
+                &record.value,
+                Some(&OffsetDateTime::now_utc()),
+                None,
+            )
+            .await?;
+
+        let is_new = match result {
+            None => false,
+            Some(result) => {
+                result.old_xrpc_create_at.is_none()
+                    && result.new_xrpc_create_at.is_some()
+                    && result.old_jetstream_at.is_none()
+                    && result.new_jetstream_at.is_none()
+            }
+        };
+
+        Ok(is_new)
+    }
+
+    pub async fn update_repository(
+        &self,
+        did: &Did,
+        rkey: &str,
+        cid: &str,
+        repository: &Repo<'_>,
+    ) -> Result<(), DataStoreError> {
+        self.inner
+            .update_repository(did, rkey, cid, repository)
+            .await?;
+        Ok(())
+    }
+
+    pub async fn delete_repository(&self, did: &Did, rkey: &str) -> Result<(), DataStoreError> {
+        self.inner.delete_repository(did, rkey).await?;
+        Ok(())
+    }
+
+    pub async fn resolve_repository(
+        &self,
+        did: &Did,
+        name_or_rkey: &str,
+    ) -> Result<Option<Box<str>>, DataStoreError> {
+        let rkey = self.inner.resolve_repository(did, name_or_rkey).await?;
+        Ok(rkey)
+    }
+
+    pub async fn is_repository_member(
+        &self,
+        repo_did: &Did,
+        repo_rkey: &str,
+        member_did: &Did,
+    ) -> bool {
+        self.inner
+            .repository_members(repo_did, repo_rkey)
+            .any(|did| async move { did.is_ok_and(|value| value == member_did) })
+            .await
+    }
+
+    pub async fn is_knot_member(&self, instance_name: &str, did: &Did) -> bool {
+        self.inner
+            .knot_members(instance_name)
+            .any(|member_did| async move { member_did.is_ok_and(|value| value == did) })
+            .await
     }
 }

crates/knot/src/services/database/pg_impl.rs

···
             .map(|label| label.clone().into())
             .collect();
 
+        match (xrpc_create_at, jetstream_at) {
+            (Some(xrpc_ts), None) => async move {
+                let result = sqlx::query_file_as!(
+                    InsertRepositoryResult,
+                    "sql/insert_repository_xrpc.sql",
+                    did.as_str(),
+                    rkey,
+                    cid,
+                    &repo.name,
+                    &repo.knot,
+                    repo.spindle.as_deref(),
+                    repo.description.as_deref(),
+                    repo.website.as_deref(),
+                    topics.as_slice(),
+                    repo.source.as_deref(),
+                    labels.as_slice(),
+                    &repo.created_at,
+                    Some(xrpc_ts),
+                    None::<OffsetDateTime>,
+                )
+                .fetch_optional(&self.pool)
+                .await?;
+
+                Ok(result)
+            }
+            .boxed(),
+            (None, Some(jetstream_ts)) => async move {
+                let result = sqlx::query_file_as!(
+                    InsertRepositoryResult,
+                    "sql/insert_repository_jetstream.sql",
+                    did.as_str(),
+                    rkey,
+                    cid,
+                    &repo.name,
+                    &repo.knot,
+                    repo.spindle.as_deref(),
+                    repo.description.as_deref(),
+                    repo.website.as_deref(),
+                    topics.as_slice(),
+                    repo.source.as_deref(),
+                    labels.as_slice(),
+                    &repo.created_at,
+                    None::<OffsetDateTime>,
+                    Some(jetstream_ts),
+                )
+                .fetch_optional(&self.pool)
+                .await?;
+
+                Ok(result)
+            }
+            .boxed(),
+            (Some(_), Some(_)) | (None, None) => panic!(),
+        }
+    }
+
+    fn update_repository<'d: 'a, 'a>(
+        &'d self,
+        did: &'a Did,
+        rkey: &'a str,
+        cid: &'a str,
+        repo: &'a Repo<'a>,
+    ) -> BoxFuture<'a, Result<(), Self::Error>> {
+        let topics: Vec<String> = repo
+            .topics
+            .iter()
+            .map(|topic| topic.clone().into())
+            .collect();
+
+        let labels: Vec<String> = repo
+            .labels
+            .iter()
+            .map(|label| label.clone().into())
+            .collect();
+
         async move {
-            let result = sqlx::query_file_as!(
-                InsertRepositoryResult,
-                "sql/insert_repository.sql",
+            sqlx::query_file!(
+                "sql/update_repository.sql",
                 did.as_str(),
                 rkey,
                 cid,
···
                 repo.source.as_deref(),
                 labels.as_slice(),
                 &repo.created_at,
-                xrpc_create_at,
-                jetstream_at,
+            )
+            .execute(&self.pool)
+            .await?;
+
+            Ok(())
+        }
+        .boxed()
+    }
+
+    fn delete_repository<'d: 'a, 'a>(
+        &'d self,
+        did: &'a Did,
+        rkey: &'a str,
+    ) -> BoxFuture<'a, Result<(), Self::Error>> {
+        async move {
+            sqlx::query!(
+                "DELETE FROM repository WHERE did = $1 AND rkey = $2",
+                did.as_str(),
+                rkey
+            )
+            .execute(&self.pool)
+            .await?;
+            Ok(())
+        }
+        .boxed()
+    }
+
+    fn resolve_repository<'d: 'a, 'a>(
+        &'d self,
+        did: &'a Did,
+        name_or_rkey: &'a str,
+    ) -> BoxFuture<'a, Result<Option<Box<str>>, Self::Error>> {
+        async move {
+            #[derive(sqlx::FromRow)]
+            struct Record {
+                rkey: Box<str>,
+            }
+
+            let result: Option<Record> = sqlx::query_as!(
+                Record,
+                "SELECT rkey FROM repository WHERE did = $1 AND (rkey = $2 OR name = $2)",
+                did.as_str(),
+                name_or_rkey
             )
             .fetch_optional(&self.pool)
             .await?;
 
-            Ok(result)
+            Ok(result.map(|record| record.rkey))
         }
+        .boxed()
+    }
+
+    fn repository_members<'d: 'a, 'a>(
+        &'d self,
+        did: &'a Did,
+        rkey: &'a str,
+    ) -> BoxStream<'a, Result<Box<Did>, Self::Error>> {
+        #[derive(sqlx::FromRow)]
+        struct Record {
+            member_did: Box<str>,
+        }
+
+        sqlx::query_as!(
+            Record,
+            "SELECT member_did FROM repository_member WHERE repo_did = $1 AND repo_rkey = $2",
+            did.as_str(),
+            rkey
+        )
+        .fetch(&self.pool)
+        .map(|record| {
+            let did = record?.member_did.parse()?;
+            Ok(did)
+        })
+        .boxed()
+    }
+
+    fn knot_members<'d: 'a, 'a>(
+        &'d self,
+        instance_name: &'a str,
+    ) -> BoxStream<'a, Result<Box<Did>, Self::Error>> {
+        #[derive(sqlx::FromRow)]
+        struct Record {
+            member_did: Box<str>,
+        }
+
+        sqlx::query_as!(
+            Record,
+            "SELECT member_did FROM knot_member WHERE instance_name = $1",
+            instance_name
+        )
+        .fetch(&self.pool)
+        .map(|record| {
+            let did = record?.member_did.parse()?;
+            Ok(did)
+        })
         .boxed()
     }
 }

crates/knot/src/services/database/types.rs

···
 #[derive(Debug, sqlx::FromRow)]
 pub struct InsertRepositoryResult {
     pub name: String,
-    pub xrpc_create_at: Option<OffsetDateTime>,
+    pub old_xrpc_create_at: Option<OffsetDateTime>,
+    pub new_xrpc_create_at: Option<OffsetDateTime>,
     pub old_jetstream_at: Option<OffsetDateTime>,
     pub new_jetstream_at: Option<OffsetDateTime>,
 }

crates/knot/src/services/jetstream.rs

···
 use super::database::DataStore;
-use crate::model::KnotState;
+use crate::{
+    model::KnotState,
+    services::rbac::{Policy, RepositoryCreatePolicy, RepositoryDeletePolicy, RepositoryRef},
+};
 use jetstream::{CommitEvent, Event};
 use lexicon::Lexicon;
 use std::sync::Arc;
···
 }
 
 async fn process_repo(state: &KnotState, event: &CommitEvent<'_>) -> anyhow::Result<()> {
+    use crate::services::rbac::{Action, PolicyResult::*};
+
     match event {
         CommitEvent::Create(commit) => {
             let Lexicon::Repo(repository) = serde_json::from_str(commit.record.get())? else {
                 return Err(anyhow::anyhow!("expected a 'sh.tangled.repo' record"));
             };
+
+            let policy = RepositoryCreatePolicy;
+            let can_create = policy
+                .evaluate_access(&commit.did, &Action::RepositoryCreate, state, state)
+                .await;
+
+            if !matches!(can_create, Granted) {
+                tracing::warn!(?commit, "RepositoryCreate permission denied");
+                return Ok(());
+            }
 
             let is_new = state
                 .store()
···
                 return Ok(());
             }
 
-            // @TODO Verify DID in commit has permission to create repositories on this knot.
-
-            let path = state
-                .repository_path()
-                .join(commit.did.as_str())
-                .join(commit.rkey);
-
-            let repo = gix::init_bare(&path)?;
-            tracing::info!(?repo, "created repository");
-
-            // Create a symlink to map the repository name -> rkey.
-            let symlink_path = state
-                .repository_path()
-                .join(commit.did.as_str())
-                .join(&*repository.name);
-
-            std::os::unix::fs::symlink(commit.rkey, &symlink_path)?;
+            state.create_repo(commit.did, commit.rkey, &repository.name)?;
         }
         CommitEvent::Update(commit) => {
-            tracing::info!("would update repo: {commit:?}");
-            //
+            let Lexicon::Repo(repository) = serde_json::from_str(commit.record.get())? else {
+                return Err(anyhow::anyhow!("expected a 'sh.tangled.repo' record"));
+            };
+
+            // @TODO Does this need auth?
+
+            state
+                .store()
+                .update_repository(commit.did, commit.rkey, commit.cid, &repository)
+                .await?;
         }
         CommitEvent::Delete(delete) => {
-            tracing::info!("would delete repo: {delete:?}");
+            let policy = RepositoryDeletePolicy;
+            let repository = RepositoryRef::new(delete.did, delete.rkey);
+            let can_create = policy
+                .evaluate_access(&delete.did, &Action::RepositoryDelete, &repository, state)
+                .await;
+
+            if !matches!(can_create, Granted) {
+                tracing::warn!(?delete, "RepositoryDelete permission denied");
+                return Ok(());
+            }
+
+            state
+                .store()
+                .delete_repository(delete.did, delete.rkey)
+                .await?;
+
+            state.delete_repo(delete.did, delete.rkey)?;
         }
     }
 

crates/knot/src/services/rbac.rs

···
+use atproto::Did;
+use futures_util::{FutureExt, future::BoxFuture};
+
+use crate::model::{Knot, KnotState};
+
+pub trait Policy<Subject, Resource, Action, Context>: Send + Sync {
+    /// Evaluates whether access should be granted.
+    ///
+    /// # Arguments
+    ///
+    /// * `subject` - The entity requesting access.
+    /// * `action` - The action being performed.
+    /// * `resource` - The target resource.
+    /// * `context` - Additional context that may affect the decision.
+    ///
+    /// # Returns
+    ///
+    /// A [`PolicyResult`] indicating whether access is granted or denied.
+    fn evaluate_access<'s: 'a, 'a>(
+        &'s self,
+        subject: &'a Subject,
+        action: &'a Action,
+        resource: &'a Resource,
+        context: &'a Context,
+    ) -> BoxFuture<'a, PolicyResult>;
+}
+
+impl<S, R, A, C> Policy<S, R, A, C> for Box<dyn Policy<S, R, A, C>> {
+    #[inline]
+    fn evaluate_access<'s: 'a, 'a>(
+        &'s self,
+        subject: &'a S,
+        action: &'a A,
+        resource: &'a R,
+        context: &'a C,
+    ) -> BoxFuture<'a, PolicyResult> {
+        (**self).evaluate_access(subject, action, resource, context)
+    }
+}
+
+pub enum PolicyResult {
+    Granted,
+    Denied,
+}
+
+pub enum Action {
+    RepositoryCreate,
+    RepositoryDelete,
+    RepositoryPush,
+}
+
+pub struct RepositoryPushPolicy;
+
+pub struct RepositoryRef<'a> {
+    pub did: &'a Did,
+    pub rkey: &'a str,
+}
+
+impl<'a> RepositoryRef<'a> {
+    pub fn new(did: &'a Did, rkey: &'a str) -> Self {
+        Self { did, rkey }
+    }
+}
+
+impl Policy<&Did, RepositoryRef<'_>, Action, Knot> for RepositoryPushPolicy {
+    fn evaluate_access<'s: 'a, 'a>(
+        &'s self,
+        &subject: &'a &Did,
+        action: &'a Action,
+        resource: &'a RepositoryRef<'_>,
+        context: &'a Knot,
+    ) -> BoxFuture<'a, PolicyResult> {
+        async move {
+            let is_member = subject == resource.did
+                || context
+                    .store()
+                    .is_repository_member(resource.did, resource.rkey, subject)
+                    .await;
+
+            match (action, is_member) {
+                (Action::RepositoryPush, true) => PolicyResult::Granted,
+                (_, _) => PolicyResult::Denied,
+            }
+        }
+        .boxed()
+    }
+}
+
+pub struct RepositoryCreatePolicy;
+
+impl Policy<&Did, KnotState, Action, KnotState> for RepositoryCreatePolicy {
+    fn evaluate_access<'s: 'a, 'a>(
+        &'s self,
+        &subject: &'a &Did,
+        action: &'a Action,
+        resource: &'a KnotState,
+        context: &'a KnotState,
+    ) -> BoxFuture<'a, PolicyResult> {
+        async move {
+            let is_member = subject == resource.owner_did()
+                || context
+                    .store()
+                    .is_knot_member(context.instance_name(), subject)
+                    .await;
+
+            match (action, is_member) {
+                (Action::RepositoryCreate, true) => PolicyResult::Granted,
+                (_, _) => PolicyResult::Denied,
+            }
+        }
+        .boxed()
+    }
+}
+
+pub struct RepositoryDeletePolicy;
+
+impl Policy<&Did, RepositoryRef<'_>, Action, KnotState> for RepositoryDeletePolicy {
+    fn evaluate_access<'s: 'a, 'a>(
+        &'s self,
+        &subject: &'a &Did,
+        action: &'a Action,
+        resource: &'a RepositoryRef<'_>,
+        context: &'a KnotState,
+    ) -> BoxFuture<'a, PolicyResult> {
+        async move {
+            let is_owner = subject == context.owner_did() || resource.did == subject;
+            match (action, is_owner) {
+                (Action::RepositoryDelete, true) => PolicyResult::Granted,
+                (_, _) => PolicyResult::Denied,
+            }
+        }
+        .boxed()
+    }
+}

crates/lexicon/src/sh/tangled/repo.rs

···
 pub mod blob;
 pub mod branches;
 pub mod create;
+pub mod delete;
 pub mod diff;
 pub mod get_default_branch;
 pub mod issue;

crates/lexicon/src/sh/tangled/repo/delete.rs

···
+//!
+//! <https://tangled.org/@tangled.org/core/blob/master/lexicons/repo/delete.json>
+//!
+use std::borrow::Cow;
+
+use atproto::Did;
+use serde::{Deserialize, Serialize};
+
+/// Parameters for the `sh.tangled.repo.delete` procedure.
+///
+/// <https://tangled.org/@tangled.org/core/blob/master/lexicons/repo/delete.json>
+#[derive(Debug, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Input<'a> {
+    /// DID of the repository owner.
+    pub did: Cow<'a, Did>,
+
+    /// Record key of the repository record.
+    pub rkey: Cow<'a, str>,
+
+    /// Name of the repository to delete.
+    pub name: Cow<'a, str>,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::Input;
+
+    #[test]
+    fn can_deserialize_required() {
+        const REQUEST: &str = r#"{"rkey":"3m4xo72lo4322"}"#;
+
+        let input: Input = serde_json::from_str(REQUEST).expect("should deserialize");
+        assert_eq!(input.rkey, "3m4xo72lo4322");
+    }
+
+    #[test]
+    fn can_deserialize_complete() {
+        const REQUEST: &str = r#"{"rkey":"3m4xo72lo4322"}"#;
+
+        let input: Input = serde_json::from_str(REQUEST).expect("should deserialize");
+        assert_eq!(input.rkey, "3m4xo72lo4322");
+    }
+}