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

Come back to some functions I made as stubs earlier

Browse Source
This commit is contained in:
Lewis
2025-12-08 17:58:53 +02:00
parent a2b2c5b4c9
commit 93422875ab
10 changed files with 428 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+32
View File
@@ -914,6 +914,7 @@ dependencies = [
"bytes",
"chrono",
"cid",
"ctor",
"dotenvy",
"jacquard",
"jacquard-axum",
@@ -1372,6 +1373,22 @@ dependencies = [
"typenum",
]
[[package]]
name = "ctor"
version = "0.6.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "424e0138278faeb2b401f174ad17e715c829512d74f3d1e81eb43365c2e0590e"
dependencies = [
"ctor-proc-macro",
"dtor",
]
[[package]]
name = "ctor-proc-macro"
version = "0.0.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "52560adf09603e58c9a7ee1fe1dcb95a16927b17c127f0ac02d6e768a0e25bc1"
[[package]]
name = "curve25519-dalek"
version = "4.1.3"
@@ -1613,6 +1630,21 @@ version = "0.15.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b"
[[package]]
name = "dtor"
version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "404d02eeb088a82cfd873006cb713fe411306c7d182c344905e101fb1167d301"
dependencies = [
"dtor-proc-macro",
]
[[package]]
name = "dtor-proc-macro"
version = "0.0.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f678cf4a922c215c63e0de95eb1ff08a958a81d47e485cf9da1e27bf6305cfa5"
[[package]]
name = "dunce"
version = "1.0.5"
Cargo.toml
+1
View File
@@ -36,6 +36,7 @@ tracing-subscriber = "0.3.22"
uuid = { version = "1.19.0", features = ["v4", "fast-rng"] }
[dev-dependencies]
ctor = "0.6.3"
testcontainers = "0.26.0"
testcontainers-modules = { version = "0.14.0", features = ["postgres"] }
wiremock = "0.6.5"
migrations/202512211600_moderation_and_status.sql
+11
View File
@@ -0,0 +1,11 @@
ALTER TABLE users ADD COLUMN deactivated_at TIMESTAMPTZ;
-- * reports u *
CREATE TABLE reports (
id BIGINT PRIMARY KEY,
reason_type TEXT NOT NULL,
reason TEXT,
subject_json JSONB NOT NULL,
reported_by_did TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
src/api/moderation/mod.rs
+24 -3
View File
@@ -110,8 +110,29 @@ pub async fn create_report(
.into_response();
}
let created_at = chrono::Utc::now().to_rfc3339();
let report_id = chrono::Utc::now().timestamp_millis();
let created_at = chrono::Utc::now();
let report_id = created_at.timestamp_millis();
let insert = sqlx::query(
"INSERT INTO reports (id, reason_type, reason, subject_json, reported_by_did, created_at) VALUES ($1, $2, $3, $4, $5, $6)"
)
.bind(report_id)
.bind(&input.reason_type)
.bind(&input.reason)
.bind(json!(input.subject))
.bind(&did)
.bind(created_at)
.execute(&state.db)
.await;
if let Err(e) = insert {
error!("Failed to insert report: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
(
StatusCode::OK,
@@ -121,7 +142,7 @@ pub async fn create_report(
reason: input.reason,
subject: input.subject,
reported_by: did,
created_at,
created_at: created_at.to_rfc3339(),
}),
)
.into_response()
src/api/repo/blob.rs
+171 -4
View File
@@ -7,11 +7,13 @@ use axum::{
response::{IntoResponse, Response},
};
use cid::Cid;
use jacquard_repo::storage::BlockStore;
use multihash::Multihash;
use serde::{Deserialize, Serialize};
use serde_json::json;
use sha2::{Digest, Sha256};
use sqlx::Row;
use std::str::FromStr;
use tracing::error;
pub async fn upload_blob(
@@ -157,10 +159,33 @@ pub struct ListMissingBlobsOutput {
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") {
if type_val == "blob" {
if let Some(r) = obj.get("ref") {
if let Some(link) = r.get("$link") {
if let Some(s) = link.as_str() {
blobs.push(s.to_string());
}
}
}
}
}
for (_, v) in obj {
find_blobs(v, blobs);
}
} else if let Some(arr) = val.as_array() {
for v in arr {
find_blobs(v, blobs);
}
}
}
pub async fn list_missing_blobs(
State(_state): State<AppState>,
State(state): State<AppState>,
headers: axum::http::HeaderMap,
Query(_params): Query<ListMissingBlobsParams>,
Query(params): Query<ListMissingBlobsParams>,
) -> Response {
let auth_header = headers.get("Authorization");
if auth_header.is_none() {
@@ -171,11 +196,153 @@ pub async fn list_missing_blobs(
.into_response();
}
let token = auth_header
.unwrap()
.to_str()
.unwrap_or("")
.replace("Bearer ", "");
let session = sqlx::query(
"SELECT s.did, k.key_bytes FROM sessions s JOIN users u ON s.did = u.did JOIN user_keys k ON u.id = k.user_id WHERE s.access_jwt = $1"
)
.bind(&token)
.fetch_optional(&state.db)
.await
.unwrap_or(None);
let (did, key_bytes) = match session {
Some(row) => (
row.get::<String, _>("did"),
row.get::<Vec<u8>, _>("key_bytes"),
),
None => {
return (
StatusCode::UNAUTHORIZED,
Json(json!({"error": "AuthenticationFailed"})),
)
.into_response();
}
};
if let Err(_) = crate::auth::verify_token(&token, &key_bytes) {
return (
StatusCode::UNAUTHORIZED,
Json(json!({"error": "AuthenticationFailed", "message": "Invalid token signature"})),
)
.into_response();
}
let user_query = sqlx::query("SELECT id FROM users WHERE did = $1")
.bind(&did)
.fetch_optional(&state.db)
.await;
let user_id: uuid::Uuid = match user_query {
Ok(Some(row)) => row.get("id"),
_ => {
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
};
let limit = params.limit.unwrap_or(500).min(1000);
let cursor_str = params.cursor.unwrap_or_default();
let (cursor_collection, cursor_rkey) = if cursor_str.contains('|') {
let parts: Vec<&str> = cursor_str.split('|').collect();
(parts[0].to_string(), parts[1].to_string())
} 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"
)
.bind(user_id)
.bind(cursor_collection)
.bind(cursor_rkey)
.bind(limit)
.fetch_all(&state.db)
.await;
let records = match records_query {
Ok(r) => r,
Err(e) => {
error!("DB error fetching records: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
};
let mut missing_blobs = Vec::new();
let mut last_cursor = None;
for row in &records {
let collection: String = row.get("collection");
let rkey: String = row.get("rkey");
let record_cid_str: String = row.get("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 FROM blobs WHERE cid = $1 AND created_by_user = $2")
.bind(&blob_cid_str)
.bind(user_id)
.fetch_optional(&state.db)
.await;
match exists {
Ok(None) => {
missing_blobs.push(RecordBlob {
cid: blob_cid_str,
record_uri: format!("at://{}/{}/{}", did, collection, rkey),
});
}
Err(e) => {
error!("DB error checking blob existence: {:?}", e);
}
_ => {}
}
}
}
// if we fetched fewer records than limit, we are done, so cursor is None.
// otherwise, cursor is the last one we saw.
// ...right?
let next_cursor = if records.len() < limit as usize {
None
} else {
last_cursor
};
(
StatusCode::OK,
Json(ListMissingBlobsOutput {
cursor: None,
blobs: vec![],
cursor: next_cursor,
blobs: missing_blobs,
}),
)
.into_response()
src/api/repo/record/write.rs
+24
View File
@@ -182,6 +182,18 @@ pub async fn create_record(
.rkey
.unwrap_or_else(|| Utc::now().format("%Y%m%d%H%M%S%f").to_string());
if input.validate.unwrap_or(true) {
if input.collection == "app.bsky.feed.post" {
if input.record.get("text").is_none() || input.record.get("createdAt").is_none() {
return (
StatusCode::BAD_REQUEST,
Json(json!({"error": "InvalidRecord", "message": "Record validation failed"})),
)
.into_response();
}
}
}
let mut record_bytes = Vec::new();
if let Err(e) = serde_ipld_dagcbor::to_writer(&mut record_bytes, &input.record) {
error!("Error serializing record: {:?}", e);
@@ -472,6 +484,18 @@ pub async fn put_record(
let rkey = input.rkey.clone();
if input.validate.unwrap_or(true) {
if input.collection == "app.bsky.feed.post" {
if input.record.get("text").is_none() || input.record.get("createdAt").is_none() {
return (
StatusCode::BAD_REQUEST,
Json(json!({"error": "InvalidRecord", "message": "Record validation failed"})),
)
.into_response();
}
}
}
let mut record_bytes = Vec::new();
if let Err(e) = serde_ipld_dagcbor::to_writer(&mut record_bytes, &input.record) {
error!("Error serializing record: {:?}", e);
src/api/server/session.rs
+143 -5
View File
@@ -561,6 +561,16 @@ pub async fn check_account_status(
.into_response();
}
let user_status = sqlx::query("SELECT deactivated_at FROM users WHERE did = $1")
.bind(&did)
.fetch_optional(&state.db)
.await;
let deactivated_at: Option<chrono::DateTime<chrono::Utc>> = match user_status {
Ok(Some(row)) => row.get("deactivated_at"),
_ => None,
};
let repo_result = sqlx::query("SELECT repo_root_cid FROM repos WHERE user_id = $1")
.bind(user_id)
.fetch_optional(&state.db)
@@ -589,7 +599,7 @@ pub async fn check_account_status(
(
StatusCode::OK,
Json(CheckAccountStatusOutput {
activated: true,
activated: deactivated_at.is_none(),
valid_did,
repo_commit: repo_commit.clone(),
repo_rev: chrono::Utc::now().timestamp_millis().to_string(),
@@ -604,7 +614,7 @@ pub async fn check_account_status(
}
pub async fn activate_account(
State(_state): State<AppState>,
State(state): State<AppState>,
headers: axum::http::HeaderMap,
) -> Response {
let auth_header = headers.get("Authorization");
@@ -616,7 +626,71 @@ pub async fn activate_account(
.into_response();
}
(StatusCode::OK, Json(json!({}))).into_response()
let token = auth_header
.unwrap()
.to_str()
.unwrap_or("")
.replace("Bearer ", "");
let session = sqlx::query(
r#"
SELECT s.did, k.key_bytes
FROM sessions s
JOIN users u ON s.did = u.did
JOIN user_keys k ON u.id = k.user_id
WHERE s.access_jwt = $1
"#,
)
.bind(&token)
.fetch_optional(&state.db)
.await;
let (did, key_bytes) = match session {
Ok(Some(row)) => (
row.get::<String, _>("did"),
row.get::<Vec<u8>, _>("key_bytes"),
),
Ok(None) => {
return (
StatusCode::UNAUTHORIZED,
Json(json!({"error": "AuthenticationFailed"})),
)
.into_response();
}
Err(e) => {
error!("DB error in activate_account: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
};
if let Err(_) = crate::auth::verify_token(&token, &key_bytes) {
return (
StatusCode::UNAUTHORIZED,
Json(json!({"error": "AuthenticationFailed", "message": "Invalid token signature"})),
)
.into_response();
}
let result = sqlx::query("UPDATE users SET deactivated_at = NULL WHERE did = $1")
.bind(&did)
.execute(&state.db)
.await;
match result {
Ok(_) => (StatusCode::OK, Json(json!({}))).into_response(),
Err(e) => {
error!("DB error activating account: {:?}", e);
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response()
}
}
}
#[derive(Deserialize)]
@@ -626,7 +700,7 @@ pub struct DeactivateAccountInput {
}
pub async fn deactivate_account(
State(_state): State<AppState>,
State(state): State<AppState>,
headers: axum::http::HeaderMap,
Json(_input): Json<DeactivateAccountInput>,
) -> Response {
@@ -639,7 +713,71 @@ pub async fn deactivate_account(
.into_response();
}
(StatusCode::OK, Json(json!({}))).into_response()
let token = auth_header
.unwrap()
.to_str()
.unwrap_or("")
.replace("Bearer ", "");
let session = sqlx::query(
r#"
SELECT s.did, k.key_bytes
FROM sessions s
JOIN users u ON s.did = u.did
JOIN user_keys k ON u.id = k.user_id
WHERE s.access_jwt = $1
"#,
)
.bind(&token)
.fetch_optional(&state.db)
.await;
let (did, key_bytes) = match session {
Ok(Some(row)) => (
row.get::<String, _>("did"),
row.get::<Vec<u8>, _>("key_bytes"),
),
Ok(None) => {
return (
StatusCode::UNAUTHORIZED,
Json(json!({"error": "AuthenticationFailed"})),
)
.into_response();
}
Err(e) => {
error!("DB error in deactivate_account: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
};
if let Err(_) = crate::auth::verify_token(&token, &key_bytes) {
return (
StatusCode::UNAUTHORIZED,
Json(json!({"error": "AuthenticationFailed", "message": "Invalid token signature"})),
)
.into_response();
}
let result = sqlx::query("UPDATE users SET deactivated_at = NOW() WHERE did = $1")
.bind(&did)
.execute(&state.db)
.await;
match result {
Ok(_) => (StatusCode::OK, Json(json!({}))).into_response(),
Err(e) => {
error!("DB error deactivating account: {:?}", e);
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response()
}
}
}
#[derive(Serialize)]
src/sync/mod.rs
+4
View File
@@ -461,6 +461,8 @@ pub async fn notify_of_update(
Query(params): Query<NotifyOfUpdateParams>,
) -> Response {
info!("Received notifyOfUpdate from hostname: {}", params.hostname);
// TODO: Queue job for crawler interaction or relay notification
info!("TODO: Queue job for notifyOfUpdate (not implemented)");
(StatusCode::OK, Json(json!({}))).into_response()
}
@@ -475,6 +477,8 @@ pub async fn request_crawl(
Json(input): Json<RequestCrawlInput>,
) -> Response {
info!("Received requestCrawl for hostname: {}", input.hostname);
// TODO: Queue job for crawling
info!("TODO: Queue job for requestCrawl (not implemented)");
(StatusCode::OK, Json(json!({}))).into_response()
}
tests/common/mod.rs
+18
View File
@@ -32,6 +32,22 @@ pub const AUTH_DID: &str = "did:plc:fake";
#[allow(dead_code)]
pub const TARGET_DID: &str = "did:plc:target";
#[cfg(test)]
#[ctor::dtor]
fn cleanup() {
// my attempt to force clean up containers created by this test binary.
// this is a fallback in case ryuk fails or is not supported
if std::env::var("XDG_RUNTIME_DIR").is_ok() {
let _ = std::process::Command::new("podman")
.args(&["rm", "-f", "--filter", "label=bspds_test=true"])
.output();
}
let _ = std::process::Command::new("docker")
.args(&["container", "prune", "-f", "--filter", "label=bspds_test=true"])
.output();
}
#[allow(dead_code)]
pub fn client() -> Client {
Client::new()
@@ -63,6 +79,7 @@ pub async fn base_url() -> &'static str {
.with_env_var("MINIO_ROOT_USER", "minioadmin")
.with_env_var("MINIO_ROOT_PASSWORD", "minioadmin")
.with_cmd(vec!["server".to_string(), "/data".to_string()])
.with_label("bspds_test", "true")
.start()
.await
.expect("Failed to start MinIO");
@@ -131,6 +148,7 @@ pub async fn base_url() -> &'static str {
let container = Postgres::default()
.with_tag("18-alpine")
.with_label("bspds_test", "true")
.start()
.await
.expect("Failed to start Postgres");
tests/repo.rs
-1
View File
@@ -204,7 +204,6 @@ async fn test_put_record_mismatched_repo() {
}
#[tokio::test]
#[ignore]
async fn test_put_record_invalid_schema() {
let client = client();
let (token, did) = create_account_and_login(&client).await;