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

Sync conformance fixes vs ref

Browse Source
This commit is contained in:
lewis
2025-12-28 23:41:17 +02:00
parent 013b80adc2
commit cd4780af41
30 changed files with 1928 additions and 351 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/admin/account/info.rs
+1 -19
View File
@@ -88,30 +88,12 @@ pub async fn get_account_info(
}
}
fn parse_repeated_param(query: Option<&str>, key: &str) -> Vec<String> {
query
.map(|q| {
q.split('&')
.filter_map(|pair| {
let (k, v) = pair.split_once('=')?;
if k == key {
Some(urlencoding::decode(v).ok()?.into_owned())
} else {
None
}
})
.collect()
})
.unwrap_or_default()
}
pub async fn get_account_infos(
State(state): State<AppState>,
_auth: BearerAuthAdmin,
RawQuery(raw_query): RawQuery,
) -> Response {
let dids = parse_repeated_param(raw_query.as_deref(), "dids");
let dids = crate::util::parse_repeated_query_param(raw_query.as_deref(), "dids");
if dids.is_empty() {
return (
StatusCode::BAD_REQUEST,
src/moderation/mod.rs
+3 -1
View File
@@ -107,7 +107,9 @@ mod tests {
use base64::Engine;
fn d(b64: &str) -> String {
let bytes = base64::engine::general_purpose::STANDARD.decode(b64).unwrap();
let bytes = base64::engine::general_purpose::STANDARD
.decode(b64)
.unwrap();
String::from_utf8(bytes).unwrap()
}
src/sync/blob.rs
+33 -56
View File
@@ -1,4 +1,5 @@
use crate::state::AppState;
use crate::sync::util::assert_repo_availability;
use axum::{
Json,
body::Body,
@@ -37,29 +38,14 @@ pub async fn get_blob(
)
.into_response();
}
let user_exists = sqlx::query!("SELECT id FROM users WHERE did = $1", did)
.fetch_optional(&state.db)
.await;
match user_exists {
Ok(None) => {
return (
StatusCode::NOT_FOUND,
Json(json!({"error": "RepoNotFound", "message": "Could not find repo for DID"})),
)
.into_response();
}
Err(e) => {
error!("DB error in get_blob: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
Ok(Some(_)) => {}
}
let _account = match assert_repo_availability(&state.db, did, false).await {
Ok(a) => a,
Err(e) => return e.into_response(),
};
let blob_result = sqlx::query!(
"SELECT storage_key, mime_type FROM blobs WHERE cid = $1",
"SELECT storage_key, mime_type, size_bytes FROM blobs WHERE cid = $1",
cid
)
.fetch_optional(&state.db)
@@ -68,10 +54,14 @@ pub async fn get_blob(
Ok(Some(row)) => {
let storage_key = &row.storage_key;
let mime_type = &row.mime_type;
let size_bytes = row.size_bytes;
match state.blob_store.get(storage_key).await {
Ok(data) => Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, mime_type)
.header(header::CONTENT_LENGTH, size_bytes.to_string())
.header("x-content-type-options", "nosniff")
.header("content-security-policy", "default-src 'none'; sandbox")
.body(Body::from(data))
.unwrap(),
Err(e) => {
@@ -127,48 +117,35 @@ pub async fn list_blobs(
)
.into_response();
}
let account = match assert_repo_availability(&state.db, did, false).await {
Ok(a) => a,
Err(e) => return e.into_response(),
};
let limit = params.limit.unwrap_or(500).clamp(1, 1000);
let cursor_cid = params.cursor.as_deref().unwrap_or("");
let user_result = sqlx::query!("SELECT id FROM users WHERE did = $1", did)
.fetch_optional(&state.db)
.await;
let user_id = match user_result {
Ok(Some(row)) => row.id,
Ok(None) => {
return (
StatusCode::NOT_FOUND,
Json(json!({"error": "RepoNotFound", "message": "Could not find repo for DID"})),
)
.into_response();
}
Err(e) => {
error!("DB error in list_blobs: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
};
let user_id = account.user_id;
let cids_result: Result<Vec<String>, sqlx::Error> = if let Some(since) = &params.since {
let since_time = chrono::DateTime::parse_from_rfc3339(since)
.map(|dt| dt.with_timezone(&chrono::Utc))
.unwrap_or_else(|_| chrono::Utc::now());
sqlx::query!(
sqlx::query_scalar!(
r#"
SELECT cid FROM blobs
WHERE created_by_user = $1 AND cid > $2 AND created_at > $3
ORDER BY cid ASC
LIMIT $4
SELECT DISTINCT unnest(blobs) as "cid!"
FROM repo_seq
WHERE did = $1 AND rev > $2 AND blobs IS NOT NULL
"#,
user_id,
cursor_cid,
since_time,
limit + 1
did,
since
)
.fetch_all(&state.db)
.await
.map(|rows| rows.into_iter().map(|r| r.cid).collect())
.map(|mut cids| {
cids.sort();
cids.into_iter()
.filter(|c| c.as_str() > cursor_cid)
.take((limit + 1) as usize)
.collect()
})
} else {
sqlx::query!(
r#"
src/sync/car.rs
+12
View File
@@ -34,3 +34,15 @@ pub fn encode_car_header(root_cid: &Cid) -> Result<Vec<u8>, String> {
result.extend_from_slice(&header_cbor);
Ok(result)
}
pub fn encode_car_header_null_root() -> Result<Vec<u8>, String> {
let header = CarHeader::new_v1(vec![]);
let header_cbor = header
.encode()
.map_err(|e| format!("Failed to encode CAR header: {:?}", e))?;
let mut result = Vec::new();
write_varint(&mut result, header_cbor.len() as u64)
.expect("Writing to Vec<u8> should never fail");
result.extend_from_slice(&header_cbor);
Ok(result)
}
src/sync/commit.rs
+97 -68
View File
@@ -1,4 +1,5 @@
use crate::state::AppState;
use crate::sync::util::{AccountStatus, assert_repo_availability, get_account_with_status};
use axum::{
Json,
extract::{Query, State},
@@ -43,45 +44,46 @@ pub async fn get_latest_commit(
)
.into_response();
}
let result = sqlx::query!(
r#"
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;
match result {
Ok(Some(row)) => {
let rev = get_rev_from_commit(&state, &row.repo_root_cid)
.await
.unwrap_or_else(|| chrono::Utc::now().timestamp_millis().to_string());
(
StatusCode::OK,
Json(GetLatestCommitOutput {
cid: row.repo_root_cid,
rev,
}),
let account = match assert_repo_availability(&state.db, did, false).await {
Ok(a) => a,
Err(e) => return e.into_response(),
};
let repo_root_cid = match account.repo_root_cid {
Some(cid) => cid,
None => {
return (
StatusCode::BAD_REQUEST,
Json(json!({"error": "RepoNotFound", "message": "Repo not initialized"})),
)
.into_response()
.into_response();
}
Ok(None) => (
StatusCode::NOT_FOUND,
Json(json!({"error": "RepoNotFound", "message": "Could not find repo for DID"})),
)
.into_response(),
Err(e) => {
error!("DB error in get_latest_commit: {:?}", e);
(
};
let rev = match get_rev_from_commit(&state, &repo_root_cid).await {
Some(r) => r,
None => {
error!(
"Failed to parse commit for DID {}: CID {}",
did, repo_root_cid
);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
Json(json!({"error": "InternalError", "message": "Failed to read repo commit"})),
)
.into_response()
.into_response();
}
}
};
(
StatusCode::OK,
Json(GetLatestCommitOutput {
cid: repo_root_cid,
rev,
}),
)
.into_response()
}
#[derive(Deserialize)]
@@ -97,6 +99,8 @@ pub struct RepoInfo {
pub head: String,
pub rev: String,
pub active: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub status: Option<String>,
}
#[derive(Serialize)]
@@ -114,7 +118,7 @@ pub async fn list_repos(
let cursor_did = params.cursor.as_deref().unwrap_or("");
let result = sqlx::query!(
r#"
SELECT u.did, r.repo_root_cid
SELECT u.did, u.deactivated_at, u.takedown_ref, r.repo_root_cid, r.repo_rev
FROM repos r
JOIN users u ON r.user_id = u.id
WHERE u.did > $1
@@ -131,14 +135,34 @@ pub async fn list_repos(
let has_more = rows.len() as i64 > limit;
let mut repos: Vec<RepoInfo> = Vec::new();
for row in rows.iter().take(limit as usize) {
let rev = get_rev_from_commit(&state, &row.repo_root_cid)
.await
.unwrap_or_else(|| chrono::Utc::now().timestamp_millis().to_string());
let rev = match get_rev_from_commit(&state, &row.repo_root_cid).await {
Some(r) => r,
None => {
if let Some(ref stored_rev) = row.repo_rev {
stored_rev.clone()
} else {
tracing::warn!(
"Failed to parse commit for DID {} in list_repos: CID {}",
row.did,
row.repo_root_cid
);
continue;
}
}
};
let status = if row.takedown_ref.is_some() {
AccountStatus::Takendown
} else if row.deactivated_at.is_some() {
AccountStatus::Deactivated
} else {
AccountStatus::Active
};
repos.push(RepoInfo {
did: row.did.clone(),
head: row.repo_root_cid.clone(),
rev,
active: true,
active: status.is_active(),
status: status.as_str().map(String::from),
});
}
let next_cursor = if has_more {
@@ -175,6 +199,9 @@ pub struct GetRepoStatusParams {
pub struct GetRepoStatusOutput {
pub did: String,
pub active: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub status: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub rev: Option<String>,
}
@@ -190,42 +217,44 @@ pub async fn get_repo_status(
)
.into_response();
}
let result = sqlx::query!(
r#"
SELECT u.did, r.repo_root_cid
FROM users u
LEFT JOIN repos r ON u.id = r.user_id
WHERE u.did = $1
"#,
did
)
.fetch_optional(&state.db)
.await;
match result {
Ok(Some(row)) => {
let rev = get_rev_from_commit(&state, &row.repo_root_cid).await;
(
StatusCode::OK,
Json(GetRepoStatusOutput {
did: row.did,
active: true,
rev,
}),
let account = match get_account_with_status(&state.db, did).await {
Ok(Some(a)) => a,
Ok(None) => {
return (
StatusCode::BAD_REQUEST,
Json(json!({"error": "RepoNotFound", "message": format!("Could not find repo for DID: {}", did)})),
)
.into_response()
}
Ok(None) => (
StatusCode::NOT_FOUND,
Json(json!({"error": "RepoNotFound", "message": "Could not find repo for DID"})),
)
.into_response(),
Err(e) => {
error!("DB error in get_repo_status: {:?}", e);
(
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response()
.into_response();
}
}
};
let rev = if account.status.is_active() {
if let Some(ref cid) = account.repo_root_cid {
get_rev_from_commit(&state, cid).await
} else {
None
}
} else {
None
};
(
StatusCode::OK,
Json(GetRepoStatusOutput {
did: account.did,
active: account.status.is_active(),
status: account.status.as_str().map(String::from),
rev,
}),
)
.into_response()
}
src/sync/frame.rs
+19
View File
@@ -74,6 +74,25 @@ pub struct SyncFrame {
pub time: String,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct InfoFrame {
pub name: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub message: Option<String>,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct ErrorFrameHeader {
pub op: i64,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct ErrorFrameBody {
pub error: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub message: Option<String>,
}
pub struct CommitFrameBuilder {
pub seq: i64,
pub did: String,
src/sync/mod.rs
+4
View File
@@ -18,4 +18,8 @@ pub use crawl::{notify_of_update, request_crawl};
pub use deprecated::{get_checkout, get_head};
pub use repo::{get_blocks, get_record, get_repo};
pub use subscribe_repos::subscribe_repos;
pub use util::{
AccountStatus, RepoAccount, RepoAvailabilityError, assert_repo_availability,
get_account_with_status,
};
pub use verify::{CarVerifier, VerifiedCar, VerifyError};
src/sync/repo.rs
+210 -72
View File
@@ -1,8 +1,9 @@
use crate::state::AppState;
use crate::sync::car::encode_car_header;
use crate::sync::util::assert_repo_availability;
use axum::{
Json,
extract::{Query, State},
extract::{Query, RawQuery, State},
http::StatusCode,
response::{IntoResponse, Response},
};
@@ -17,48 +18,102 @@ use tracing::error;
const MAX_REPO_BLOCKS_TRAVERSAL: usize = 20_000;
#[derive(Deserialize)]
pub struct GetBlocksQuery {
pub did: String,
pub cids: String,
fn parse_get_blocks_query(query_string: &str) -> Result<(String, Vec<String>), String> {
let did = crate::util::parse_repeated_query_param(Some(query_string), "did")
.into_iter()
.next()
.ok_or("Missing required parameter: did")?;
let cids = crate::util::parse_repeated_query_param(Some(query_string), "cids");
Ok((did, cids))
}
pub async fn get_blocks(
State(state): State<AppState>,
Query(query): Query<GetBlocksQuery>,
) -> Response {
let user_exists = sqlx::query!("SELECT id FROM users WHERE did = $1", query.did)
.fetch_optional(&state.db)
.await
.unwrap_or(None);
if user_exists.is_none() {
return (StatusCode::NOT_FOUND, "Repo not found").into_response();
}
let cids_str: Vec<&str> = query.cids.split(',').collect();
pub async fn get_blocks(State(state): State<AppState>, RawQuery(query): RawQuery) -> Response {
let query_string = match query {
Some(q) => q,
None => {
return (
StatusCode::BAD_REQUEST,
Json(json!({"error": "InvalidRequest", "message": "Missing query parameters"})),
)
.into_response();
}
};
let (did, cid_strings) = match parse_get_blocks_query(&query_string) {
Ok(parsed) => parsed,
Err(msg) => {
return (
StatusCode::BAD_REQUEST,
Json(json!({"error": "InvalidRequest", "message": msg})),
)
.into_response();
}
};
let _account = match assert_repo_availability(&state.db, &did, false).await {
Ok(a) => a,
Err(e) => return e.into_response(),
};
let mut cids = Vec::new();
for s in cids_str {
for s in &cid_strings {
match Cid::from_str(s) {
Ok(cid) => cids.push(cid),
Err(_) => return (StatusCode::BAD_REQUEST, "Invalid CID").into_response(),
Err(_) => return (
StatusCode::BAD_REQUEST,
Json(json!({"error": "InvalidRequest", "message": format!("Invalid CID: {}", s)})),
)
.into_response(),
}
}
if cids.is_empty() {
return (
StatusCode::BAD_REQUEST,
Json(json!({"error": "InvalidRequest", "message": "No CIDs provided"})),
)
.into_response();
}
let blocks_res = state.block_store.get_many(&cids).await;
let blocks = match blocks_res {
Ok(blocks) => blocks,
Err(e) => {
error!("Failed to get blocks: {}", e);
return (StatusCode::INTERNAL_SERVER_ERROR, "Failed to get blocks").into_response();
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError", "message": "Failed to get blocks"})),
)
.into_response();
}
};
if cids.is_empty() {
return (StatusCode::BAD_REQUEST, "No CIDs provided").into_response();
let mut missing_cids: Vec<String> = Vec::new();
for (i, block_opt) in blocks.iter().enumerate() {
if block_opt.is_none() {
missing_cids.push(cids[i].to_string());
}
}
let root_cid = cids[0];
let header = match encode_car_header(&root_cid) {
if !missing_cids.is_empty() {
return (
StatusCode::BAD_REQUEST,
Json(json!({
"error": "InvalidRequest",
"message": format!("Could not find blocks: {}", missing_cids.join(", "))
})),
)
.into_response();
}
let header = match crate::sync::car::encode_car_header_null_root() {
Ok(h) => h,
Err(e) => {
error!("Failed to encode CAR header: {}", e);
return (StatusCode::INTERNAL_SERVER_ERROR, "Failed to encode CAR").into_response();
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError", "message": "Failed to encode CAR"})),
)
.into_response();
}
};
let mut car_bytes = header;
@@ -97,40 +152,22 @@ pub async fn get_repo(
State(state): State<AppState>,
Query(query): Query<GetRepoQuery>,
) -> Response {
let repo_row = sqlx::query!(
r#"
SELECT r.repo_root_cid
FROM repos r
JOIN users u ON u.id = r.user_id
WHERE u.did = $1
"#,
query.did
)
.fetch_optional(&state.db)
.await
.unwrap_or(None);
let head_str = match repo_row {
Some(r) => r.repo_root_cid,
let account = match assert_repo_availability(&state.db, &query.did, false).await {
Ok(a) => a,
Err(e) => return e.into_response(),
};
let head_str = match account.repo_root_cid {
Some(cid) => cid,
None => {
let user_exists = sqlx::query!("SELECT id FROM users WHERE did = $1", query.did)
.fetch_optional(&state.db)
.await
.unwrap_or(None);
if user_exists.is_none() {
return (
StatusCode::NOT_FOUND,
Json(json!({"error": "RepoNotFound", "message": "Repo not found"})),
)
.into_response();
} else {
return (
StatusCode::NOT_FOUND,
Json(json!({"error": "RepoNotFound", "message": "Repo not initialized"})),
)
.into_response();
}
return (
StatusCode::BAD_REQUEST,
Json(json!({"error": "RepoNotFound", "message": "Repo not initialized"})),
)
.into_response();
}
};
let head_cid = match Cid::from_str(&head_str) {
Ok(c) => c,
Err(_) => {
@@ -141,6 +178,11 @@ pub async fn get_repo(
.into_response();
}
};
if let Some(since) = &query.since {
return get_repo_since(&state, &query.did, &head_cid, since).await;
}
let mut car_bytes = match encode_car_header(&head_cid) {
Ok(h) => h,
Err(e) => {
@@ -189,6 +231,109 @@ pub async fn get_repo(
.into_response()
}
async fn get_repo_since(state: &AppState, did: &str, head_cid: &Cid, since: &str) -> Response {
let events = sqlx::query!(
r#"
SELECT blocks_cids, commit_cid
FROM repo_seq
WHERE did = $1 AND rev > $2
ORDER BY seq DESC
"#,
did,
since
)
.fetch_all(&state.db)
.await;
let events = match events {
Ok(e) => e,
Err(e) => {
error!("DB error in get_repo_since: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError", "message": "Database error"})),
)
.into_response();
}
};
let mut block_cids: Vec<Cid> = Vec::new();
for event in &events {
if let Some(cids) = &event.blocks_cids {
for cid_str in cids {
if let Ok(cid) = Cid::from_str(cid_str)
&& !block_cids.contains(&cid)
{
block_cids.push(cid);
}
}
}
if let Some(commit_cid_str) = &event.commit_cid
&& let Ok(cid) = Cid::from_str(commit_cid_str)
&& !block_cids.contains(&cid)
{
block_cids.push(cid);
}
}
let mut car_bytes = match encode_car_header(head_cid) {
Ok(h) => h,
Err(e) => {
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError", "message": format!("Failed to encode CAR header: {}", e)})),
)
.into_response();
}
};
if block_cids.is_empty() {
return (
StatusCode::OK,
[(axum::http::header::CONTENT_TYPE, "application/vnd.ipld.car")],
car_bytes,
)
.into_response();
}
let blocks = match state.block_store.get_many(&block_cids).await {
Ok(b) => b,
Err(e) => {
error!("Block store error in get_repo_since: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError", "message": "Failed to get blocks"})),
)
.into_response();
}
};
for (i, block_opt) in blocks.into_iter().enumerate() {
if let Some(block) = block_opt {
let cid = block_cids[i];
let cid_bytes = cid.to_bytes();
let total_len = cid_bytes.len() + block.len();
let mut writer = Vec::new();
crate::sync::car::write_varint(&mut writer, total_len as u64)
.expect("Writing to Vec<u8> should never fail");
writer
.write_all(&cid_bytes)
.expect("Writing to Vec<u8> should never fail");
writer
.write_all(&block)
.expect("Writing to Vec<u8> should never fail");
car_bytes.extend_from_slice(&writer);
}
}
(
StatusCode::OK,
[(axum::http::header::CONTENT_TYPE, "application/vnd.ipld.car")],
car_bytes,
)
.into_response()
}
fn extract_links_ipld(value: &Ipld, stack: &mut Vec<Cid>) {
match value {
Ipld::Link(cid) => {
@@ -224,24 +369,17 @@ pub async fn get_record(
use std::collections::BTreeMap;
use std::sync::Arc;
let repo_row = sqlx::query!(
r#"
SELECT r.repo_root_cid
FROM repos r
JOIN users u ON u.id = r.user_id
WHERE u.did = $1
"#,
query.did
)
.fetch_optional(&state.db)
.await
.unwrap_or(None);
let commit_cid_str = match repo_row {
Some(r) => r.repo_root_cid,
let account = match assert_repo_availability(&state.db, &query.did, false).await {
Ok(a) => a,
Err(e) => return e.into_response(),
};
let commit_cid_str = match account.repo_root_cid {
Some(cid) => cid,
None => {
return (
StatusCode::NOT_FOUND,
Json(json!({"error": "RepoNotFound", "message": "Repo not found"})),
StatusCode::BAD_REQUEST,
Json(json!({"error": "RepoNotFound", "message": "Repo not initialized"})),
)
.into_response();
}
src/sync/subscribe_repos.rs
+120 -2
View File
@@ -1,7 +1,8 @@
use crate::state::AppState;
use crate::sync::firehose::SequencedEvent;
use crate::sync::util::{
format_event_for_sending, format_event_with_prefetched_blocks, prefetch_blocks_for_events,
format_error_frame, format_event_for_sending, format_event_with_prefetched_blocks,
format_info_frame, prefetch_blocks_for_events,
};
use axum::{
extract::{Query, State, ws::Message, ws::WebSocket, ws::WebSocketUpgrade},
@@ -55,13 +56,85 @@ async fn handle_socket(mut socket: WebSocket, state: AppState, params: Subscribe
info!(subscribers = count, "Firehose subscriber disconnected");
}
fn get_backfill_hours() -> i64 {
std::env::var("FIREHOSE_BACKFILL_HOURS")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(72)
}
async fn handle_socket_inner(
socket: &mut WebSocket,
state: &AppState,
params: SubscribeReposParams,
) -> Result<(), ()> {
let mut rx = state.firehose_tx.subscribe();
let mut last_seen: i64 = -1;
if let Some(cursor) = params.cursor {
let current_seq = sqlx::query_scalar!("SELECT MAX(seq) FROM repo_seq")
.fetch_one(&state.db)
.await
.ok()
.flatten()
.unwrap_or(0);
if cursor > current_seq {
if let Ok(error_bytes) =
format_error_frame("FutureCursor", Some("Cursor in the future."))
{
let _ = socket.send(Message::Binary(error_bytes.into())).await;
}
socket.close().await.ok();
return Err(());
}
let backfill_time = chrono::Utc::now() - chrono::Duration::hours(get_backfill_hours());
let first_event = sqlx::query_as!(
SequencedEvent,
r#"
SELECT seq, did, created_at, event_type, commit_cid, prev_cid, prev_data_cid, ops, blobs, blocks_cids, handle, active, status, rev
FROM repo_seq
WHERE seq > $1
ORDER BY seq ASC
LIMIT 1
"#,
cursor
)
.fetch_optional(&state.db)
.await
.ok()
.flatten();
let mut current_cursor = cursor;
if let Some(ref event) = first_event
&& event.created_at < backfill_time
{
if let Ok(info_bytes) = format_info_frame(
"OutdatedCursor",
Some("Requested cursor exceeded limit. Possibly missing events"),
) {
let _ = socket.send(Message::Binary(info_bytes.into())).await;
}
let earliest = sqlx::query_scalar!(
"SELECT MIN(seq) FROM repo_seq WHERE created_at >= $1",
backfill_time
)
.fetch_one(&state.db)
.await
.ok()
.flatten();
if let Some(earliest_seq) = earliest {
current_cursor = earliest_seq - 1;
}
}
last_seen = current_cursor;
loop {
let events = sqlx::query_as!(
SequencedEvent,
@@ -93,6 +166,7 @@ async fn handle_socket_inner(
};
for event in events {
current_cursor = event.seq;
last_seen = event.seq;
let bytes =
match format_event_with_prefetched_blocks(event, &prefetched).await {
Ok(b) => b,
@@ -118,8 +192,48 @@ async fn handle_socket_inner(
}
}
}
let cutover_events = sqlx::query_as!(
SequencedEvent,
r#"
SELECT seq, did, created_at, event_type, commit_cid, prev_cid, prev_data_cid, ops, blobs, blocks_cids, handle, active, status, rev
FROM repo_seq
WHERE seq > $1
ORDER BY seq ASC
"#,
last_seen
)
.fetch_all(&state.db)
.await;
if let Ok(events) = cutover_events
&& !events.is_empty()
{
let prefetched = match prefetch_blocks_for_events(state, &events).await {
Ok(blocks) => blocks,
Err(e) => {
error!("Failed to prefetch blocks for cutover: {}", e);
socket.close().await.ok();
return Err(());
}
};
for event in events {
last_seen = event.seq;
let bytes = match format_event_with_prefetched_blocks(event, &prefetched).await {
Ok(b) => b,
Err(e) => {
warn!("Failed to format cutover event: {}", e);
return Err(());
}
};
if let Err(e) = socket.send(Message::Binary(bytes.into())).await {
warn!("Failed to send cutover event: {}", e);
return Err(());
}
crate::metrics::record_firehose_event();
}
}
}
let mut rx = state.firehose_tx.subscribe();
let max_lag_before_disconnect: u64 = std::env::var("FIREHOSE_MAX_LAG")
.ok()
.and_then(|v| v.parse().ok())
@@ -129,6 +243,10 @@ async fn handle_socket_inner(
result = rx.recv() => {
match result {
Ok(event) => {
if event.seq <= last_seen {
continue;
}
last_seen = event.seq;
if let Err(e) = send_event(socket, state, event).await {
warn!("Failed to send event: {}", e);
break;
src/sync/util.rs
+179 -1
View File
@@ -1,16 +1,167 @@
use crate::state::AppState;
use crate::sync::firehose::SequencedEvent;
use crate::sync::frame::{AccountFrame, CommitFrame, FrameHeader, IdentityFrame, SyncFrame};
use crate::sync::frame::{
AccountFrame, CommitFrame, ErrorFrameBody, ErrorFrameHeader, FrameHeader, IdentityFrame,
InfoFrame, SyncFrame,
};
use axum::Json;
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use bytes::Bytes;
use cid::Cid;
use iroh_car::{CarHeader, CarWriter};
use jacquard_repo::commit::Commit;
use jacquard_repo::storage::BlockStore;
use serde::Serialize;
use serde_json::json;
use sqlx::PgPool;
use std::collections::{BTreeMap, HashMap};
use std::io::Cursor;
use std::str::FromStr;
use tokio::io::AsyncWriteExt;
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum AccountStatus {
Active,
Takendown,
Suspended,
Deactivated,
Deleted,
}
impl AccountStatus {
pub fn as_str(&self) -> Option<&'static str> {
match self {
AccountStatus::Active => None,
AccountStatus::Takendown => Some("takendown"),
AccountStatus::Suspended => Some("suspended"),
AccountStatus::Deactivated => Some("deactivated"),
AccountStatus::Deleted => Some("deleted"),
}
}
pub fn is_active(&self) -> bool {
matches!(self, AccountStatus::Active)
}
}
pub struct RepoAccount {
pub did: String,
pub user_id: uuid::Uuid,
pub status: AccountStatus,
pub repo_root_cid: Option<String>,
}
pub enum RepoAvailabilityError {
NotFound(String),
Takendown(String),
Deactivated(String),
Internal(String),
}
impl IntoResponse for RepoAvailabilityError {
fn into_response(self) -> Response {
match self {
RepoAvailabilityError::NotFound(did) => (
StatusCode::BAD_REQUEST,
Json(json!({
"error": "RepoNotFound",
"message": format!("Could not find repo for DID: {}", did)
})),
)
.into_response(),
RepoAvailabilityError::Takendown(did) => (
StatusCode::BAD_REQUEST,
Json(json!({
"error": "RepoTakendown",
"message": format!("Repo has been takendown: {}", did)
})),
)
.into_response(),
RepoAvailabilityError::Deactivated(did) => (
StatusCode::BAD_REQUEST,
Json(json!({
"error": "RepoDeactivated",
"message": format!("Repo has been deactivated: {}", did)
})),
)
.into_response(),
RepoAvailabilityError::Internal(msg) => (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({
"error": "InternalError",
"message": msg
})),
)
.into_response(),
}
}
}
pub async fn get_account_with_status(
db: &PgPool,
did: &str,
) -> Result<Option<RepoAccount>, sqlx::Error> {
let row = sqlx::query!(
r#"
SELECT u.id, u.did, u.deactivated_at, u.takedown_ref, r.repo_root_cid
FROM users u
LEFT JOIN repos r ON r.user_id = u.id
WHERE u.did = $1
"#,
did
)
.fetch_optional(db)
.await?;
Ok(row.map(|r| {
let status = if r.takedown_ref.is_some() {
AccountStatus::Takendown
} else if r.deactivated_at.is_some() {
AccountStatus::Deactivated
} else {
AccountStatus::Active
};
RepoAccount {
did: r.did,
user_id: r.id,
status,
repo_root_cid: Some(r.repo_root_cid),
}
}))
}
pub async fn assert_repo_availability(
db: &PgPool,
did: &str,
is_admin_or_self: bool,
) -> Result<RepoAccount, RepoAvailabilityError> {
let account = get_account_with_status(db, did)
.await
.map_err(|e| RepoAvailabilityError::Internal(e.to_string()))?;
let account = match account {
Some(a) => a,
None => return Err(RepoAvailabilityError::NotFound(did.to_string())),
};
if is_admin_or_self {
return Ok(account);
}
match account.status {
AccountStatus::Takendown => return Err(RepoAvailabilityError::Takendown(did.to_string())),
AccountStatus::Deactivated => {
return Err(RepoAvailabilityError::Deactivated(did.to_string()));
}
_ => {}
}
Ok(account)
}
fn extract_rev_from_commit_bytes(commit_bytes: &[u8]) -> Option<String> {
Commit::from_cbor(commit_bytes)
.ok()
@@ -351,3 +502,30 @@ pub async fn format_event_with_prefetched_blocks(
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
Ok(bytes)
}
pub fn format_info_frame(name: &str, message: Option<&str>) -> Result<Vec<u8>, anyhow::Error> {
let header = FrameHeader {
op: 1,
t: "#info".to_string(),
};
let frame = InfoFrame {
name: name.to_string(),
message: message.map(String::from),
};
let mut bytes = Vec::new();
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
Ok(bytes)
}
pub fn format_error_frame(error: &str, message: Option<&str>) -> Result<Vec<u8>, anyhow::Error> {
let header = ErrorFrameHeader { op: -1 };
let frame = ErrorFrameBody {
error: error.to_string(),
message: message.map(String::from),
};
let mut bytes = Vec::new();
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
Ok(bytes)
}
src/util.rs
+69
View File
@@ -73,6 +73,33 @@ pub async fn get_user_by_identifier(
.ok_or(DbLookupError::NotFound)
}
pub fn parse_repeated_query_param(query: Option<&str>, key: &str) -> Vec<String> {
query
.map(|q| {
let mut values = Vec::new();
for pair in q.split('&') {
if let Some((k, v)) = pair.split_once('=')
&& k == key
&& let Ok(decoded) = urlencoding::decode(v)
{
let decoded = decoded.into_owned();
if decoded.contains(',') {
for part in decoded.split(',') {
let trimmed = part.trim();
if !trimmed.is_empty() {
values.push(trimmed.to_string());
}
}
} else if !decoded.is_empty() {
values.push(decoded);
}
}
}
values
})
.unwrap_or_default()
}
pub fn extract_client_ip(headers: &HeaderMap) -> String {
if let Some(forwarded) = headers.get("x-forwarded-for")
&& let Ok(value) = forwarded.to_str()
@@ -92,6 +119,48 @@ pub fn extract_client_ip(headers: &HeaderMap) -> String {
mod tests {
use super::*;
#[test]
fn test_parse_repeated_query_param_repeated() {
let query = "did=test&cids=a&cids=b&cids=c";
let result = parse_repeated_query_param(Some(query), "cids");
assert_eq!(result, vec!["a", "b", "c"]);
}
#[test]
fn test_parse_repeated_query_param_comma_separated() {
let query = "did=test&cids=a,b,c";
let result = parse_repeated_query_param(Some(query), "cids");
assert_eq!(result, vec!["a", "b", "c"]);
}
#[test]
fn test_parse_repeated_query_param_mixed() {
let query = "did=test&cids=a,b&cids=c";
let result = parse_repeated_query_param(Some(query), "cids");
assert_eq!(result, vec!["a", "b", "c"]);
}
#[test]
fn test_parse_repeated_query_param_single() {
let query = "did=test&cids=a";
let result = parse_repeated_query_param(Some(query), "cids");
assert_eq!(result, vec!["a"]);
}
#[test]
fn test_parse_repeated_query_param_empty() {
let query = "did=test";
let result = parse_repeated_query_param(Some(query), "cids");
assert!(result.is_empty());
}
#[test]
fn test_parse_repeated_query_param_url_encoded() {
let query = "did=test&cids=bafyreib%2Btest";
let result = parse_repeated_query_param(Some(query), "cids");
assert_eq!(result, vec!["bafyreib+test"]);
}
#[test]
fn test_generate_token_code() {
let code = generate_token_code();