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

SQLx query macro the remaining ones

Browse Source
This commit is contained in:
Lewis
2025-12-08 20:14:23 +02:00
parent b19013c6ab
commit 48477d8ea3
26 changed files with 595 additions and 517 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/feed/timeline.rs
+11 -23
View File
@@ -11,7 +11,6 @@ use axum::{
use jacquard_repo::storage::BlockStore;
use serde::Serialize;
use serde_json::{Value, json};
use sqlx::Row;
use tracing::error;
#[derive(Serialize)]
@@ -153,29 +152,18 @@ pub async fn get_timeline(
.into_response();
}
let placeholders: Vec<String> = followed_dids
.iter()
.enumerate()
.map(|(i, _)| format!("${}", i + 1))
.collect();
let posts_query = format!(
let posts_result = sqlx::query!(
"SELECT r.record_cid, r.rkey, r.created_at, u.did, u.handle
FROM records r
JOIN repos rp ON r.repo_id = rp.user_id
JOIN users u ON rp.user_id = u.id
WHERE u.did IN ({}) AND r.collection = 'app.bsky.feed.post'
WHERE u.did = ANY($1) AND r.collection = 'app.bsky.feed.post'
ORDER BY r.created_at DESC
LIMIT 50",
placeholders.join(", ")
);
let mut query = sqlx::query(&posts_query);
for did in &followed_dids {
query = query.bind(did);
}
let posts_result = query.fetch_all(&state.db).await;
&followed_dids
)
.fetch_all(&state.db)
.await;
let posts = match posts_result {
Ok(rows) => rows,
@@ -192,11 +180,11 @@ pub async fn get_timeline(
let mut feed: Vec<FeedViewPost> = Vec::new();
for row in posts {
let record_cid: String = row.get("record_cid");
let rkey: String = row.get("rkey");
let created_at: chrono::DateTime<chrono::Utc> = row.get("created_at");
let author_did: String = row.get("did");
let author_handle: String = row.get("handle");
let record_cid: String = row.record_cid;
let rkey: String = row.rkey;
let created_at: chrono::DateTime<chrono::Utc> = row.created_at;
let author_did: String = row.did;
let author_handle: String = row.handle;
let cid = match record_cid.parse::<cid::Cid>() {
Ok(c) => c,
src/api/repo/meta.rs
+7 -13
View File
@@ -7,7 +7,6 @@ use axum::{
};
use serde::Deserialize;
use serde_json::json;
use sqlx::Row;
#[derive(Deserialize)]
pub struct DescribeRepoInput {
@@ -19,23 +18,19 @@ pub async fn describe_repo(
Query(input): Query<DescribeRepoInput>,
) -> Response {
let user_row = if input.repo.starts_with("did:") {
sqlx::query("SELECT id, handle, did FROM users WHERE did = $1")
.bind(&input.repo)
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 {
sqlx::query("SELECT id, handle, did FROM users WHERE handle = $1")
.bind(&input.repo)
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(row)) => (
row.get::<uuid::Uuid, _>("id"),
row.get::<String, _>("handle"),
row.get::<String, _>("did"),
),
Ok(Some((id, handle, did))) => (id, handle, did),
_ => {
return (
StatusCode::NOT_FOUND,
@@ -46,13 +41,12 @@ pub async fn describe_repo(
};
let collections_query =
sqlx::query("SELECT DISTINCT collection FROM records WHERE repo_id = $1")
.bind(user_id)
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.get("collection")).collect(),
Ok(rows) => rows.iter().map(|r| r.collection.clone()).collect(),
Err(_) => Vec::new(),
};
src/api/repo/record/batch.rs
+19 -24
View File
@@ -15,7 +15,6 @@ use jacquard::types::{
use jacquard_repo::{commit::Commit, mst::Mst, storage::BlockStore};
use serde::{Deserialize, Serialize};
use serde_json::json;
use sqlx::Row;
use std::str::FromStr;
use std::sync::Arc;
use tracing::error;
@@ -90,18 +89,18 @@ pub async fn apply_writes(
.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"
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",
token
)
.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"),
row.did,
row.key_bytes,
),
None => {
return (
@@ -144,13 +143,12 @@ pub async fn apply_writes(
.into_response();
}
let user_query = sqlx::query("SELECT id FROM users WHERE did = $1")
.bind(&did)
let user_query = sqlx::query!("SELECT id FROM users WHERE did = $1", did)
.fetch_optional(&state.db)
.await;
let user_id: uuid::Uuid = match user_query {
Ok(Some(row)) => row.get("id"),
Ok(Some(row)) => row.id,
_ => {
return (
StatusCode::INTERNAL_SERVER_ERROR,
@@ -160,14 +158,13 @@ pub async fn apply_writes(
}
};
let repo_root_query = sqlx::query("SELECT repo_root_cid FROM repos WHERE user_id = $1")
.bind(user_id)
let repo_root_query = sqlx::query!("SELECT repo_root_cid FROM repos WHERE user_id = $1", user_id)
.fetch_optional(&state.db)
.await;
let current_root_cid = match repo_root_query {
Ok(Some(row)) => {
let cid_str: String = row.get("repo_root_cid");
let cid_str: String = row.repo_root_cid;
match Cid::from_str(&cid_str) {
Ok(c) => c,
Err(_) => {
@@ -449,9 +446,7 @@ pub async fn apply_writes(
}
};
let update_repo = sqlx::query("UPDATE repos SET repo_root_cid = $1 WHERE user_id = $2")
.bind(new_root_cid.to_string())
.bind(user_id)
let update_repo = sqlx::query!("UPDATE repos SET repo_root_cid = $1 WHERE user_id = $2", new_root_cid.to_string(), user_id)
.execute(&state.db)
.await;
@@ -467,24 +462,24 @@ pub async fn apply_writes(
for (collection, rkey, record_cid) in record_ops {
match record_cid {
Some(cid) => {
let _ = sqlx::query(
let _ = sqlx::query!(
"INSERT INTO records (repo_id, collection, rkey, record_cid) VALUES ($1, $2, $3, $4)
ON CONFLICT (repo_id, collection, rkey) DO UPDATE SET record_cid = $4, created_at = NOW()",
user_id,
collection,
rkey,
cid
)
.bind(user_id)
.bind(&collection)
.bind(&rkey)
.bind(&cid)
.execute(&state.db)
.await;
}
None => {
let _ = sqlx::query(
let _ = sqlx::query!(
"DELETE FROM records WHERE repo_id = $1 AND collection = $2 AND rkey = $3",
user_id,
collection,
rkey
)
.bind(user_id)
.bind(&collection)
.bind(&rkey)
.execute(&state.db)
.await;
}
src/api/repo/record/delete.rs
+11 -19
View File
@@ -14,7 +14,6 @@ use jacquard::types::{
use jacquard_repo::{commit::Commit, mst::Mst, storage::BlockStore};
use serde::Deserialize;
use serde_json::json;
use sqlx::Row;
use std::str::FromStr;
use std::sync::Arc;
use tracing::error;
@@ -49,18 +48,18 @@ pub async fn delete_record(
.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"
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",
token
)
.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"),
row.did,
row.key_bytes,
),
None => {
return (
@@ -83,13 +82,12 @@ pub async fn delete_record(
return (StatusCode::FORBIDDEN, Json(json!({"error": "InvalidRepo", "message": "Repo does not match authenticated user"}))).into_response();
}
let user_query = sqlx::query("SELECT id FROM users WHERE did = $1")
.bind(&did)
let user_query = sqlx::query!("SELECT id FROM users WHERE did = $1", did)
.fetch_optional(&state.db)
.await;
let user_id: uuid::Uuid = match user_query {
Ok(Some(row)) => row.get("id"),
Ok(Some(row)) => row.id,
_ => {
return (
StatusCode::INTERNAL_SERVER_ERROR,
@@ -99,14 +97,13 @@ pub async fn delete_record(
}
};
let repo_root_query = sqlx::query("SELECT repo_root_cid FROM repos WHERE user_id = $1")
.bind(user_id)
let repo_root_query = sqlx::query!("SELECT repo_root_cid FROM repos WHERE user_id = $1", user_id)
.fetch_optional(&state.db)
.await;
let current_root_cid = match repo_root_query {
Ok(Some(row)) => {
let cid_str: String = row.get("repo_root_cid");
let cid_str: String = row.repo_root_cid;
Cid::from_str(&cid_str).ok()
}
_ => None,
@@ -209,9 +206,7 @@ pub async fn delete_record(
}
};
let update_repo = sqlx::query("UPDATE repos SET repo_root_cid = $1 WHERE user_id = $2")
.bind(new_root_cid.to_string())
.bind(user_id)
let update_repo = sqlx::query!("UPDATE repos SET repo_root_cid = $1 WHERE user_id = $2", new_root_cid.to_string(), user_id)
.execute(&state.db)
.await;
@@ -225,10 +220,7 @@ pub async fn delete_record(
}
let record_delete =
sqlx::query("DELETE FROM records WHERE repo_id = $1 AND collection = $2 AND rkey = $3")
.bind(user_id)
.bind(&input.collection)
.bind(&input.rkey)
sqlx::query!("DELETE FROM records WHERE repo_id = $1 AND collection = $2 AND rkey = $3", user_id, input.collection, input.rkey)
.execute(&state.db)
.await;
src/api/repo/record/read.rs
+68 -45
View File
@@ -9,7 +9,6 @@ use cid::Cid;
use jacquard_repo::storage::BlockStore;
use serde::{Deserialize, Serialize};
use serde_json::json;
use sqlx::Row;
use std::str::FromStr;
use tracing::error;
@@ -25,20 +24,20 @@ pub async fn get_record(
State(state): State<AppState>,
Query(input): Query<GetRecordInput>,
) -> Response {
let user_row = if input.repo.starts_with("did:") {
sqlx::query("SELECT id FROM users WHERE did = $1")
.bind(&input.repo)
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)
.await
.map(|opt| opt.map(|r| r.id))
} else {
sqlx::query("SELECT id FROM users WHERE handle = $1")
.bind(&input.repo)
sqlx::query!("SELECT id FROM users WHERE handle = $1", input.repo)
.fetch_optional(&state.db)
.await
.map(|opt| opt.map(|r| r.id))
};
let user_id: uuid::Uuid = match user_row {
Ok(Some(row)) => row.get("id"),
let user_id: uuid::Uuid = match user_id_opt {
Ok(Some(id)) => id,
_ => {
return (
StatusCode::NOT_FOUND,
@@ -48,17 +47,17 @@ pub async fn get_record(
}
};
let record_row = sqlx::query(
let record_row = sqlx::query!(
"SELECT record_cid FROM records WHERE repo_id = $1 AND collection = $2 AND rkey = $3",
user_id,
input.collection,
input.rkey
)
.bind(user_id)
.bind(&input.collection)
.bind(&input.rkey)
.fetch_optional(&state.db)
.await;
let record_cid_str: String = match record_row {
Ok(Some(row)) => row.get("record_cid"),
Ok(Some(row)) => row.record_cid,
_ => {
return (
StatusCode::NOT_FOUND,
@@ -143,20 +142,20 @@ pub async fn list_records(
State(state): State<AppState>,
Query(input): Query<ListRecordsInput>,
) -> Response {
let user_row = if input.repo.starts_with("did:") {
sqlx::query("SELECT id FROM users WHERE did = $1")
.bind(&input.repo)
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)
.await
.map(|opt| opt.map(|r| r.id))
} else {
sqlx::query("SELECT id FROM users WHERE handle = $1")
.bind(&input.repo)
sqlx::query!("SELECT id FROM users WHERE handle = $1", input.repo)
.fetch_optional(&state.db)
.await
.map(|opt| opt.map(|r| r.id))
};
let user_id: uuid::Uuid = match user_row {
Ok(Some(row)) => row.get("id"),
let user_id: uuid::Uuid = match user_id_opt {
Ok(Some(id)) => id,
_ => {
return (
StatusCode::NOT_FOUND,
@@ -172,30 +171,56 @@ pub async fn list_records(
// Simplistic query construction - no sophisticated cursor handling or rkey ranges for now, just basic pagination
// TODO: Implement rkeyStart/End and correct cursor logic
let query_str = format!(
"SELECT rkey, record_cid FROM records WHERE repo_id = $1 AND collection = $2 {} ORDER BY rkey {} LIMIT {}",
if let Some(_c) = &input.cursor {
if reverse {
"AND rkey < $3"
} else {
"AND rkey > $3"
}
let limit_i64 = limit as i64;
let rows_res = if let Some(cursor) = &input.cursor {
if reverse {
sqlx::query!(
"SELECT rkey, record_cid FROM records WHERE repo_id = $1 AND collection = $2 AND rkey < $3 ORDER BY rkey DESC LIMIT $4",
user_id,
input.collection,
cursor,
limit_i64
)
.fetch_all(&state.db)
.await
.map(|rows| rows.into_iter().map(|r| (r.rkey, r.record_cid)).collect::<Vec<_>>())
} else {
""
},
if reverse { "DESC" } else { "ASC" },
limit
);
sqlx::query!(
"SELECT rkey, record_cid FROM records WHERE repo_id = $1 AND collection = $2 AND rkey > $3 ORDER BY rkey ASC LIMIT $4",
user_id,
input.collection,
cursor,
limit_i64
)
.fetch_all(&state.db)
.await
.map(|rows| rows.into_iter().map(|r| (r.rkey, r.record_cid)).collect::<Vec<_>>())
}
} else {
if reverse {
sqlx::query!(
"SELECT rkey, record_cid FROM records WHERE repo_id = $1 AND collection = $2 ORDER BY rkey DESC LIMIT $3",
user_id,
input.collection,
limit_i64
)
.fetch_all(&state.db)
.await
.map(|rows| rows.into_iter().map(|r| (r.rkey, r.record_cid)).collect::<Vec<_>>())
} else {
sqlx::query!(
"SELECT rkey, record_cid FROM records WHERE repo_id = $1 AND collection = $2 ORDER BY rkey ASC LIMIT $3",
user_id,
input.collection,
limit_i64
)
.fetch_all(&state.db)
.await
.map(|rows| rows.into_iter().map(|r| (r.rkey, r.record_cid)).collect::<Vec<_>>())
}
};
let mut query = sqlx::query(&query_str)
.bind(user_id)
.bind(&input.collection);
if let Some(c) = &input.cursor {
query = query.bind(c);
}
let rows = match query.fetch_all(&state.db).await {
let rows = match rows_res {
Ok(r) => r,
Err(e) => {
error!("Error listing records: {:?}", e);
@@ -210,9 +235,7 @@ pub async fn list_records(
let mut records = Vec::new();
let mut last_rkey = None;
for row in rows {
let rkey: String = row.get("rkey");
let cid_str: String = row.get("record_cid");
for (rkey, cid_str) in rows {
last_rkey = Some(rkey.clone());
if let Ok(cid) = Cid::from_str(&cid_str) {
src/api/repo/record/write.rs
+30 -39
View File
@@ -15,7 +15,6 @@ use jacquard::types::{
use jacquard_repo::{commit::Commit, mst::Mst, storage::BlockStore};
use serde::{Deserialize, Serialize};
use serde_json::json;
use sqlx::Row;
use std::str::FromStr;
use std::sync::Arc;
use tracing::error;
@@ -58,18 +57,18 @@ pub async fn create_record(
.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"
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",
token
)
.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"),
row.did,
row.key_bytes,
),
None => {
return (
@@ -92,13 +91,12 @@ pub async fn create_record(
return (StatusCode::FORBIDDEN, Json(json!({"error": "InvalidRepo", "message": "Repo does not match authenticated user"}))).into_response();
}
let user_query = sqlx::query("SELECT id FROM users WHERE did = $1")
.bind(&did)
let user_query = sqlx::query!("SELECT id FROM users WHERE did = $1", did)
.fetch_optional(&state.db)
.await;
let user_id: uuid::Uuid = match user_query {
Ok(Some(row)) => row.get("id"),
Ok(Some(row)) => row.id,
_ => {
return (
StatusCode::INTERNAL_SERVER_ERROR,
@@ -108,14 +106,13 @@ pub async fn create_record(
}
};
let repo_root_query = sqlx::query("SELECT repo_root_cid FROM repos WHERE user_id = $1")
.bind(user_id)
let repo_root_query = sqlx::query!("SELECT repo_root_cid FROM repos WHERE user_id = $1", user_id)
.fetch_optional(&state.db)
.await;
let current_root_cid = match repo_root_query {
Ok(Some(row)) => {
let cid_str: String = row.get("repo_root_cid");
let cid_str: String = row.repo_root_cid;
Cid::from_str(&cid_str).ok()
}
_ => None,
@@ -280,9 +277,7 @@ pub async fn create_record(
}
};
let update_repo = sqlx::query("UPDATE repos SET repo_root_cid = $1 WHERE user_id = $2")
.bind(new_root_cid.to_string())
.bind(user_id)
let update_repo = sqlx::query!("UPDATE repos SET repo_root_cid = $1 WHERE user_id = $2", new_root_cid.to_string(), user_id)
.execute(&state.db)
.await;
@@ -295,14 +290,14 @@ pub async fn create_record(
.into_response();
}
let record_insert = sqlx::query(
let record_insert = sqlx::query!(
"INSERT INTO records (repo_id, collection, rkey, record_cid) VALUES ($1, $2, $3, $4)
ON CONFLICT (repo_id, collection, rkey) DO UPDATE SET record_cid = $4, created_at = NOW()",
user_id,
input.collection,
rkey,
record_cid.to_string()
)
.bind(user_id)
.bind(&input.collection)
.bind(&rkey)
.bind(record_cid.to_string())
.execute(&state.db)
.await;
@@ -362,18 +357,18 @@ pub async fn put_record(
.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"
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",
token
)
.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"),
row.did,
row.key_bytes,
),
None => {
return (
@@ -396,13 +391,12 @@ pub async fn put_record(
return (StatusCode::FORBIDDEN, Json(json!({"error": "InvalidRepo", "message": "Repo does not match authenticated user"}))).into_response();
}
let user_query = sqlx::query("SELECT id FROM users WHERE did = $1")
.bind(&did)
let user_query = sqlx::query!("SELECT id FROM users WHERE did = $1", did)
.fetch_optional(&state.db)
.await;
let user_id: uuid::Uuid = match user_query {
Ok(Some(row)) => row.get("id"),
Ok(Some(row)) => row.id,
_ => {
return (
StatusCode::INTERNAL_SERVER_ERROR,
@@ -412,14 +406,13 @@ pub async fn put_record(
}
};
let repo_root_query = sqlx::query("SELECT repo_root_cid FROM repos WHERE user_id = $1")
.bind(user_id)
let repo_root_query = sqlx::query!("SELECT repo_root_cid FROM repos WHERE user_id = $1", user_id)
.fetch_optional(&state.db)
.await;
let current_root_cid = match repo_root_query {
Ok(Some(row)) => {
let cid_str: String = row.get("repo_root_cid");
let cid_str: String = row.repo_root_cid;
Cid::from_str(&cid_str).ok()
}
_ => None,
@@ -637,9 +630,7 @@ pub async fn put_record(
}
};
let update_repo = sqlx::query("UPDATE repos SET repo_root_cid = $1 WHERE user_id = $2")
.bind(new_root_cid.to_string())
.bind(user_id)
let update_repo = sqlx::query!("UPDATE repos SET repo_root_cid = $1 WHERE user_id = $2", new_root_cid.to_string(), user_id)
.execute(&state.db)
.await;
@@ -652,14 +643,14 @@ pub async fn put_record(
.into_response();
}
let record_insert = sqlx::query(
let record_insert = sqlx::query!(
"INSERT INTO records (repo_id, collection, rkey, record_cid) VALUES ($1, $2, $3, $4)
ON CONFLICT (repo_id, collection, rkey) DO UPDATE SET record_cid = $4, created_at = NOW()",
user_id,
input.collection,
rkey,
record_cid.to_string()
)
.bind(user_id)
.bind(&input.collection)
.bind(&rkey)
.bind(record_cid.to_string())
.execute(&state.db)
.await;