Something went wrong. Try again.
An atproto AppView for tracking a Magic: The Gathering collection github.com/eth0net/manaweb
appview atproto mtg rust typescript
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365//! The S3-compatible bucket every static artifact of ours is served from.//!//! An *artifact set* is a directory holding a `manifest.json` and the files it//! names, each under its own prefix in the bucket: the catalog today, a//! scanner index and precomputed recommendations later. A prune takes the//! replaced files away before the new ones go up, never after.//!//! Configured entirely from the environment, and unconfigured means nothing//! is uploaded://!//! | Variable | Default |//! |---|---|//! | `MANAWEB_S3_ENDPOINT` | unset |//! | `MANAWEB_S3_BUCKET` | unset |//! | `MANAWEB_S3_KEY_ID` | unset |//! | `MANAWEB_S3_SECRET` | unset |//! | `MANAWEB_S3_REGION` | `auto` |
pub mod verify;
use std::collections::{BTreeMap, HashSet};use std::env;use std::path::Path;use std::sync::Arc;use std::time::{Duration, SystemTime};
use futures_util::{StreamExt, TryStreamExt, stream};use object_store::aws::{AmazonS3, AmazonS3Builder};use object_store::path::Path as Key;use object_store::{ Attribute, AttributeValue, Attributes, ObjectStore, ObjectStoreExt as _, PutOptions, PutPayload,};use serde::Deserialize;use serde_json::Value;
/// Content-addressed files are never rewritten under the same name.const IMMUTABLE: &str = "public, max-age=31536000, immutable";
/// The manifest is the only part a client re-reads, so it must not be held.const REVALIDATE: &str = "no-cache";
const MANIFEST: &str = "manifest.json";const JSON: &str = "application/json";/// Hashes and the like, which a CDN compressing by content type will leave/// alone rather than spend on incompressible bytes.const BINARY: &str = "application/octet-stream";
/// R2 has no regions, and signing still wants one.const REGION: &str = "auto";
#[derive(Debug, thiserror::Error)]pub enum Error { #[error("{0} is set, so {1} must be too")] Partial(&'static str, &'static str), #[error("object storage: {0}")] Storage(#[from] object_store::Error), #[error("{0}: {1}")] Read(String, #[source] std::io::Error), #[error("{MANIFEST} does not parse: {0}")] Manifest(#[from] serde_json::Error), #[error("{MANIFEST} gives `{0}` no `name`, so there is no file to send")] Part(String),}
pub type Result<T> = std::result::Result<T, Error>;
/// Where a bucket is and what opens it.#[derive(Clone)]pub struct Config { pub endpoint: String, pub bucket: String, pub key_id: String, pub secret: String, /// R2 ignores this; another S3 will not. pub region: String,}
// Written out rather than derived, so the secret cannot reach a log through// one `{:?}` somebody adds while looking at something else.impl std::fmt::Debug for Config { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("Config") .field("endpoint", &self.endpoint) .field("bucket", &self.bucket) .field("key_id", &self.key_id) .field("secret", &"<redacted>") .field("region", &self.region) .finish() }}
/// What a client fetches first: a version, and an entry per file naming it.////// An entry is an object naming its file; any other field is something else/// the set carries, and is passed over.#[derive(Debug, Deserialize)]struct Manifest { version: String, #[serde(flatten)] parts: BTreeMap<String, Value>,}
impl Manifest { /// The file each entry names, in a fixed order. fn names(&self) -> Result<Vec<String>> { self.parts .iter() .filter(|(_, part)| part.is_object()) .map(|(field, part)| { part.get("name") .and_then(Value::as_str) .map(str::to_owned) .ok_or_else(|| Error::Part(field.clone())) }) .collect() }}
/// How long a manifest must have stood before what it replaced can go. A/// client reads the manifest and then spends megabytes fetching the names it/// gave, so the run that takes those names away must never be the run that/// stopped naming them.const SETTLED: Duration = Duration::from_hours(24);
/// One artifact set, as it went down.#[derive(Debug, Default)]pub struct Pruned { /// Gone from the bucket, named as the manifest would have. pub removed: Vec<String>, /// Left: named by the manifest, or too young to be sure of. pub kept: Vec<String>,}
/// One artifact set, as it went up.#[derive(Debug)]pub struct Uploaded { pub version: String, /// Named in upload order, the manifest last. pub sent: Vec<String>, /// Already there under the same name, so not sent again. pub held: Vec<String>,}
/// A bucket to upload artifact sets into.#[derive(Debug, Clone)]pub struct Bucket { s3: Arc<AmazonS3>,}
impl Bucket { /// Reads the configuration, returning `None` when none is set. /// /// # Errors /// /// Fails when the configuration is half-written, or the bucket won't /// build from it. pub fn from_env() -> Result<Option<Self>> { let Some(endpoint) = var("MANAWEB_S3_ENDPOINT") else { return Ok(None); };
let want = |name: &'static str| var(name).ok_or(Error::Partial("MANAWEB_S3_ENDPOINT", name));
Ok(Some(Self::new(&Config { endpoint, bucket: want("MANAWEB_S3_BUCKET")?, key_id: want("MANAWEB_S3_KEY_ID")?, secret: want("MANAWEB_S3_SECRET")?, region: var("MANAWEB_S3_REGION").unwrap_or_else(|| REGION.to_owned()), })?)) }
/// # Errors /// /// Fails if a bucket won't build from these. pub fn new(config: &Config) -> Result<Self> { Ok(Self { s3: Arc::new( AmazonS3Builder::new() .with_endpoint(&config.endpoint) // Naming a plain-http endpoint is how you ask for one. .with_allow_http(config.endpoint.starts_with("http://")) .with_bucket_name(&config.bucket) .with_access_key_id(&config.key_id) .with_secret_access_key(&config.secret) .with_region(&config.region) .build()?, ), }) }
/// Uploads the files the manifest names, then the manifest. /// /// The manifest goes last, so it never names an object that has not /// landed. A name already present is skipped rather than rewritten. /// /// # Errors /// /// Fails if the directory has no readable manifest, if it names a file it /// does not hold, or if an upload is refused. pub async fn upload(&self, prefix: &str, dir: &Path) -> Result<Uploaded> { let manifest = read(&dir.join(MANIFEST)).await?; let named: Manifest = serde_json::from_slice(&manifest)?;
let mut sent = Vec::new(); let mut held = Vec::new(); for name in named.names()? { let key = key(prefix, &name); if self.s3.head(&key).await.is_ok() { held.push(name); continue; } let body = read(&dir.join(&name)).await?; self.put(&key, body, IMMUTABLE).await?; sent.push(name); }
// Rewriting a manifest nobody changed would be a write for nothing, // and worse: a prune measures how long this one has stood, so a // service restarting often would keep pushing that clock back. let at = key(prefix, MANIFEST); if sent.is_empty() && self.holds(&at, &manifest).await { held.push(MANIFEST.to_owned()); } else { self.put(&at, manifest, REVALIDATE).await?; sent.push(MANIFEST.to_owned()); }
Ok(Uploaded { version: named.version, sent, held, }) }
/// Removes what the set's own manifest no longer names. /// /// Belongs before an upload rather than after: what the manifest names is /// what a client is being told to fetch, and a name this has stopped /// carrying is one the previous upload replaced. A manifest that changed /// within the last day takes nothing at all, since a client told about /// the names it replaced may still be reading them. /// /// Reads the manifest from the bucket rather than from a directory, so /// this answers for what is published rather than what was last built. /// Only objects directly under `prefix` are considered, so one set never /// reaches into another's. /// /// # Errors /// /// Fails if the manifest is there and doesn't parse, if the listing is /// refused, or if a delete is. pub async fn prune(&self, prefix: &str) -> Result<Pruned> { let manifest = match self.s3.get(&key(prefix, MANIFEST)).await { Ok(manifest) => manifest, // A prefix nothing has published to yet. Sweeping is what makes // room for an upload, so refusing here would leave the first one // with no way through. Err(object_store::Error::NotFound { .. }) => return Ok(Pruned::default()), Err(error) => return Err(error.into()), }; // A bucket clock running ahead of ours reads as no age at all, which // holds everything rather than failing the sweep. let stood = SystemTime::now() .duration_since(SystemTime::from(manifest.meta.last_modified)) .unwrap_or_default(); let named: Manifest = serde_json::from_slice(&manifest.bytes().await?)?; let mut keep: HashSet<String> = named.names()?.into_iter().collect(); keep.insert(MANIFEST.to_owned());
let at = prefix.trim_matches('/'); let at = (!at.is_empty()).then(|| Key::from(at)); let listing = self.s3.list_with_delimiter(at.as_ref()).await?;
let settled = stood >= SETTLED; let (mut kept, mut going) = (Vec::new(), Vec::new()); for object in listing.objects { let Some(name) = object.location.filename().map(str::to_owned) else { continue; }; if !settled || keep.contains(&name) { kept.push(name); } else { going.push(object.location); } }
// One `DeleteObjects` call rather than one request each. A bucket // without it wants `disable_bulk_delete` on the builder. let gone: Vec<Key> = self .s3 .delete_stream(stream::iter(going).map(Ok).boxed()) .try_collect() .await?; for key in &gone { tracing::debug!(%key, "removed"); }
Ok(Pruned { removed: gone .iter() .filter_map(|key| key.filename().map(str::to_owned)) .collect(), kept, }) }
/// Whether the bucket already holds exactly these bytes under `key`. /// /// A bucket that will not answer is treated as not holding them, since /// the cost of being wrong is one write. async fn holds(&self, key: &Key, body: &[u8]) -> bool { match self.s3.get(key).await { Ok(got) => got.bytes().await.is_ok_and(|there| there == body), Err(_) => false, } }
async fn put(&self, key: &Key, body: Vec<u8>, cache: &str) -> Result<()> { let bytes = body.len(); let mut attributes = Attributes::new(); let kind = match key.extension() { Some("json") => JSON, _ => BINARY, }; attributes.insert(Attribute::ContentType, AttributeValue::from(kind)); attributes.insert( Attribute::CacheControl, AttributeValue::from(cache.to_owned()), );
let options = PutOptions { attributes, ..PutOptions::default() }; self.s3 .put_opts(key, PutPayload::from(body), options) .await?; tracing::debug!(%key, bytes, "uploaded"); Ok(()) }}
/// Each set lives under its own prefix, so two of them can both call the file/// a client reads first `manifest.json`.fn key(prefix: &str, name: &str) -> Key { let prefix = prefix.trim_matches('/'); if prefix.is_empty() { Key::from(name) } else { Key::from(format!("{prefix}/{name}")) }}
async fn read(path: &Path) -> Result<Vec<u8>> { tokio::fs::read(path) .await .map_err(|error| Error::Read(path.display().to_string(), error))}
fn var(name: &str) -> Option<String> { env::var(name).ok().filter(|one| !one.is_empty())}