use std::collections::BTreeMap;
use std::fs::File;
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use anyhow::{anyhow, bail};
use decdn_bao_range::{EncodedOutboard, encode_outboard, shard_prefix};
use decdn_common::cli::{OriginArgs, OriginCommand, OriginImportArgs};
use serde::Serialize;
use super::chunker::{ChunkSizes, chunk_file};
use super::manifest::{
BundleEntry, Chunk, b3_hex_str, build_excluder, hash_file_at, serialize_canonical,
validate_relpath, walk_and_collect, write_bundle,
};
pub async fn origin_dispatch(args: &OriginArgs) -> anyhow::Result<()> {
match &args.cmd {
OriginCommand::Import(import_args) => origin_import(import_args).await,
}
}
#[derive(Debug)]
enum ImportTarget {
Fs(PathBuf),
}
fn parse_target(raw: &Path) -> anyhow::Result<ImportTarget> {
if let Some(s) = raw.to_str() {
if s.is_empty() {
bail!("--to needs a directory, e.g. --to /var/lib/decdn/origin");
}
if s.starts_with("s3://") {
bail!(
"--to {s} is not a write target: `origin import` writes a local \
filesystem store only. The fs and S3 object layouts are identical, so \
import to a directory and sync it up:\n \
decdn origin import -i <input> --to <dir>\n \
aws s3 sync <dir> {s}"
);
}
if s.starts_with("http://") || s.starts_with("https://") {
bail!(
"--to {s} is not a write target: an HTTP origin is a read-only static \
server — seed a local directory and serve it from that disk instead"
);
}
if s.starts_with("fs:") {
bail!(
"--to takes a directory path, not an `fs:` URI: pass the path directly, \
e.g. --to /var/lib/decdn/origin"
);
}
}
Ok(ImportTarget::Fs(raw.to_path_buf()))
}
#[derive(Serialize)]
struct ImportReport {
imported: u64,
bytes: u64,
origin: String,
files: BTreeMap<String, String>,
bundle_hash: Option<String>,
moved: bool,
optimized: bool,
chunks_total: u64,
skipped_symlinks: u64,
}
#[expect(
clippy::struct_excessive_bools,
reason = "independent per-invocation toggles resolved from CLI flags, not modelled state"
)]
struct ImportCtx {
base: Option<PathBuf>,
write: bool,
dry_run: bool,
move_source: bool,
force: bool,
bundle_out: Option<PathBuf>,
origin_label: String,
optimize: bool,
sizes: Option<ChunkSizes>,
subfolder: Option<String>,
}
fn resolve_subfolder(raw: Option<&str>) -> anyhow::Result<Option<String>> {
let Some(raw) = raw else { return Ok(None) };
if raw.contains('\\') {
bail!(
"invalid --subfolder {raw:?}: use '/' as the path separator; \
'\\' is not allowed (a POSIX-relative subfolder keeps the \
manifest identical and safe across platforms)"
);
}
let normalized = validate_relpath(Path::new(raw))
.map_err(|e| anyhow!("invalid --subfolder {raw:?}: {e}"))?;
Ok(Some(normalized))
}
fn place_under(prefix: Option<&str>, rel: &str) -> String {
match prefix {
Some(sub) => format!("{sub}/{rel}"),
None => rel.to_string(),
}
}
pub async fn origin_import(args: &OriginImportArgs) -> anyhow::Result<()> {
let target = match (&args.to, args.dry_run) {
(Some(t), _) => Some(parse_target(t)?),
(None, true) => None,
(None, false) => bail!("--to <target> is required unless --dry-run"),
};
if !args.optimize
&& (args.chunk_avg.is_some() || args.chunk_min.is_some() || args.chunk_max.is_some())
{
bail!("--chunk-avg/--chunk-min/--chunk-max require --optimize");
}
let avg = args.chunk_avg.unwrap_or(4 * 1024 * 1024);
let sizes = if args.optimize {
Some(ChunkSizes::resolve(avg, args.chunk_min, args.chunk_max)?)
} else {
None
};
let subfolder = resolve_subfolder(args.subfolder.as_deref())?;
let write = !args.dry_run;
let origin_label = args
.to
.as_ref()
.map_or_else(|| "(dry-run)".to_string(), |p| p.display().to_string());
let input = args.input.clone();
let follow = args.follow_symlinks;
let exclude = args.exclude.clone();
let json = args.json;
let dry_run = args.dry_run;
let ctx = ImportCtx {
base: None,
write,
dry_run,
move_source: args.move_source,
force: args.force,
bundle_out: args.bundle.clone(),
origin_label,
optimize: args.optimize,
sizes,
subfolder,
};
let report = tokio::task::spawn_blocking(move || -> anyhow::Result<ImportReport> {
let mut ctx = ctx;
if ctx.write {
let ImportTarget::Fs(base) =
target.ok_or_else(|| anyhow!("internal: write without a --to target"))?;
std::fs::create_dir_all(&base)
.map_err(|e| anyhow!("create origin dir {}: {e}", base.display()))?;
let base = std::fs::canonicalize(&base)
.map_err(|e| anyhow!("canonicalize origin dir {}: {e}", base.display()))?;
ctx.base = Some(base);
}
let meta = std::fs::symlink_metadata(&input)
.map_err(|e| anyhow!("--input {}: {e}", input.display()))?;
if meta.is_dir() {
import_directory(&ctx, &input, follow, &exclude)
} else if meta.is_file() {
if ctx.optimize {
import_single_optimized(&ctx, &input)
} else {
import_single_plain(&ctx, &input)
}
} else {
bail!(
"--input {} is neither a regular file nor a directory",
input.display()
);
}
})
.await
.map_err(|e| {
let note = if e.is_panic() {
"origin import task panicked"
} else if e.is_cancelled() {
"origin import task cancelled"
} else {
"origin import task failed to join"
};
anyhow::Error::from(e).context(note)
})??;
if dry_run {
let mut stderr = std::io::stderr().lock();
write_import_report(&mut stderr, &report, json)
.map_err(|e| anyhow!("failed to write status report: {e}"))?;
} else {
let mut stdout = std::io::stdout().lock();
write_import_report(&mut stdout, &report, json)
.map_err(|e| anyhow!("failed to write status report: {e}"))?;
}
Ok(())
}
fn import_directory(
ctx: &ImportCtx,
input: &Path,
follow: bool,
exclude: &[String],
) -> anyhow::Result<ImportReport> {
let excluder = build_excluder(exclude)?;
let root = std::fs::canonicalize(input)
.map_err(|e| anyhow!("canonicalize --input {}: {e}", input.display()))?;
if ctx.optimize {
let sizes = ctx
.sizes
.as_ref()
.ok_or_else(|| anyhow!("internal: optimize without chunk sizes"))?;
let mut chunks_total: u64 = 0;
let collected = walk_and_collect(&root, follow, &excluder, |canonical, _rel| {
let (hash_hex, size, chunks) = import_optimized_file(ctx, canonical, sizes)?;
let n = u64::try_from(chunks.len())
.map_err(|_| anyhow!("chunk count {} exceeds u64", chunks.len()))?;
chunks_total = chunks_total
.checked_add(n)
.ok_or_else(|| anyhow!("chunk count overflow"))?;
Ok((b3_hex_str_from_hex(&hash_hex), size, Some(chunks)))
})?;
emit_manifest(
ctx,
collected.entries,
collected.total_size,
true,
chunks_total,
collected.skipped_symlinks,
)
} else {
let collected = walk_and_collect(&root, follow, &excluder, |canonical, _rel| {
if ctx.write {
let base = ctx
.base
.as_deref()
.ok_or_else(|| anyhow!("internal: write without a base"))?;
let blob = import_one_file(base, canonical, ctx.move_source, ctx.force)?;
Ok((b3_hex_str_from_hex(&blob.hash_hex), blob.size, None))
} else {
let (h, s) = hash_file_at(canonical)
.map_err(|e| anyhow!("hash {}: {e}", canonical.display()))?;
Ok((b3_hex_str(h), s, None))
}
})?;
emit_manifest(
ctx,
collected.entries,
collected.total_size,
false,
0,
collected.skipped_symlinks,
)
}
}
fn import_optimized_file(
ctx: &ImportCtx,
path: &Path,
sizes: &ChunkSizes,
) -> anyhow::Result<(String, u64, Vec<Chunk>)> {
if ctx.write {
let base = ctx
.base
.as_deref()
.ok_or_else(|| anyhow!("internal: write without a base"))?;
let blob = import_one_file(base, path, ctx.move_source, ctx.force)?;
let stored = base.join(shard_prefix(&blob.hash_hex)).join(&blob.hash_hex);
let file = File::open(&stored)
.map_err(|e| anyhow!("open stored object {}: {e}", stored.display()))?;
let cf = chunk_file(file, sizes, |_chash, _data| Ok(()))?;
if cf.whole_hash.to_hex().as_str() != blob.hash_hex || cf.total_size != blob.size {
bail!(
"stored object {} is corrupt: whole-file pass wrote {} bytes at {}, \
chunk pass read {} bytes at {}",
stored.display(),
blob.size,
blob.hash_hex,
cf.total_size,
cf.whole_hash.to_hex()
);
}
Ok((blob.hash_hex, blob.size, cf.chunks))
} else {
let file = File::open(path).map_err(|e| anyhow!("open {}: {e}", path.display()))?;
let cf = chunk_file(file, sizes, |_chash, _data| Ok(()))?;
Ok((cf.whole_hash.to_hex().to_string(), cf.total_size, cf.chunks))
}
}
fn import_single_optimized(ctx: &ImportCtx, input: &Path) -> anyhow::Result<ImportReport> {
let sizes = ctx
.sizes
.as_ref()
.ok_or_else(|| anyhow!("internal: optimize without chunk sizes"))?;
let (hash_hex, total_size, chunks) = import_optimized_file(ctx, input, sizes)?;
let name = input
.file_name()
.ok_or_else(|| anyhow!("--input {} has no file name", input.display()))?;
let path = validate_relpath(Path::new(name))?;
let chunks_total =
u64::try_from(chunks.len()).map_err(|_| anyhow!("chunk count exceeds u64"))?;
let entry = BundleEntry {
path,
hash: b3_hex_str_from_hex(&hash_hex),
size: total_size,
chunks: Some(chunks),
};
emit_manifest(ctx, vec![entry], total_size, true, chunks_total, 0)
}
fn import_single_plain(ctx: &ImportCtx, input: &Path) -> anyhow::Result<ImportReport> {
let (hash_hex, size) = if ctx.write {
let base = ctx
.base
.as_deref()
.ok_or_else(|| anyhow!("internal: write without a base"))?;
let blob = import_one_file(base, input, ctx.move_source, ctx.force)?;
(blob.hash_hex, blob.size)
} else {
let (h, s) = hash_file_at(input).map_err(|e| anyhow!("hash {}: {e}", input.display()))?;
(h.to_hex().to_string(), s)
};
let name = input.file_name().map_or_else(
|| input.display().to_string(),
|n| n.to_string_lossy().into_owned(),
);
let files = BTreeMap::from([(
place_under(ctx.subfolder.as_deref(), &name),
b3_hex_str_from_hex(&hash_hex),
)]);
Ok(ImportReport {
imported: 1,
bytes: size,
origin: ctx.origin_label.clone(),
files,
bundle_hash: None,
moved: ctx.move_source && ctx.write,
optimized: false,
chunks_total: 0,
skipped_symlinks: 0,
})
}
fn emit_manifest(
ctx: &ImportCtx,
mut entries: Vec<BundleEntry>,
total_size: u64,
optimized: bool,
chunks_total: u64,
skipped_symlinks: u64,
) -> anyhow::Result<ImportReport> {
if let Some(sub) = ctx.subfolder.as_deref() {
for entry in &mut entries {
entry.path = place_under(Some(sub), &entry.path);
}
}
let bundle_bytes = serialize_canonical(&entries)?;
if ctx.write
&& let Some(base) = ctx.base.as_deref()
{
import_bytes(base, &bundle_bytes, ctx.force)?;
}
if let Some(out) = ctx.bundle_out.as_deref() {
write_bundle(out, &bundle_bytes)
.map_err(|e| anyhow!("write --bundle {}: {e}", out.display()))?;
}
if ctx.dry_run {
let mut stdout = std::io::stdout().lock();
stdout
.write_all(&bundle_bytes)
.map_err(|e| anyhow!("write manifest to stdout: {e}"))?;
}
let bundle_hash = Some(b3_hex_str(blake3::hash(&bundle_bytes)));
let imported = u64::try_from(entries.len())
.map_err(|_| anyhow!("entry count {} exceeds u64", entries.len()))?;
let files = entries.into_iter().map(|e| (e.path, e.hash)).collect();
Ok(ImportReport {
imported,
bytes: total_size,
origin: ctx.origin_label.clone(),
files,
bundle_hash,
moved: ctx.move_source && ctx.write,
optimized,
chunks_total,
skipped_symlinks,
})
}
struct ImportedBlob {
hash_hex: String,
size: u64,
}
fn b3_hex_str_from_hex(hex: &str) -> String {
format!("b3:{hex}")
}
fn import_one_file(
base: &Path,
source: &Path,
move_source: bool,
force: bool,
) -> anyhow::Result<ImportedBlob> {
let file = File::open(source).map_err(|e| anyhow!("open {}: {e}", source.display()))?;
let size = file
.metadata()
.map_err(|e| anyhow!("stat {}: {e}", source.display()))?
.len();
let (eo, staged): (EncodedOutboard, Option<tempfile::NamedTempFile>) = if move_source {
let eo = encode_outboard(file, size)
.map_err(|e| anyhow!("encode outboard for {}: {e}", source.display()))?;
(eo, None)
} else {
let tmp = tempfile::NamedTempFile::new_in(base)
.map_err(|e| anyhow!("stage temp data object in {}: {e}", base.display()))?;
let eo = {
let sink = tmp.as_file();
let tee = TeeReader::new(file, sink);
encode_outboard(tee, size)
.map_err(|e| anyhow!("encode outboard for {}: {e}", source.display()))?
};
tmp.as_file()
.sync_all()
.map_err(|e| anyhow!("sync staged data object: {e}"))?;
(eo, Some(tmp))
};
let (shard_dir, data_path, obao4_path) = object_paths(base, &eo);
std::fs::create_dir_all(&shard_dir)
.map_err(|e| anyhow!("create shard dir {}: {e}", shard_dir.display()))?;
if data_slot_needs_write(&data_path, size, force, &eo.hash_hex)? {
if move_source {
move_into_place(source, base, &data_path)?;
} else if let Some(tmp) = staged {
tmp.persist(&data_path)
.map_err(|e| anyhow!("persist data object {}: {}", data_path.display(), e.error))?;
}
} else if move_source {
let (existing_hash, _) = hash_file_at(&data_path)
.map_err(|e| anyhow!("verify existing object {}: {e}", data_path.display()))?;
if existing_hash.to_hex().as_str() != eo.hash_hex.as_str() {
bail!(
"origin object {} already exists at {} but its bytes do not match \
their hash; refusing to delete the moved source (pass --force to overwrite)",
eo.hash_hex,
data_path.display()
);
}
std::fs::remove_file(source)
.map_err(|e| anyhow!("remove moved source {}: {e}", source.display()))?;
}
ensure_obao4(&shard_dir, &obao4_path, &eo.outboard)?;
Ok(ImportedBlob {
hash_hex: eo.hash_hex,
size,
})
}
fn import_bytes(base: &Path, bytes: &[u8], force: bool) -> anyhow::Result<ImportedBlob> {
let size = u64::try_from(bytes.len()).map_err(|_| anyhow!("blob length exceeds u64"))?;
let eo = encode_outboard(bytes, size).map_err(|e| anyhow!("encode outboard for blob: {e}"))?;
let (shard_dir, data_path, obao4_path) = object_paths(base, &eo);
std::fs::create_dir_all(&shard_dir)
.map_err(|e| anyhow!("create shard dir {}: {e}", shard_dir.display()))?;
if data_slot_needs_write(&data_path, size, force, &eo.hash_hex)? {
let mut tmp = tempfile::NamedTempFile::new_in(&shard_dir)
.map_err(|e| anyhow!("stage temp data object in {}: {e}", shard_dir.display()))?;
tmp.write_all(bytes)
.map_err(|e| anyhow!("write data object: {e}"))?;
tmp.as_file()
.sync_all()
.map_err(|e| anyhow!("sync data object: {e}"))?;
tmp.persist(&data_path)
.map_err(|e| anyhow!("persist data object {}: {}", data_path.display(), e.error))?;
}
ensure_obao4(&shard_dir, &obao4_path, &eo.outboard)?;
Ok(ImportedBlob {
hash_hex: eo.hash_hex,
size,
})
}
fn object_paths(base: &Path, eo: &EncodedOutboard) -> (PathBuf, PathBuf, PathBuf) {
let shard_dir = base.join(eo.shard());
let data_path = shard_dir.join(&eo.hash_hex);
let obao4_path = shard_dir.join(eo.obao4_name());
(shard_dir, data_path, obao4_path)
}
fn data_slot_needs_write(
data_path: &Path,
size: u64,
force: bool,
hex: &str,
) -> anyhow::Result<bool> {
match std::fs::metadata(data_path) {
Ok(_) if force => Ok(true),
Ok(meta) => {
if meta.len() != size {
bail!(
"origin object {hex} already exists at {} with a different size \
({} vs {size}); pass --force to overwrite",
data_path.display(),
meta.len()
);
}
Ok(false)
}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(true),
Err(e) => Err(anyhow!("stat {}: {e}", data_path.display())),
}
}
fn ensure_obao4(shard_dir: &Path, obao4_path: &Path, outboard: &[u8]) -> anyhow::Result<()> {
match std::fs::read(obao4_path) {
Ok(existing) if existing == outboard => return Ok(()),
Ok(_) => {}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => return Err(anyhow!("read outboard {}: {e}", obao4_path.display())),
}
let mut tmp = tempfile::NamedTempFile::new_in(shard_dir)
.map_err(|e| anyhow!("stage temp outboard in {}: {e}", shard_dir.display()))?;
tmp.write_all(outboard)
.map_err(|e| anyhow!("write outboard: {e}"))?;
tmp.as_file()
.sync_all()
.map_err(|e| anyhow!("sync outboard: {e}"))?;
tmp.persist(obao4_path)
.map_err(|e| anyhow!("persist outboard {}: {}", obao4_path.display(), e.error))?;
Ok(())
}
fn move_into_place(source: &Path, base: &Path, data_path: &Path) -> anyhow::Result<()> {
match std::fs::rename(source, data_path) {
Ok(()) => Ok(()),
Err(e) if is_cross_device(&e) => {
let tmp = tempfile::NamedTempFile::new_in(base)
.map_err(|e| anyhow!("stage cross-device copy in {}: {e}", base.display()))?;
std::fs::copy(source, tmp.path())
.map_err(|e| anyhow!("copy {} across filesystems: {e}", source.display()))?;
tmp.as_file()
.sync_all()
.map_err(|e| anyhow!("sync cross-device copy: {e}"))?;
tmp.persist(data_path)
.map_err(|e| anyhow!("persist data object {}: {}", data_path.display(), e.error))?;
std::fs::remove_file(source)
.map_err(|e| anyhow!("remove moved source {}: {e}", source.display()))?;
Ok(())
}
Err(e) => Err(anyhow!(
"move {} into {}: {e}",
source.display(),
data_path.display()
)),
}
}
fn is_cross_device(err: &std::io::Error) -> bool {
err.kind() == std::io::ErrorKind::CrossesDevices
}
struct TeeReader<R, W> {
inner: R,
sink: W,
}
impl<R: Read, W: Write> TeeReader<R, W> {
const fn new(inner: R, sink: W) -> Self {
Self { inner, sink }
}
}
impl<R: Read, W: Write> Read for TeeReader<R, W> {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let n = self.inner.read(buf)?;
if let Some(chunk) = buf.get(..n) {
self.sink.write_all(chunk)?;
}
Ok(n)
}
}
fn single_file_hash(report: &ImportReport) -> Option<&str> {
if report.bundle_hash.is_some() {
return None;
}
let mut it = report.files.values();
match (it.next(), it.next()) {
(Some(h), None) => Some(h.as_str()),
_ => None,
}
}
fn write_import_report(
w: &mut impl Write,
report: &ImportReport,
json: bool,
) -> std::io::Result<()> {
if json {
let line = serde_json::to_string(report)
.map_err(|e| std::io::Error::other(format!("serialize status report: {e}")))?;
writeln!(w, "{line}")
} else {
let verb = if report.moved { "moved" } else { "imported" };
write!(
w,
"{verb} {} file(s), {} bytes, into {}",
report.imported, report.bytes, report.origin
)?;
if report.optimized {
write!(w, ", {} chunk hint(s)", report.chunks_total)?;
}
match (&report.bundle_hash, single_file_hash(report)) {
(Some(b), _) => writeln!(w, " (bundle {b})")?,
(None, Some(h)) => writeln!(w, " ({h})")?,
(None, None) => writeln!(w)?,
}
if report.skipped_symlinks > 0 {
writeln!(
w,
"skipped {} symlink(s); pass --follow-symlinks to include",
report.skipped_symlinks
)?;
}
Ok(())
}
}
#[cfg(test)]
#[allow(
clippy::unwrap_used,
clippy::expect_used,
clippy::indexing_slicing,
clippy::panic
)]
mod tests;