use std::path::Path;
use anyhow::Context as _;
use serde::{Deserialize, Serialize};
use crate::atproto::{ChainState, POOLED, Pool, Prior, SettledRepo, short_hash};
use crate::cas::atomic_write;
use crate::nginx::cached;
use crate::nginx::conf::{Block, NONE};
pub mod com_atproto_repo_describe_repo;
pub mod com_atproto_repo_get_record;
pub mod com_atproto_repo_list_records;
pub mod com_atproto_sync_get_blob;
pub mod com_atproto_sync_get_latest_commit;
pub mod com_atproto_sync_get_repo;
pub mod com_atproto_sync_get_repo_status;
pub mod com_atproto_sync_list_repos;
pub mod org_tangled_knot_describe_knot;
pub mod org_tangled_repo_list_record_versions;
pub(crate) mod social;
#[cfg(test)]
pub(crate) mod tests;
pub const ATPROTO_DIR: &str = "www/_atproto";
pub(crate) const MAPS_FILE: &str = "60_atproto.conf";
pub(crate) const RECORDS_MAPS_FILE: &str = "62_records.conf";
pub(crate) fn seal_repo(
pool: &mut Pool,
state: &ChainState,
prior: Option<&Prior>,
) -> anyhow::Result<Rows> {
com_atproto_sync_get_repo::seal(pool, state)?;
com_atproto_sync_get_latest_commit::seal(pool, state)?;
com_atproto_sync_get_repo_status::seal(pool, state)?;
com_atproto_repo_describe_repo::seal(pool, state)?;
let head = pool.dir.join(state.commit.to_string());
for shared in POOLED {
std::fs::create_dir_all(pool.dir.join(shared))?;
match std::os::unix::fs::symlink(Path::new("..").join(shared), head.join(shared)) {
Err(error) if error.kind() != std::io::ErrorKind::AlreadyExists => {
return Err(error).with_context(|| format!("link {shared} in {}", head.display()));
}
_ => {}
}
}
let mut rows = Rows {
social: social::digest(state),
..Rows::default()
};
rows.add(pool, state, prior)?;
Ok(rows)
}
#[cfg(test)]
pub fn emit(generation_dir: &Path, repos: &[SettledRepo]) -> anyhow::Result<()> {
emit_with(generation_dir, repos, &social::Social::default())
}
pub fn emit_with(
generation_dir: &Path,
repos: &[SettledRepo],
social: &social::Social,
) -> anyhow::Result<()> {
let root = generation_dir.join(ATPROTO_DIR);
let mut sorted: Vec<&SettledRepo> = repos.iter().collect();
sorted.sort_unstable_by_key(|r| r.did.as_str());
let mut rows = Rows::paging();
std::fs::create_dir_all(&root)?;
for repo in &sorted {
let head = std::fs::canonicalize(repo.pool.join(repo.commit.to_string()))
.with_context(|| format!("atproto files of {}", repo.did))?;
std::os::unix::fs::symlink(&head, root.join(repo.dir_name()))?;
rows.extend(&repo.rows);
}
com_atproto_repo_list_records::write_empty(&root)?;
social::emit(generation_dir, &sorted, social)?;
let (list_keys, list_one) = com_atproto_sync_list_repos::emit_pages(&root, &sorted)?;
let by_did: Vec<(String, String)> = sorted
.iter()
.flat_map(|repo| wire_spellings(repo.did.as_str(), repo.dir_name()))
.collect();
let mut by_repo = by_did.clone();
by_repo.extend(social::repo_spellings(&sorted, social));
let maps = Block::file()
.block(map("$arg_did", "$ks_atp_did", by_did))
.block(map("$arg_repo", "$ks_atp_repo", by_repo))
.blocks(com_atproto_sync_list_repos::maps(list_keys, list_one))
.block(com_atproto_sync_get_repo::since_map());
let maps_path = generation_dir.join("nginx/maps").join(MAPS_FILE);
atomic_write(&maps_path, maps.to_string().as_bytes())
.with_context(|| format!("write {}", maps_path.display()))?;
rows.write_maps(generation_dir)
}
pub fn locations(generation_path: &str, limits: &crate::config::ServeLimits) -> Vec<Block> {
let files = Block::new("location", ["/_ks_atproto/"])
.directive("internal", NONE)
.block(
Block::new("types", NONE)
.directive("application/json", ["json"])
.directive("application/vnd.ipld.car", ["car"]),
);
let mut types = Block::new("types", NONE);
for (mime, extension) in social::avatar_types() {
types = types.directive(mime, [extension]);
}
let avatars = Block::new("location", ["/_ks_avatars/"])
.directive("internal", NONE)
.block(types);
vec![
com_atproto_sync_get_repo::location(),
com_atproto_sync_get_latest_commit::location(),
com_atproto_sync_get_repo_status::location(),
com_atproto_repo_describe_repo::location(),
com_atproto_repo_get_record::location(),
com_atproto_repo_list_records::location(),
org_tangled_repo_list_record_versions::location(),
com_atproto_sync_list_repos::location(),
]
.into_iter()
.chain(social::locations())
.chain([
crate::nginx::read_limited(
cached(files, "application/json", crate::nginx::REVALIDATE)
.directive(
"error_page",
["404", "=400", "/_ks_error_body/atp_record_not_found.json"],
)
.directive("alias", [format!("{generation_path}/{ATPROTO_DIR}/")]),
limits,
),
cached(
avatars,
"application/octet-stream",
crate::nginx::REVALIDATE,
)
.directive(
"add_header",
["X-Content-Type-Options", "nosniff", "always"],
)
.directive(
"alias",
[format!("{generation_path}/{}/", social::AVATARS_WWW)],
),
reroute(
Block::new("location", ["^~", "/xrpc/com.atproto."]),
"/_ks_error/atp_not_implemented",
),
])
.collect()
}
#[derive(Default, Serialize, Deserialize)]
pub(crate) struct Rows {
pub record: Vec<(String, String)>,
#[serde(skip)]
pub record_page: Vec<(String, String)>,
#[serde(skip)]
pub record_dir: Vec<(String, String)>,
pub list: Vec<(String, String)>,
pub list_one: Vec<(String, String)>,
#[serde(skip)]
pub version_page: Vec<(String, String)>,
pub version: Vec<(String, String)>,
pub version_uri: Vec<(String, String)>,
#[serde(default)]
pub social: Vec<social::SocialRecord>,
}
impl Rows {
pub(crate) fn paging() -> Self {
let (record_page, record_dir) = com_atproto_repo_list_records::record_page_rows();
Self {
record_page,
record_dir,
list_one: com_atproto_repo_list_records::empty_collection_rows(),
version_page: org_tangled_repo_list_record_versions::version_page_rows(),
..Self::default()
}
}
pub(crate) fn extend(&mut self, other: &Rows) {
self.record.extend_from_slice(&other.record);
self.record_page.extend_from_slice(&other.record_page);
self.record_dir.extend_from_slice(&other.record_dir);
self.list.extend_from_slice(&other.list);
self.list_one.extend_from_slice(&other.list_one);
self.version_page.extend_from_slice(&other.version_page);
self.version.extend_from_slice(&other.version);
self.version_uri.extend_from_slice(&other.version_uri);
}
pub(crate) fn write_maps(self, generation_dir: &Path) -> anyhow::Result<()> {
let maps = Block::file()
.blocks(com_atproto_repo_get_record::maps(self.record))
.blocks(com_atproto_repo_list_records::maps(
self.record_page,
self.record_dir,
self.list,
self.list_one,
))
.blocks(org_tangled_repo_list_record_versions::maps(
self.version_page,
self.version_uri,
self.version,
));
let path = generation_dir.join("nginx/maps").join(RECORDS_MAPS_FILE);
atomic_write(&path, maps.to_string().as_bytes())
.with_context(|| format!("write {}", path.display()))
}
pub(crate) fn add(
&mut self,
pool: &mut Pool,
repo: &ChainState,
prior: Option<&Prior>,
) -> anyhow::Result<()> {
com_atproto_repo_get_record::add(&mut self.record, pool, repo)?;
com_atproto_repo_list_records::add(&mut self.list, &mut self.list_one, pool, repo)?;
org_tangled_repo_list_record_versions::add(
&mut self.version,
&mut self.version_uri,
pool,
repo,
prior,
)
}
}
pub(crate) fn put_page(
pool: &mut Pool,
repo_dir_name: &str,
subdir: &str,
body: &impl Serialize,
) -> anyhow::Result<String> {
let bytes = serde_json::to_vec(body)?;
let file = format!("{subdir}/{}.json", short_hash(&bytes));
pool.put(&file, &bytes)?;
Ok(format!("{repo_dir_name}/{file}"))
}
pub(crate) fn form_encoded(value: &str) -> String {
url::form_urlencoded::byte_serialize(value.as_bytes()).collect()
}
/// `$arg_` is raw, which means the raw and the form-encoded value both need a row.
pub fn wire_spellings(value: &str, target: String) -> [(String, String); 2] {
[
(value.to_string(), target.clone()),
(form_encoded(value), target),
]
}
/// the js encodeURIComponent spelling, which browser clients send
pub(crate) fn uri_component(value: &str) -> String {
let mut out = String::with_capacity(value.len());
for &byte in value.as_bytes() {
match byte {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' => out.push(byte as char),
b'-' | b'_' | b'.' | b'!' | b'~' | b'*' | b'\'' | b'(' | b')' => out.push(byte as char),
_ => out.push_str(&format!("%{byte:02X}")),
}
}
out
}
pub(crate) fn spellings(value: &str) -> Vec<String> {
let mut all = vec![value.to_string()];
for spelling in [form_encoded(value), uri_component(value)] {
if !all.contains(&spelling) {
all.push(spelling);
}
}
all
}
pub(crate) fn page_rows(
sizes: &[(usize, &[&str])],
directions: &[(char, &[&str])],
) -> Vec<(String, String)> {
let mut rows = Vec::new();
for &(size, limits) in sizes {
for &(label, spellings) in directions {
for limit in limits {
for spelling in spellings {
rows.push((format!("{limit}|{spelling}"), format!("{size}{label}")));
}
}
}
}
rows
}
pub fn map(source: &str, target: &str, rows: Vec<(String, String)>) -> Block {
map_with_default("", source, target, rows)
}
pub(crate) fn map_with_default(
default: &str,
source: &str,
target: &str,
mut rows: Vec<(String, String)>,
) -> Block {
rows.sort_unstable();
rows.dedup();
Block::new("map", [source, target])
.directive("default", [default])
.entries(rows)
}
pub fn reroute(from: Block, to: impl std::fmt::Display) -> Block {
from.directive("rewrite", ["^".to_string(), to.to_string(), "last".into()])
}
pub(crate) fn empty_to(variable: &str, target: &str) -> Block {
reroute(Block::when(variable, "=", ""), target)
}
pub(crate) fn empty(variable: &str, error: &str) -> Block {
empty_to(variable, &format!("/_ks_error/{error}"))
}
/// reads `$args`, not `$arg_name`, since `$arg_name` only sees the first of a repeated param.
#[derive(Clone, Copy)]
pub(crate) enum Refused {
Unless(&'static str, &'static str),
Distinct(&'static str),
}
impl Refused {
pub(crate) fn block(self) -> Block {
let pattern = match self {
Self::Unless(param, allowed) => format!("(^|&){param}=(?!(?:{allowed})(&|$))"),
Self::Distinct(param) => format!("(^|&){param}=([^&]*)&(.*&)?{param}=(?!\\2(&|$))"),
};
reroute(
Block::when("$args", "~*", pattern),
"/_ks_error/git_not_projected",
)
}
}
pub(crate) fn by_key_location(nsid: &str, arg: &str, key: &str, file: &str) -> Block {
reroute(
Block::exact(format!("/xrpc/com.atproto.{nsid}"))
.block(empty(arg, "invalid_request"))
.block(empty(key, "atp_repo_not_found")),
format!("/_ks_atproto/{key}/{file}"),
)
}