OAuth scopes full impl.

src/api/repo/blob.rs

@@ -62,6 +62,17 @@ pub async fn upload_blob(
     } else {
         match crate::auth::validate_bearer_token_allow_deactivated(&state.db, &token).await {
             Ok(user) => {
+                let mime_type_for_check = headers
+                    .get("content-type")
+                    .and_then(|h| h.to_str().ok())
+                    .unwrap_or("application/octet-stream");
+                if let Err(e) = crate::auth::scope_check::check_blob_scope(
+                    user.is_oauth,
+                    user.scope.as_deref(),
+                    mime_type_for_check,
+                ) {
+                    return e;
+                }
                 let deactivated = sqlx::query_scalar!(
                     "SELECT deactivated_at FROM users WHERE did = $1",
                     user.did
@@ -171,23 +182,22 @@ pub async fn upload_blob(
             .blob_store
             .put_bytes(&storage_key, bytes::Bytes::from(data))
             .await
-        {
-            error!("Failed to upload blob to storage: {:?}", e);
-            return (
-                StatusCode::INTERNAL_SERVER_ERROR,
-                Json(json!({"error": "InternalError", "message": "Failed to store blob"})),
-            )
-                .into_response();
-        }
+    {
+        error!("Failed to upload blob to storage: {:?}", e);
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError", "message": "Failed to store blob"})),
+        )
+            .into_response();
+    }
     if let Err(e) = tx.commit().await {
         error!("Failed to commit blob transaction: {:?}", e);
-        if was_inserted
-            && let Err(cleanup_err) = state.blob_store.delete(&storage_key).await {
-                error!(
-                    "Failed to cleanup orphaned blob {}: {:?}",
-                    storage_key, cleanup_err
-                );
-            }
+        if was_inserted && let Err(cleanup_err) = state.blob_store.delete(&storage_key).await {
+            error!(
+                "Failed to cleanup orphaned blob {}: {:?}",
+                storage_key, cleanup_err
+            );
+        }
         return (
             StatusCode::INTERNAL_SERVER_ERROR,
             Json(json!({"error": "InternalError"})),
@@ -231,11 +241,12 @@ fn find_blobs(val: &serde_json::Value, blobs: &mut Vec<String>) {
     if let Some(obj) = val.as_object() {
         if let Some(type_val) = obj.get("$type")
             && type_val == "blob"
-                && let Some(r) = obj.get("ref")
-                    && let Some(link) = r.get("$link")
-                        && let Some(s) = link.as_str() {
-                            blobs.push(s.to_string());
-                        }
+            && let Some(r) = obj.get("ref")
+            && let Some(link) = r.get("$link")
+            && let Some(s) = link.as_str()
+        {
+            blobs.push(s.to_string());
+        }
         for (_, v) in obj {
             find_blobs(v, blobs);
         }

src/api/repo/import.rs

@@ -53,13 +53,14 @@ pub async fn import_repo(
         Some(t) => t,
         None => return ApiError::AuthenticationRequired.into_response(),
     };
-    let auth_user = match crate::auth::validate_bearer_token_allow_deactivated(&state.db, &token).await {
-        Ok(user) => user,
-        Err(e) => return ApiError::from(e).into_response(),
-    };
+    let auth_user =
+        match crate::auth::validate_bearer_token_allow_deactivated(&state.db, &token).await {
+            Ok(user) => user,
+            Err(e) => return ApiError::from(e).into_response(),
+        };
     let did = &auth_user.did;
     let user = match sqlx::query!(
-        "SELECT id, deactivated_at, takedown_ref FROM users WHERE did = $1",
+        "SELECT id, handle, deactivated_at, takedown_ref FROM users WHERE did = $1",
         did
     )
     .fetch_optional(&state.db)
@@ -317,6 +318,30 @@ pub async fn import_repo(
                 records.len(),
                 did
             );
+            if is_migration {
+                if let Err(e) =
+                    sqlx::query!("UPDATE users SET deactivated_at = NULL WHERE did = $1", did)
+                        .execute(&state.db)
+                        .await
+                {
+                    error!("Failed to reactivate account after import: {:?}", e);
+                }
+                let _ = state.cache.delete(&format!("handle:{}", user.handle)).await;
+                if let Err(e) = crate::api::repo::record::sequence_identity_event(
+                    &state,
+                    did,
+                    Some(&user.handle),
+                )
+                .await
+                {
+                    warn!("Failed to sequence identity event after import: {:?}", e);
+                }
+                if let Err(e) =
+                    crate::api::repo::record::sequence_account_event(&state, did, true, None).await
+                {
+                    warn!("Failed to sequence account event after import: {:?}", e);
+                }
+            }
             if let Err(e) = sequence_import_event(&state, did, &root.to_string()).await {
                 warn!("Failed to sequence import event: {:?}", e);
             }

src/api/repo/record/batch.rs

@@ -101,7 +101,9 @@ pub async fn apply_writes(
                 .into_response();
         }
     };
-    let did = auth_user.did;
+    let did = auth_user.did.clone();
+    let is_oauth = auth_user.is_oauth;
+    let scope = auth_user.scope;
     if input.repo != did {
         return (
             StatusCode::FORBIDDEN,
@@ -144,6 +146,75 @@ pub async fn apply_writes(
         )
             .into_response();
     }
+
+    if is_oauth {
+        use std::collections::HashSet;
+        let create_collections: HashSet<&str> = input
+            .writes
+            .iter()
+            .filter_map(|w| {
+                if let WriteOp::Create { collection, .. } = w {
+                    Some(collection.as_str())
+                } else {
+                    None
+                }
+            })
+            .collect();
+        let update_collections: HashSet<&str> = input
+            .writes
+            .iter()
+            .filter_map(|w| {
+                if let WriteOp::Update { collection, .. } = w {
+                    Some(collection.as_str())
+                } else {
+                    None
+                }
+            })
+            .collect();
+        let delete_collections: HashSet<&str> = input
+            .writes
+            .iter()
+            .filter_map(|w| {
+                if let WriteOp::Delete { collection, .. } = w {
+                    Some(collection.as_str())
+                } else {
+                    None
+                }
+            })
+            .collect();
+
+        for collection in create_collections {
+            if let Err(e) = crate::auth::scope_check::check_repo_scope(
+                is_oauth,
+                scope.as_deref(),
+                crate::oauth::RepoAction::Create,
+                collection,
+            ) {
+                return e;
+            }
+        }
+        for collection in update_collections {
+            if let Err(e) = crate::auth::scope_check::check_repo_scope(
+                is_oauth,
+                scope.as_deref(),
+                crate::oauth::RepoAction::Update,
+                collection,
+            ) {
+                return e;
+            }
+        }
+        for collection in delete_collections {
+            if let Err(e) = crate::auth::scope_check::check_repo_scope(
+                is_oauth,
+                scope.as_deref(),
+                crate::oauth::RepoAction::Delete,
+                collection,
+            ) {
+                return e;
+            }
+        }
+    }
+
     let user_id: uuid::Uuid = match sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", did)
         .fetch_optional(&state.db)
         .await
@@ -184,13 +255,14 @@ pub async fn apply_writes(
         }
     };
     if let Some(swap_commit) = &input.swap_commit
-        && Cid::from_str(swap_commit).ok() != Some(current_root_cid) {
-            return (
-                StatusCode::CONFLICT,
-                Json(json!({"error": "InvalidSwap", "message": "Repo has been modified"})),
-            )
-                .into_response();
-        }
+        && Cid::from_str(swap_commit).ok() != Some(current_root_cid)
+    {
+        return (
+            StatusCode::CONFLICT,
+            Json(json!({"error": "InvalidSwap", "message": "Repo has been modified"})),
+        )
+            .into_response();
+    }
     let tracking_store = TrackingBlockStore::new(state.block_store.clone());
     let commit_bytes = match tracking_store.get(&current_root_cid).await {
         Ok(Some(b)) => b,
@@ -225,9 +297,10 @@ pub async fn apply_writes(
                 value,
             } => {
                 if input.validate.unwrap_or(true)
-                    && let Err(err_response) = validate_record(value, collection) {
-                        return *err_response;
-                    }
+                    && let Err(err_response) = validate_record(value, collection)
+                {
+                    return *err_response;
+                }
                 let rkey = rkey
                     .clone()
                     .unwrap_or_else(|| Tid::now(LimitedU32::MIN).to_string());
@@ -276,9 +349,10 @@ pub async fn apply_writes(
                 value,
             } => {
                 if input.validate.unwrap_or(true)
-                    && let Err(err_response) = validate_record(value, collection) {
-                        return *err_response;
-                    }
+                    && let Err(err_response) = validate_record(value, collection)
+                {
+                    return *err_response;
+                }
                 let mut record_bytes = Vec::new();
                 if serde_ipld_dagcbor::to_writer(&mut record_bytes, value).is_err() {
                     return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidRecord", "message": "Failed to serialize record"}))).into_response();
@@ -353,7 +427,11 @@ pub async fn apply_writes(
     };
     let mut relevant_blocks = std::collections::BTreeMap::new();
     for key in &modified_keys {
-        if mst.blocks_for_path(key, &mut relevant_blocks).await.is_err() {
+        if mst
+            .blocks_for_path(key, &mut relevant_blocks)
+            .await
+            .is_err()
+        {
             return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to get new MST blocks for path"}))).into_response();
         }
         if original_mst

src/api/repo/record/delete.rs

@@ -34,19 +34,34 @@ pub async fn delete_record(
     axum::extract::OriginalUri(uri): axum::extract::OriginalUri,
     Json(input): Json<DeleteRecordInput>,
 ) -> Response {
-    let (did, user_id, current_root_cid) =
+    let auth =
         match prepare_repo_write(&state, &headers, &input.repo, "POST", &uri.to_string()).await {
             Ok(res) => res,
             Err(err_res) => return err_res,
         };
+
+    if let Err(e) = crate::auth::scope_check::check_repo_scope(
+        auth.is_oauth,
+        auth.scope.as_deref(),
+        crate::oauth::RepoAction::Delete,
+        &input.collection,
+    ) {
+        return e;
+    }
+
+    let did = auth.did;
+    let user_id = auth.user_id;
+    let current_root_cid = auth.current_root_cid;
+
     if let Some(swap_commit) = &input.swap_commit
-        && Cid::from_str(swap_commit).ok() != Some(current_root_cid) {
-            return (
-                StatusCode::CONFLICT,
-                Json(json!({"error": "InvalidSwap", "message": "Repo has been modified"})),
-            )
-                .into_response();
-        }
+        && Cid::from_str(swap_commit).ok() != Some(current_root_cid)
+    {
+        return (
+            StatusCode::CONFLICT,
+            Json(json!({"error": "InvalidSwap", "message": "Repo has been modified"})),
+        )
+            .into_response();
+    }
     let tracking_store = TrackingBlockStore::new(state.block_store.clone());
     let commit_bytes = match tracking_store.get(&current_root_cid).await {
         Ok(Some(b)) => b,
@@ -115,10 +130,18 @@ pub async fn delete_record(
         prev: prev_record_cid,
     };
     let mut relevant_blocks = std::collections::BTreeMap::new();
-    if new_mst.blocks_for_path(&key, &mut relevant_blocks).await.is_err() {
+    if new_mst
+        .blocks_for_path(&key, &mut relevant_blocks)
+        .await
+        .is_err()
+    {
         return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to get new MST blocks for path"}))).into_response();
     }
-    if mst.blocks_for_path(&key, &mut relevant_blocks).await.is_err() {
+    if mst
+        .blocks_for_path(&key, &mut relevant_blocks)
+        .await
+        .is_err()
+    {
         return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to get old MST blocks for path"}))).into_response();
     }
     let mut written_cids = tracking_store.get_all_relevant_cids();

src/api/repo/record/read.rs

@@ -48,10 +48,7 @@ pub async fn get_record(
     let user_id: uuid::Uuid = match user_id_opt {
         Ok(Some(id)) => id,
         Ok(None) => {
-            if let Some(proxy_header) = headers
-                .get("atproto-proxy")
-                .and_then(|h| h.to_str().ok())
-            {
+            if let Some(proxy_header) = headers.get("atproto-proxy").and_then(|h| h.to_str().ok()) {
                 let did = proxy_header.split('#').next().unwrap_or(proxy_header);
                 if let Some(resolved) = state.did_resolver.resolve_did(did).await {
                     let mut url = format!(
@@ -84,7 +81,8 @@ pub async fn get_record(
                                 .header("content-type", "application/json")
                                 .body(axum::body::Body::from(body))
                                 .unwrap_or_else(|_| {
-                                    (StatusCode::INTERNAL_SERVER_ERROR, "Internal error").into_response()
+                                    (StatusCode::INTERNAL_SERVER_ERROR, "Internal error")
+                                        .into_response()
                                 });
                         }
                         Err(e) => {
@@ -138,13 +136,14 @@ pub async fn get_record(
         }
     };
     if let Some(expected_cid) = &input.cid
-        && &record_cid_str != expected_cid {
-            return (
-                StatusCode::NOT_FOUND,
-                Json(json!({"error": "NotFound", "message": "Record CID mismatch"})),
-            )
-                .into_response();
-        }
+        && &record_cid_str != expected_cid
+    {
+        return (
+            StatusCode::NOT_FOUND,
+            Json(json!({"error": "NotFound", "message": "Record CID mismatch"})),
+        )
+            .into_response();
+    }
     let cid = match Cid::from_str(&record_cid_str) {
         Ok(c) => c,
         Err(_) => {
@@ -326,13 +325,14 @@ pub async fn list_records(
     for (cid, block_opt) in cids.iter().zip(blocks.into_iter()) {
         if let Some(block) = block_opt
             && let Some((rkey, cid_str)) = cid_to_rkey.get(cid)
-                && let Ok(value) = serde_ipld_dagcbor::from_slice::<serde_json::Value>(&block) {
-                    records.push(json!({
-                        "uri": format!("at://{}/{}/{}", input.repo, input.collection, rkey),
-                        "cid": cid_str,
-                        "value": value
-                    }));
-                }
+            && let Ok(value) = serde_ipld_dagcbor::from_slice::<serde_json::Value>(&block)
+        {
+            records.push(json!({
+                "uri": format!("at://{}/{}/{}", input.repo, input.collection, rkey),
+                "cid": cid_str,
+                "value": value
+            }));
+        }
     }
     Json(ListRecordsOutput {
         cursor: last_rkey,

src/api/repo/record/utils.rs

@@ -151,27 +151,36 @@ pub async fn commit_and_log(
     match lock_result {
         Err(e) => {
             if let Some(db_err) = e.as_database_error()
-                && db_err.code().as_deref() == Some("55P03") {
-                    return Err(
-                        "ConcurrentModification: Another request is modifying this repo"
-                            .to_string(),
-                    );
-                }
+                && db_err.code().as_deref() == Some("55P03")
+            {
+                return Err(
+                    "ConcurrentModification: Another request is modifying this repo".to_string(),
+                );
+            }
             return Err(format!("Failed to acquire repo lock: {}", e));
         }
         Ok(Some(row)) => {
             if let Some(expected_root) = &current_root_cid
-                && row.repo_root_cid != expected_root.to_string() {
-                    return Err(
-                        "ConcurrentModification: Repo has been modified since last read"
-                            .to_string(),
-                    );
-                }
+                && row.repo_root_cid != expected_root.to_string()
+            {
+                return Err(
+                    "ConcurrentModification: Repo has been modified since last read".to_string(),
+                );
+            }
         }
         Ok(None) => {
             return Err("Repo not found".to_string());
         }
     }
+    let is_account_active = sqlx::query_scalar!(
+        "SELECT deactivated_at IS NULL FROM users WHERE id = $1",
+        user_id
+    )
+    .fetch_optional(&mut *tx)
+    .await
+    .map_err(|e| format!("Failed to check account status: {}", e))?
+    .flatten()
+    .unwrap_or(false);
     sqlx::query!(
         "UPDATE repos SET repo_root_cid = $1 WHERE user_id = $2",
         new_root_cid.to_string(),
@@ -289,35 +298,39 @@ pub async fn commit_and_log(
             }
         })
         .collect::<Vec<_>>();
-    let event_type = "commit";
-    let prev_cid_str = current_root_cid.map(|c| c.to_string());
-    let prev_data_cid_str = prev_data_cid.map(|c| c.to_string());
-    let seq_row = sqlx::query!(
-        r#"
-        INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, blocks_cids, prev_data_cid)
-        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
-        RETURNING seq
-        "#,
-        did,
-        event_type,
-        new_root_cid.to_string(),
-        prev_cid_str,
-        json!(ops_json),
-        &[] as &[String],
-        blocks_cids,
-        prev_data_cid_str,
-    )
-    .fetch_one(&mut *tx)
-    .await
-    .map_err(|e| format!("DB Error (repo_seq): {}", e))?;
-    sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq_row.seq))
-        .execute(&mut *tx)
+    if is_account_active {
+        let event_type = "commit";
+        let prev_cid_str = current_root_cid.map(|c| c.to_string());
+        let prev_data_cid_str = prev_data_cid.map(|c| c.to_string());
+        let seq_row = sqlx::query!(
+            r#"
+            INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, blocks_cids, prev_data_cid)
+            VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+            RETURNING seq
+            "#,
+            did,
+            event_type,
+            new_root_cid.to_string(),
+            prev_cid_str,
+            json!(ops_json),
+            &[] as &[String],
+            blocks_cids,
+            prev_data_cid_str,
+        )
+        .fetch_one(&mut *tx)
         .await
-        .map_err(|e| format!("DB Error (notify): {}", e))?;
+        .map_err(|e| format!("DB Error (repo_seq): {}", e))?;
+        sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq_row.seq))
+            .execute(&mut *tx)
+            .await
+            .map_err(|e| format!("DB Error (notify): {}", e))?;
+    }
     tx.commit()
         .await
         .map_err(|e| format!("Failed to commit transaction: {}", e))?;
-    let _ = sequence_sync_event(state, did, &new_root_cid.to_string()).await;
+    if is_account_active {
+        let _ = sequence_sync_event(state, did, &new_root_cid.to_string()).await;
+    }
     Ok(CommitResult {
         commit_cid: new_root_cid,
         rev: rev_str,
@@ -482,3 +495,37 @@ pub async fn sequence_sync_event(
         .map_err(|e| format!("DB Error (notify): {}", e))?;
     Ok(seq_row.seq)
 }
+
+pub async fn sequence_empty_commit_event(state: &AppState, did: &str) -> Result<i64, String> {
+    let repo_root = sqlx::query_scalar!(
+        "SELECT r.repo_root_cid FROM repos r JOIN users u ON r.user_id = u.id WHERE u.did = $1",
+        did
+    )
+    .fetch_optional(&state.db)
+    .await
+    .map_err(|e| format!("DB Error fetching repo root: {}", e))?
+    .ok_or_else(|| "Repo not found".to_string())?;
+    let ops = serde_json::json!([]);
+    let blobs: Vec<String> = vec![];
+    let blocks_cids: Vec<String> = vec![];
+    let seq_row = sqlx::query!(
+        r#"
+        INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, blocks_cids)
+        VALUES ($1, 'commit', $2, $2, $3, $4, $5)
+        RETURNING seq
+        "#,
+        did,
+        repo_root,
+        ops,
+        &blobs,
+        &blocks_cids
+    )
+    .fetch_one(&state.db)
+    .await
+    .map_err(|e| format!("DB Error (repo_seq empty commit): {}", e))?;
+    sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq_row.seq))
+        .execute(&state.db)
+        .await
+        .map_err(|e| format!("DB Error (notify): {}", e))?;
+    Ok(seq_row.seq)
+}

src/api/repo/record/write.rs

@@ -22,10 +22,7 @@ use std::sync::Arc;
 use tracing::error;
 use uuid::Uuid;
 
-pub async fn has_verified_comms_channel(
-    db: &PgPool,
-    did: &str,
-) -> Result<bool, sqlx::Error> {
+pub async fn has_verified_comms_channel(db: &PgPool, did: &str) -> Result<bool, sqlx::Error> {
     let row = sqlx::query(
         r#"
         SELECT
@@ -52,13 +49,21 @@ pub async fn has_verified_comms_channel(
     }
 }
 
+pub struct RepoWriteAuth {
+    pub did: String,
+    pub user_id: Uuid,
+    pub current_root_cid: Cid,
+    pub is_oauth: bool,
+    pub scope: Option<String>,
+}
+
 pub async fn prepare_repo_write(
     state: &AppState,
     headers: &HeaderMap,
     repo_did: &str,
     http_method: &str,
     http_uri: &str,
-) -> Result<(String, Uuid, Cid), Response> {
+) -> Result<RepoWriteAuth, Response> {
     let extracted = crate::auth::extract_auth_token_from_header(
         headers.get("Authorization").and_then(|h| h.to_str().ok()),
     )
@@ -69,9 +74,7 @@ pub async fn prepare_repo_write(
         )
             .into_response()
     })?;
-    let dpop_proof = headers
-        .get("DPoP")
-        .and_then(|h| h.to_str().ok());
+    let dpop_proof = headers.get("DPoP").and_then(|h| h.to_str().ok());
     let auth_user = crate::auth::validate_token_with_dpop(
         &state.db,
         &extracted.token,
@@ -163,7 +166,13 @@ pub async fn prepare_repo_write(
         )
             .into_response()
     })?;
-    Ok((auth_user.did, user_id, current_root_cid))
+    Ok(RepoWriteAuth {
+        did: auth_user.did,
+        user_id,
+        current_root_cid,
+        is_oauth: auth_user.is_oauth,
+        scope: auth_user.scope,
+    })
 }
 #[derive(Deserialize)]
 #[allow(dead_code)]
@@ -188,19 +197,34 @@ pub async fn create_record(
     axum::extract::OriginalUri(uri): axum::extract::OriginalUri,
     Json(input): Json<CreateRecordInput>,
 ) -> Response {
-    let (did, user_id, current_root_cid) =
+    let auth =
         match prepare_repo_write(&state, &headers, &input.repo, "POST", &uri.to_string()).await {
             Ok(res) => res,
             Err(err_res) => return err_res,
         };
+
+    if let Err(e) = crate::auth::scope_check::check_repo_scope(
+        auth.is_oauth,
+        auth.scope.as_deref(),
+        crate::oauth::RepoAction::Create,
+        &input.collection,
+    ) {
+        return e;
+    }
+
+    let did = auth.did;
+    let user_id = auth.user_id;
+    let current_root_cid = auth.current_root_cid;
+
     if let Some(swap_commit) = &input.swap_commit
-        && Cid::from_str(swap_commit).ok() != Some(current_root_cid) {
-            return (
-                StatusCode::CONFLICT,
-                Json(json!({"error": "InvalidSwap", "message": "Repo has been modified"})),
-            )
-                .into_response();
-        }
+        && Cid::from_str(swap_commit).ok() != Some(current_root_cid)
+    {
+        return (
+            StatusCode::CONFLICT,
+            Json(json!({"error": "InvalidSwap", "message": "Repo has been modified"})),
+        )
+            .into_response();
+    }
     let tracking_store = TrackingBlockStore::new(state.block_store.clone());
     let commit_bytes = match tracking_store.get(&current_root_cid).await {
         Ok(Some(b)) => b,
@@ -234,9 +258,10 @@ pub async fn create_record(
         }
     };
     if input.validate.unwrap_or(true)
-        && let Err(err_response) = validate_record(&input.record, &input.collection) {
-            return *err_response;
-        }
+        && let Err(err_response) = validate_record(&input.record, &input.collection)
+    {
+        return *err_response;
+    }
     let rkey = input
         .rkey
         .unwrap_or_else(|| Tid::now(LimitedU32::MIN).to_string());
@@ -285,10 +310,18 @@ pub async fn create_record(
         cid: record_cid,
     };
     let mut relevant_blocks = std::collections::BTreeMap::new();
-    if new_mst.blocks_for_path(&key, &mut relevant_blocks).await.is_err() {
+    if new_mst
+        .blocks_for_path(&key, &mut relevant_blocks)
+        .await
+        .is_err()
+    {
         return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to get new MST blocks for path"}))).into_response();
     }
-    if mst.blocks_for_path(&key, &mut relevant_blocks).await.is_err() {
+    if mst
+        .blocks_for_path(&key, &mut relevant_blocks)
+        .await
+        .is_err()
+    {
         return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to get old MST blocks for path"}))).into_response();
     }
     relevant_blocks.insert(record_cid, bytes::Bytes::from(record_bytes));
@@ -356,19 +389,42 @@ pub async fn put_record(
     axum::extract::OriginalUri(uri): axum::extract::OriginalUri,
     Json(input): Json<PutRecordInput>,
 ) -> Response {
-    let (did, user_id, current_root_cid) =
+    let auth =
         match prepare_repo_write(&state, &headers, &input.repo, "POST", &uri.to_string()).await {
             Ok(res) => res,
             Err(err_res) => return err_res,
         };
+
+    if let Err(e) = crate::auth::scope_check::check_repo_scope(
+        auth.is_oauth,
+        auth.scope.as_deref(),
+        crate::oauth::RepoAction::Create,
+        &input.collection,
+    ) {
+        return e;
+    }
+    if let Err(e) = crate::auth::scope_check::check_repo_scope(
+        auth.is_oauth,
+        auth.scope.as_deref(),
+        crate::oauth::RepoAction::Update,
+        &input.collection,
+    ) {
+        return e;
+    }
+
+    let did = auth.did;
+    let user_id = auth.user_id;
+    let current_root_cid = auth.current_root_cid;
+
     if let Some(swap_commit) = &input.swap_commit
-        && Cid::from_str(swap_commit).ok() != Some(current_root_cid) {
-            return (
-                StatusCode::CONFLICT,
-                Json(json!({"error": "InvalidSwap", "message": "Repo has been modified"})),
-            )
-                .into_response();
-        }
+        && Cid::from_str(swap_commit).ok() != Some(current_root_cid)
+    {
+        return (
+            StatusCode::CONFLICT,
+            Json(json!({"error": "InvalidSwap", "message": "Repo has been modified"})),
+        )
+            .into_response();
+    }
     let tracking_store = TrackingBlockStore::new(state.block_store.clone());
     let commit_bytes = match tracking_store.get(&current_root_cid).await {
         Ok(Some(b)) => b,
@@ -403,9 +459,10 @@ pub async fn put_record(
     };
     let key = format!("{}/{}", collection_nsid, input.rkey);
     if input.validate.unwrap_or(true)
-        && let Err(err_response) = validate_record(&input.record, &input.collection) {
-            return *err_response;
-        }
+        && let Err(err_response) = validate_record(&input.record, &input.collection)
+    {
+        return *err_response;
+    }
     if let Some(swap_record_str) = &input.swap_record {
         let expected_cid = Cid::from_str(swap_record_str).ok();
         let actual_cid = mst.get(&key).await.ok().flatten();
@@ -480,10 +537,18 @@ pub async fn put_record(
         }
     };
     let mut relevant_blocks = std::collections::BTreeMap::new();
-    if new_mst.blocks_for_path(&key, &mut relevant_blocks).await.is_err() {
+    if new_mst
+        .blocks_for_path(&key, &mut relevant_blocks)
+        .await
+        .is_err()
+    {
         return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to get new MST blocks for path"}))).into_response();
     }
-    if mst.blocks_for_path(&key, &mut relevant_blocks).await.is_err() {
+    if mst
+        .blocks_for_path(&key, &mut relevant_blocks)
+        .await
+        .is_err()
+    {
         return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to get old MST blocks for path"}))).into_response();
     }
     relevant_blocks.insert(record_cid, bytes::Bytes::from(record_bytes));