From fc45ea8867743f323bb2ede8c0ef14ef13272b87 Mon Sep 17 00:00:00 2001 From: Nikolay Govorov Date: Thu, 15 Jan 2026 22:51:56 +0000 Subject: Update project structure --- Cargo.lock | 22 ++ Cargo.toml | 3 +- src/backends/mod.rs | 4 + src/backends/zig.rs | 218 ++++++++++++++++++ src/config.rs | 93 ++++++++ src/controller_web.rs | 36 --- src/controller_zig.rs | 212 ----------------- src/index.html | 100 -------- src/main.rs | 270 +++++++++++++++++++--- src/service_config.rs | 93 -------- src/service_storage.rs | 497 ---------------------------------------- src/service_upstream.rs | 60 ----- src/storage.rs | 497 ++++++++++++++++++++++++++++++++++++++++ src/upstream.rs | 60 +++++ 14 files changed, 1129 insertions(+), 1036 deletions(-) create mode 100644 src/backends/mod.rs create mode 100644 src/backends/zig.rs create mode 100644 src/config.rs delete mode 100644 src/controller_web.rs delete mode 100644 src/controller_zig.rs delete mode 100644 src/index.html delete mode 100644 src/service_config.rs delete mode 100644 src/service_storage.rs delete mode 100644 src/service_upstream.rs create mode 100644 src/storage.rs create mode 100644 src/upstream.rs diff --git a/Cargo.lock b/Cargo.lock index bd1d6bc..e9ea97e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -199,6 +199,27 @@ dependencies = [ "tracing", ] +[[package]] +name = "axum-extra" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fef252edff26ddba56bbcdf2ee3307b8129acb86f5749b68990c168a6fcc9c76" +dependencies = [ + "axum", + "axum-core", + "bytes", + "futures-core", + "futures-util", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "tower-layer", + "tower-service", + "tracing", +] + [[package]] name = "axum-macros" version = "0.5.0" @@ -4944,6 +4965,7 @@ name = "zorian" version = "0.1.0" dependencies = [ "axum", + "axum-extra", "bytes", "cargo-deny", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 11e5f9a..85f596c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,6 +10,7 @@ repository = "https://github.com/mrdimidium/Zorian" [dependencies] axum = { version = "0.8.8", features = ["http2", "macros", "multipart"] } +axum-extra = "0.12.5" bytes = "1.11.0" chrono = "0.4" crc32fast = "1.4.2" @@ -25,10 +26,10 @@ sqlx = { version = "0.8.6", features = ["sqlite", "runtime-tokio", "macros", "de thiserror = "2.0" tokio = { version = "1.49.0", features = ["full"] } toml = "0.9.8" +tower = "0.5.3" tower-http = { version = "0.6.8", features = ["full"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } -tower = "0.5.3" uuid = { version = "1.19.0", features = ["v4","v5"] } [dev-dependencies] diff --git a/src/backends/mod.rs b/src/backends/mod.rs new file mode 100644 index 0000000..1fb3a4b --- /dev/null +++ b/src/backends/mod.rs @@ -0,0 +1,4 @@ +// SPDX-FileCopyrightText: 2026 Nikolay Govorov +// SPDX-License-Identifier: AGPL-3.0-or-later + +pub mod zig; diff --git a/src/backends/zig.rs b/src/backends/zig.rs new file mode 100644 index 0000000..44ac3bc --- /dev/null +++ b/src/backends/zig.rs @@ -0,0 +1,218 @@ +// SPDX-FileCopyrightText: 2026 Nikolay Govorov +// SPDX-License-Identifier: AGPL-3.0-or-later + +use std::sync::Arc; + +use axum::{Router, body, extract, http, response, routing}; +use semver::Version; +use thiserror::Error; +use tracing::error; + +use crate::config; +use crate::storage; +use crate::upstream; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Archive { + Zip, + TarXz, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +enum TarballType<'a> { + Source, + Bootstrap, + Binary { os: &'a str, arch: &'a str }, +} + +#[derive(Debug, Clone, PartialEq, Eq, Error)] +#[error("invalid tarball filename")] +struct ParseError; + +/// Describes a single file stored at `ziglang.org/download/`. +/// +/// The tarball naming has changed several times. When parsing, +/// we standardize the files, but for the reverse operation +/// (getting a string from a tarball), we preserve the original path. +#[derive(Debug, Clone, PartialEq, Eq)] +struct Tarball<'a> { + filename: &'a str, + tarball_type: TarballType<'a>, + minisig: bool, + archive: Archive, + version: Version, + development: bool, +} + +impl<'a> Tarball<'a> { + pub fn parse(filename: &'a str) -> Result { + let mut buffer = filename; + let mut minisig = false; + let archive; + let tarball_type; + + // (?:|-bootstrap|-[a-zA-Z0-9_]+-[a-zA-Z0-9_]+)-( + // \d+\.\d+\.\d+(?:-dev\.\d+\+[0-9a-f]+)? + // )\.(?:tar\.xz|zip)(?:\.minisig)? + buffer = buffer.strip_prefix("zig-").ok_or(ParseError)?; + + // (?:|bootstrap|[a-zA-Z0-9_]+-[a-zA-Z0-9_]+)-( + // \d+\.\d+\.\d+(?:-dev\.\d+\+[0-9a-f]+)? + // )\.(?:tar\.xz|zip) + if let Some(it) = buffer.strip_suffix(".minisig") { + buffer = it; + minisig = true; + } + + // (?:|bootstrap|[a-zA-Z0-9_]+-[a-zA-Z0-9_]+)-( + // \d+\.\d+\.\d+(?:-dev\.\d+\+[0-9a-f]+)? + // ) + if let Some(it) = buffer.strip_suffix(".zip") { + buffer = it; + archive = Archive::Zip; + } else if let Some(it) = buffer.strip_suffix(".tar.xz") { + buffer = it; + archive = Archive::TarXz; + } else { + return Err(ParseError); + } + + if buffer.is_empty() { + return Err(ParseError); + } + + let mut it = buffer.rsplit('-'); + let last = it.next().ok_or(ParseError)?; + + let development = last.starts_with("dev"); + + let version = if !development { + Version::parse(last).map_err(|_| ParseError)? + } else { + let semver = it.next().ok_or(ParseError)?; + let devver = last; + let version_str = format!("{}-{}", semver, devver); + Version::parse(&version_str).map_err(|_| ParseError)? + }; + + if let Some(payload) = it.next() { + if payload == "bootstrap" { + tarball_type = TarballType::Bootstrap; + } else { + // Version 0.14.0 is the last one to use the OS-ARCH format in names; newer versions use ARCH-OS. + let min_version = Version::new(0, 14, 0); + if version > min_version { + tarball_type = TarballType::Binary { + os: payload, + arch: it.next().ok_or(ParseError)?, + }; + } else { + tarball_type = TarballType::Binary { + arch: payload, + os: it.next().ok_or(ParseError)?, + }; + } + } + } else { + tarball_type = TarballType::Source; + } + + if it.next().is_some() { + return Err(ParseError); + } + + Ok(Tarball { + filename, + tarball_type, + minisig, + archive, + version, + development, + }) + } + + /// Builds the upstream URL for this tarball. + pub fn upstream_url(&self, source: &str) -> String { + if self.development { + format!( + "https://ziglang.org/builds/{}?source={}", + self.filename, source + ) + } else { + format!( + "https://ziglang.org/download/{}/{}?source={}", + self.version, self.filename, source, + ) + } + } +} + +pub struct ZigController { + config: Arc, + storage: Arc, + upstream: Arc, +} + +impl ZigController { + pub fn new( + config: Arc, + storage: Arc, + upstream: Arc, + ) -> Self { + Self { + config, + storage, + upstream, + } + } + + pub fn router(self: Arc) -> Router { + Router::new() + .route("/zig/{filename}", routing::get(Self::handle)) + .with_state(self) + } + + async fn handle( + extract::State(controller): extract::State>, + extract::Path(filename): extract::Path, + ) -> Result { + let tarball = Tarball::parse(&filename).map_err(|_| http::StatusCode::NOT_FOUND)?; + let url = tarball.upstream_url(controller.config.appname()); + + match controller.storage.get("zig", &filename).await { + Ok(Some(entry)) => { + return Ok(Self::build_response( + http::StatusCode::OK, + entry.file_bytes.0, + )); + } + Ok(None) => {} + Err(err) => { + error!("failed get file from storage: {err}"); + return Err(http::StatusCode::INTERNAL_SERVER_ERROR); + } + } + + let entry = controller + .upstream + .fetch(upstream::DownloadRequest { url }) + .await?; + + match controller.storage.put("zig", &filename, &entry.bytes).await { + Ok(()) => {} + Err(_) => { + return Err(http::StatusCode::INTERNAL_SERVER_ERROR); + } + } + + Ok(Self::build_response(http::StatusCode::OK, entry.bytes)) + } + + fn build_response(status: http::StatusCode, bytes: bytes::Bytes) -> response::Response { + response::Response::builder() + .status(status) + .header(http::header::CONTENT_TYPE, "application/octet-stream") + .body(body::Body::from(bytes)) + .unwrap() + } +} diff --git a/src/config.rs b/src/config.rs new file mode 100644 index 0000000..dd55bfc --- /dev/null +++ b/src/config.rs @@ -0,0 +1,93 @@ +// SPDX-FileCopyrightText: 2026 Nikolay Govorov +// SPDX-License-Identifier: AGPL-3.0-or-later + +use std::fs; +use std::path::{Path, PathBuf}; + +use serde::Deserialize; +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum ConfigError { + #[error("failed to read config file: {0}")] + Io(#[from] std::io::Error), + + #[error("failed to parse config file: {0}")] + Parse(#[from] toml::de::Error), + + #[error("appname '{0}' contains invalid characters (only a-z, A-Z, 0-9, -, _ allowed)")] + InvalidAppname(String), + + #[error("dirname '{0}' does not exist")] + DirNotFound(PathBuf), + + #[error("dirname '{0}' is not a directory")] + NotADirectory(PathBuf), + + #[error("dirname '{0}' is not writable: {1}")] + NotWritable(PathBuf, std::io::Error), +} + +#[derive(Debug, Deserialize)] +#[serde(default)] +pub struct ConfigService { + listen: String, + appname: String, + dirname: PathBuf, +} + +impl Default for ConfigService { + fn default() -> Self { + Self { + listen: "0.0.0.0:3000".to_string(), + appname: "zorian".to_string(), + dirname: PathBuf::from("./.zorian-state"), + } + } +} + +impl ConfigService { + pub fn from_file(path: &Path) -> Result { + let content = fs::read_to_string(path)?; + let config: Self = toml::from_str(&content)?; + Ok(config) + } + + pub fn validate(&self) -> Result<(), ConfigError> { + let mut chars = self.appname.chars(); + if !chars.all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') { + return Err(ConfigError::InvalidAppname(self.appname.clone())); + } + + let metadata = fs::metadata(&self.dirname).map_err(|e| { + if e.kind() == std::io::ErrorKind::NotFound { + ConfigError::DirNotFound(self.dirname.clone()) + } else { + ConfigError::Io(e) + } + })?; + + if !metadata.is_dir() { + return Err(ConfigError::NotADirectory(self.dirname.clone())); + } + + let testfile = self.dirname.join(".health"); + fs::write(&testfile, std::process::id().to_string()) + .map_err(|e| ConfigError::NotWritable(self.dirname.clone(), e))?; + fs::remove_file(&testfile)?; + + Ok(()) + } + + pub fn appname(&self) -> &str { + &self.appname + } + + pub fn listen(&self) -> &str { + &self.listen + } + + pub fn dirname(&self) -> &Path { + &self.dirname + } +} diff --git a/src/controller_web.rs b/src/controller_web.rs deleted file mode 100644 index ab62cc1..0000000 --- a/src/controller_web.rs +++ /dev/null @@ -1,36 +0,0 @@ -// SPDX-FileCopyrightText: 2026 Nikolay Govorov -// SPDX-License-Identifier: AGPL-3.0-or-later - -use axum::{Router, extract, http, response, routing}; -use std::sync::Arc; - -/// Handles html pages rendering and static files -pub struct WebController { - jinja: minijinja::Environment<'static>, -} - -impl WebController { - pub fn new() -> Self { - let mut jinja = minijinja::Environment::new(); - jinja - .add_template("index", include_str!("./index.html")) - .unwrap(); - - Self { jinja } - } - - pub fn router(self: Arc) -> Router { - Router::new() - .route("/", routing::get(Self::index)) - .with_state(self) - } - - async fn index( - extract::State(controller): extract::State>, - ) -> Result, http::StatusCode> { - let template = controller.jinja.get_template("index").unwrap(); - let rendered = template.render(minijinja::context! {}).unwrap(); - - Ok(response::Html(rendered)) - } -} diff --git a/src/controller_zig.rs b/src/controller_zig.rs deleted file mode 100644 index 7e9e6a8..0000000 --- a/src/controller_zig.rs +++ /dev/null @@ -1,212 +0,0 @@ -// SPDX-FileCopyrightText: 2026 Nikolay Govorov -// SPDX-License-Identifier: AGPL-3.0-or-later - -use std::sync::Arc; - -use axum::{Router, body, extract, http, response, routing}; -use semver::Version; -use thiserror::Error; -use tracing::error; - -use crate::service_config; -use crate::service_storage; -use crate::service_upstream; - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum Archive { - Zip, - TarXz, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -enum TarballType<'a> { - Source, - Bootstrap, - Binary { os: &'a str, arch: &'a str }, -} - -#[derive(Debug, Clone, PartialEq, Eq, Error)] -#[error("invalid tarball filename")] -struct ParseError; - -/// Describes a single file stored at `ziglang.org/download/`. -/// -/// The tarball naming has changed several times. When parsing, -/// we standardize the files, but for the reverse operation -/// (getting a string from a tarball), we preserve the original path. -#[derive(Debug, Clone, PartialEq, Eq)] -struct Tarball<'a> { - filename: &'a str, - tarball_type: TarballType<'a>, - minisig: bool, - archive: Archive, - version: Version, - development: bool, -} - -impl<'a> Tarball<'a> { - pub fn parse(filename: &'a str) -> Result { - let mut buffer = filename; - let mut minisig = false; - let archive; - let tarball_type; - - // (?:|-bootstrap|-[a-zA-Z0-9_]+-[a-zA-Z0-9_]+)-(\d+\.\d+\.\d+(?:-dev\.\d+\+[0-9a-f]+)?)\.(?:tar\.xz|zip)(?:\.minisig)? - buffer = buffer.strip_prefix("zig-").ok_or(ParseError)?; - - // (?:|bootstrap|[a-zA-Z0-9_]+-[a-zA-Z0-9_]+)-(\d+\.\d+\.\d+(?:-dev\.\d+\+[0-9a-f]+)?)\.(?:tar\.xz|zip) - if let Some(it) = buffer.strip_suffix(".minisig") { - buffer = it; - minisig = true; - } - - // (?:|bootstrap|[a-zA-Z0-9_]+-[a-zA-Z0-9_]+)-(\d+\.\d+\.\d+(?:-dev\.\d+\+[0-9a-f]+)?) - if let Some(it) = buffer.strip_suffix(".zip") { - buffer = it; - archive = Archive::Zip; - } else if let Some(it) = buffer.strip_suffix(".tar.xz") { - buffer = it; - archive = Archive::TarXz; - } else { - return Err(ParseError); - } - - if buffer.is_empty() { - return Err(ParseError); - } - - let mut it = buffer.rsplit('-'); - let last = it.next().ok_or(ParseError)?; - - let development = last.starts_with("dev"); - - let version = if !development { - Version::parse(last).map_err(|_| ParseError)? - } else { - let semver = it.next().ok_or(ParseError)?; - let devver = last; - let version_str = format!("{}-{}", semver, devver); - Version::parse(&version_str).map_err(|_| ParseError)? - }; - - if let Some(payload) = it.next() { - if payload == "bootstrap" { - tarball_type = TarballType::Bootstrap; - } else { - // Version 0.14.0 is the last one to use the OS-ARCH format in names; newer versions use ARCH-OS. - let min_version = Version::new(0, 14, 0); - if version > min_version { - tarball_type = TarballType::Binary { - os: payload, - arch: it.next().ok_or(ParseError)?, - }; - } else { - tarball_type = TarballType::Binary { - arch: payload, - os: it.next().ok_or(ParseError)?, - }; - } - } - } else { - tarball_type = TarballType::Source; - } - - if it.next().is_some() { - return Err(ParseError); - } - - Ok(Tarball { - filename, - tarball_type, - minisig, - archive, - version, - development, - }) - } - - /// Builds the upstream URL for this tarball. - pub fn upstream_url(&self, source: &str) -> String { - if self.development { - format!( - "https://ziglang.org/builds/{}?source={}", - self.filename, source - ) - } else { - format!( - "https://ziglang.org/download/{}/{}?source={}", - self.version, self.filename, source, - ) - } - } -} - -pub struct ZigController { - config: Arc, - storage: Arc, - upstream: Arc, -} - -impl ZigController { - pub fn new( - config: Arc, - storage: Arc, - upstream: Arc, - ) -> Self { - Self { - config, - storage, - upstream, - } - } - - pub fn router(self: Arc) -> Router { - Router::new() - .route("/zig/{filename}", routing::get(Self::handle)) - .with_state(self) - } - - async fn handle( - extract::State(controller): extract::State>, - extract::Path(filename): extract::Path, - ) -> Result { - let tarball = Tarball::parse(&filename).map_err(|_| http::StatusCode::NOT_FOUND)?; - let url = tarball.upstream_url(controller.config.appname()); - - match controller.storage.get("zig", &filename).await { - Ok(Some(entry)) => { - return Ok(Self::build_response( - http::StatusCode::OK, - entry.file_bytes.0, - )); - } - Ok(None) => {} - Err(err) => { - error!("failed get file from storage: {err}"); - return Err(http::StatusCode::INTERNAL_SERVER_ERROR); - } - } - - let entry = controller - .upstream - .fetch(service_upstream::DownloadRequest { url }) - .await?; - - match controller.storage.put("zig", &filename, &entry.bytes).await { - Ok(()) => {} - Err(_) => { - return Err(http::StatusCode::OK); - } - } - - Ok(Self::build_response(http::StatusCode::OK, entry.bytes)) - } - - fn build_response(status: http::StatusCode, bytes: bytes::Bytes) -> response::Response { - response::Response::builder() - .status(status) - .header(http::header::CONTENT_TYPE, "application/octet-stream") - .body(body::Body::from(bytes)) - .unwrap() - } -} diff --git a/src/index.html b/src/index.html deleted file mode 100644 index 6cfc683..0000000 --- a/src/index.html +++ /dev/null @@ -1,100 +0,0 @@ - - - - - - - - - Earth PKG — tiny & opinionated packages mirror - - - - -

- Earth PKG — tiny & opinionated packages mirror. -

- - -

This site provides a proxy for downloading zig installation files and dependencies. - On the one hand, this reduces the load on the original project's site - and makes your infrastructure more reliable by adding redundancy.

- -

Read more about community mirrors in the blog post. - Information on how to deploy your own mirror is available - in the documentation.

- -

Direct usage:

- -

For simplicity, you can use tools like prantlf/zigup and - mlugg/setup-zig.

- -

- To install manually: -

    -
  1. download zig dist file:
    wget https://pkg.earth/zig/zig-x86_64-linux-0.15.1.tar.xz;
    2. -
  2. download zig minisign file:
    wget https://pkg.earth/zig/zig-x86_64-linux-0.15.1.tar.xz.minisig;
    3. -
  3. check archive integrity:
    minisign -Vm zig-x86_64-linux-0.15.1.tar.xz -P RWSGOq2NVecA2UPNdBUZykf1CCb147pkmdtYxgb3Ti+JO/wCYvhbAb/U;
    4. -
  4. unpack arhive:
    tar -xf "zig-x86_64-linux-0.15.1.tar.xz";
    5. -
  5. check installed zig:
    ./zig-x86_64-linux-0.15.1/zig --version.
    6. -
- - You can take actual minisign public key in download page. -

- -

Privacy policy

- -

This mirror is a non-profit project available on a voluntary basis. The author has no plans to fund it.

- -

Since the mirror is hosted on hardware, we collect access logs to combat bots and brute-force attacks. - The logs are used for security purposes and load planning, are not shared with third parties, - and are deleted after 30 days.

- -

Third-party analytics systems are not used same as client-side trackers.

- - diff --git a/src/main.rs b/src/main.rs index 6d74b09..d2317c8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,10 +1,25 @@ // SPDX-FileCopyrightText: 2026 Nikolay Govorov // SPDX-License-Identifier: AGPL-3.0-or-later +mod backends; +mod config; +mod storage; +mod upstream; + +use std::future::Future; use std::path::PathBuf; +use std::pin::Pin; use std::sync::Arc; +use std::task::{Context, Poll}; -use axum::{Router, http}; +use axum::{ + Router, + body::Body, + extract, + http::{self, Request, Response}, + response::{self, IntoResponse}, + routing, +}; use tower_http::{ request_id, trace::{DefaultOnFailure, TraceLayer}, @@ -12,11 +27,7 @@ use tower_http::{ use tracing::{error, info}; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; -mod controller_web; -mod controller_zig; -mod service_config; -mod service_storage; -mod service_upstream; +use crate::backends::zig::ZigController; static REQUEST_ID_HEADER: http::HeaderName = http::HeaderName::from_static("x-request-id"); @@ -66,14 +77,14 @@ async fn main() { Some(path) => { info!("use config file from {}", path.to_str().unwrap()); - service_config::ConfigService::from_file(&path).unwrap_or_else(|e| { + config::ConfigService::from_file(&path).unwrap_or_else(|e| { error!("{e}"); std::process::exit(1); }) } None => { info!("configuration file path not provided"); - service_config::ConfigService::default() + config::ConfigService::default() } }); config.validate().unwrap_or_else(|e| { @@ -81,21 +92,17 @@ async fn main() { std::process::exit(1); }); - let storage = Arc::new( - service_storage::StorageService::new(config.clone()) - .await - .unwrap(), - ); - let upstream = Arc::new(service_upstream::UpstreamService::new()); + let storage = Arc::new(storage::StorageService::new(config.clone()).await.unwrap()); + let upstream = Arc::new(upstream::UpstreamService::new()); - let web_controller = Arc::new(controller_web::WebController::new()); - let zig_controller = Arc::new(controller_zig::ZigController::new( + let web_controller = Arc::new(WebController::default()); + let zig_controller = Arc::new(ZigController::new( config.clone(), storage.clone(), upstream.clone(), )); - let accept_logger = TraceLayer::new_for_http() + let trace_layer = TraceLayer::new_for_http() .make_span_with(|req: &http::Request<_>| { let request_id = req .headers() @@ -105,14 +112,15 @@ async fn main() { tracing::info_span!("http_request", request_id = %request_id) }) - .on_request(log_request) - .on_response(log_response) + .on_request(()) + .on_response(()) .on_failure(DefaultOnFailure::new().level(tracing::Level::ERROR)); let app = Router::new() .merge(web_controller.router()) .merge(zig_controller.router()) - .layer(accept_logger) + .layer(LoggingLayer) + .layer(trace_layer) .layer(request_id::PropagateRequestIdLayer::new( REQUEST_ID_HEADER.clone(), )) @@ -129,31 +137,68 @@ async fn main() { axum::serve(listener, app).await.unwrap(); } -fn log_request(req: &http::Request, _span: &tracing::Span) { - let headers = req.headers(); +#[derive(Clone)] +pub struct LoggingLayer; + +impl tower::Layer for LoggingLayer { + type Service = LoggingService; + + fn layer(&self, inner: S) -> Self::Service { + LoggingService { inner } + } +} + +#[derive(Clone)] +pub struct LoggingService { + inner: S, +} + +impl tower::Service> for LoggingService +where + S: tower::Service, Response = Response> + Clone + Send + 'static, + S::Future: Send, +{ + type Response = S::Response; + type Error = S::Error; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.inner.poll_ready(cx) + } + + fn call(&mut self, req: Request) -> Self::Future { + Box::pin(log_request(self.inner.clone(), req)) + } +} + +async fn log_request(mut inner: S, req: Request) -> Result, S::Error> +where + S: tower::Service, Response = Response>, +{ + let start = std::time::Instant::now(); + + let method = req.method().clone(); + let uri = req.uri().clone(); + let headers = req.headers(); let client_ip = headers .get("x-forwarded-for") .or_else(|| headers.get("x-real-ip")) - .and_then(|v| v.to_str().ok()); + .and_then(|v| v.to_str().ok()) + .map(String::from); let user_agent = headers .get(http::header::USER_AGENT) - .and_then(|v| v.to_str().ok()); + .and_then(|v| v.to_str().ok()) + .map(String::from); let referer = headers .get(http::header::REFERER) - .and_then(|v| v.to_str().ok()); + .and_then(|v| v.to_str().ok()) + .map(String::from); - info!( - method = %req.method(), - uri = %req.uri(), - client_ip, - user_agent, - referer, - "request started" - ); -} + let res = inner.call(req).await?; -fn log_response(res: &http::Response, latency: std::time::Duration, _span: &tracing::Span) { + let latency = start.elapsed(); + let status = res.status(); let headers = res.headers(); let content_length = headers .get(http::header::CONTENT_LENGTH) @@ -163,10 +208,161 @@ fn log_response(res: &http::Response, latency: std::time::Duration, _span: .and_then(|v| v.to_str().ok()); info!( - status = %res.status(), + method = %method, + uri = %uri, + status = %status, latency = ?latency, + client_ip = client_ip.as_deref(), + user_agent = user_agent.as_deref(), + referer = referer.as_deref(), content_length, content_type, - "request finished" + "request" ); + + Ok(res) +} + +/// Handles html pages rendering and static files +pub struct WebController { + jinja: minijinja::Environment<'static>, +} + +impl Default for WebController { + fn default() -> Self { + let mut jinja = minijinja::Environment::new(); + jinja.add_template("index", HTML).unwrap(); + + Self { jinja } + } } + +impl WebController { + pub fn router(self: Arc) -> Router { + Router::new() + .route("/index.css", routing::get(Self::styles)) + .route("/", routing::get(Self::index)) + .with_state(self) + } + + async fn index( + extract::State(controller): extract::State>, + ) -> Result { + let template = controller.jinja.get_template("index").unwrap(); + let rendered = template.render(minijinja::context! {}).unwrap(); + + let mut response = response::Html(rendered).into_response(); + response.headers_mut().insert( + http::header::CONTENT_SECURITY_POLICY, + http::HeaderValue::from_static( + "img-src 'self'; base-uri 'none'; font-src 'self'; style-src 'self'; script-src 'self'; object-src 'none'; default-src 'self'; frame-ancestors 'none'", + ), + ); + + Ok(response) + } + + async fn styles() -> Result, http::StatusCode> { + Ok(axum_extra::response::Css(CSS)) + } +} + +const HTML: &str = r#" + + + + + + + + Earth PKG — tiny & opinionated packages mirror + + + + +

Earth PKG — tiny & opinionated packages mirror.

+ +
+

This site provides a proxy for downloading zig installation files and dependencies. + On the one hand, this reduces the load on the original project's site + and makes your infrastructure more reliable by adding redundancy.

+ +

Read more about community mirrors in the blog post. + Information on how to deploy your own mirror is available + in the documentation.

+ +

Direct usage:

+ +

For simplicity, you can use tools like prantlf/zigup and + mlugg/setup-zig.

+ +

+ To install manually: +

    +
  1. download zig dist file:
    wget https://pkg.earth/zig/zig-x86_64-linux-0.15.1.tar.xz;
    2. +
  2. download zig minisign file:
    wget https://pkg.earth/zig/zig-x86_64-linux-0.15.1.tar.xz.minisig;
    3. +
  3. check archive integrity:
    minisign -Vm zig-x86_64-linux-0.15.1.tar.xz -P RWSGOq2NVecA2UPNdBUZykf1CCb147pkmdtYxgb3Ti+JO/wCYvhbAb/U;
    4. +
  4. unpack archive:
    tar -xf "zig-x86_64-linux-0.15.1.tar.xz";
    5. +
  5. check installed zig:
    ./zig-x86_64-linux-0.15.1/zig --version.
    6. +
+ + You can take actual minisign public key in download page. +

+ +

Privacy policy

+ +

This mirror is a non-profit project available on a voluntary basis. The author has no plans to fund it.

+ +

Since the mirror is hosted on hardware, we collect access logs to combat bots and brute-force attacks. + The logs are used for security purposes and load planning, are not shared with third parties, + and are deleted after 30 days.

+ +

Third-party analytics systems are not used, same as client-side trackers.

+
+ + +"#; + +const CSS: &str = " +:root { + font-size: 1.125rem; + line-height: 1.4; + font-family: + 'Alegreya Sans', -apple-system, BlinkMacSystemFont, + 'Segoe UI', Roboto, Oxygen, Ubuntu, Cantarell, + 'Open Sans', 'Helvetica Neue', sans-serif; +} + +html { + margin: 0; + padding: 0; +} + +body { + margin: 0 auto; + padding: 1.5em 2em; + max-width: 680px; +} + +code { + font-size: .75rem; +} + +h1, h2, h3, h4, h5, h6 { + font-weight: 700; + line-height: 1.2; + margin: 0; +} + +h1 { + font-size: 2.75rem; +} + +h1:first-child { + margin-top: 0; +} + +th { + text-align: start; +} +"; diff --git a/src/service_config.rs b/src/service_config.rs deleted file mode 100644 index dd55bfc..0000000 --- a/src/service_config.rs +++ /dev/null @@ -1,93 +0,0 @@ -// SPDX-FileCopyrightText: 2026 Nikolay Govorov -// SPDX-License-Identifier: AGPL-3.0-or-later - -use std::fs; -use std::path::{Path, PathBuf}; - -use serde::Deserialize; -use thiserror::Error; - -#[derive(Debug, Error)] -pub enum ConfigError { - #[error("failed to read config file: {0}")] - Io(#[from] std::io::Error), - - #[error("failed to parse config file: {0}")] - Parse(#[from] toml::de::Error), - - #[error("appname '{0}' contains invalid characters (only a-z, A-Z, 0-9, -, _ allowed)")] - InvalidAppname(String), - - #[error("dirname '{0}' does not exist")] - DirNotFound(PathBuf), - - #[error("dirname '{0}' is not a directory")] - NotADirectory(PathBuf), - - #[error("dirname '{0}' is not writable: {1}")] - NotWritable(PathBuf, std::io::Error), -} - -#[derive(Debug, Deserialize)] -#[serde(default)] -pub struct ConfigService { - listen: String, - appname: String, - dirname: PathBuf, -} - -impl Default for ConfigService { - fn default() -> Self { - Self { - listen: "0.0.0.0:3000".to_string(), - appname: "zorian".to_string(), - dirname: PathBuf::from("./.zorian-state"), - } - } -} - -impl ConfigService { - pub fn from_file(path: &Path) -> Result { - let content = fs::read_to_string(path)?; - let config: Self = toml::from_str(&content)?; - Ok(config) - } - - pub fn validate(&self) -> Result<(), ConfigError> { - let mut chars = self.appname.chars(); - if !chars.all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') { - return Err(ConfigError::InvalidAppname(self.appname.clone())); - } - - let metadata = fs::metadata(&self.dirname).map_err(|e| { - if e.kind() == std::io::ErrorKind::NotFound { - ConfigError::DirNotFound(self.dirname.clone()) - } else { - ConfigError::Io(e) - } - })?; - - if !metadata.is_dir() { - return Err(ConfigError::NotADirectory(self.dirname.clone())); - } - - let testfile = self.dirname.join(".health"); - fs::write(&testfile, std::process::id().to_string()) - .map_err(|e| ConfigError::NotWritable(self.dirname.clone(), e))?; - fs::remove_file(&testfile)?; - - Ok(()) - } - - pub fn appname(&self) -> &str { - &self.appname - } - - pub fn listen(&self) -> &str { - &self.listen - } - - pub fn dirname(&self) -> &Path { - &self.dirname - } -} diff --git a/src/service_storage.rs b/src/service_storage.rs deleted file mode 100644 index f1f04b2..0000000 --- a/src/service_storage.rs +++ /dev/null @@ -1,497 +0,0 @@ -// SPDX-FileCopyrightText: 2026 Nikolay Govorov -// SPDX-License-Identifier: AGPL-3.0-or-later - -// This class stores uploaded files and associated metadata. -// Files are immutable but can be deleted. -// -// A unique file is defined by a `scope` and a `filename`. -// The filename can be any valid UTF-8 string up to 1KB -// in size and does not have to be a valid filesystem name. -// -// The list of files and their metadata are stored in a single -// SQLite database table. Small files are stored in a BLOB column -// in SQLite; for larger files, a checksum is stored in the database, -// and the file itself is stored on disk. -// -// On disk new files are written in two steps: -// - the file is written to a temporary file in the same directory; -// - a transaction is opened and a new file entry is created -// - temp file is renamed to the final filename -// - the transaction is committed. -// -// This scheme guarantees that if a record exists in the table, the file is written. -// However if the server crashes after rename but before commit, an orphan file will remain on the disk. -// To combat this, when the server starts, we check that there is an entry -// in the database for each file on the disk, and we also delete all temporary (non-renamed) files. - -use std::path::{Path, PathBuf}; -use std::sync; - -use bytes::Bytes; -use crc32fast::Hasher as Crc32Hasher; -use sqlx::encode::{Encode, IsNull}; -use sqlx::error::BoxDynError; -use sqlx::{FromRow, Pool, Sqlite, query, query_as, query_scalar, sqlite}; -use thiserror::Error; -use tokio::fs; -use tokio::io::AsyncWriteExt; -use tracing::{debug, instrument, warn}; -use uuid::Uuid; - -use super::service_config::ConfigService; - -const SQLITE_POOL_SIZE: u32 = 16; -const INLINE_THRESHOLD: usize = 256 * 1024; // 256 KB - -// You cannot change this ID, this will desynchronize the records in the database and the files on the disk. -const UUID_ROOT_NAMESPACE: Uuid = Uuid::from_bytes([ - 0x8b, 0x06, 0x3c, 0x4c, 0x6b, 0x5c, 0x4a, 0x8b, 0x92, 0x8f, 0x75, 0x8b, 0x0e, 0x63, 0xc3, 0x5d, -]); - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct Id(pub Uuid); -impl std::ops::Deref for Id { - type Target = Uuid; - fn deref(&self) -> &Self::Target { - &self.0 - } -} -impl sqlx::Type for Id { - fn type_info() -> sqlite::SqliteTypeInfo { - as sqlx::Type>::type_info() - } - - fn compatible(ty: &sqlite::SqliteTypeInfo) -> bool { - as sqlx::Type>::compatible(ty) - } -} -impl<'r> sqlx::decode::Decode<'r, Sqlite> for Id { - fn decode( - value: sqlite::SqliteValueRef<'r>, - ) -> Result> { - let value: Vec = as sqlx::decode::Decode>::decode(value)?; - let uuid = Uuid::from_slice(&value)?; - Ok(Id(uuid)) - } -} -impl<'q> Encode<'q, Sqlite> for Id { - fn encode_by_ref( - &self, - buf: &mut ::ArgumentBuffer<'q>, - ) -> Result { - let value = self.0.as_bytes().to_vec(); - as Encode>::encode(value, buf) - } -} - -#[derive(Debug, Clone)] -pub struct Blob(pub Bytes); -impl std::ops::Deref for Blob { - type Target = Bytes; - fn deref(&self) -> &Self::Target { - &self.0 - } -} -impl sqlx::Type for Blob { - fn type_info() -> sqlite::SqliteTypeInfo { - as sqlx::Type>::type_info() // BLOB - } - - fn compatible(ty: &sqlite::SqliteTypeInfo) -> bool { - as sqlx::Type>::compatible(ty) - } -} -impl<'r> sqlx::decode::Decode<'r, Sqlite> for Blob { - fn decode( - value: sqlite::SqliteValueRef<'r>, - ) -> Result> { - let slice: &'r [u8] = <&'r [u8] as sqlx::decode::Decode>::decode(value)?; - Ok(Blob(Bytes::copy_from_slice(slice))) - } -} - -#[derive(Debug, Error)] -pub enum StorageError { - #[error("file already exists: {0}/{1}")] - AlreadyExists(String, String), - - #[error("blob integrity check failed")] - IntegrityError, - - #[error("blob file missing on disk: {0}")] - BlobNotFound(PathBuf), - - #[error("failed to connect index db: {0}")] - DbError(#[from] sqlx::Error), - - #[error("io error: {0}")] - IoError(#[from] std::io::Error), -} - -#[allow(unused)] -#[derive(Debug, Clone, FromRow)] -pub struct File { - pub id: Id, - pub scope: String, - pub created_at: chrono::DateTime, - - pub file_name: String, - pub file_size: i64, - pub file_bytes: Blob, - pub inlined: bool, -} - -impl File { - fn uuid(scope: &str, file: &str) -> Id { - let scope_ns = Uuid::new_v5(&UUID_ROOT_NAMESPACE, scope.as_bytes()); - Id(Uuid::new_v5(&scope_ns, file.as_bytes())) - } - - fn hash(scope: &str, file: &str, blob: &[u8]) -> Vec { - let mut hasher = Crc32Hasher::new(); - - hasher.update(b"/"); - hasher.update(scope.as_bytes()); - - hasher.update(b"/"); - hasher.update(file.as_bytes()); - - hasher.update(b"/"); - hasher.update(blob); - hasher.update(b"\0"); - - hasher.finalize().to_be_bytes().to_vec() - } -} - -struct FileSystem { - objects: PathBuf, - database: PathBuf, -} - -impl FileSystem { - fn new(root: &Path) -> Self { - Self { - objects: root.join("objects"), - database: root.join("index.sqlite"), - } - } - - fn database(&self) -> &Path { - &self.database - } - - fn objects_root(&self) -> &Path { - &self.objects - } - - fn object(&self, scope: &str, file: &str) -> PathBuf { - let id = File::uuid(scope, file); - let hash_hex = hex::encode(id.0.as_bytes()); - self.objects_root() - .join(&hash_hex[0..2]) - .join(&hash_hex[2..4]) - .join(&hash_hex) - } - - async fn objects_walk(&self, mut f: F) -> Result<(), StorageError> - where - F: FnMut(PathBuf) -> Fut, - Fut: std::future::Future>, - { - if !self.objects.exists() { - return Ok(()); - } - - let mut stack = vec![self.objects.to_path_buf()]; - while let Some(dir) = stack.pop() { - let mut entries = fs::read_dir(&dir).await?; - while let Some(entry) = entries.next_entry().await? { - let path = entry.path(); - let file_type = entry.file_type().await?; - - if file_type.is_dir() { - stack.push(path); - } else if file_type.is_file() { - f(path).await?; - } - } - } - - Ok(()) - } -} - -pub struct StorageService { - blobfs: FileSystem, - sqlite: Pool, -} - -impl StorageService { - pub async fn new(config: sync::Arc) -> Result { - let blobfs = FileSystem::new(config.dirname()); - - let connection = format!("sqlite:{}?mode=rwc", blobfs.database().to_str().unwrap()); - let sqlite: Pool = sqlite::SqlitePoolOptions::new() - .max_connections(SQLITE_POOL_SIZE) - .after_connect(|conn, _meta| { - Box::pin(async move { - // Connection-specific PRAGMAs (must be set on each connection) - sqlx::query("PRAGMA foreign_keys = ON;") - .execute(&mut *conn) - .await?; - sqlx::query("PRAGMA busy_timeout = 5000;") - .execute(&mut *conn) - .await?; - Ok(()) - }) - }) - .connect(&connection) - .await?; - - // WAL mode is database-wide and persists, only needs to be set once - query("PRAGMA journal_mode = WAL;").execute(&sqlite).await?; - - let storage = Self { sqlite, blobfs }; - storage.migrations().await?; - storage.doctor().await?; - - Ok(storage) - } - - /// Synchronously traverses the tree and removes temporary files. - /// Must run before the application starts. - async fn doctor(&self) -> Result<(), StorageError> { - let result = self.blobfs.objects_walk(|path| async move { - let name = match path.file_name().and_then(|name| name.to_str()) { - Some(name) => name.to_string(), - None => { - warn!(path = %path.display(), "cleanup: invalid blob name"); - let _ = fs::remove_file(path).await; - return Ok(()); - } - }; - - if name.ends_with(".part") { - warn!(path = %path.display(), "cleanup: removing temp file"); - let _ = fs::remove_file(path).await; - return Ok(()); - } - - if name.len() != 32 { - warn!(path = %path.display(), "cleanup: invalid blob name length"); - let _ = fs::remove_file(path).await; - return Ok(()); - } - - let id_bytes = match hex::decode(&name) { - Ok(bytes) => bytes, - Err(_) => { - warn!(path = %path.display(), "cleanup: invalid blob name hex"); - let _ = fs::remove_file(path).await; - return Ok(()); - } - }; - - let id = match Uuid::from_slice(&id_bytes) { - Ok(uuid) => Id(uuid), - Err(_) => { - warn!(path = %path.display(), "cleanup: invalid blob uuid"); - let _ = fs::remove_file(path).await; - return Ok(()); - } - }; - - let exists: Option = query_scalar("SELECT 1 FROM datafiles WHERE id = ?1") - .bind(id) - .fetch_optional(&self.sqlite) - .await?; - - if exists.is_none() { - warn!(path = %path.display(), "cleanup: removing orphan blob"); - let _ = fs::remove_file(path).await; - } - - Ok(()) - }); - result.await?; - - Ok(()) - } - - async fn migrations(&self) -> Result<(), StorageError> { - query( - " - CREATE TABLE IF NOT EXISTS datafiles( - id BLOB PRIMARY KEY CHECK (length(id) = 16), - scope TEXT NOT NULL, - created_at TEXT DEFAULT (datetime('now')), - file_name TEXT NOT NULL, - file_size INTEGER NOT NULL, - file_bytes BLOB, - inlined INTEGER NOT NULL, - - UNIQUE (scope, file_name) - ) STRICT; - ", - ) - .execute(&self.sqlite) - .await?; - - Ok(()) - } - - #[instrument(skip(self))] - pub async fn get(&self, scope: &str, filename: &str) -> Result, StorageError> { - let file: Option = - query_as("SELECT * FROM datafiles WHERE scope = ?1 AND file_name = ?2") - .bind(scope) - .bind(filename) - .fetch_optional(&self.sqlite) - .await?; - - match file { - None => { - debug!("get: file not found"); - Ok(None) - } - Some(mut file) if !file.inlined => { - let obj = self.blobfs.object(scope, filename); - let bytes = match fs::read(&obj).await { - Ok(b) => Bytes::from(b), - Err(e) if e.kind() == std::io::ErrorKind::NotFound => { - return Err(StorageError::BlobNotFound(obj)); - } - Err(e) => return Err(e.into()), - }; - - let hash = File::hash(scope, filename, &bytes); - if hash != file.file_bytes.to_vec() { - return Err(StorageError::IntegrityError); - } - - file.file_bytes = Blob(bytes); - debug!(size = file.file_size, "get: loaded from disk"); - Ok(Some(file)) - } - Some(file) => { - debug!(size = file.file_size, "get: loaded inline"); - Ok(Some(file)) - } - } - } - - #[instrument(skip(self, bytes), fields(size = bytes.len()))] - pub async fn put( - &self, - scope: &str, - filename: &str, - bytes: &Bytes, - ) -> Result<(), StorageError> { - let inlined = bytes.len() <= INLINE_THRESHOLD; - - let obj = self.blobfs.object(scope, filename); - let tmp = obj.with_extension(format!("{}.part", Uuid::new_v4())); - - // write temp file - if !inlined { - if let Some(parent) = obj.parent() { - fs::create_dir_all(parent).await?; - } - - let mut file = fs::OpenOptions::new() - .write(true) - .create_new(true) - .open(&tmp) - .await?; - - file.write_all(bytes).await?; - file.sync_all().await?; - } - - // bytes for small files or hash for large ones - let payload: Vec = if inlined { - bytes.to_vec() - } else { - File::hash(scope, filename, bytes) - }; - - let result = async { - let mut tx = self.sqlite.begin().await?; - let id = File::uuid(scope, filename); - - let result = query( - " - INSERT INTO datafiles ( - id, scope, file_name, file_size, file_bytes, inlined - ) VALUES ( - ?1, ?2, ?3, ?4, ?5, ?6 - ); - ", - ) - .bind(id) - .bind(scope) - .bind(filename) - .bind(bytes.len() as i64) - .bind(&payload) - .bind(inlined) - .execute(tx.as_mut()) - .await; - - match result { - Ok(_) => { - if !inlined { - fs::rename(&tmp, &obj).await?; - if let Some(parent) = obj.parent() { - let dir = fs::File::open(parent).await?; - dir.sync_all().await?; - } - } - - tx.commit().await?; - debug!("put: a new file has been commited"); - Ok(()) - } - Err(sqlx::Error::Database(ref db_err)) if db_err.is_unique_violation() => { - let existing: Option = - query_as("SELECT * FROM datafiles WHERE scope = ?1 AND file_name = ?2") - .bind(scope) - .bind(filename) - .fetch_optional(tx.as_mut()) - .await?; - - match existing { - Some(file) if file.file_bytes.to_vec() == payload => { - if !inlined { - let _ = fs::remove_file(&tmp).await; - tx.rollback().await?; - } - - debug!("put: identical file already exists"); - return Ok(()); - } - Some(_) => { - warn!("put: file already exists with different content"); - } - None => { - // unique_violation but row doesn't exist - shouldn't happen - warn!("corrupted! 'is_unique_violation' received, but data cannot be selected"); - } - } - - Err(StorageError::AlreadyExists( - scope.to_string(), - filename.to_string(), - )) - } - Err(e) => Err(e.into()), - } - }.await; - - if result.is_err() && !inlined { - let _ = fs::remove_file(&tmp).await; - } - - result - } -} diff --git a/src/service_upstream.rs b/src/service_upstream.rs deleted file mode 100644 index da25a7d..0000000 --- a/src/service_upstream.rs +++ /dev/null @@ -1,60 +0,0 @@ -// SPDX-FileCopyrightText: 2026 Nikolay Govorov -// SPDX-License-Identifier: AGPL-3.0-or-later - -use bytes::Bytes; -use http_body_util::{BodyExt, Empty}; -use hyper::{Request, http}; -use hyper_tls::HttpsConnector; -use hyper_util::client::legacy::{Client, connect::HttpConnector}; -use hyper_util::rt::TokioExecutor; - -#[derive(Clone)] -pub struct DownloadRequest { - pub url: String, -} - -#[derive(Clone)] -pub struct File { - pub bytes: Bytes, -} - -pub struct UpstreamService { - client: Client, Empty>, -} - -impl UpstreamService { - pub fn new() -> Self { - let https = HttpsConnector::new(); - let client = Client::builder(TokioExecutor::new()).build(https); - Self { client } - } - - pub async fn fetch(&self, request: DownloadRequest) -> Result { - let request = Request::builder() - .method(http::Method::GET) - .uri(&request.url) - .header(http::header::USER_AGENT, "zorian/0.1") - .body(Empty::::new()) - .unwrap(); - - let response = self - .client - .request(request) - .await - .map_err(|_| http::StatusCode::GATEWAY_TIMEOUT)?; - - let (parts, body) = response.into_parts(); - let status = parts.status; - if !status.is_success() { - return Err(status); - } - - let bytes = body - .collect() - .await - .map_err(|_| http::StatusCode::GATEWAY_TIMEOUT)? - .to_bytes(); - - Ok(File { bytes }) - } -} diff --git a/src/storage.rs b/src/storage.rs new file mode 100644 index 0000000..960afc5 --- /dev/null +++ b/src/storage.rs @@ -0,0 +1,497 @@ +// SPDX-FileCopyrightText: 2026 Nikolay Govorov +// SPDX-License-Identifier: AGPL-3.0-or-later + +// This class stores uploaded files and associated metadata. +// Files are immutable but can be deleted. +// +// A unique file is defined by a `scope` and a `filename`. +// The filename can be any valid UTF-8 string up to 1KB +// in size and does not have to be a valid filesystem name. +// +// The list of files and their metadata are stored in a single +// SQLite database table. Small files are stored in a BLOB column +// in SQLite; for larger files, a checksum is stored in the database, +// and the file itself is stored on disk. +// +// On disk new files are written in two steps: +// - the file is written to a temporary file in the same directory; +// - a transaction is opened and a new file entry is created +// - temp file is renamed to the final filename +// - the transaction is committed. +// +// This scheme guarantees that if a record exists in the table, the file is written. +// However if the server crashes after rename but before commit, an orphan file will remain on the disk. +// To combat this, when the server starts, we check that there is an entry +// in the database for each file on the disk, and we also delete all temporary (non-renamed) files. + +use std::path::{Path, PathBuf}; +use std::sync; + +use bytes::Bytes; +use crc32fast::Hasher as Crc32Hasher; +use sqlx::encode::{Encode, IsNull}; +use sqlx::error::BoxDynError; +use sqlx::{FromRow, Pool, Sqlite, query, query_as, query_scalar, sqlite}; +use thiserror::Error; +use tokio::fs; +use tokio::io::AsyncWriteExt; +use tracing::{debug, instrument, warn}; +use uuid::Uuid; + +use super::config::ConfigService; + +const SQLITE_POOL_SIZE: u32 = 16; +const INLINE_THRESHOLD: usize = 256 * 1024; // 256 KB + +// You cannot change this ID, this will desynchronize the records in the database and the files on the disk. +const UUID_ROOT_NAMESPACE: Uuid = Uuid::from_bytes([ + 0x8b, 0x06, 0x3c, 0x4c, 0x6b, 0x5c, 0x4a, 0x8b, 0x92, 0x8f, 0x75, 0x8b, 0x0e, 0x63, 0xc3, 0x5d, +]); + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct Id(pub Uuid); +impl std::ops::Deref for Id { + type Target = Uuid; + fn deref(&self) -> &Self::Target { + &self.0 + } +} +impl sqlx::Type for Id { + fn type_info() -> sqlite::SqliteTypeInfo { + as sqlx::Type>::type_info() + } + + fn compatible(ty: &sqlite::SqliteTypeInfo) -> bool { + as sqlx::Type>::compatible(ty) + } +} +impl<'r> sqlx::decode::Decode<'r, Sqlite> for Id { + fn decode( + value: sqlite::SqliteValueRef<'r>, + ) -> Result> { + let value: Vec = as sqlx::decode::Decode>::decode(value)?; + let uuid = Uuid::from_slice(&value)?; + Ok(Id(uuid)) + } +} +impl<'q> Encode<'q, Sqlite> for Id { + fn encode_by_ref( + &self, + buf: &mut ::ArgumentBuffer<'q>, + ) -> Result { + let value = self.0.as_bytes().to_vec(); + as Encode>::encode(value, buf) + } +} + +#[derive(Debug, Clone)] +pub struct Blob(pub Bytes); +impl std::ops::Deref for Blob { + type Target = Bytes; + fn deref(&self) -> &Self::Target { + &self.0 + } +} +impl sqlx::Type for Blob { + fn type_info() -> sqlite::SqliteTypeInfo { + as sqlx::Type>::type_info() // BLOB + } + + fn compatible(ty: &sqlite::SqliteTypeInfo) -> bool { + as sqlx::Type>::compatible(ty) + } +} +impl<'r> sqlx::decode::Decode<'r, Sqlite> for Blob { + fn decode( + value: sqlite::SqliteValueRef<'r>, + ) -> Result> { + let slice: &'r [u8] = <&'r [u8] as sqlx::decode::Decode>::decode(value)?; + Ok(Blob(Bytes::copy_from_slice(slice))) + } +} + +#[derive(Debug, Error)] +pub enum StorageError { + #[error("file already exists: {0}/{1}")] + AlreadyExists(String, String), + + #[error("blob integrity check failed")] + IntegrityError, + + #[error("blob file missing on disk: {0}")] + BlobNotFound(PathBuf), + + #[error("failed to connect index db: {0}")] + DbError(#[from] sqlx::Error), + + #[error("io error: {0}")] + IoError(#[from] std::io::Error), +} + +#[allow(unused)] +#[derive(Debug, Clone, FromRow)] +pub struct File { + pub id: Id, + pub scope: String, + pub created_at: chrono::DateTime, + + pub file_name: String, + pub file_size: i64, + pub file_bytes: Blob, + pub inlined: bool, +} + +impl File { + fn uuid(scope: &str, file: &str) -> Id { + let scope_ns = Uuid::new_v5(&UUID_ROOT_NAMESPACE, scope.as_bytes()); + Id(Uuid::new_v5(&scope_ns, file.as_bytes())) + } + + fn hash(scope: &str, file: &str, blob: &[u8]) -> Vec { + let mut hasher = Crc32Hasher::new(); + + hasher.update(b"/"); + hasher.update(scope.as_bytes()); + + hasher.update(b"/"); + hasher.update(file.as_bytes()); + + hasher.update(b"/"); + hasher.update(blob); + hasher.update(b"\0"); + + hasher.finalize().to_be_bytes().to_vec() + } +} + +struct FileSystem { + objects: PathBuf, + database: PathBuf, +} + +impl FileSystem { + fn new(root: &Path) -> Self { + Self { + objects: root.join("objects"), + database: root.join("index.sqlite"), + } + } + + fn database(&self) -> &Path { + &self.database + } + + fn objects_root(&self) -> &Path { + &self.objects + } + + fn object(&self, scope: &str, file: &str) -> PathBuf { + let id = File::uuid(scope, file); + let hash_hex = hex::encode(id.0.as_bytes()); + self.objects_root() + .join(&hash_hex[0..2]) + .join(&hash_hex[2..4]) + .join(&hash_hex) + } + + async fn objects_walk(&self, mut f: F) -> Result<(), StorageError> + where + F: FnMut(PathBuf) -> Fut, + Fut: std::future::Future>, + { + if !self.objects.exists() { + return Ok(()); + } + + let mut stack = vec![self.objects.to_path_buf()]; + while let Some(dir) = stack.pop() { + let mut entries = fs::read_dir(&dir).await?; + while let Some(entry) = entries.next_entry().await? { + let path = entry.path(); + let file_type = entry.file_type().await?; + + if file_type.is_dir() { + stack.push(path); + } else if file_type.is_file() { + f(path).await?; + } + } + } + + Ok(()) + } +} + +pub struct StorageService { + blobfs: FileSystem, + sqlite: Pool, +} + +impl StorageService { + pub async fn new(config: sync::Arc) -> Result { + let blobfs = FileSystem::new(config.dirname()); + + let connection = format!("sqlite:{}?mode=rwc", blobfs.database().to_str().unwrap()); + let sqlite: Pool = sqlite::SqlitePoolOptions::new() + .max_connections(SQLITE_POOL_SIZE) + .after_connect(|conn, _meta| { + Box::pin(async move { + // Connection-specific PRAGMAs (must be set on each connection) + sqlx::query("PRAGMA foreign_keys = ON;") + .execute(&mut *conn) + .await?; + sqlx::query("PRAGMA busy_timeout = 5000;") + .execute(&mut *conn) + .await?; + Ok(()) + }) + }) + .connect(&connection) + .await?; + + // WAL mode is database-wide and persists, only needs to be set once + query("PRAGMA journal_mode = WAL;").execute(&sqlite).await?; + + let storage = Self { sqlite, blobfs }; + storage.migrations().await?; + storage.doctor().await?; + + Ok(storage) + } + + /// Synchronously traverses the tree and removes temporary files. + /// Must run before the application starts. + async fn doctor(&self) -> Result<(), StorageError> { + let result = self.blobfs.objects_walk(|path| async move { + let name = match path.file_name().and_then(|name| name.to_str()) { + Some(name) => name.to_string(), + None => { + warn!(path = %path.display(), "cleanup: invalid blob name"); + let _ = fs::remove_file(path).await; + return Ok(()); + } + }; + + if name.ends_with(".part") { + warn!(path = %path.display(), "cleanup: removing temp file"); + let _ = fs::remove_file(path).await; + return Ok(()); + } + + if name.len() != 32 { + warn!(path = %path.display(), "cleanup: invalid blob name length"); + let _ = fs::remove_file(path).await; + return Ok(()); + } + + let id_bytes = match hex::decode(&name) { + Ok(bytes) => bytes, + Err(_) => { + warn!(path = %path.display(), "cleanup: invalid blob name hex"); + let _ = fs::remove_file(path).await; + return Ok(()); + } + }; + + let id = match Uuid::from_slice(&id_bytes) { + Ok(uuid) => Id(uuid), + Err(_) => { + warn!(path = %path.display(), "cleanup: invalid blob uuid"); + let _ = fs::remove_file(path).await; + return Ok(()); + } + }; + + let exists: Option = query_scalar("SELECT 1 FROM datafiles WHERE id = ?1") + .bind(id) + .fetch_optional(&self.sqlite) + .await?; + + if exists.is_none() { + warn!(path = %path.display(), "cleanup: removing orphan blob"); + let _ = fs::remove_file(path).await; + } + + Ok(()) + }); + result.await?; + + Ok(()) + } + + async fn migrations(&self) -> Result<(), StorageError> { + query( + " + CREATE TABLE IF NOT EXISTS datafiles( + id BLOB PRIMARY KEY CHECK (length(id) = 16), + scope TEXT NOT NULL, + created_at TEXT DEFAULT (datetime('now')), + file_name TEXT NOT NULL, + file_size INTEGER NOT NULL, + file_bytes BLOB, + inlined INTEGER NOT NULL, + + UNIQUE (scope, file_name) + ) STRICT; + ", + ) + .execute(&self.sqlite) + .await?; + + Ok(()) + } + + #[instrument(skip(self))] + pub async fn get(&self, scope: &str, filename: &str) -> Result, StorageError> { + let file: Option = + query_as("SELECT * FROM datafiles WHERE scope = ?1 AND file_name = ?2") + .bind(scope) + .bind(filename) + .fetch_optional(&self.sqlite) + .await?; + + match file { + None => { + debug!("get: file not found"); + Ok(None) + } + Some(mut file) if !file.inlined => { + let obj = self.blobfs.object(scope, filename); + let bytes = match fs::read(&obj).await { + Ok(b) => Bytes::from(b), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + return Err(StorageError::BlobNotFound(obj)); + } + Err(e) => return Err(e.into()), + }; + + let hash = File::hash(scope, filename, &bytes); + if hash != file.file_bytes.to_vec() { + return Err(StorageError::IntegrityError); + } + + file.file_bytes = Blob(bytes); + debug!(size = file.file_size, "get: loaded from disk"); + Ok(Some(file)) + } + Some(file) => { + debug!(size = file.file_size, "get: loaded inline"); + Ok(Some(file)) + } + } + } + + #[instrument(skip(self, bytes), fields(size = bytes.len()))] + pub async fn put( + &self, + scope: &str, + filename: &str, + bytes: &Bytes, + ) -> Result<(), StorageError> { + let inlined = bytes.len() <= INLINE_THRESHOLD; + + let obj = self.blobfs.object(scope, filename); + let tmp = obj.with_extension(format!("{}.part", Uuid::new_v4())); + + // write temp file + if !inlined { + if let Some(parent) = obj.parent() { + fs::create_dir_all(parent).await?; + } + + let mut file = fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&tmp) + .await?; + + file.write_all(bytes).await?; + file.sync_all().await?; + } + + // bytes for small files or hash for large ones + let payload: Vec = if inlined { + bytes.to_vec() + } else { + File::hash(scope, filename, bytes) + }; + + let result = async { + let mut tx = self.sqlite.begin().await?; + let id = File::uuid(scope, filename); + + let result = query( + " + INSERT INTO datafiles ( + id, scope, file_name, file_size, file_bytes, inlined + ) VALUES ( + ?1, ?2, ?3, ?4, ?5, ?6 + ); + ", + ) + .bind(id) + .bind(scope) + .bind(filename) + .bind(bytes.len() as i64) + .bind(&payload) + .bind(inlined) + .execute(tx.as_mut()) + .await; + + match result { + Ok(_) => { + if !inlined { + fs::rename(&tmp, &obj).await?; + if let Some(parent) = obj.parent() { + let dir = fs::File::open(parent).await?; + dir.sync_all().await?; + } + } + + tx.commit().await?; + debug!("put: a new file has been commited"); + Ok(()) + } + Err(sqlx::Error::Database(ref db_err)) if db_err.is_unique_violation() => { + let existing: Option = + query_as("SELECT * FROM datafiles WHERE scope = ?1 AND file_name = ?2") + .bind(scope) + .bind(filename) + .fetch_optional(tx.as_mut()) + .await?; + + match existing { + Some(file) if file.file_bytes.to_vec() == payload => { + if !inlined { + let _ = fs::remove_file(&tmp).await; + tx.rollback().await?; + } + + debug!("put: identical file already exists"); + return Ok(()); + } + Some(_) => { + warn!("put: file already exists with different content"); + } + None => { + // unique_violation but row doesn't exist - shouldn't happen + warn!("corrupted! 'is_unique_violation' received, but data cannot be selected"); + } + } + + Err(StorageError::AlreadyExists( + scope.to_string(), + filename.to_string(), + )) + } + Err(e) => Err(e.into()), + } + }.await; + + if result.is_err() && !inlined { + let _ = fs::remove_file(&tmp).await; + } + + result + } +} diff --git a/src/upstream.rs b/src/upstream.rs new file mode 100644 index 0000000..da25a7d --- /dev/null +++ b/src/upstream.rs @@ -0,0 +1,60 @@ +// SPDX-FileCopyrightText: 2026 Nikolay Govorov +// SPDX-License-Identifier: AGPL-3.0-or-later + +use bytes::Bytes; +use http_body_util::{BodyExt, Empty}; +use hyper::{Request, http}; +use hyper_tls::HttpsConnector; +use hyper_util::client::legacy::{Client, connect::HttpConnector}; +use hyper_util::rt::TokioExecutor; + +#[derive(Clone)] +pub struct DownloadRequest { + pub url: String, +} + +#[derive(Clone)] +pub struct File { + pub bytes: Bytes, +} + +pub struct UpstreamService { + client: Client, Empty>, +} + +impl UpstreamService { + pub fn new() -> Self { + let https = HttpsConnector::new(); + let client = Client::builder(TokioExecutor::new()).build(https); + Self { client } + } + + pub async fn fetch(&self, request: DownloadRequest) -> Result { + let request = Request::builder() + .method(http::Method::GET) + .uri(&request.url) + .header(http::header::USER_AGENT, "zorian/0.1") + .body(Empty::::new()) + .unwrap(); + + let response = self + .client + .request(request) + .await + .map_err(|_| http::StatusCode::GATEWAY_TIMEOUT)?; + + let (parts, body) = response.into_parts(); + let status = parts.status; + if !status.is_success() { + return Err(status); + } + + let bytes = body + .collect() + .await + .map_err(|_| http::StatusCode::GATEWAY_TIMEOUT)? + .to_bytes(); + + Ok(File { bytes }) + } +} -- Gilti