use std::collections::BTreeSet;
use std::path::Path;
use std::time::SystemTime;
use anyhow::Context as _;
use knot_record::attachment::OCTET_STREAM;
use knot_types::BlobCid;
use crate::atproto::SettledRepo;
use crate::blobs::{BLOBS_DIR, SERVED, Store};
use crate::cas::atomic_write;
use crate::config::ServeLimits;
use crate::nginx::cached;
use crate::nginx::conf::{Block, NONE};
use crate::plan::repo_token;
use super::{map, reroute, wire_spellings};
const MAPS_FILE: &str = "61_blobs.conf";
const BLOB_CSP: &str = "default-src 'none'; sandbox";
/// only captured repos are swept, and touching a referenced blob restarts its grace.
pub fn emit(
output_root: &Path,
generation_dir: &Path,
repos: &[SettledRepo],
captured: &BTreeSet<&str>,
) -> anyhow::Result<()> {
let mut rows = Vec::new();
for repo in repos {
let store = Store::of(output_root, &repo.did);
let referenced = &repo.blobs;
if captured.contains(repo.did.as_str()) {
store
.sweep(referenced, SystemTime::now())
.with_context(|| format!("sweep blobs of {}", repo.did))?;
}
let token = repo_token(repo.did.as_str());
let dir = generation_dir.join(BLOBS_DIR).join(&token);
for link in referenced {
let Some(cid) = link.parse().ok().and_then(BlobCid::new) else {
continue;
};
if store.size(&cid)?.is_none() {
tracing::warn!(repo = %repo.did, %link, "a record references a blob that isn't stored");
continue;
}
if captured.contains(repo.did.as_str()) {
store
.touch(&cid)
.with_context(|| format!("touch blob {link} of {}", repo.did))?;
}
let link = cid.to_string();
let served = store.sniffed(&cid).unwrap_or(OCTET_STREAM);
let extension = SERVED
.iter()
.find(|(mime, _)| *mime == served)
.map_or("bin", |(_, extension)| extension);
std::fs::create_dir_all(&dir)?;
let file = format!("{link}.{extension}");
std::os::unix::fs::symlink(std::fs::canonicalize(store.path(&cid))?, dir.join(&file))?;
let target = format!("{token}/{file}");
rows.extend(
wire_spellings(repo.did.as_str(), target)
.map(|(did, target)| (format!("{did}|{link}"), target)),
);
}
}
let maps = Block::file().block(map("$arg_did|$arg_cid", "$ks_blob", rows));
let path = generation_dir.join("nginx/maps").join(MAPS_FILE);
atomic_write(&path, maps.to_string().as_bytes())
.with_context(|| format!("write {}", path.display()))
}
pub fn locations(generation_path: &str, limits: &ServeLimits) -> Vec<Block> {
let empty = |variable: &str, error: &str| {
reroute(
Block::when(variable, "=", ""),
format!("/_ks_error/{error}"),
)
};
let mut types = Block::new("types", NONE);
for (mime, extension) in SERVED {
types = types.directive(mime, [extension]);
}
// attachments run to 200 MB with public cids, which is why connections and rate are limited.
let mut files = Block::new("location", ["/_ks_blob/"])
.directive("internal", NONE)
.block(types);
if let Some(connections) = limits.blob_connections {
files = files.directive("limit_conn", ["ks_blobs", &connections.to_string()]);
}
if let Some(rate) = &limits.blob_rate {
if let Some(after) = &limits.blob_rate_after {
files = files.directive("limit_rate_after", [after]);
}
files = files.directive("limit_rate", [rate]);
}
vec![
reroute(
Block::exact("/xrpc/com.atproto.sync.getBlob")
.block(empty("$ks_atp_did", "atp_repo_not_found"))
.block(empty("$ks_blob", "blob_not_found")),
"/_ks_blob/$ks_blob",
),
cached(files, OCTET_STREAM, "public, max-age=31536000, immutable")
.directive(
"add_header",
["X-Content-Type-Options", "nosniff", "always"],
)
.directive(
"add_header",
["Content-Security-Policy", BLOB_CSP, "always"],
)
.directive(
"error_page",
["404", "=404", "/_ks_error_body/blob_not_found.json"],
)
.directive("alias", [format!("{generation_path}/{BLOBS_DIR}/")]),
]
}