Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384mod claim;mod kvmap;mod serviceauth;mod sites;mod xrpc;
use claim::SitesStore;use kvmap::DomainMapping;use worker::*;
fn r2_key(did: &str, rkey: &str, site_path: &str) -> String { let base = format!("{}/{}/", did, rkey); if site_path.is_empty() || site_path == "/" { format!("{}index.html", base) } else { let trimmed = site_path.trim_start_matches('/'); if trimmed.is_empty() || trimmed.ends_with('/') { format!("{}{}index.html", base, trimmed) } else { format!("{}{}", base, trimmed) } }}
fn needs_trailing_slash(path: &str) -> bool { if path == "/" || path.ends_with('/') { return false; } let last_segment = path.rsplit('/').next().unwrap_or(path); !last_segment.contains('.')}
fn with_trailing_slash(url: &Url) -> String { let mut url = url.clone(); url.set_path(&format!("{}/", url.path())); url.to_string()}
async fn fetch_from_r2(bucket: &Bucket, key: &str) -> Result<Option<Object>> { if let Some(obj) = bucket.get(key).execute().await? { return Ok(Some(obj)); }
let last_segment = key.rsplit('/').next().unwrap_or(key); if !last_segment.contains('.') { let index_key = format!("{}/index.html", key.trim_end_matches('/')); if let Some(obj) = bucket.get(&index_key).execute().await? { return Ok(Some(obj)); } }
Ok(None)}
fn response_from_object(obj: Object) -> Result<Response> { let content_type = obj .http_metadata() .content_type .unwrap_or_else(|| "application/octet-stream".to_string());
let body = obj .body() .ok_or_else(|| Error::RustError("empty R2 body".into()))?; let mut resp = Response::from_body(body.response_body()?)?; resp.headers_mut().set("Content-Type", &content_type)?; resp.headers_mut() .set("Cache-Control", "public, max-age=60")?; Ok(resp)}
#[event(fetch)]async fn fetch(req: Request, env: Env, ctx: Context) -> Result<Response> { let origin = req.headers().get("Origin")?.filter(|o| !o.is_empty());
if req.method() == Method::Options { return preflight_response(&origin); }
let mut resp = fetch_inner(req, env, ctx).await?; if let Some(o) = &origin { resp.headers_mut().set("Access-Control-Allow-Origin", o)?; resp.headers_mut().set("Vary", "Origin")?; } Ok(resp)}
fn preflight_response(origin: &Option<String>) -> Result<Response> { let mut resp = Response::empty()?.with_status(204); if let Some(o) = origin { resp.headers_mut().set("Access-Control-Allow-Origin", o)?; resp.headers_mut().set("Access-Control-Allow-Methods", "GET, POST, OPTIONS")?; resp.headers_mut().set( "Access-Control-Allow-Headers", "authorization, content-type, accept", )?; resp.headers_mut().set("Access-Control-Max-Age", "86400")?; } Ok(resp)}
const HANDLE_TITLE_TTL: u64 = 300;
// shorter negative cache so fresh signups aren't delayedconst UNKNOWN_HANDLE_TTL: u64 = 60;
async fn handle_title(kv: &KvStore, host: &str) -> Result<Option<String>, String> { let key = format!("handle-title:{}", host); if let Some(memo) = kv.get(&key).text().await.map_err(|e| e.to_string())? { let memo = memo.trim(); return Ok((!memo.is_empty()).then(|| memo.to_string())); }
let resolved = claim::resolve_handle(host.to_string()).await?; kv.put(&key, resolved.as_deref().unwrap_or("")) .map_err(|e| e.to_string())? .expiration_ttl(if resolved.is_none() { UNKNOWN_HANDLE_TTL } else { HANDLE_TITLE_TTL }) .execute() .await .map_err(|e| e.to_string())?; Ok(resolved)}
fn local_r2_enabled(mode: &str) -> bool { mode == "true"}
fn local_r2_authorized(token: &str, authorization: &str) -> bool { !token.is_empty() && authorization == format!("Bearer {}", token)}
async fn local_r2(mut req: Request, env: &Env, path: &str) -> Result<Response> { let enabled = env .var("SITES_LOCAL_R2_ENABLED") .map(|v| v.to_string()) .unwrap_or_default(); if !local_r2_enabled(&enabled) { return Response::error("Not Found", 404); } let token = env .var("SITES_LOCAL_R2_TOKEN") .map(|v| v.to_string()) .unwrap_or_default(); let auth = req.headers().get("Authorization")?.unwrap_or_default(); if !local_r2_authorized(&token, &auth) { return Response::error("Unauthorized", 401); } if path != "/__tangled_r2/tangled-sites" { return Response::error("Not Found", 404); }
let url = req.url()?; let bucket = env.bucket("SITES_BUCKET")?; match req.method() { Method::Get => { let prefix = url .query_pairs() .find(|(key, _)| key == "prefix") .map(|(_, value)| value.into_owned()) .unwrap_or_default(); let mut cursor = None; let mut objects = Vec::new(); loop { let mut list = bucket.list().prefix(&prefix).limit(1000); if let Some(current) = cursor { list = list.cursor(current); } let page = list.execute().await?; objects.extend(page.objects().iter().map(|object| { serde_json::json!({ "key": object.key(), "size": object.size(), "etag": object.etag() }) })); if !page.truncated() { break; } cursor = page.cursor(); if cursor.is_none() { return Response::error("Incomplete R2 list", 500); } } Response::from_json(&objects) } Method::Put => { let key = url .query_pairs() .find(|(name, _)| name == "key") .map(|(_, value)| value.into_owned()) .unwrap_or_default(); if key.is_empty() { return Response::error("Bad Request", 400); } let bytes = req.bytes().await?; let content_type = req .headers() .get("Content-Type")? .unwrap_or_else(|| "application/octet-stream".into()); bucket .put(&key, bytes) .http_metadata(HttpMetadata { content_type: Some(content_type), ..Default::default() }) .execute() .await?; Ok(Response::empty()?.with_status(204)) } Method::Delete => { let key = url .query_pairs() .find(|(name, _)| name == "key") .map(|(_, value)| value.into_owned()) .unwrap_or_default(); if key.is_empty() { return Response::error("Bad Request", 400); } bucket.delete(&key).await?; Ok(Response::empty()?.with_status(204)) } _ => Response::error("Method Not Allowed", 405), }}
async fn fetch_inner(req: Request, env: Env, ctx: Context) -> Result<Response> { let url = req.url()?; let path = url.path();
if path.starts_with("/__tangled_r2/") { return local_r2(req, &env, path).await; }
if path.starts_with("/xrpc/") { return xrpc::route(req, env, ctx).await; }
let kv = env.kv("SITES")?; let bucket = env.bucket("SITES_BUCKET")?;
let host = req.headers().get("host")?.unwrap_or_default(); let host = host.split(':').next().unwrap_or("").to_string();
if host.is_empty() { return Response::error("Bad Request: missing host", 400); }
if path == "/.well-known/atproto-did" { // from the domain claim, never the site if req.method() != Method::Get { let mut resp = Response::error("Method Not Allowed", 405)?; resp.headers_mut().set("Allow", "GET")?; return Ok(resp); } let did = if claim::is_handle_domain(&host) { match handle_title(&kv, &host).await { Ok(resolved) => resolved, Err(e) => { console_warn!("handle authority lookup for {} failed: {}", host, e); return Response::error("Service Unavailable", 503); } } } else { let store = SitesStore::d1(env.d1("SITES_DB")?); match store.select_claim_by_domain(&host).await { Ok(Some(claim)) if claim.deleted.is_none() => Some(claim.did), Ok(_) => None, Err(e) => { console_warn!("atproto-did claim lookup for {} failed: {}", host, e); return Response::error("Internal Error", 500); } } };
return match did { Some(did) => { let mut resp = Response::ok(did)?; resp.headers_mut() .set("Cache-Control", "public, max-age=300")?; Ok(resp) } None => Response::error("Not Found", 404), }; }
if needs_trailing_slash(path) { let redirect_url = with_trailing_slash(&url); return Response::redirect_with_status(redirect_url.parse()?, 308); }
let mapping = match kv.get(&host).text().await? { Some(raw) => { let trimmed = raw.trim(); if trimmed.is_empty() || trimmed == "{}" { return Response::error("site not found!", 404); } match serde_json::from_str::<DomainMapping>(&raw) { Ok(m) => m, Err(_) => return Response::error("Internal Error: bad mapping", 500), } } None => return Response::error("site not found!", 404), };
if claim::is_handle_domain(&host) { match handle_title(&kv, &host).await { Ok(resolved) if resolved.as_deref() == Some(mapping.did.as_str()) => {} Ok(_) => { console_warn!( "refusing to serve {}: the zone does not resolve it to its mapping's owner", host ); return Response::error("Not Found", 404); } Err(e) => { console_warn!("handle authority lookup for {} failed: {}", host, e); return Response::error("Service Unavailable", 503); } } }
let path = url.path();
let first_segment = path .trim_start_matches('/') .split('/') .next() .unwrap_or("") .to_string();
if !first_segment.is_empty() { if let Some(entry) = mapping.repos.get(&first_segment) { if !entry.is_index() { let site_path = path .trim_start_matches('/') .trim_start_matches(&first_segment) .to_string();
let key = r2_key(&mapping.did, entry.rkey(&first_segment), &site_path); return match fetch_from_r2(&bucket, &key).await? { Some(obj) => response_from_object(obj), None => Response::error("Not Found", 404), }; } } }
if let Some((name, entry)) = mapping.index_repo() { let key = r2_key(&mapping.did, entry.rkey(name), path); return match fetch_from_r2(&bucket, &key).await? { Some(obj) => response_from_object(obj), None => Response::error("Not Found", 404), }; }
Response::error("Not Found", 404)}#[cfg(test)]mod local_r2_tests { use super::{local_r2_authorized, local_r2_enabled, r2_key}; #[test] fn adapter_requires_the_explicit_dev_binding() { assert!(local_r2_enabled("true")); assert!(!local_r2_enabled("")); assert!(!local_r2_enabled("false")); }
#[test] fn adapter_requires_a_configured_bearer_token() { assert!(local_r2_authorized("local", "Bearer local")); assert!(!local_r2_authorized("local", "")); assert!(!local_r2_authorized("", "Bearer local")); }
#[test] fn site_paths_map_to_the_worker_bucket_key_space() { assert_eq!(r2_key("did:plc:abc", "repo", "index.html"), "did:plc:abc/repo/index.html"); assert_eq!(r2_key("did:plc:abc", "repo", "/"), "did:plc:abc/repo/index.html"); }}