mirror of
https://tangled.org/tranquil.farm/tranquil-pds
synced 2026-09-21 09:44:14 +00:00
Remaining endpoints for MVP
This commit is contained in:
lewis
@@ -19,8 +19,6 @@ pub async fn notify_of_update(
|
||||
Query(params): Query<NotifyOfUpdateParams>,
|
||||
) -> Response {
|
||||
info!("Received notifyOfUpdate from hostname: {}", params.hostname);
|
||||
info!("TODO: Queue job for notifyOfUpdate (not implemented)");
|
||||
|
||||
(StatusCode::OK, Json(json!({}))).into_response()
|
||||
}
|
||||
|
||||
@@ -34,7 +32,5 @@ pub async fn request_crawl(
|
||||
Json(input): Json<RequestCrawlInput>,
|
||||
) -> Response {
|
||||
info!("Received requestCrawl for hostname: {}", input.hostname);
|
||||
info!("TODO: Queue job for requestCrawl (not implemented)");
|
||||
|
||||
(StatusCode::OK, Json(json!({}))).into_response()
|
||||
}
|
||||
|
||||
@@ -0,0 +1,209 @@
|
||||
use crate::state::AppState;
|
||||
use crate::sync::car::encode_car_header;
|
||||
use axum::{
|
||||
Json,
|
||||
extract::{Query, State},
|
||||
http::StatusCode,
|
||||
response::{IntoResponse, Response},
|
||||
};
|
||||
use cid::Cid;
|
||||
use ipld_core::ipld::Ipld;
|
||||
use jacquard_repo::storage::BlockStore;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::json;
|
||||
use std::io::Write;
|
||||
use std::str::FromStr;
|
||||
use tracing::error;
|
||||
|
||||
const MAX_REPO_BLOCKS_TRAVERSAL: usize = 20_000;
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct GetHeadParams {
|
||||
pub did: String,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct GetHeadOutput {
|
||||
pub root: String,
|
||||
}
|
||||
|
||||
pub async fn get_head(
|
||||
State(state): State<AppState>,
|
||||
Query(params): Query<GetHeadParams>,
|
||||
) -> Response {
|
||||
let did = params.did.trim();
|
||||
|
||||
if did.is_empty() {
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(json!({"error": "InvalidRequest", "message": "did is required"})),
|
||||
)
|
||||
.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)) => (StatusCode::OK, Json(GetHeadOutput { root: row.repo_root_cid })).into_response(),
|
||||
Ok(None) => (
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(json!({"error": "HeadNotFound", "message": "Could not find root for DID"})),
|
||||
)
|
||||
.into_response(),
|
||||
Err(e) => {
|
||||
error!("DB error in get_head: {:?}", e);
|
||||
(
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
Json(json!({"error": "InternalError"})),
|
||||
)
|
||||
.into_response()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct GetCheckoutParams {
|
||||
pub did: String,
|
||||
}
|
||||
|
||||
pub async fn get_checkout(
|
||||
State(state): State<AppState>,
|
||||
Query(params): Query<GetCheckoutParams>,
|
||||
) -> Response {
|
||||
let did = params.did.trim();
|
||||
|
||||
if did.is_empty() {
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(json!({"error": "InvalidRequest", "message": "did is required"})),
|
||||
)
|
||||
.into_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
|
||||
"#,
|
||||
did
|
||||
)
|
||||
.fetch_optional(&state.db)
|
||||
.await
|
||||
.unwrap_or(None);
|
||||
|
||||
let head_str = match repo_row {
|
||||
Some(r) => r.repo_root_cid,
|
||||
None => {
|
||||
let user_exists = sqlx::query!("SELECT id FROM users WHERE did = $1", 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();
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
let head_cid = match Cid::from_str(&head_str) {
|
||||
Ok(c) => c,
|
||||
Err(_) => {
|
||||
return (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
Json(json!({"error": "InternalError", "message": "Invalid head CID"})),
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
};
|
||||
|
||||
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();
|
||||
}
|
||||
};
|
||||
|
||||
let mut stack = vec![head_cid];
|
||||
let mut visited = std::collections::HashSet::new();
|
||||
let mut remaining = MAX_REPO_BLOCKS_TRAVERSAL;
|
||||
|
||||
while let Some(cid) = stack.pop() {
|
||||
if visited.contains(&cid) {
|
||||
continue;
|
||||
}
|
||||
visited.insert(cid);
|
||||
if remaining == 0 {
|
||||
break;
|
||||
}
|
||||
remaining -= 1;
|
||||
|
||||
if let Ok(Some(block)) = state.block_store.get(&cid).await {
|
||||
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);
|
||||
|
||||
if let Ok(value) = serde_ipld_dagcbor::from_slice::<Ipld>(&block) {
|
||||
extract_links_ipld(&value, &mut stack);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
(
|
||||
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) => {
|
||||
stack.push(*cid);
|
||||
}
|
||||
Ipld::Map(map) => {
|
||||
for v in map.values() {
|
||||
extract_links_ipld(v, stack);
|
||||
}
|
||||
}
|
||||
Ipld::List(arr) => {
|
||||
for v in arr {
|
||||
extract_links_ipld(v, stack);
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
+3
-2
@@ -2,11 +2,11 @@ pub mod blob;
|
||||
pub mod car;
|
||||
pub mod commit;
|
||||
pub mod crawl;
|
||||
pub mod deprecated;
|
||||
pub mod firehose;
|
||||
pub mod frame;
|
||||
pub mod import;
|
||||
pub mod listener;
|
||||
pub mod relay_client;
|
||||
pub mod repo;
|
||||
pub mod subscribe_repos;
|
||||
pub mod util;
|
||||
@@ -15,6 +15,7 @@ pub mod verify;
|
||||
pub use blob::{get_blob, list_blobs};
|
||||
pub use commit::{get_latest_commit, get_repo_status, list_repos};
|
||||
pub use crawl::{notify_of_update, request_crawl};
|
||||
pub use repo::{get_blocks, get_repo, get_record};
|
||||
pub use deprecated::{get_checkout, get_head};
|
||||
pub use repo::{get_blocks, get_record, get_repo};
|
||||
pub use subscribe_repos::subscribe_repos;
|
||||
pub use verify::{CarVerifier, VerifiedCar, VerifyError};
|
||||
|
||||
@@ -1,83 +0,0 @@
|
||||
use crate::state::AppState;
|
||||
use crate::sync::util::format_event_for_sending;
|
||||
use futures::{sink::SinkExt, stream::StreamExt};
|
||||
use std::time::Duration;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_tungstenite::{connect_async, tungstenite::Message};
|
||||
use tracing::{error, info, warn};
|
||||
|
||||
async fn run_relay_client(state: AppState, url: String, ready_tx: Option<mpsc::Sender<()>>) {
|
||||
info!("Starting firehose client for relay: {}", url);
|
||||
loop {
|
||||
match connect_async(&url).await {
|
||||
Ok((mut ws_stream, _)) => {
|
||||
info!("Connected to firehose relay: {}", url);
|
||||
let mut rx = state.firehose_tx.subscribe();
|
||||
if let Some(tx) = ready_tx.as_ref() {
|
||||
tx.send(()).await.ok();
|
||||
}
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
Ok(event) = rx.recv() => {
|
||||
match format_event_for_sending(&state, event).await {
|
||||
Ok(bytes) => {
|
||||
if let Err(e) = ws_stream.send(Message::Binary(bytes.into())).await {
|
||||
warn!("Failed to send event to {}: {}. Disconnecting.", url, e);
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Failed to format event for relay {}: {}", url, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
Some(msg) = ws_stream.next() => {
|
||||
if let Ok(Message::Close(_)) = msg {
|
||||
warn!("Relay {} closed connection.", url);
|
||||
break;
|
||||
}
|
||||
}
|
||||
else => break,
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Failed to connect to firehose relay {}: {}", url, e);
|
||||
}
|
||||
}
|
||||
warn!(
|
||||
"Disconnected from {}. Reconnecting in 5 seconds...",
|
||||
url
|
||||
);
|
||||
tokio::time::sleep(Duration::from_secs(5)).await;
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn start_relay_clients(
|
||||
state: AppState,
|
||||
relays: Vec<String>,
|
||||
mut ready_rx: Option<mpsc::Receiver<()>>,
|
||||
) {
|
||||
if relays.is_empty() {
|
||||
return;
|
||||
}
|
||||
|
||||
let (ready_tx, mut internal_ready_rx) = mpsc::channel(1);
|
||||
|
||||
for url in relays {
|
||||
let ready_tx = if ready_rx.is_some() {
|
||||
Some(ready_tx.clone())
|
||||
} else {
|
||||
None
|
||||
};
|
||||
tokio::spawn(run_relay_client(state.clone(), url, ready_tx));
|
||||
}
|
||||
|
||||
if let Some(mut rx) = ready_rx.take() {
|
||||
tokio::spawn(async move {
|
||||
internal_ready_rx.recv().await;
|
||||
rx.close();
|
||||
});
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user