mirrors/tranquil-pds
1
0
Fork 0
mirror of https://tangled.org/tranquil.farm/tranquil-pds synced 2026-09-26 04:04:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

Creating & posting records works. Also messed up newlines but will fix later.

Browse Source
This commit is contained in:
lewis
2025-12-14 23:55:04 +02:00
parent 86db6617af
commit c6f9062979
227 changed files with 2122 additions and 7842 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/repo/blob.rs
-39
View File
@@ -14,9 +14,7 @@ use serde_json::json;
use sha2::{Digest, Sha256};
use std::str::FromStr;
use tracing::error;
const MAX_BLOB_SIZE: usize = 1_000_000;
pub async fn upload_blob(
State(state): State<AppState>,
headers: axum::http::HeaderMap,
@@ -29,7 +27,6 @@ pub async fn upload_blob(
)
.into_response();
}
let token = match crate::auth::extract_bearer_token_from_header(
headers.get("Authorization").and_then(|h| h.to_str().ok())
) {
@@ -42,7 +39,6 @@ pub async fn upload_blob(
.into_response();
}
};
let auth_user = match crate::auth::validate_bearer_token(&state.db, &token).await {
Ok(user) => user,
Err(_) => {
@@ -54,16 +50,13 @@ pub async fn upload_blob(
}
};
let did = auth_user.did;
let mime_type = headers
.get("content-type")
.and_then(|h| h.to_str().ok())
.unwrap_or("application/octet-stream")
.to_string();
let size = body.len() as i64;
let data = body.to_vec();
let mut hasher = Sha256::new();
hasher.update(&data);
let hash = hasher.finalize();
@@ -80,13 +73,10 @@ pub async fn upload_blob(
};
let cid = Cid::new_v1(0x55, multihash);
let cid_str = cid.to_string();
let storage_key = format!("blobs/{}", cid_str);
let user_query = sqlx::query!("SELECT id FROM users WHERE did = $1", did)
.fetch_optional(&state.db)
.await;
let user_id = match user_query {
Ok(Some(row)) => row.id,
_ => {
@@ -97,7 +87,6 @@ pub async fn upload_blob(
.into_response();
}
};
let mut tx = match state.db.begin().await {
Ok(tx) => tx,
Err(e) => {
@@ -109,7 +98,6 @@ pub async fn upload_blob(
.into_response();
}
};
let insert = sqlx::query!(
"INSERT INTO blobs (cid, mime_type, size_bytes, created_by_user, storage_key) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (cid) DO NOTHING RETURNING cid",
cid_str,
@@ -120,7 +108,6 @@ pub async fn upload_blob(
)
.fetch_optional(&mut *tx)
.await;
let was_inserted = match insert {
Ok(Some(_)) => true,
Ok(None) => false,
@@ -133,7 +120,6 @@ pub async fn upload_blob(
.into_response();
}
};
if was_inserted {
if let Err(e) = state.blob_store.put_bytes(&storage_key, bytes::Bytes::from(data)).await {
error!("Failed to upload blob to storage: {:?}", e);
@@ -144,7 +130,6 @@ pub async fn upload_blob(
.into_response();
}
}
if let Err(e) = tx.commit().await {
error!("Failed to commit blob transaction: {:?}", e);
if was_inserted {
@@ -158,7 +143,6 @@ pub async fn upload_blob(
)
.into_response();
}
Json(json!({
"blob": {
"ref": {
@@ -170,26 +154,22 @@ pub async fn upload_blob(
}))
.into_response()
}
#[derive(Deserialize)]
pub struct ListMissingBlobsParams {
pub limit: Option<i64>,
pub cursor: Option<String>,
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct RecordBlob {
pub cid: String,
pub record_uri: String,
}
#[derive(Serialize)]
pub struct ListMissingBlobsOutput {
pub cursor: Option<String>,
pub blobs: Vec<RecordBlob>,
}
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") {
@@ -212,7 +192,6 @@ fn find_blobs(val: &serde_json::Value, blobs: &mut Vec<String>) {
}
}
}
pub async fn list_missing_blobs(
State(state): State<AppState>,
headers: axum::http::HeaderMap,
@@ -230,7 +209,6 @@ pub async fn list_missing_blobs(
.into_response();
}
};
let auth_user = match crate::auth::validate_bearer_token(&state.db, &token).await {
Ok(user) => user,
Err(_) => {
@@ -241,13 +219,10 @@ pub async fn list_missing_blobs(
.into_response();
}
};
let did = auth_user.did;
let user_query = sqlx::query!("SELECT id FROM users WHERE did = $1", did)
.fetch_optional(&state.db)
.await;
let user_id = match user_query {
Ok(Some(row)) => row.id,
_ => {
@@ -258,7 +233,6 @@ pub async fn list_missing_blobs(
.into_response();
}
};
let limit = params.limit.unwrap_or(500).clamp(1, 1000);
let cursor_str = params.cursor.unwrap_or_default();
let (cursor_collection, cursor_rkey) = if cursor_str.contains('|') {
@@ -267,7 +241,6 @@ pub async fn list_missing_blobs(
} else {
(String::new(), String::new())
};
let records_query = sqlx::query!(
"SELECT collection, rkey, record_cid FROM records WHERE repo_id = $1 AND (collection, rkey) > ($2, $3) ORDER BY collection, rkey LIMIT $4",
user_id,
@@ -277,7 +250,6 @@ pub async fn list_missing_blobs(
)
.fetch_all(&state.db)
.await;
let records = match records_query {
Ok(r) => r,
Err(e) => {
@@ -289,40 +261,31 @@ pub async fn list_missing_blobs(
.into_response();
}
};
let mut missing_blobs = Vec::new();
let mut last_cursor = None;
for row in &records {
let collection = &row.collection;
let rkey = &row.rkey;
let record_cid_str = &row.record_cid;
last_cursor = Some(format!("{}|{}", collection, rkey));
let record_cid = match Cid::from_str(&record_cid_str) {
Ok(c) => c,
Err(_) => continue,
};
let block_bytes = match state.block_store.get(&record_cid).await {
Ok(Some(b)) => b,
_ => continue,
};
let record_val: serde_json::Value = match serde_ipld_dagcbor::from_slice(&block_bytes) {
Ok(v) => v,
Err(_) => continue,
};
let mut blobs = Vec::new();
find_blobs(&record_val, &mut blobs);
for blob_cid_str in blobs {
let exists = sqlx::query!("SELECT 1 as one FROM blobs WHERE cid = $1 AND created_by_user = $2", blob_cid_str, user_id)
.fetch_optional(&state.db)
.await;
match exists {
Ok(None) => {
missing_blobs.push(RecordBlob {
@@ -337,7 +300,6 @@ pub async fn list_missing_blobs(
}
}
}
// if we fetched fewer records than limit, we are done, so cursor is None.
// otherwise, cursor is the last one we saw.
// ...right?
@@ -346,7 +308,6 @@ pub async fn list_missing_blobs(
} else {
last_cursor
};
(
StatusCode::OK,
Json(ListMissingBlobsOutput {
src/api/repo/import.rs
+4 -30
View File
@@ -11,10 +11,8 @@ use axum::{
};
use serde_json::json;
use tracing::{debug, error, info, warn};
const DEFAULT_MAX_IMPORT_SIZE: usize = 100 * 1024 * 1024;
const DEFAULT_MAX_BLOCKS: usize = 50000;
pub async fn import_repo(
State(state): State<AppState>,
headers: axum::http::HeaderMap,
@@ -23,7 +21,6 @@ pub async fn import_repo(
let accepting_imports = std::env::var("ACCEPTING_REPO_IMPORTS")
.map(|v| v != "false" && v != "0")
.unwrap_or(true);
if !accepting_imports {
return (
StatusCode::BAD_REQUEST,
@@ -34,12 +31,10 @@ pub async fn import_repo(
)
.into_response();
}
let max_size: usize = std::env::var("MAX_IMPORT_SIZE")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(DEFAULT_MAX_IMPORT_SIZE);
if body.len() > max_size {
return (
StatusCode::PAYLOAD_TOO_LARGE,
@@ -50,21 +45,17 @@ pub async fn import_repo(
)
.into_response();
}
let token = match crate::auth::extract_bearer_token_from_header(
headers.get("Authorization").and_then(|h| h.to_str().ok()),
) {
Some(t) => t,
None => return ApiError::AuthenticationRequired.into_response(),
};
let auth_user = match crate::auth::validate_bearer_token(&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",
did
@@ -89,7 +80,6 @@ pub async fn import_repo(
.into_response();
}
};
if user.deactivated_at.is_some() {
return (
StatusCode::FORBIDDEN,
@@ -100,7 +90,6 @@ pub async fn import_repo(
)
.into_response();
}
if user.takedown_ref.is_some() {
return (
StatusCode::FORBIDDEN,
@@ -111,9 +100,7 @@ pub async fn import_repo(
)
.into_response();
}
let user_id = user.id;
let (root, blocks) = match parse_car(&body).await {
Ok((r, b)) => (r, b),
Err(ImportError::InvalidRootCount) => {
@@ -148,14 +135,12 @@ pub async fn import_repo(
.into_response();
}
};
info!(
"Importing repo for user {}: {} blocks, root {}",
did,
blocks.len(),
root
);
let root_block = match blocks.get(&root) {
Some(b) => b,
None => {
@@ -169,7 +154,6 @@ pub async fn import_repo(
.into_response();
}
};
let commit_did = match jacquard_repo::commit::Commit::from_cbor(root_block) {
Ok(commit) => commit.did().to_string(),
Err(e) => {
@@ -183,7 +167,6 @@ pub async fn import_repo(
.into_response();
}
};
if commit_did != *did {
return (
StatusCode::FORBIDDEN,
@@ -197,15 +180,12 @@ pub async fn import_repo(
)
.into_response();
}
let skip_verification = std::env::var("SKIP_IMPORT_VERIFICATION")
.map(|v| v == "true" || v == "1")
.unwrap_or(false);
if !skip_verification {
debug!("Verifying CAR file signature and structure for DID {}", did);
let verifier = CarVerifier::new();
match verifier.verify_car(did, &root, &blocks).await {
Ok(verified) => {
debug!(
@@ -285,12 +265,10 @@ pub async fn import_repo(
} else {
warn!("Skipping CAR signature verification for import (SKIP_IMPORT_VERIFICATION=true)");
}
let max_blocks: usize = std::env::var("MAX_IMPORT_BLOCKS")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(DEFAULT_MAX_BLOCKS);
match apply_import(&state.db, user_id, root, blocks, max_blocks).await {
Ok(records) => {
info!(
@@ -298,11 +276,9 @@ pub async fn import_repo(
records.len(),
did
);
if let Err(e) = sequence_import_event(&state, did, &root.to_string()).await {
warn!("Failed to sequence import event: {:?}", e);
}
(StatusCode::OK, Json(json!({}))).into_response()
}
Err(ImportError::SizeLimitExceeded) => (
@@ -379,36 +355,34 @@ pub async fn import_repo(
}
}
}
async fn sequence_import_event(
state: &AppState,
did: &str,
commit_cid: &str,
) -> Result<(), sqlx::Error> {
let prev_cid: Option<String> = None;
let prev_data_cid: Option<String> = None;
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, $3, $4, $5, $6)
INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, prev_data_cid, ops, blobs, blocks_cids)
VALUES ($1, 'commit', $2, $3, $4, $5, $6, $7)
RETURNING seq
"#,
did,
commit_cid,
prev_cid,
prev_data_cid,
ops,
&blobs,
&blocks_cids
)
.fetch_one(&state.db)
.await?;
sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq_row.seq))
.execute(&state.db)
.await?;
Ok(())
}
src/api/repo/meta.rs
+3 -19
View File
@@ -7,36 +7,25 @@ use axum::{
};
use serde::Deserialize;
use serde_json::json;
#[derive(Deserialize)]
pub struct DescribeRepoInput {
pub repo: String,
}
pub async fn describe_repo(
State(state): State<AppState>,
Query(input): Query<DescribeRepoInput>,
) -> Response {
let hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
let user_row = if input.repo.starts_with("did:") {
sqlx::query!("SELECT id, handle, did FROM users WHERE did = $1", input.repo)
.fetch_optional(&state.db)
.await
.map(|opt| opt.map(|r| (r.id, r.handle, r.did)))
} else {
let suffix = format!(".{}", hostname);
let short_handle = if input.repo.ends_with(&suffix) {
input.repo.strip_suffix(&suffix).unwrap_or(&input.repo)
} else {
&input.repo
};
sqlx::query!("SELECT id, handle, did FROM users WHERE handle = $1", short_handle)
sqlx::query!("SELECT id, handle, did FROM users WHERE handle = $1", input.repo)
.fetch_optional(&state.db)
.await
.map(|opt| opt.map(|r| (r.id, r.handle, r.did)))
};
let (user_id, handle, did) = match user_row {
Ok(Some((id, handle, did))) => (id, handle, did),
_ => {
@@ -47,25 +36,20 @@ pub async fn describe_repo(
.into_response();
}
};
let collections_query =
sqlx::query!("SELECT DISTINCT collection FROM records WHERE repo_id = $1", user_id)
.fetch_all(&state.db)
.await;
let collections: Vec<String> = match collections_query {
Ok(rows) => rows.iter().map(|r| r.collection.clone()).collect(),
Err(_) => Vec::new(),
};
let full_handle = format!("{}.{}", handle, hostname);
let did_doc = json!({
"id": did,
"alsoKnownAs": [format!("at://{}", full_handle)]
"alsoKnownAs": [format!("at://{}", handle)]
});
Json(json!({
"handle": full_handle,
"handle": handle,
"did": did,
"didDoc": did_doc,
"collections": collections,
src/api/repo/mod.rs
-1
View File
@@ -2,7 +2,6 @@ pub mod blob;
pub mod import;
pub mod meta;
pub mod record;
pub use blob::{list_missing_blobs, upload_blob};
pub use import::import_repo;
pub use meta::describe_repo;
src/api/repo/record/batch.rs
+27 -35
View File
@@ -9,18 +9,15 @@ use axum::{
response::{IntoResponse, Response},
Json,
};
use chrono::Utc;
use cid::Cid;
use jacquard::types::string::Nsid;
use jacquard::types::{integer::LimitedU32, string::{Nsid, Tid}};
use jacquard_repo::{commit::Commit, mst::Mst, storage::BlockStore};
use serde::{Deserialize, Serialize};
use serde_json::json;
use std::str::FromStr;
use std::sync::Arc;
use tracing::error;
const MAX_BATCH_WRITES: usize = 200;
#[derive(Deserialize)]
#[serde(tag = "$type")]
pub enum WriteOp {
@@ -39,7 +36,6 @@ pub enum WriteOp {
#[serde(rename = "com.atproto.repo.applyWrites#delete")]
Delete { collection: String, rkey: String },
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ApplyWritesInput {
@@ -48,7 +44,6 @@ pub struct ApplyWritesInput {
pub writes: Vec<WriteOp>,
pub swap_commit: Option<String>,
}
#[derive(Serialize)]
#[serde(tag = "$type")]
pub enum WriteResult {
@@ -59,19 +54,16 @@ pub enum WriteResult {
#[serde(rename = "com.atproto.repo.applyWrites#deleteResult")]
DeleteResult {},
}
#[derive(Serialize)]
pub struct ApplyWritesOutput {
pub commit: CommitInfo,
pub results: Vec<WriteResult>,
}
#[derive(Serialize)]
pub struct CommitInfo {
pub cid: String,
pub rev: String,
}
pub async fn apply_writes(
State(state): State<AppState>,
headers: axum::http::HeaderMap,
@@ -89,7 +81,6 @@ pub async fn apply_writes(
.into_response();
}
};
let auth_user = match crate::auth::validate_bearer_token(&state.db, &token).await {
Ok(user) => user,
Err(_) => {
@@ -100,9 +91,7 @@ pub async fn apply_writes(
.into_response();
}
};
let did = auth_user.did;
if input.repo != did {
return (
StatusCode::FORBIDDEN,
@@ -110,7 +99,6 @@ pub async fn apply_writes(
)
.into_response();
}
match has_verified_notification_channel(&state.db, &did).await {
Ok(true) => {}
Ok(false) => {
@@ -132,7 +120,6 @@ pub async fn apply_writes(
.into_response();
}
}
if input.writes.is_empty() {
return (
StatusCode::BAD_REQUEST,
@@ -140,7 +127,6 @@ pub async fn apply_writes(
)
.into_response();
}
if input.writes.len() > MAX_BATCH_WRITES {
return (
StatusCode::BAD_REQUEST,
@@ -148,7 +134,6 @@ pub async fn apply_writes(
)
.into_response();
}
let user_id: uuid::Uuid = match sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", did)
.fetch_optional(&state.db)
.await
@@ -162,7 +147,6 @@ pub async fn apply_writes(
.into_response();
}
};
let root_cid_str: String =
match sqlx::query_scalar!("SELECT repo_root_cid FROM repos WHERE user_id = $1", user_id)
.fetch_optional(&state.db)
@@ -177,7 +161,6 @@ pub async fn apply_writes(
.into_response();
}
};
let current_root_cid = match Cid::from_str(&root_cid_str) {
Ok(c) => c,
Err(_) => {
@@ -188,7 +171,6 @@ pub async fn apply_writes(
.into_response();
}
};
if let Some(swap_commit) = &input.swap_commit {
if Cid::from_str(swap_commit).ok() != Some(current_root_cid) {
return (
@@ -198,9 +180,7 @@ pub async fn apply_writes(
.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,
_ => {
@@ -211,7 +191,6 @@ pub async fn apply_writes(
.into_response()
}
};
let commit = match Commit::from_cbor(&commit_bytes) {
Ok(c) => c,
_ => {
@@ -222,12 +201,11 @@ pub async fn apply_writes(
.into_response()
}
};
let original_mst = Mst::load(Arc::new(tracking_store.clone()), commit.data, None);
let mut mst = Mst::load(Arc::new(tracking_store.clone()), commit.data, None);
let mut results: Vec<WriteResult> = Vec::new();
let mut ops: Vec<RecordOp> = Vec::new();
let mut modified_keys: Vec<String> = Vec::new();
for write in &input.writes {
match write {
WriteOp::Create {
@@ -242,7 +220,7 @@ pub async fn apply_writes(
}
let rkey = rkey
.clone()
.unwrap_or_else(|| Utc::now().format("%Y%m%d%H%M%S%f").to_string());
.unwrap_or_else(|| Tid::now(LimitedU32::MIN).to_string());
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();
@@ -251,17 +229,16 @@ pub async fn apply_writes(
Ok(c) => c,
Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to store record"}))).into_response(),
};
let collection_nsid = match collection.parse::<Nsid>() {
Ok(n) => n,
Err(_) => return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidCollection", "message": "Invalid collection NSID"}))).into_response(),
};
let key = format!("{}/{}", collection_nsid, rkey);
modified_keys.push(key.clone());
mst = match mst.add(&key, record_cid).await {
Ok(m) => m,
Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to add to MST"}))).into_response(),
};
let uri = format!("at://{}/{}/{}", did, collection, rkey);
results.push(WriteResult::CreateResult {
uri,
@@ -291,17 +268,17 @@ pub async fn apply_writes(
Ok(c) => c,
Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to store record"}))).into_response(),
};
let collection_nsid = match collection.parse::<Nsid>() {
Ok(n) => n,
Err(_) => return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidCollection", "message": "Invalid collection NSID"}))).into_response(),
};
let key = format!("{}/{}", collection_nsid, rkey);
modified_keys.push(key.clone());
let prev_record_cid = mst.get(&key).await.ok().flatten();
mst = match mst.update(&key, record_cid).await {
Ok(m) => m,
Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to update MST"}))).into_response(),
};
let uri = format!("at://{}/{}/{}", did, collection, rkey);
results.push(WriteResult::UpdateResult {
uri,
@@ -311,6 +288,7 @@ pub async fn apply_writes(
collection: collection.clone(),
rkey: rkey.clone(),
cid: record_cid,
prev: prev_record_cid,
});
}
WriteOp::Delete { collection, rkey } => {
@@ -319,35 +297,50 @@ pub async fn apply_writes(
Err(_) => return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidCollection", "message": "Invalid collection NSID"}))).into_response(),
};
let key = format!("{}/{}", collection_nsid, rkey);
modified_keys.push(key.clone());
let prev_record_cid = mst.get(&key).await.ok().flatten();
mst = match mst.delete(&key).await {
Ok(m) => m,
Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to delete from MST"}))).into_response(),
};
results.push(WriteResult::DeleteResult {});
ops.push(RecordOp::Delete {
collection: collection.clone(),
rkey: rkey.clone(),
prev: prev_record_cid,
});
}
}
}
let new_mst_root = match mst.persist().await {
Ok(c) => c,
Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to persist MST"}))).into_response(),
};
let written_cids = tracking_store.get_written_cids();
let mut relevant_blocks = std::collections::BTreeMap::new();
for key in &modified_keys {
if let Err(_) = mst.blocks_for_path(key, &mut relevant_blocks).await {
return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to get new MST blocks for path"}))).into_response();
}
if let Err(_) = original_mst.blocks_for_path(key, &mut relevant_blocks).await {
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();
for cid in relevant_blocks.keys() {
if !written_cids.contains(cid) {
written_cids.push(*cid);
}
}
let written_cids_str = written_cids
.iter()
.map(|c| c.to_string())
.collect::<Vec<_>>();
let commit_res = match commit_and_log(
&state,
&did,
user_id,
Some(current_root_cid),
Some(commit.data),
new_mst_root,
ops,
&written_cids_str,
@@ -364,7 +357,6 @@ pub async fn apply_writes(
.into_response();
}
};
(
StatusCode::OK,
Json(ApplyWritesOutput {
src/api/repo/record/delete.rs
+17 -17
View File
@@ -16,7 +16,6 @@ use serde_json::json;
use std::str::FromStr;
use std::sync::Arc;
use tracing::error;
#[derive(Deserialize)]
pub struct DeleteRecordInput {
pub repo: String,
@@ -27,7 +26,6 @@ pub struct DeleteRecordInput {
#[serde(rename = "swapCommit")]
pub swap_commit: Option<String>,
}
pub async fn delete_record(
State(state): State<AppState>,
headers: HeaderMap,
@@ -38,7 +36,6 @@ pub async fn delete_record(
Ok(res) => res,
Err(err_res) => return err_res,
};
if let Some(swap_commit) = &input.swap_commit {
if Cid::from_str(swap_commit).ok() != Some(current_root_cid) {
return (
@@ -48,9 +45,7 @@ pub async fn delete_record(
.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,
_ => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Commit block not found"}))).into_response(),
@@ -59,7 +54,6 @@ pub async fn delete_record(
Ok(c) => c,
_ => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to parse commit"}))).into_response(),
};
let mst = Mst::load(
Arc::new(tracking_store.clone()),
commit.data,
@@ -70,7 +64,6 @@ pub async fn delete_record(
Err(_) => return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidCollection"}))).into_response(),
};
let key = format!("{}/{}", collection_nsid, input.rkey);
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();
@@ -78,11 +71,10 @@ pub async fn delete_record(
return (StatusCode::CONFLICT, Json(json!({"error": "InvalidSwap", "message": "Record has been modified or does not exist"}))).into_response();
}
}
if mst.get(&key).await.ok().flatten().is_none() {
let prev_record_cid = mst.get(&key).await.ok().flatten();
if prev_record_cid.is_none() {
return (StatusCode::OK, Json(json!({}))).into_response();
}
let new_mst = match mst.delete(&key).await {
Ok(m) => m,
Err(e) => {
@@ -90,7 +82,6 @@ pub async fn delete_record(
return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": format!("Failed to delete from MST: {:?}", e)}))).into_response();
}
};
let new_mst_root = match new_mst.persist().await {
Ok(c) => c,
Err(e) => {
@@ -98,14 +89,23 @@ pub async fn delete_record(
return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to persist MST"}))).into_response();
}
};
let op = RecordOp::Delete { collection: input.collection, rkey: input.rkey };
let written_cids = tracking_store.get_written_cids();
let op = RecordOp::Delete { collection: input.collection, rkey: input.rkey, prev: prev_record_cid };
let mut relevant_blocks = std::collections::BTreeMap::new();
if let Err(_) = new_mst.blocks_for_path(&key, &mut relevant_blocks).await {
return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to get new MST blocks for path"}))).into_response();
}
if let Err(_) = mst.blocks_for_path(&key, &mut relevant_blocks).await {
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();
for cid in relevant_blocks.keys() {
if !written_cids.contains(cid) {
written_cids.push(*cid);
}
}
let written_cids_str = written_cids.iter().map(|c| c.to_string()).collect::<Vec<_>>();
if let Err(e) = commit_and_log(&state, &did, user_id, Some(current_root_cid), new_mst_root, vec![op], &written_cids_str).await {
if let Err(e) = commit_and_log(&state, &did, user_id, Some(current_root_cid), Some(commit.data), new_mst_root, vec![op], &written_cids_str).await {
return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": e}))).into_response();
};
(StatusCode::OK, Json(json!({}))).into_response()
}
src/api/repo/record/mod.rs
-1
View File
@@ -4,7 +4,6 @@ pub mod read;
pub mod utils;
pub mod validation;
pub mod write;
pub use batch::apply_writes;
pub use delete::{DeleteRecordInput, delete_record};
pub use read::{GetRecordInput, ListRecordsInput, ListRecordsOutput, get_record, list_records};
src/api/repo/record/read.rs
-32
View File
@@ -12,7 +12,6 @@ use serde_json::json;
use std::collections::HashMap;
use std::str::FromStr;
use tracing::error;
#[derive(Deserialize)]
pub struct GetRecordInput {
pub repo: String,
@@ -20,13 +19,11 @@ pub struct GetRecordInput {
pub rkey: String,
pub cid: Option<String>,
}
pub async fn get_record(
State(state): State<AppState>,
Query(input): Query<GetRecordInput>,
) -> Response {
let hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
let user_id_opt = if input.repo.starts_with("did:") {
sqlx::query!("SELECT id FROM users WHERE did = $1", input.repo)
.fetch_optional(&state.db)
@@ -44,7 +41,6 @@ pub async fn get_record(
.await
.map(|opt| opt.map(|r| r.id))
};
let user_id: uuid::Uuid = match user_id_opt {
Ok(Some(id)) => id,
_ => {
@@ -55,7 +51,6 @@ pub async fn get_record(
.into_response();
}
};
let record_row = sqlx::query!(
"SELECT record_cid FROM records WHERE repo_id = $1 AND collection = $2 AND rkey = $3",
user_id,
@@ -64,7 +59,6 @@ pub async fn get_record(
)
.fetch_optional(&state.db)
.await;
let record_cid_str: String = match record_row {
Ok(Some(row)) => row.record_cid,
_ => {
@@ -75,7 +69,6 @@ pub async fn get_record(
.into_response();
}
};
if let Some(expected_cid) = &input.cid {
if &record_cid_str != expected_cid {
return (
@@ -85,7 +78,6 @@ pub async fn get_record(
.into_response();
}
}
let cid = match Cid::from_str(&record_cid_str) {
Ok(c) => c,
Err(_) => {
@@ -96,7 +88,6 @@ pub async fn get_record(
.into_response();
}
};
let block = match state.block_store.get(&cid).await {
Ok(Some(b)) => b,
_ => {
@@ -107,7 +98,6 @@ pub async fn get_record(
.into_response();
}
};
let value: serde_json::Value = match serde_ipld_dagcbor::from_slice(&block) {
Ok(v) => v,
Err(e) => {
@@ -119,7 +109,6 @@ pub async fn get_record(
.into_response();
}
};
Json(json!({
"uri": format!("at://{}/{}/{}", input.repo, input.collection, input.rkey),
"cid": record_cid_str,
@@ -127,7 +116,6 @@ pub async fn get_record(
}))
.into_response()
}
#[derive(Deserialize)]
pub struct ListRecordsInput {
pub repo: String,
@@ -140,19 +128,16 @@ pub struct ListRecordsInput {
pub rkey_end: Option<String>,
pub reverse: Option<bool>,
}
#[derive(Serialize)]
pub struct ListRecordsOutput {
pub cursor: Option<String>,
pub records: Vec<serde_json::Value>,
}
pub async fn list_records(
State(state): State<AppState>,
Query(input): Query<ListRecordsInput>,
) -> Response {
let hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
let user_id_opt = if input.repo.starts_with("did:") {
sqlx::query!("SELECT id FROM users WHERE did = $1", input.repo)
.fetch_optional(&state.db)
@@ -170,7 +155,6 @@ pub async fn list_records(
.await
.map(|opt| opt.map(|r| r.id))
};
let user_id: uuid::Uuid = match user_id_opt {
Ok(Some(id)) => id,
_ => {
@@ -181,12 +165,10 @@ pub async fn list_records(
.into_response();
}
};
let limit = input.limit.unwrap_or(50).clamp(1, 100);
let reverse = input.reverse.unwrap_or(false);
let limit_i64 = limit as i64;
let order = if reverse { "ASC" } else { "DESC" };
let rows_res: Result<Vec<(String, String)>, sqlx::Error> = if let Some(cursor) = &input.cursor {
let comparator = if reverse { ">" } else { "<" };
let query = format!(
@@ -203,40 +185,32 @@ pub async fn list_records(
} else {
let mut conditions = vec!["repo_id = $1", "collection = $2"];
let mut param_idx = 3;
if input.rkey_start.is_some() {
conditions.push("rkey > $3");
param_idx += 1;
}
if input.rkey_end.is_some() {
conditions.push(if param_idx == 3 { "rkey < $3" } else { "rkey < $4" });
param_idx += 1;
}
let limit_idx = param_idx;
let query = format!(
"SELECT rkey, record_cid FROM records WHERE {} ORDER BY rkey {} LIMIT ${}",
conditions.join(" AND "),
order,
limit_idx
);
let mut query_builder = sqlx::query_as::<_, (String, String)>(&query)
.bind(user_id)
.bind(&input.collection);
if let Some(start) = &input.rkey_start {
query_builder = query_builder.bind(start);
}
if let Some(end) = &input.rkey_end {
query_builder = query_builder.bind(end);
}
query_builder.bind(limit_i64).fetch_all(&state.db).await
};
let rows = match rows_res {
Ok(r) => r,
Err(e) => {
@@ -248,19 +222,15 @@ pub async fn list_records(
.into_response();
}
};
let last_rkey = rows.last().map(|(rkey, _)| rkey.clone());
let mut cid_to_rkey: HashMap<Cid, (String, String)> = HashMap::new();
let mut cids: Vec<Cid> = Vec::with_capacity(rows.len());
for (rkey, cid_str) in &rows {
if let Ok(cid) = Cid::from_str(cid_str) {
cid_to_rkey.insert(cid, (rkey.clone(), cid_str.clone()));
cids.push(cid);
}
}
let blocks = match state.block_store.get_many(&cids).await {
Ok(b) => b,
Err(e) => {
@@ -272,7 +242,6 @@ pub async fn list_records(
.into_response();
}
};
let mut records = Vec::new();
for (cid, block_opt) in cids.iter().zip(blocks.into_iter()) {
if let Some(block) = block_opt {
@@ -287,7 +256,6 @@ pub async fn list_records(
}
}
}
Json(ListRecordsOutput {
cursor: last_rkey,
records,
src/api/repo/record/utils.rs
+249 -53
View File
@@ -1,28 +1,88 @@
use crate::state::AppState;
use bytes::Bytes;
use cid::Cid;
use jacquard::types::{did::Did, integer::LimitedU32, string::Tid};
use jacquard_repo::commit::Commit;
use jacquard::types::{integer::LimitedU32, string::Tid};
use jacquard_repo::storage::BlockStore;
use k256::ecdsa::SigningKey;
use k256::ecdsa::{signature::Signer, Signature, SigningKey};
use serde::Serialize;
use serde_json::json;
use uuid::Uuid;
/*
* Why am I making custom commit objects instead of jacquard's Commit::sign(), you ask?
*
* At time of writing, jacquard has a bug in how it creates unsigned bytes for signing.
* Jacquard sets sig to empty bytes and serializes (6-field CBOR map)
* Indigo/ATProto creates a struct *without* the sig field (5-field CBOR map)
*
* These produce different CBOR bytes, so signatures created with jacquard
* don't verify with the relay's algorithm. The relay silently rejects commits
* with invalid signatures.
*
* If you have it downloaded, see: reference-relay-indigo/atproto/repo/commit.go UnsignedBytes()
*/
#[derive(Serialize)]
struct UnsignedCommit<'a> {
data: Cid,
did: &'a str,
prev: Option<Cid>,
rev: &'a str,
version: i64,
}
fn create_signed_commit(
did: &str,
data: Cid,
rev: &str,
prev: Option<Cid>,
signing_key: &SigningKey,
) -> Result<(Vec<u8>, Bytes), String> {
let unsigned = UnsignedCommit {
data,
did,
prev,
rev,
version: 3,
};
let unsigned_bytes = serde_ipld_dagcbor::to_vec(&unsigned)
.map_err(|e| format!("Failed to serialize unsigned commit: {:?}", e))?;
let sig: Signature = signing_key.sign(&unsigned_bytes);
let sig_bytes = Bytes::copy_from_slice(&sig.to_bytes());
#[derive(Serialize)]
struct SignedCommit<'a> {
data: Cid,
did: &'a str,
prev: Option<Cid>,
rev: &'a str,
#[serde(with = "serde_bytes")]
sig: &'a [u8],
version: i64,
}
let signed = SignedCommit {
data,
did,
prev,
rev,
sig: &sig_bytes,
version: 3,
};
let signed_bytes = serde_ipld_dagcbor::to_vec(&signed)
.map_err(|e| format!("Failed to serialize signed commit: {:?}", e))?;
Ok((signed_bytes, sig_bytes))
}
pub enum RecordOp {
Create { collection: String, rkey: String, cid: Cid },
Update { collection: String, rkey: String, cid: Cid },
Delete { collection: String, rkey: String },
Update { collection: String, rkey: String, cid: Cid, prev: Option<Cid> },
Delete { collection: String, rkey: String, prev: Option<Cid> },
}
pub struct CommitResult {
pub commit_cid: Cid,
pub rev: String,
}
pub async fn commit_and_log(
state: &AppState,
did: &str,
user_id: Uuid,
current_root_cid: Option<Cid>,
prev_data_cid: Option<Cid>,
new_mst_root: Cid,
ops: Vec<RecordOp>,
blocks_cids: &[String],
@@ -34,37 +94,29 @@ pub async fn commit_and_log(
.fetch_one(&state.db)
.await
.map_err(|e| format!("Failed to fetch signing key: {}", e))?;
let key_bytes = crate::config::decrypt_key(&key_row.key_bytes, key_row.encryption_version)
.map_err(|e| format!("Failed to decrypt signing key: {}", e))?;
let signing_key = SigningKey::from_slice(&key_bytes)
.map_err(|e| format!("Invalid signing key: {}", e))?;
let did_obj = Did::new(did).map_err(|e| format!("Invalid DID: {}", e))?;
let rev = Tid::now(LimitedU32::MIN);
let unsigned_commit = Commit::new_unsigned(did_obj, new_mst_root, rev.clone(), current_root_cid);
let signed_commit = unsigned_commit
.sign(&signing_key)
.map_err(|e| format!("Failed to sign commit: {:?}", e))?;
let new_commit_bytes = signed_commit.to_cbor().map_err(|e| format!("Failed to serialize commit: {:?}", e))?;
let rev_str = rev.to_string();
let (new_commit_bytes, _sig) = create_signed_commit(
did,
new_mst_root,
&rev_str,
current_root_cid,
&signing_key,
)?;
let new_root_cid = state.block_store.put(&new_commit_bytes).await
.map_err(|e| format!("Failed to save commit block: {:?}", e))?;
let mut tx = state.db.begin().await
.map_err(|e| format!("Failed to begin transaction: {}", e))?;
let lock_result = sqlx::query!(
"SELECT repo_root_cid FROM repos WHERE user_id = $1 FOR UPDATE NOWAIT",
user_id
)
.fetch_optional(&mut *tx)
.await;
match lock_result {
Err(e) => {
if let Some(db_err) = e.as_database_error() {
@@ -85,35 +137,28 @@ pub async fn commit_and_log(
return Err("Repo not found".to_string());
}
}
sqlx::query!("UPDATE repos SET repo_root_cid = $1 WHERE user_id = $2", new_root_cid.to_string(), user_id)
.execute(&mut *tx)
.await
.map_err(|e| format!("DB Error (repos): {}", e))?;
let rev_str = rev.to_string();
let mut upsert_collections: Vec<String> = Vec::new();
let mut upsert_rkeys: Vec<String> = Vec::new();
let mut upsert_cids: Vec<String> = Vec::new();
let mut delete_collections: Vec<String> = Vec::new();
let mut delete_rkeys: Vec<String> = Vec::new();
for op in &ops {
match op {
RecordOp::Create { collection, rkey, cid } | RecordOp::Update { collection, rkey, cid } => {
RecordOp::Create { collection, rkey, cid } | RecordOp::Update { collection, rkey, cid, .. } => {
upsert_collections.push(collection.clone());
upsert_rkeys.push(rkey.clone());
upsert_cids.push(cid.to_string());
}
RecordOp::Delete { collection, rkey } => {
RecordOp::Delete { collection, rkey, .. } => {
delete_collections.push(collection.clone());
delete_rkeys.push(rkey.clone());
}
}
}
if !upsert_collections.is_empty() {
sqlx::query!(
r#"
@@ -133,7 +178,6 @@ pub async fn commit_and_log(
.await
.map_err(|e| format!("DB Error (records batch upsert): {}", e))?;
}
if !delete_collections.is_empty() {
sqlx::query!(
r#"
@@ -149,7 +193,6 @@ pub async fn commit_and_log(
.await
.map_err(|e| format!("DB Error (records batch delete): {}", e))?;
}
let ops_json = ops.iter().map(|op| {
match op {
RecordOp::Create { collection, rkey, cid } => json!({
@@ -157,26 +200,37 @@ pub async fn commit_and_log(
"path": format!("{}/{}", collection, rkey),
"cid": cid.to_string()
}),
RecordOp::Update { collection, rkey, cid } => json!({
"action": "update",
"path": format!("{}/{}", collection, rkey),
"cid": cid.to_string()
}),
RecordOp::Delete { collection, rkey } => json!({
"action": "delete",
"path": format!("{}/{}", collection, rkey),
"cid": null
}),
RecordOp::Update { collection, rkey, cid, prev } => {
let mut obj = json!({
"action": "update",
"path": format!("{}/{}", collection, rkey),
"cid": cid.to_string()
});
if let Some(prev_cid) = prev {
obj["prev"] = json!(prev_cid.to_string());
}
obj
},
RecordOp::Delete { collection, rkey, prev } => {
let mut obj = json!({
"action": "delete",
"path": format!("{}/{}", collection, rkey),
"cid": null
});
if let Some(prev_cid) = prev {
obj["prev"] = json!(prev_cid.to_string());
}
obj
},
}
}).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)
VALUES ($1, $2, $3, $4, $5, $6, $7)
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,
@@ -186,23 +240,165 @@ pub async fn commit_and_log(
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)
.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;
Ok(CommitResult {
commit_cid: new_root_cid,
rev: rev.to_string(),
rev: rev_str,
})
}
pub async fn create_record_internal(
state: &AppState,
did: &str,
collection: &str,
rkey: &str,
record: &serde_json::Value,
) -> Result<(String, Cid), String> {
use crate::repo::tracking::TrackingBlockStore;
use jacquard_repo::mst::Mst;
use std::sync::Arc;
let user_id: Uuid = sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", did)
.fetch_optional(&state.db)
.await
.map_err(|e| format!("DB error: {}", e))?
.ok_or_else(|| "User not found".to_string())?;
let root_cid_str: String =
sqlx::query_scalar!("SELECT repo_root_cid FROM repos WHERE user_id = $1", user_id)
.fetch_optional(&state.db)
.await
.map_err(|e| format!("DB error: {}", e))?
.ok_or_else(|| "Repo not found".to_string())?;
let current_root_cid = Cid::from_str(&root_cid_str)
.map_err(|_| "Invalid repo root CID".to_string())?;
let tracking_store = TrackingBlockStore::new(state.block_store.clone());
let commit_bytes = tracking_store.get(&current_root_cid).await
.map_err(|e| format!("Failed to fetch commit: {:?}", e))?
.ok_or_else(|| "Commit block not found".to_string())?;
let commit = jacquard_repo::commit::Commit::from_cbor(&commit_bytes)
.map_err(|e| format!("Failed to parse commit: {:?}", e))?;
let mst = Mst::load(Arc::new(tracking_store.clone()), commit.data, None);
let mut record_bytes = Vec::new();
serde_ipld_dagcbor::to_writer(&mut record_bytes, record)
.map_err(|e| format!("Failed to serialize record: {:?}", e))?;
let record_cid = tracking_store.put(&record_bytes).await
.map_err(|e| format!("Failed to save record block: {:?}", e))?;
let key = format!("{}/{}", collection, rkey);
let new_mst = mst.add(&key, record_cid).await
.map_err(|e| format!("Failed to add to MST: {:?}", e))?;
let new_mst_root = new_mst.persist().await
.map_err(|e| format!("Failed to persist MST: {:?}", e))?;
let op = RecordOp::Create {
collection: collection.to_string(),
rkey: rkey.to_string(),
cid: record_cid,
};
let mut relevant_blocks = std::collections::BTreeMap::new();
new_mst.blocks_for_path(&key, &mut relevant_blocks).await
.map_err(|e| format!("Failed to get new MST blocks for path: {:?}", e))?;
mst.blocks_for_path(&key, &mut relevant_blocks).await
.map_err(|e| format!("Failed to get old MST blocks for path: {:?}", e))?;
relevant_blocks.insert(record_cid, bytes::Bytes::from(record_bytes));
let mut written_cids = tracking_store.get_all_relevant_cids();
for cid in relevant_blocks.keys() {
if !written_cids.contains(cid) {
written_cids.push(*cid);
}
}
let written_cids_str: Vec<String> = written_cids.iter().map(|c| c.to_string()).collect();
let result = commit_and_log(
state,
did,
user_id,
Some(current_root_cid),
Some(commit.data),
new_mst_root,
vec![op],
&written_cids_str,
).await?;
let uri = format!("at://{}/{}/{}", did, collection, rkey);
Ok((uri, result.commit_cid))
}
use std::str::FromStr;
pub async fn sequence_identity_event(
state: &AppState,
did: &str,
handle: Option<&str>,
) -> Result<i64, String> {
let seq_row = sqlx::query!(
r#"
INSERT INTO repo_seq (did, event_type, handle)
VALUES ($1, 'identity', $2)
RETURNING seq
"#,
did,
handle,
)
.fetch_one(&state.db)
.await
.map_err(|e| format!("DB Error (repo_seq identity): {}", 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)
}
pub async fn sequence_account_event(
state: &AppState,
did: &str,
active: bool,
status: Option<&str>,
) -> Result<i64, String> {
let seq_row = sqlx::query!(
r#"
INSERT INTO repo_seq (did, event_type, active, status)
VALUES ($1, 'account', $2, $3)
RETURNING seq
"#,
did,
active,
status,
)
.fetch_one(&state.db)
.await
.map_err(|e| format!("DB Error (repo_seq account): {}", 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)
}
pub async fn sequence_sync_event(
state: &AppState,
did: &str,
commit_cid: &str,
) -> Result<i64, String> {
let seq_row = sqlx::query!(
r#"
INSERT INTO repo_seq (did, event_type, commit_cid)
VALUES ($1, 'sync', $2)
RETURNING seq
"#,
did,
commit_cid,
)
.fetch_one(&state.db)
.await
.map_err(|e| format!("DB Error (repo_seq sync): {}", 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/validation.rs
-1
View File
@@ -5,7 +5,6 @@ use axum::{
Json,
};
use serde_json::json;
pub fn validate_record(record: &serde_json::Value, collection: &str) -> Result<(), Response> {
let validator = RecordValidator::new();
match validator.validate(record, collection) {
src/api/repo/record/write.rs
+33 -49
View File
@@ -8,9 +8,8 @@ use axum::{
response::{IntoResponse, Response},
Json,
};
use chrono::Utc;
use cid::Cid;
use jacquard::types::string::Nsid;
use jacquard::types::{integer::LimitedU32, string::{Nsid, Tid}};
use jacquard_repo::{commit::Commit, mst::Mst, storage::BlockStore};
use serde::{Deserialize, Serialize};
use serde_json::json;
@@ -19,7 +18,6 @@ use std::str::FromStr;
use std::sync::Arc;
use tracing::error;
use uuid::Uuid;
pub async fn has_verified_notification_channel(db: &PgPool, did: &str) -> Result<bool, sqlx::Error> {
let row = sqlx::query(
r#"
@@ -35,7 +33,6 @@ pub async fn has_verified_notification_channel(db: &PgPool, did: &str) -> Result
.bind(did)
.fetch_optional(db)
.await?;
match row {
Some(r) => {
let email_confirmed: bool = r.get("email_confirmed");
@@ -47,7 +44,6 @@ pub async fn has_verified_notification_channel(db: &PgPool, did: &str) -> Result
None => Ok(false),
}
}
pub async fn prepare_repo_write(
state: &AppState,
headers: &HeaderMap,
@@ -62,7 +58,6 @@ pub async fn prepare_repo_write(
)
.into_response()
})?;
let auth_user = crate::auth::validate_bearer_token(&state.db, &token)
.await
.map_err(|_| {
@@ -72,7 +67,6 @@ pub async fn prepare_repo_write(
)
.into_response()
})?;
if repo_did != auth_user.did {
return Err((
StatusCode::FORBIDDEN,
@@ -80,7 +74,6 @@ pub async fn prepare_repo_write(
)
.into_response());
}
match has_verified_notification_channel(&state.db, &auth_user.did).await {
Ok(true) => {}
Ok(false) => {
@@ -102,7 +95,6 @@ pub async fn prepare_repo_write(
.into_response());
}
}
let user_id = sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", auth_user.did)
.fetch_optional(&state.db)
.await
@@ -117,7 +109,6 @@ pub async fn prepare_repo_write(
)
.into_response()
})?;
let root_cid_str: String =
sqlx::query_scalar!("SELECT repo_root_cid FROM repos WHERE user_id = $1", user_id)
.fetch_optional(&state.db)
@@ -133,7 +124,6 @@ pub async fn prepare_repo_write(
)
.into_response()
})?;
let current_root_cid = Cid::from_str(&root_cid_str).map_err(|_| {
(
StatusCode::INTERNAL_SERVER_ERROR,
@@ -141,10 +131,8 @@ pub async fn prepare_repo_write(
)
.into_response()
})?;
Ok((auth_user.did, user_id, current_root_cid))
}
#[derive(Deserialize)]
#[allow(dead_code)]
pub struct CreateRecordInput {
@@ -156,14 +144,12 @@ pub struct CreateRecordInput {
#[serde(rename = "swapCommit")]
pub swap_commit: Option<String>,
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CreateRecordOutput {
pub uri: String,
pub cid: String,
}
pub async fn create_record(
State(state): State<AppState>,
headers: HeaderMap,
@@ -174,7 +160,6 @@ pub async fn create_record(
Ok(res) => res,
Err(err_res) => return err_res,
};
if let Some(swap_commit) = &input.swap_commit {
if Cid::from_str(swap_commit).ok() != Some(current_root_cid) {
return (
@@ -184,9 +169,7 @@ pub async fn create_record(
.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,
_ => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Commit block not found"}))).into_response(),
@@ -195,26 +178,21 @@ pub async fn create_record(
Ok(c) => c,
_ => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to parse commit"}))).into_response(),
};
let mst = Mst::load(
Arc::new(tracking_store.clone()),
commit.data,
None,
);
let collection_nsid = match input.collection.parse::<Nsid>() {
Ok(n) => n,
Err(_) => return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidCollection"}))).into_response(),
};
if input.validate.unwrap_or(true) {
if let Err(err_response) = validate_record(&input.record, &input.collection) {
return err_response;
}
}
let rkey = input.rkey.unwrap_or_else(|| Utc::now().format("%Y%m%d%H%M%S%f").to_string());
let rkey = input.rkey.unwrap_or_else(|| Tid::now(LimitedU32::MIN).to_string());
let mut record_bytes = Vec::new();
if serde_ipld_dagcbor::to_writer(&mut record_bytes, &input.record).is_err() {
return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidRecord", "message": "Failed to serialize record"}))).into_response();
@@ -223,7 +201,6 @@ pub async fn create_record(
Ok(c) => c,
_ => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to save record block"}))).into_response(),
};
let key = format!("{}/{}", collection_nsid, rkey);
let new_mst = match mst.add(&key, record_cid).await {
Ok(m) => m,
@@ -233,21 +210,30 @@ pub async fn create_record(
Ok(c) => c,
_ => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to persist MST"}))).into_response(),
};
let op = RecordOp::Create { collection: input.collection.clone(), rkey: rkey.clone(), cid: record_cid };
let written_cids = tracking_store.get_written_cids();
let mut relevant_blocks = std::collections::BTreeMap::new();
if let Err(_) = new_mst.blocks_for_path(&key, &mut relevant_blocks).await {
return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to get new MST blocks for path"}))).into_response();
}
if let Err(_) = mst.blocks_for_path(&key, &mut relevant_blocks).await {
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));
let mut written_cids = tracking_store.get_all_relevant_cids();
for cid in relevant_blocks.keys() {
if !written_cids.contains(cid) {
written_cids.push(*cid);
}
}
let written_cids_str = written_cids.iter().map(|c| c.to_string()).collect::<Vec<_>>();
if let Err(e) = commit_and_log(&state, &did, user_id, Some(current_root_cid), new_mst_root, vec![op], &written_cids_str).await {
if let Err(e) = commit_and_log(&state, &did, user_id, Some(current_root_cid), Some(commit.data), new_mst_root, vec![op], &written_cids_str).await {
return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": e}))).into_response();
};
(StatusCode::OK, Json(CreateRecordOutput {
uri: format!("at://{}/{}/{}", did, input.collection, rkey),
cid: record_cid.to_string(),
})).into_response()
}
#[derive(Deserialize)]
#[allow(dead_code)]
pub struct PutRecordInput {
@@ -261,14 +247,12 @@ pub struct PutRecordInput {
#[serde(rename = "swapRecord")]
pub swap_record: Option<String>,
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct PutRecordOutput {
pub uri: String,
pub cid: String,
}
pub async fn put_record(
State(state): State<AppState>,
headers: HeaderMap,
@@ -279,15 +263,12 @@ pub async fn put_record(
Ok(res) => res,
Err(err_res) => return err_res,
};
if let Some(swap_commit) = &input.swap_commit {
if 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,
_ => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Commit block not found"}))).into_response(),
@@ -296,7 +277,6 @@ pub async fn put_record(
Ok(c) => c,
_ => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to parse commit"}))).into_response(),
};
let mst = Mst::load(
Arc::new(tracking_store.clone()),
commit.data,
@@ -307,13 +287,11 @@ pub async fn put_record(
Err(_) => return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidCollection"}))).into_response(),
};
let key = format!("{}/{}", collection_nsid, input.rkey);
if input.validate.unwrap_or(true) {
if 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();
@@ -321,9 +299,7 @@ pub async fn put_record(
return (StatusCode::CONFLICT, Json(json!({"error": "InvalidSwap", "message": "Record has been modified or does not exist"}))).into_response();
}
}
let existing_cid = mst.get(&key).await.ok().flatten();
let mut record_bytes = Vec::new();
if serde_ipld_dagcbor::to_writer(&mut record_bytes, &input.record).is_err() {
return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidRecord", "message": "Failed to serialize record"}))).into_response();
@@ -332,7 +308,6 @@ pub async fn put_record(
Ok(c) => c,
_ => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to save record block"}))).into_response(),
};
let new_mst = if existing_cid.is_some() {
match mst.update(&key, record_cid).await {
Ok(m) => m,
@@ -348,20 +323,29 @@ pub async fn put_record(
Ok(c) => c,
Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to persist MST"}))).into_response(),
};
let op = if existing_cid.is_some() {
RecordOp::Update { collection: input.collection.clone(), rkey: input.rkey.clone(), cid: record_cid }
RecordOp::Update { collection: input.collection.clone(), rkey: input.rkey.clone(), cid: record_cid, prev: existing_cid }
} else {
RecordOp::Create { collection: input.collection.clone(), rkey: input.rkey.clone(), cid: record_cid }
};
let written_cids = tracking_store.get_written_cids();
let mut relevant_blocks = std::collections::BTreeMap::new();
if let Err(_) = new_mst.blocks_for_path(&key, &mut relevant_blocks).await {
return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Failed to get new MST blocks for path"}))).into_response();
}
if let Err(_) = mst.blocks_for_path(&key, &mut relevant_blocks).await {
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));
let mut written_cids = tracking_store.get_all_relevant_cids();
for cid in relevant_blocks.keys() {
if !written_cids.contains(cid) {
written_cids.push(*cid);
}
}
let written_cids_str = written_cids.iter().map(|c| c.to_string()).collect::<Vec<_>>();
if let Err(e) = commit_and_log(&state, &did, user_id, Some(current_root_cid), new_mst_root, vec![op], &written_cids_str).await {
if let Err(e) = commit_and_log(&state, &did, user_id, Some(current_root_cid), Some(commit.data), new_mst_root, vec![op], &written_cids_str).await {
return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": e}))).into_response();
};
(StatusCode::OK, Json(PutRecordOutput {
uri: format!("at://{}/{}/{}", did, input.collection, input.rkey),
cid: record_cid.to_string(),