aven_core/attachments/

lifecycle.rs

#![allow(dead_code)]

use std::collections::{HashMap, HashSet};
use std::fs;
use std::io;
use std::path::{Path, PathBuf};
use std::sync::{LazyLock, Mutex};
use std::time::{Duration, SystemTime};

use anyhow::{Result, bail};
use chrono::{DateTime, SecondsFormat, Utc};
use sqlx::{Row, SqliteConnection};

use crate::attachments::storage::object_path;
use crate::attachments::validation::validate_sha256;
use crate::db::{begin_immediate, get_meta, set_meta};
use crate::ids::new_id;

pub const DEFAULT_LOCAL_GRACE: Duration = Duration::from_secs(7 * 24 * 60 * 60);
pub const DEFAULT_ORIGINAL_QUOTA_BYTES: i64 = 10 * 1024 * 1024 * 1024;
pub const DEFAULT_PREVIEW_QUOTA_BYTES: u64 = 512 * 1024 * 1024;
pub const DEFAULT_MAINTENANCE_LIMIT: usize = 128;
const LEASE_TTL: Duration = Duration::from_secs(10 * 60);
const LIVENESS_CURSOR_META_KEY: &str = "attachment_liveness_cursor";

pub trait Clock: Send + Sync {
    fn now(&self) -> DateTime<Utc>;
}

pub struct SystemClock;

impl Clock for SystemClock {
    fn now(&self) -> DateTime<Utc> {
        Utc::now()
    }
}

#[derive(Debug, Clone, Copy)]
pub struct LifecyclePolicy {
    pub grace: Duration,
    pub quota_bytes: i64,
    pub preview_quota_bytes: u64,
    pub maintenance_limit: usize,
}

impl Default for LifecyclePolicy {
    fn default() -> Self {
        Self {
            grace: DEFAULT_LOCAL_GRACE,
            quota_bytes: DEFAULT_ORIGINAL_QUOTA_BYTES,
            preview_quota_bytes: DEFAULT_PREVIEW_QUOTA_BYTES,
            maintenance_limit: DEFAULT_MAINTENANCE_LIMIT,
        }
    }
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ByteCount {
    pub count: u64,
    pub bytes: u64,
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct LifecycleReport {
    pub referenced: ByteCount,
    pub protected: ByteCount,
    pub grace_period: ByteCount,
    pub eligible: ByteCount,
    pub staging: ByteCount,
    pub trash: ByteCount,
    pub reservations: ByteCount,
    pub quota: ByteCount,
    pub inconsistencies: ByteCount,
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct PruneSummary {
    pub eligible: ByteCount,
    pub pruned: ByteCount,
}

fn timestamp(now: DateTime<Utc>) -> String {
    now.to_rfc3339_opts(SecondsFormat::Secs, true)
}

fn cutoff(now: DateTime<Utc>, grace: Duration) -> Result<String> {
    let grace = chrono::Duration::from_std(grace)?;
    Ok(timestamp(now - grace))
}

fn trash_dir(blob_dir: &Path) -> PathBuf {
    blob_dir.join("trash")
}

fn staging_dir(blob_dir: &Path) -> PathBuf {
    blob_dir.join("objects").join("sha256")
}

struct ScannedFile {
    path: PathBuf,
    name: String,
    len: u64,
    modified: SystemTime,
}

#[derive(Default)]
struct DirectoryCursor {
    entries: Option<fs::ReadDir>,
}

static DIRECTORY_CURSORS: LazyLock<Mutex<HashMap<PathBuf, DirectoryCursor>>> =
    LazyLock::new(|| Mutex::new(HashMap::new()));

async fn scan_directory_all(dir: PathBuf) -> Result<Vec<ScannedFile>> {
    crate::attachments::blocking::run(move || {
        let entries = match fs::read_dir(&dir) {
            Ok(entries) => entries,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(error) => return Err(error.into()),
        };
        let mut files = Vec::new();
        for entry in entries {
            let entry = entry?;
            if !entry.file_type()?.is_file() {
                continue;
            }
            let metadata = entry.metadata()?;
            files.push(ScannedFile {
                path: entry.path(),
                name: entry.file_name().to_string_lossy().into_owned(),
                len: metadata.len(),
                modified: metadata.modified()?,
            });
        }
        Ok(files)
    })
    .await
}

async fn scan_directory_page(dir: PathBuf, limit: usize) -> Result<Vec<ScannedFile>> {
    scan_directory_page_with(dir, limit, |entries| entries.next().transpose()).await
}

async fn scan_directory_page_with<F>(
    dir: PathBuf,
    limit: usize,
    mut next_entry: F,
) -> Result<Vec<ScannedFile>>
where
    F: FnMut(&mut fs::ReadDir) -> io::Result<Option<fs::DirEntry>> + Send + 'static,
{
    if limit == 0 {
        return Ok(Vec::new());
    }
    crate::attachments::blocking::run(move || {
        let key = dir.clone();
        let mut cursors = DIRECTORY_CURSORS
            .lock()
            .map_err(|_| anyhow::anyhow!("attachment directory cursor poisoned"))?;
        let mut cursor = match cursors.remove(&key) {
            Some(cursor) => cursor,
            None => {
                let entries = match fs::read_dir(&dir) {
                    Ok(entries) => entries,
                    Err(error) if error.kind() == io::ErrorKind::NotFound => {
                        return Ok(Vec::new());
                    }
                    Err(error) => return Err(error.into()),
                };
                DirectoryCursor {
                    entries: Some(entries),
                }
            }
        };
        let mut files = Vec::new();
        let mut exhausted = false;
        for _ in 0..limit {
            let Some(entry) = next_entry(
                cursor
                    .entries
                    .as_mut()
                    .expect("attachment directory cursor should be open"),
            )?
            else {
                exhausted = true;
                break;
            };
            if !entry.file_type()?.is_file() {
                continue;
            }
            let metadata = entry.metadata()?;
            files.push(ScannedFile {
                path: entry.path(),
                name: entry.file_name().to_string_lossy().into_owned(),
                len: metadata.len(),
                modified: metadata.modified()?,
            });
        }
        if !exhausted {
            cursors.insert(key, cursor);
        }
        Ok(files)
    })
    .await
}

async fn remove_files(files: Vec<PathBuf>) -> Result<()> {
    crate::attachments::blocking::run(move || {
        for path in files {
            match fs::remove_file(path) {
                Ok(()) => {}
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                Err(error) => return Err(error.into()),
            }
        }
        Ok(())
    })
    .await
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum FileMove {
    Moved,
    SourceMissing,
    TargetExists,
}

fn sync_parent(path: &Path) -> io::Result<()> {
    let parent = path
        .parent()
        .ok_or_else(|| io::Error::other("attachment move target has no parent"))?;
    fs::File::open(parent)?.sync_all()
}

fn finish_file_move(source: &Path, target: &Path) -> Result<FileMove> {
    if let Err(error) = sync_parent(target) {
        let _ = fs::remove_file(target);
        return Err(error.into());
    }
    if let Err(error) = fs::remove_file(source) {
        let _ = fs::remove_file(target);
        let _ = sync_parent(target);
        return Err(error.into());
    }
    Ok(FileMove::Moved)
}

fn copy_file_without_replacing(source: &Path, target: &Path) -> Result<FileMove> {
    let mut source_file = match fs::File::open(source) {
        Ok(file) => file,
        Err(error) if error.kind() == io::ErrorKind::NotFound => {
            return Ok(FileMove::SourceMissing);
        }
        Err(error) => return Err(error.into()),
    };
    let parent = target
        .parent()
        .ok_or_else(|| io::Error::other("attachment move target has no parent"))?;
    let mut staging = tempfile::Builder::new()
        .prefix(".aven-move-")
        .tempfile_in(parent)?;
    io::copy(&mut source_file, &mut staging)?;
    staging.as_file().sync_all()?;
    match staging.persist_noclobber(target) {
        Ok(_) => finish_file_move(source, target),
        Err(error) if error.error.kind() == io::ErrorKind::AlreadyExists => {
            Ok(FileMove::TargetExists)
        }
        Err(error) => Err(error.error.into()),
    }
}

fn move_file_without_replacing(source: &Path, target: &Path) -> Result<FileMove> {
    move_file_without_replacing_with(source, target, |source, target| {
        fs::hard_link(source, target)
    })
}

fn move_file_without_replacing_with<F>(
    source: &Path,
    target: &Path,
    hard_link: F,
) -> Result<FileMove>
where
    F: FnOnce(&Path, &Path) -> io::Result<()>,
{
    match hard_link(source, target) {
        Ok(()) => finish_file_move(source, target),
        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => Ok(FileMove::TargetExists),
        Err(error) if error.kind() == io::ErrorKind::NotFound && !source.exists() => {
            Ok(FileMove::SourceMissing)
        }
        Err(_) => copy_file_without_replacing(source, target),
    }
}

async fn restore_trashed_files(files: Vec<(PathBuf, PathBuf)>) {
    let _ = crate::attachments::blocking::run(move || {
        for (source, target) in files {
            if !target.exists() {
                continue;
            }
            if source.exists() {
                fs::remove_file(target)?;
            } else {
                let _ = move_file_without_replacing(&target, &source)?;
            }
        }
        Ok(())
    })
    .await;
}

fn live_blob_references_sql(sha256_expr: &str) -> String {
    format!(
        "EXISTS(
           SELECT 1 FROM task_attachments ta
           JOIN tasks t ON t.workspace_id = ta.workspace_id AND t.id = ta.task_id
           WHERE ta.sha256 = {sha256_expr} AND ta.deleted = 0 AND t.deleted = 0
         ) OR EXISTS(
           SELECT 1 FROM server_blob_references sbr
           LEFT JOIN server_task_tombstones st
             ON st.workspace_id = sbr.workspace_id AND st.task_id = sbr.task_id
           WHERE sbr.sha256 = {sha256_expr} AND sbr.deleted = 0
             AND COALESCE(st.deleted, 0) = 0
         )"
    )
}

pub async fn reconcile_liveness(conn: &mut SqliteConnection, clock: &dyn Clock) -> Result<()> {
    let mut tx = begin_immediate(conn).await?;
    reconcile_liveness_in_transaction(&mut tx, clock).await?;
    tx.commit().await?;
    Ok(())
}

pub(crate) async fn reconcile_liveness_in_transaction(
    conn: &mut SqliteConnection,
    clock: &dyn Clock,
) -> Result<()> {
    let now = timestamp(clock.now());
    sqlx::query("DELETE FROM blob_leases WHERE expires_at <= ?")
        .bind(&now)
        .execute(&mut *conn)
        .await?;
    sqlx::query("DELETE FROM blob_upload_reservations WHERE expires_at <= ?")
        .bind(&now)
        .execute(&mut *conn)
        .await?;
    sqlx::query(
        "INSERT OR IGNORE INTO blob_lifecycle(sha256, unreferenced_at)
         SELECT sha256, NULL FROM blob_inventory",
    )
    .execute(&mut *conn)
    .await?;
    let live_blob_references = live_blob_references_sql("blob_lifecycle.sha256");
    sqlx::query(sqlx::AssertSqlSafe(format!(
        "UPDATE blob_lifecycle SET unreferenced_at = NULL
         WHERE unreferenced_at IS NOT NULL
           AND ({live_blob_references})"
    )))
    .execute(&mut *conn)
    .await?;
    sqlx::query(sqlx::AssertSqlSafe(format!(
        "UPDATE blob_lifecycle SET unreferenced_at = ?
         WHERE unreferenced_at IS NULL
           AND NOT ({live_blob_references})"
    )))
    .bind(&now)
    .execute(&mut *conn)
    .await?;
    Ok(())
}

pub(crate) async fn reconcile_liveness_for_hashes_in_transaction(
    conn: &mut SqliteConnection,
    hashes: &[String],
    clock: &dyn Clock,
) -> Result<()> {
    if hashes.is_empty() {
        return Ok(());
    }
    let now = timestamp(clock.now());
    let hashes = serde_json::to_string(hashes)?;
    let live_inventory_references = live_blob_references_sql("bi.sha256");
    sqlx::query(sqlx::AssertSqlSafe(format!(
        "INSERT OR IGNORE INTO blob_lifecycle(sha256, unreferenced_at)
         SELECT bi.sha256,
                CASE WHEN ({live_inventory_references}) THEN NULL ELSE ? END
         FROM blob_inventory bi
         JOIN (SELECT DISTINCT value AS sha256 FROM json_each(?)) affected
           ON affected.sha256 = bi.sha256"
    )))
    .bind(&now)
    .bind(&hashes)
    .execute(&mut *conn)
    .await?;

    let live_blob_references = live_blob_references_sql("blob_lifecycle.sha256");
    sqlx::query(sqlx::AssertSqlSafe(format!(
        "UPDATE blob_lifecycle SET unreferenced_at = NULL
         WHERE unreferenced_at IS NOT NULL
           AND sha256 IN (SELECT value FROM json_each(?))
           AND ({live_blob_references})"
    )))
    .bind(&hashes)
    .execute(&mut *conn)
    .await?;
    sqlx::query(sqlx::AssertSqlSafe(format!(
        "UPDATE blob_lifecycle SET unreferenced_at = ?
         WHERE unreferenced_at IS NULL
           AND sha256 IN (SELECT value FROM json_each(?))
           AND NOT ({live_blob_references})"
    )))
    .bind(&now)
    .bind(&hashes)
    .execute(&mut *conn)
    .await?;
    Ok(())
}

async fn reconcile_liveness_bounded(
    conn: &mut SqliteConnection,
    limit: usize,
    clock: &dyn Clock,
) -> Result<()> {
    let mut tx = begin_immediate(conn).await?;
    let now = timestamp(clock.now());
    sqlx::query("DELETE FROM blob_leases WHERE expires_at <= ?")
        .bind(&now)
        .execute(&mut *tx)
        .await?;
    sqlx::query("DELETE FROM blob_upload_reservations WHERE expires_at <= ?")
        .bind(&now)
        .execute(&mut *tx)
        .await?;
    if limit == 0 {
        tx.commit().await?;
        return Ok(());
    }

    let cursor = get_meta(&mut tx, LIVENESS_CURSOR_META_KEY)
        .await?
        .unwrap_or_default();
    let hashes = sqlx::query_scalar::<_, String>(
        "SELECT sha256 FROM blob_inventory
         WHERE sha256 > ? ORDER BY sha256 LIMIT ?",
    )
    .bind(&cursor)
    .bind(i64::try_from(limit)?)
    .fetch_all(&mut *tx)
    .await?;
    reconcile_liveness_for_hashes_in_transaction(&mut tx, &hashes, clock).await?;
    set_meta(
        &mut tx,
        LIVENESS_CURSOR_META_KEY,
        if hashes.len() < limit {
            ""
        } else {
            hashes.last().map(String::as_str).unwrap_or_default()
        },
    )
    .await?;
    tx.commit().await?;
    Ok(())
}

async fn is_protected(conn: &mut SqliteConnection, sha256: &str, now: &str) -> Result<bool> {
    let live_blob_references = live_blob_references_sql("?");
    Ok(sqlx::query_scalar::<_, bool>(sqlx::AssertSqlSafe(format!(
        "SELECT
           {live_blob_references} OR EXISTS(
             SELECT 1 FROM changes
             WHERE server_seq IS NULL AND op_type = 'attachment_add'
               AND json_extract(payload, '$.sha256') = ?
           ) OR EXISTS(
             SELECT 1 FROM blob_leases WHERE sha256 = ? AND expires_at > ?
           ) OR EXISTS(
             SELECT 1 FROM blob_upload_reservations WHERE sha256 = ? AND expires_at > ?
           )"
    )))
    .bind(sha256)
    .bind(sha256)
    .bind(sha256)
    .bind(sha256)
    .bind(now)
    .bind(sha256)
    .bind(now)
    .fetch_one(&mut *conn)
    .await?)
}

pub async fn acquire_lease(
    conn: &mut SqliteConnection,
    sha256: &str,
    kind: &str,
    clock: &dyn Clock,
) -> Result<String> {
    validate_sha256(sha256)?;
    if !matches!(kind, "staging" | "read" | "backup" | "transfer") {
        bail!("error attachment-lease-kind-invalid");
    }
    let lease_id = new_id();
    let now = clock.now();
    let expires = now + chrono::Duration::from_std(LEASE_TTL)?;
    sqlx::query(
        "INSERT INTO blob_leases(lease_id, sha256, kind, created_at, expires_at)
         VALUES (?, ?, ?, ?, ?)",
    )
    .bind(&lease_id)
    .bind(sha256)
    .bind(kind)
    .bind(timestamp(now))
    .bind(timestamp(expires))
    .execute(&mut *conn)
    .await?;
    Ok(lease_id)
}

pub async fn release_lease(conn: &mut SqliteConnection, lease_id: &str) -> Result<()> {
    sqlx::query("DELETE FROM blob_leases WHERE lease_id = ?")
        .bind(lease_id)
        .execute(&mut *conn)
        .await?;
    Ok(())
}

pub async fn reserve_upload(
    conn: &mut SqliteConnection,
    workspace_id: &str,
    sha256: &str,
    byte_size: i64,
    quota_bytes: i64,
    clock: &dyn Clock,
) -> Result<Option<String>> {
    validate_sha256(sha256)?;
    let mut tx = begin_immediate(conn).await?;
    let existing: bool = sqlx::query_scalar(
        "SELECT EXISTS(
           SELECT 1 FROM server_blob_references sbr
           LEFT JOIN server_task_tombstones st
             ON st.workspace_id = sbr.workspace_id AND st.task_id = sbr.task_id
           WHERE sbr.workspace_id = ? AND sbr.sha256 = ? AND sbr.deleted = 0
             AND COALESCE(st.deleted, 0) = 0
         )",
    )
    .bind(workspace_id)
    .bind(sha256)
    .fetch_one(&mut *tx)
    .await?;
    if existing {
        tx.commit().await?;
        return Ok(None);
    }
    let used: i64 = sqlx::query_scalar(
        "SELECT COALESCE(SUM(byte_size), 0) FROM (
           SELECT sbr.sha256, MAX(sbr.byte_size) AS byte_size
           FROM server_blob_references sbr
           LEFT JOIN server_task_tombstones st
             ON st.workspace_id = sbr.workspace_id AND st.task_id = sbr.task_id
           WHERE sbr.workspace_id = ? AND sbr.deleted = 0 AND COALESCE(st.deleted, 0) = 0
           GROUP BY sbr.sha256
         )",
    )
    .bind(workspace_id)
    .fetch_one(&mut *tx)
    .await?;
    let reserved: i64 = sqlx::query_scalar(
        "SELECT COALESCE(SUM(byte_size), 0) FROM blob_upload_reservations
         WHERE workspace_id = ? AND sha256 != ? AND expires_at > ?",
    )
    .bind(workspace_id)
    .bind(sha256)
    .bind(timestamp(clock.now()))
    .fetch_one(&mut *tx)
    .await?;
    if used.saturating_add(reserved).saturating_add(byte_size) > quota_bytes {
        tx.rollback().await?;
        bail!("error attachment-quota-exceeded");
    }
    let reservation_id = new_id();
    let now = clock.now();
    let expires = now + chrono::Duration::from_std(LEASE_TTL)?;
    sqlx::query(
        "INSERT INTO blob_upload_reservations(
           reservation_id, workspace_id, sha256, byte_size, created_at, expires_at
         ) VALUES (?, ?, ?, ?, ?, ?)
         ON CONFLICT(workspace_id, sha256) DO UPDATE SET
           reservation_id = excluded.reservation_id,
           byte_size = excluded.byte_size,
           created_at = excluded.created_at,
           expires_at = excluded.expires_at",
    )
    .bind(&reservation_id)
    .bind(workspace_id)
    .bind(sha256)
    .bind(byte_size)
    .bind(timestamp(now))
    .bind(timestamp(expires))
    .execute(&mut *tx)
    .await?;
    tx.commit().await?;
    Ok(Some(reservation_id))
}

pub async fn release_reservation(conn: &mut SqliteConnection, reservation_id: &str) -> Result<()> {
    sqlx::query("DELETE FROM blob_upload_reservations WHERE reservation_id = ?")
        .bind(reservation_id)
        .execute(&mut *conn)
        .await?;
    Ok(())
}

pub async fn local_unique_bytes(conn: &mut SqliteConnection) -> Result<i64> {
    Ok(sqlx::query_scalar(
        "SELECT COALESCE(SUM(byte_size), 0) FROM blob_inventory WHERE available = 1",
    )
    .fetch_one(&mut *conn)
    .await?)
}

pub async fn ensure_local_capacity(
    conn: &mut SqliteConnection,
    blob_dir: &Path,
    sha256: &str,
    byte_size: i64,
    policy: LifecyclePolicy,
    clock: &dyn Clock,
) -> Result<Option<String>> {
    let used = local_unique_bytes(conn).await?;
    if used.saturating_add(byte_size) > policy.quota_bytes {
        prune(conn, blob_dir, policy, true, clock).await?;
    }
    let now = clock.now();
    let mut tx = begin_immediate(conn).await?;
    let existing: bool = sqlx::query_scalar(
        "SELECT EXISTS(SELECT 1 FROM blob_inventory WHERE sha256 = ? AND available = 1)",
    )
    .bind(sha256)
    .fetch_one(&mut *tx)
    .await?;
    if existing {
        tx.commit().await?;
        return Ok(None);
    }
    let used: i64 = sqlx::query_scalar(
        "SELECT COALESCE(SUM(byte_size), 0) FROM blob_inventory WHERE available = 1",
    )
    .fetch_one(&mut *tx)
    .await?;
    let reserved: i64 = sqlx::query_scalar(
        "SELECT COALESCE(SUM(byte_size), 0) FROM blob_upload_reservations
         WHERE workspace_id = '__local__' AND sha256 != ? AND expires_at > ?",
    )
    .bind(sha256)
    .bind(timestamp(now))
    .fetch_one(&mut *tx)
    .await?;
    if used.saturating_add(reserved).saturating_add(byte_size) > policy.quota_bytes {
        tx.rollback().await?;
        bail!("error attachment-quota-exceeded");
    }
    let reservation_id = new_id();
    let expires = now + chrono::Duration::from_std(LEASE_TTL)?;
    sqlx::query(
        "INSERT INTO blob_upload_reservations(
           reservation_id, workspace_id, sha256, byte_size, created_at, expires_at
         ) VALUES (?, '__local__', ?, ?, ?, ?)
         ON CONFLICT(workspace_id, sha256) DO UPDATE SET
           reservation_id = excluded.reservation_id,
           byte_size = excluded.byte_size,
           created_at = excluded.created_at,
           expires_at = excluded.expires_at",
    )
    .bind(&reservation_id)
    .bind(sha256)
    .bind(byte_size)
    .bind(timestamp(now))
    .bind(timestamp(expires))
    .execute(&mut *tx)
    .await?;
    tx.commit().await?;
    Ok(Some(reservation_id))
}

async fn reconcile_trash_files(
    conn: &mut SqliteConnection,
    blob_dir: &Path,
    files: Vec<ScannedFile>,
) -> Result<()> {
    let hashes = files
        .iter()
        .filter(|file| validate_sha256(&file.name).is_ok())
        .map(|file| file.name.clone())
        .collect::<Vec<_>>();
    let available = if hashes.is_empty() {
        HashSet::new()
    } else {
        sqlx::query_scalar::<_, String>(
            "SELECT sha256 FROM blob_inventory
             WHERE available = 1 AND sha256 IN (SELECT value FROM json_each(?))",
        )
        .bind(serde_json::to_string(&hashes)?)
        .fetch_all(&mut *conn)
        .await?
        .into_iter()
        .collect()
    };
    let blob_dir = blob_dir.to_path_buf();
    crate::attachments::blocking::run(move || {
        for file in files {
            if validate_sha256(&file.name).is_err() {
                continue;
            }
            let target = object_path(&blob_dir, &file.name)?;
            if available.contains(&file.name) {
                if target.exists() {
                    fs::remove_file(file.path)?;
                } else {
                    if let Some(parent) = target.parent() {
                        fs::create_dir_all(parent)?;
                    }
                    match move_file_without_replacing(&file.path, &target)? {
                        FileMove::Moved | FileMove::SourceMissing | FileMove::TargetExists => {}
                    }
                }
            } else {
                match fs::remove_file(file.path) {
                    Ok(()) => {}
                    Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                    Err(error) => return Err(error.into()),
                }
            }
        }
        Ok(())
    })
    .await
}

pub async fn reconcile_trash(conn: &mut SqliteConnection, blob_dir: &Path) -> Result<()> {
    reconcile_trash_files(
        conn,
        blob_dir,
        scan_directory_all(trash_dir(blob_dir)).await?,
    )
    .await
}

async fn reconcile_trash_page(
    conn: &mut SqliteConnection,
    blob_dir: &Path,
    limit: usize,
) -> Result<()> {
    reconcile_trash_files(
        conn,
        blob_dir,
        scan_directory_page(trash_dir(blob_dir), limit).await?,
    )
    .await
}

pub async fn reconcile_staging(blob_dir: &Path) -> Result<ByteCount> {
    let files = scan_directory_all(staging_dir(blob_dir)).await?;
    let stale = files
        .into_iter()
        .filter(|file| {
            file.name.starts_with(".aven-stage-")
                && file.modified.elapsed().is_ok_and(|age| age >= LEASE_TTL)
        })
        .collect::<Vec<_>>();
    let removed = ByteCount {
        count: u64::try_from(stale.len())?,
        bytes: stale.iter().map(|file| file.len).sum(),
    };
    remove_files(stale.into_iter().map(|file| file.path).collect()).await?;
    Ok(removed)
}

pub async fn reconcile_missing_objects(
    conn: &mut SqliteConnection,
    blob_dir: &Path,
    clock: &dyn Clock,
) -> Result<ByteCount> {
    let rows = sqlx::query(
        "SELECT sha256, byte_size FROM blob_inventory
         WHERE available = 1 ORDER BY sha256",
    )
    .fetch_all(&mut *conn)
    .await?;
    let verified_at = timestamp(clock.now());
    let mut missing = ByteCount::default();
    for row in rows {
        let sha256: String = row.get("sha256");
        if object_path(blob_dir, &sha256)?.exists() {
            continue;
        }
        sqlx::query(
            "UPDATE blob_inventory SET available = 0, last_verified_at = ?
             WHERE sha256 = ? AND available = 1",
        )
        .bind(&verified_at)
        .bind(&sha256)
        .execute(&mut *conn)
        .await?;
        missing.count += 1;
        missing.bytes += u64::try_from(row.get::<i64, _>("byte_size"))?;
    }
    Ok(missing)
}

async fn reconcile_object_directory(
    conn: &mut SqliteConnection,
    blob_dir: &Path,
    grace: Duration,
    limit: usize,
    clock: &dyn Clock,
) -> Result<ByteCount> {
    let files = scan_directory_page(staging_dir(blob_dir), limit).await?;
    let stale_staging = files
        .iter()
        .filter(|file| {
            file.name.starts_with(".aven-stage-")
                && file.modified.elapsed().is_ok_and(|age| age >= LEASE_TTL)
        })
        .map(|file| file.path.clone())
        .collect::<Vec<_>>();
    remove_files(stale_staging).await?;

    let cutoff = clock.now() - chrono::Duration::from_std(grace)?;
    let canonical = files
        .iter()
        .filter(|file| {
            validate_sha256(&file.name).is_ok() && DateTime::<Utc>::from(file.modified) <= cutoff
        })
        .collect::<Vec<_>>();
    if canonical.is_empty() {
        return Ok(ByteCount::default());
    }
    let hashes = canonical
        .iter()
        .map(|file| file.name.clone())
        .collect::<Vec<_>>();
    let candidates = serde_json::to_string(&hashes)?;
    let now = timestamp(clock.now());
    let live_blob_references = live_blob_references_sql("candidate.value");
    let protected = sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(format!(
        "SELECT candidate.value FROM json_each(?) candidate
         WHERE EXISTS(SELECT 1 FROM blob_inventory bi WHERE bi.sha256 = candidate.value)
            OR {live_blob_references}
            OR EXISTS(
              SELECT 1 FROM changes
              WHERE server_seq IS NULL AND op_type = 'attachment_add'
                AND json_extract(payload, '$.sha256') = candidate.value
            )
            OR EXISTS(
              SELECT 1 FROM blob_leases
              WHERE sha256 = candidate.value AND expires_at > ?
            )
            OR EXISTS(
              SELECT 1 FROM blob_upload_reservations
              WHERE sha256 = candidate.value AND expires_at > ?
            )"
    )))
    .bind(candidates)
    .bind(&now)
    .bind(&now)
    .fetch_all(&mut *conn)
    .await?
    .into_iter()
    .collect::<HashSet<_>>();
    let removable = canonical
        .into_iter()
        .filter(|file| !protected.contains(&file.name))
        .map(|file| (file.path.clone(), file.name.clone(), file.len))
        .collect::<Vec<_>>();
    if removable.is_empty() {
        return Ok(ByteCount::default());
    }
    let blob_dir = blob_dir.to_path_buf();
    crate::attachments::blocking::run(move || {
        let trash = trash_dir(&blob_dir);
        fs::create_dir_all(&trash)?;
        let mut removed = ByteCount::default();
        for (source, name, len) in removable {
            let target = trash.join(name);
            if move_file_without_replacing(&source, &target)? == FileMove::Moved {
                fs::remove_file(target)?;
                removed.count += 1;
                removed.bytes += len;
            }
        }
        Ok(removed)
    })
    .await
}

pub async fn reconcile_orphan_objects(
    conn: &mut SqliteConnection,
    blob_dir: &Path,
    grace: Duration,
    clock: &dyn Clock,
) -> Result<ByteCount> {
    reconcile_object_directory(conn, blob_dir, grace, usize::MAX, clock).await
}

pub async fn prune(
    conn: &mut SqliteConnection,
    blob_dir: &Path,
    policy: LifecyclePolicy,
    apply: bool,
    clock: &dyn Clock,
) -> Result<PruneSummary> {
    if apply {
        reconcile_trash_page(conn, blob_dir, policy.maintenance_limit).await?;
        reconcile_object_directory(
            conn,
            blob_dir,
            policy.grace,
            policy.maintenance_limit,
            clock,
        )
        .await?;
        reconcile_missing_objects(conn, blob_dir, clock).await?;
    }
    reconcile_liveness_bounded(conn, policy.maintenance_limit, clock).await?;
    let now = timestamp(clock.now());
    let cutoff = cutoff(clock.now(), policy.grace)?;
    let rows = sqlx::query(
        "SELECT bi.sha256, bi.byte_size
         FROM blob_inventory bi
         JOIN blob_lifecycle bl ON bl.sha256 = bi.sha256
         WHERE bi.available = 1 AND bl.unreferenced_at IS NOT NULL
           AND bl.unreferenced_at <= ?
         ORDER BY bl.unreferenced_at, bi.sha256 LIMIT ?",
    )
    .bind(&cutoff)
    .bind(i64::try_from(policy.maintenance_limit)?)
    .fetch_all(&mut *conn)
    .await?;
    let mut summary = PruneSummary::default();
    for row in rows {
        let sha256: String = row.get("sha256");
        let byte_size: i64 = row.get("byte_size");
        if is_protected(conn, &sha256, &now).await? {
            continue;
        }
        summary.eligible.count += 1;
        summary.eligible.bytes += u64::try_from(byte_size)?;
        if !apply {
            continue;
        }
        let mut tx = begin_immediate(conn).await?;
        let still_eligible: bool = sqlx::query_scalar(
            "SELECT EXISTS(
               SELECT 1 FROM blob_inventory bi
               JOIN blob_lifecycle bl ON bl.sha256 = bi.sha256
               WHERE bi.sha256 = ? AND bi.available = 1
                 AND bl.unreferenced_at IS NOT NULL AND bl.unreferenced_at <= ?
             )",
        )
        .bind(&sha256)
        .bind(&cutoff)
        .fetch_one(&mut *tx)
        .await?;
        if !still_eligible || is_protected(&mut tx, &sha256, &now).await? {
            tx.rollback().await?;
            continue;
        }
        let source = object_path(blob_dir, &sha256)?;
        let trash = trash_dir(blob_dir);
        let trashed = trash.join(&sha256);
        let source_for_move = source.clone();
        let trashed_for_move = trashed.clone();
        let moved = crate::attachments::blocking::run(move || {
            fs::create_dir_all(&trash)?;
            move_file_without_replacing(&source_for_move, &trashed_for_move)
        })
        .await?;
        if moved == FileMove::TargetExists {
            tx.rollback().await?;
            continue;
        }
        if let Err(error) = sqlx::query(
            "UPDATE blob_inventory SET available = 0, last_verified_at = ? WHERE sha256 = ?",
        )
        .bind(&now)
        .bind(&sha256)
        .execute(&mut *tx)
        .await
        {
            let _ = tx.rollback().await;
            if moved == FileMove::Moved {
                restore_trashed_files(vec![(source, trashed)]).await;
            }
            return Err(error.into());
        }
        if let Err(error) = tx.commit().await {
            if moved == FileMove::Moved {
                restore_trashed_files(vec![(source, trashed)]).await;
            }
            return Err(error.into());
        }
        if moved == FileMove::Moved {
            remove_files(vec![trashed]).await?;
        }
        summary.pruned.count += 1;
        summary.pruned.bytes += u64::try_from(byte_size)?;
    }
    let blob_dir = blob_dir.to_path_buf();
    crate::attachments::blocking::run(move || {
        prune_preview_cache(&blob_dir, policy.preview_quota_bytes)
    })
    .await?;
    Ok(summary)
}

pub fn prune_preview_cache(blob_dir: &Path, quota: u64) -> Result<ByteCount> {
    let root = blob_dir.join("cache").join("previews");
    if !root.exists() {
        return Ok(ByteCount::default());
    }
    let mut files = Vec::new();
    let mut dirs = vec![root];
    while let Some(dir) = dirs.pop() {
        for entry in fs::read_dir(dir)? {
            let entry = entry?;
            if entry.file_type()?.is_dir() {
                dirs.push(entry.path());
            } else if entry.file_type()?.is_file() {
                let metadata = entry.metadata()?;
                files.push((metadata.modified()?, metadata.len(), entry.path()));
            }
        }
    }
    let mut total: u64 = files.iter().map(|(_, size, _)| *size).sum();
    files.sort_by_key(|(modified, _, path)| (*modified, path.clone()));
    let mut removed = ByteCount::default();
    for (_, size, path) in files {
        if total <= quota {
            break;
        }
        fs::remove_file(path)?;
        total -= size;
        removed.count += 1;
        removed.bytes += size;
    }
    Ok(removed)
}

pub async fn lifecycle_report(
    conn: &mut SqliteConnection,
    blob_dir: &Path,
    policy: LifecyclePolicy,
    clock: &dyn Clock,
) -> Result<LifecycleReport> {
    reconcile_liveness(conn, clock).await?;
    let now = timestamp(clock.now());
    let cutoff = cutoff(clock.now(), policy.grace)?;
    let live_blob_references = live_blob_references_sql("bi.sha256");
    let rows = sqlx::query(sqlx::AssertSqlSafe(format!(
        "SELECT bi.sha256, bi.byte_size, bi.available, bl.unreferenced_at,
           {live_blob_references} AS referenced
         FROM blob_inventory bi LEFT JOIN blob_lifecycle bl ON bl.sha256 = bi.sha256"
    )))
    .fetch_all(&mut *conn)
    .await?;
    let mut report = LifecycleReport::default();
    for row in rows {
        let sha256: String = row.get("sha256");
        let bytes = u64::try_from(row.get::<i64, _>("byte_size"))?;
        let available = row.get::<i64, _>("available") != 0;
        let referenced = row.get::<i64, _>("referenced") != 0;
        let unreferenced_at: Option<String> = row.get("unreferenced_at");
        if available {
            report.quota.count += 1;
            report.quota.bytes += bytes;
        }
        if referenced {
            report.referenced.count += 1;
            report.referenced.bytes += bytes;
        } else if is_protected(conn, &sha256, &now).await? {
            report.protected.count += 1;
            report.protected.bytes += bytes;
        } else if unreferenced_at
            .as_deref()
            .is_some_and(|at| at <= cutoff.as_str())
        {
            report.eligible.count += 1;
            report.eligible.bytes += bytes;
        } else {
            report.grace_period.count += 1;
            report.grace_period.bytes += bytes;
        }
        if referenced && unreferenced_at.is_some() || !referenced && unreferenced_at.is_none() {
            report.inconsistencies.count += 1;
            report.inconsistencies.bytes += bytes;
        }
    }
    let (reservation_count, reservation_bytes): (i64, i64) = sqlx::query_as(
        "SELECT COUNT(*), COALESCE(SUM(byte_size), 0)
         FROM blob_upload_reservations WHERE expires_at > ?",
    )
    .bind(&now)
    .fetch_one(&mut *conn)
    .await?;
    report.reservations = ByteCount {
        count: u64::try_from(reservation_count)?,
        bytes: u64::try_from(reservation_bytes)?,
    };
    for entry in fs::read_dir(staging_dir(blob_dir))
        .into_iter()
        .flatten()
        .flatten()
    {
        let name = entry.file_name().to_string_lossy().to_string();
        let metadata = entry.metadata()?;
        if name.starts_with(".aven-stage-") {
            report.staging.count += 1;
            report.staging.bytes += metadata.len();
        } else if validate_sha256(&name).is_ok() {
            let tracked: bool =
                sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM blob_inventory WHERE sha256 = ?)")
                    .bind(&name)
                    .fetch_one(&mut *conn)
                    .await?;
            if !tracked {
                report.staging.count += 1;
                report.staging.bytes += metadata.len();
                report.inconsistencies.count += 1;
                report.inconsistencies.bytes += metadata.len();
            }
        }
    }
    for entry in fs::read_dir(trash_dir(blob_dir))
        .into_iter()
        .flatten()
        .flatten()
    {
        if entry.file_type()?.is_file() {
            report.trash.count += 1;
            report.trash.bytes += entry.metadata()?.len();
        }
    }
    Ok(report)
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, Mutex};

    use chrono::DateTime;
    use sqlx::Connection as _;
    use sqlx::sqlite::SqliteConnectOptions;

    use crate::attachments::storage::{object_path, upsert_inventory_available};
    use crate::db::open_db;

    use super::*;

    #[derive(Clone)]
    struct TestClock(Arc<Mutex<DateTime<Utc>>>);

    impl TestClock {
        fn at(value: &str) -> Self {
            Self(Arc::new(Mutex::new(
                DateTime::parse_from_rfc3339(value).unwrap().to_utc(),
            )))
        }

        fn advance(&self, duration: chrono::Duration) {
            let mut now = self.0.lock().unwrap();
            *now += duration;
        }
    }

    impl Clock for TestClock {
        fn now(&self) -> DateTime<Utc> {
            *self.0.lock().unwrap()
        }
    }

    async fn insert_task(conn: &mut SqliteConnection, task_id: &str) {
        sqlx::query(
            "INSERT INTO tasks(
               workspace_id, id, title, description, project_id, status, priority,
               created_at, updated_at, queue_activity_at
             ) VALUES ('0000000000000000', ?, 'task', '', 'project', 'inbox', 'none',
                       '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z')",
        )
        .bind(task_id)
        .execute(conn)
        .await
        .unwrap();
    }

    async fn insert_attachment(
        conn: &mut SqliteConnection,
        attachment_id: &str,
        task_id: &str,
        sha256: &str,
        deleted: bool,
    ) {
        sqlx::query(
            "INSERT INTO task_attachments(
               workspace_id, attachment_id, task_id, sha256, byte_size, media_type, width, height,
               created_at, deleted, deleted_at
             ) VALUES ('0000000000000000', ?, ?, ?, 4, 'image/png', 1, 1,
                       '2026-01-01T00:00:00Z', ?, ?)",
        )
        .bind(attachment_id)
        .bind(task_id)
        .bind(sha256)
        .bind(i64::from(deleted))
        .bind(deleted.then_some("2026-01-01T00:00:00Z"))
        .execute(conn)
        .await
        .unwrap();
    }

    #[tokio::test]
    async fn final_live_reference_starts_grace_once_and_restore_clears_it() {
        let temp = tempfile::tempdir().unwrap();
        let pool = open_db(&temp.path().join("test.sqlite")).await.unwrap();
        let mut conn = pool.acquire().await.unwrap();
        let hash = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
        upsert_inventory_available(&mut conn, hash, 4, "image/png")
            .await
            .unwrap();
        insert_task(&mut conn, "0000000000000001").await;
        insert_task(&mut conn, "0000000000000002").await;
        insert_attachment(
            &mut conn,
            "0000000000000011",
            "0000000000000001",
            hash,
            false,
        )
        .await;
        insert_attachment(
            &mut conn,
            "0000000000000012",
            "0000000000000002",
            hash,
            false,
        )
        .await;
        let clock = TestClock::at("2026-07-01T00:00:00Z");

        reconcile_liveness(&mut conn, &clock).await.unwrap();
        sqlx::query("UPDATE task_attachments SET deleted = 1, deleted_at = 'x' WHERE attachment_id = '0000000000000011'")
            .execute(&mut *conn).await.unwrap();
        reconcile_liveness(&mut conn, &clock).await.unwrap();
        let value: Option<String> =
            sqlx::query_scalar("SELECT unreferenced_at FROM blob_lifecycle WHERE sha256 = ?")
                .bind(hash)
                .fetch_one(&mut *conn)
                .await
                .unwrap();
        assert_eq!(value, None, "one live reference keeps the hash live");

        sqlx::query("UPDATE task_attachments SET deleted = 1, deleted_at = 'x' WHERE attachment_id = '0000000000000012'")
            .execute(&mut *conn).await.unwrap();
        reconcile_liveness(&mut conn, &clock).await.unwrap();
        let first: String =
            sqlx::query_scalar("SELECT unreferenced_at FROM blob_lifecycle WHERE sha256 = ?")
                .bind(hash)
                .fetch_one(&mut *conn)
                .await
                .unwrap();
        clock.advance(chrono::Duration::days(1));
        reconcile_liveness(&mut conn, &clock).await.unwrap();
        let second: String =
            sqlx::query_scalar("SELECT unreferenced_at FROM blob_lifecycle WHERE sha256 = ?")
                .bind(hash)
                .fetch_one(&mut *conn)
                .await
                .unwrap();
        assert_eq!(first, second, "grace starts exactly once");

        sqlx::query("UPDATE task_attachments SET deleted = 0, deleted_at = NULL WHERE attachment_id = '0000000000000012'")
            .execute(&mut *conn).await.unwrap();
        reconcile_liveness(&mut conn, &clock).await.unwrap();
        let restored: Option<String> =
            sqlx::query_scalar("SELECT unreferenced_at FROM blob_lifecycle WHERE sha256 = ?")
                .bind(hash)
                .fetch_one(&mut *conn)
                .await
                .unwrap();
        assert_eq!(restored, None);
    }

    #[tokio::test]
    async fn affected_liveness_reconciliation_is_scoped_and_write_minimal() {
        let temp = tempfile::tempdir().unwrap();
        let pool = open_db(&temp.path().join("test.sqlite")).await.unwrap();
        let mut conn = pool.acquire().await.unwrap();
        let affected =
            "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".to_string();
        let unrelated =
            "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".to_string();
        let clock = TestClock::at("2026-07-01T00:00:00Z");
        for hash in [&affected, &unrelated] {
            upsert_inventory_available(&mut conn, hash, 4, "image/png")
                .await
                .unwrap();
        }
        insert_task(&mut conn, "0000000000000001").await;
        insert_attachment(
            &mut conn,
            "0000000000000011",
            "0000000000000001",
            &affected,
            false,
        )
        .await;
        for hash in [&affected, &unrelated] {
            sqlx::query("INSERT INTO blob_lifecycle(sha256, unreferenced_at) VALUES (?, ?)")
                .bind(hash)
                .bind("2026-06-01T00:00:00Z")
                .execute(&mut *conn)
                .await
                .unwrap();
        }
        sqlx::query(
            "CREATE TABLE lifecycle_updates(count INTEGER NOT NULL DEFAULT 0);
             CREATE TRIGGER count_lifecycle_updates
             AFTER UPDATE OF unreferenced_at ON blob_lifecycle
             BEGIN
                 INSERT INTO lifecycle_updates(count) VALUES (1);
             END",
        )
        .execute(&mut *conn)
        .await
        .unwrap();

        reconcile_liveness_for_hashes_in_transaction(
            &mut conn,
            std::slice::from_ref(&affected),
            &clock,
        )
        .await
        .unwrap();
        let unrelated_at: String =
            sqlx::query_scalar("SELECT unreferenced_at FROM blob_lifecycle WHERE sha256 = ?")
                .bind(&unrelated)
                .fetch_one(&mut *conn)
                .await
                .unwrap();
        assert_eq!(unrelated_at, "2026-06-01T00:00:00Z");
        let updates: i64 = sqlx::query_scalar("SELECT count(*) FROM lifecycle_updates")
            .fetch_one(&mut *conn)
            .await
            .unwrap();
        assert_eq!(updates, 1, "the live affected row changes once");

        reconcile_liveness_for_hashes_in_transaction(
            &mut conn,
            std::slice::from_ref(&affected),
            &clock,
        )
        .await
        .unwrap();
        let repeated_updates: i64 = sqlx::query_scalar("SELECT count(*) FROM lifecycle_updates")
            .fetch_one(&mut *conn)
            .await
            .unwrap();
        assert_eq!(repeated_updates, 1, "an already-live row is not rewritten");

        sqlx::query(
            "UPDATE task_attachments SET deleted = 1, deleted_at = 'x'
             WHERE attachment_id = '0000000000000011'",
        )
        .execute(&mut *conn)
        .await
        .unwrap();
        reconcile_liveness_for_hashes_in_transaction(
            &mut conn,
            std::slice::from_ref(&affected),
            &clock,
        )
        .await
        .unwrap();
        let first_unreferenced: String =
            sqlx::query_scalar("SELECT unreferenced_at FROM blob_lifecycle WHERE sha256 = ?")
                .bind(&affected)
                .fetch_one(&mut *conn)
                .await
                .unwrap();
        assert_eq!(first_unreferenced, "2026-07-01T00:00:00Z");
        clock.advance(chrono::Duration::days(1));
        reconcile_liveness_for_hashes_in_transaction(
            &mut conn,
            std::slice::from_ref(&affected),
            &clock,
        )
        .await
        .unwrap();
        let second_unreferenced: String =
            sqlx::query_scalar("SELECT unreferenced_at FROM blob_lifecycle WHERE sha256 = ?")
                .bind(&affected)
                .fetch_one(&mut *conn)
                .await
                .unwrap();
        assert_eq!(first_unreferenced, second_unreferenced);

        sqlx::query(
            "UPDATE task_attachments SET deleted = 0, deleted_at = NULL
             WHERE attachment_id = '0000000000000011'",
        )
        .execute(&mut *conn)
        .await
        .unwrap();
        reconcile_liveness(&mut conn, &clock).await.unwrap();
        let restored: Option<String> =
            sqlx::query_scalar("SELECT unreferenced_at FROM blob_lifecycle WHERE sha256 = ?")
                .bind(&affected)
                .fetch_one(&mut *conn)
                .await
                .unwrap();
        assert_eq!(restored, None);
        let full_updates: i64 = sqlx::query_scalar("SELECT count(*) FROM lifecycle_updates")
            .fetch_one(&mut *conn)
            .await
            .unwrap();
        assert_eq!(full_updates, 3, "the full reset writes only the transition");
    }

    #[tokio::test]
    async fn lease_protects_expired_unreferenced_blob_until_release() {
        let temp = tempfile::tempdir().unwrap();
        let db_path = temp.path().join("test.sqlite");
        let pool = open_db(&db_path).await.unwrap();
        let mut conn = pool.acquire().await.unwrap();
        let blob_dir = temp.path().join("blobs");
        let hash = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
        upsert_inventory_available(&mut conn, hash, 4, "image/png")
            .await
            .unwrap();
        let path = object_path(&blob_dir, hash).unwrap();
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        fs::write(&path, b"blob").unwrap();
        let clock = TestClock::at("2026-07-10T00:00:00Z");
        reconcile_liveness(&mut conn, &clock).await.unwrap();
        clock.advance(chrono::Duration::days(8));
        let lease = acquire_lease(&mut conn, hash, "backup", &clock)
            .await
            .unwrap();
        let policy = LifecyclePolicy::default();
        let blocked = prune(&mut conn, &blob_dir, policy, true, &clock)
            .await
            .unwrap();
        assert_eq!(blocked.pruned.count, 0);
        assert!(path.exists());

        release_lease(&mut conn, &lease).await.unwrap();
        let pruned = prune(&mut conn, &blob_dir, policy, true, &clock)
            .await
            .unwrap();
        assert_eq!(pruned.pruned.count, 1);
        assert!(!path.exists());
    }

    #[tokio::test]
    async fn quota_is_unique_by_hash_and_reservations_are_workspace_scoped() {
        let temp = tempfile::tempdir().unwrap();
        let pool = open_db(&temp.path().join("test.sqlite")).await.unwrap();
        let mut conn = pool.acquire().await.unwrap();
        let hash = "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc";
        let clock = TestClock::at("2026-07-01T00:00:00Z");
        let first = reserve_upload(&mut conn, "workspace-a", hash, 8, 8, &clock)
            .await
            .unwrap();
        assert!(first.is_some());
        let replacement = reserve_upload(&mut conn, "workspace-a", hash, 8, 8, &clock)
            .await
            .unwrap();
        assert!(replacement.is_some());
        let other = reserve_upload(&mut conn, "workspace-b", hash, 8, 8, &clock)
            .await
            .unwrap();
        assert!(other.is_some());
        let count: i64 = sqlx::query_scalar("SELECT count(*) FROM blob_upload_reservations")
            .fetch_one(&mut *conn)
            .await
            .unwrap();
        assert_eq!(count, 2);
    }

    #[tokio::test]
    async fn concurrent_attach_wins_prune_recheck() {
        let temp = tempfile::tempdir().unwrap();
        let db_path = temp.path().join("test.sqlite");
        let pool = open_db(&db_path).await.unwrap();
        let mut conn = pool.acquire().await.unwrap();
        let blob_dir = temp.path().join("blobs");
        let hash = "abababababababababababababababababababababababababababababababab";
        upsert_inventory_available(&mut conn, hash, 4, "image/png")
            .await
            .unwrap();
        insert_task(&mut conn, "0000000000000003").await;
        let path = object_path(&blob_dir, hash).unwrap();
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        fs::write(&path, b"blob").unwrap();
        let clock = TestClock::at("2026-07-01T00:00:00Z");
        reconcile_liveness(&mut conn, &clock).await.unwrap();
        clock.advance(chrono::Duration::days(8));
        let options = SqliteConnectOptions::new()
            .filename(&db_path)
            .busy_timeout(Duration::from_secs(5));
        let prune_conn = SqliteConnection::connect_with(&options).await.unwrap();

        let mut tx = begin_immediate(&mut conn).await.unwrap();
        insert_attachment(&mut tx, "0000000000000013", "0000000000000003", hash, false).await;
        let prune_dir = blob_dir.clone();
        let prune_clock = clock.clone();
        let pruning = tokio::spawn(async move {
            let mut prune_conn = prune_conn;
            prune(
                &mut prune_conn,
                &prune_dir,
                LifecyclePolicy::default(),
                true,
                &prune_clock,
            )
            .await
            .unwrap()
        });
        tokio::time::sleep(Duration::from_millis(50)).await;
        tx.commit().await.unwrap();

        let summary = pruning.await.unwrap();
        assert_eq!(summary.pruned.count, 0);
        assert!(path.exists());
    }

    #[tokio::test]
    async fn accepted_server_reference_keeps_blob_live() {
        let temp = tempfile::tempdir().unwrap();
        let pool = open_db(&temp.path().join("test.sqlite")).await.unwrap();
        let mut conn = pool.acquire().await.unwrap();
        let hash = "acacacacacacacacacacacacacacacacacacacacacacacacacacacacacacacac";
        upsert_inventory_available(&mut conn, hash, 4, "image/png")
            .await
            .unwrap();
        sqlx::query(
            "INSERT INTO server_blob_references(
               workspace_id, attachment_id, task_id, sha256, byte_size
             ) VALUES ('workspace', '0000000000000042', '0000000000000004', ?, 4)",
        )
        .bind(hash)
        .execute(&mut *conn)
        .await
        .unwrap();
        let clock = TestClock::at("2026-07-01T00:00:00Z");

        reconcile_liveness(&mut conn, &clock).await.unwrap();
        let unreferenced_at: Option<String> =
            sqlx::query_scalar("SELECT unreferenced_at FROM blob_lifecycle WHERE sha256 = ?")
                .bind(hash)
                .fetch_one(&mut *conn)
                .await
                .unwrap();
        assert_eq!(unreferenced_at, None);
    }

    #[tokio::test]
    async fn local_quota_boundary_is_hash_idempotent() {
        let temp = tempfile::tempdir().unwrap();
        let db_path = temp.path().join("test.sqlite");
        let pool = open_db(&db_path).await.unwrap();
        let mut conn = pool.acquire().await.unwrap();
        let blob_dir = temp.path().join("blobs");
        let existing = "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee";
        let new_hash = "ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff";
        upsert_inventory_available(&mut conn, existing, 8, "image/png")
            .await
            .unwrap();
        let clock = TestClock::at("2026-07-01T00:00:00Z");
        let policy = LifecyclePolicy {
            quota_bytes: 8,
            ..LifecyclePolicy::default()
        };

        ensure_local_capacity(&mut conn, &blob_dir, existing, 8, policy, &clock)
            .await
            .unwrap();
        let error = ensure_local_capacity(&mut conn, &blob_dir, new_hash, 1, policy, &clock)
            .await
            .unwrap_err();
        assert_eq!(error.to_string(), "error attachment-quota-exceeded");
    }

    #[tokio::test]
    async fn unavailable_directory_releases_cursor_owned_by_lexical_path() {
        let temp = tempfile::tempdir().unwrap();
        let scan_root = temp.path().join("scan");
        let lexical_root = temp.path().join("anchor").join("..").join("scan");
        fs::create_dir_all(temp.path().join("anchor")).unwrap();
        fs::create_dir_all(&scan_root).unwrap();
        fs::write(scan_root.join("first"), b"first").unwrap();
        fs::write(scan_root.join("second"), b"second").unwrap();
        let canonical_root = fs::canonicalize(&lexical_root).unwrap();

        scan_directory_page(lexical_root.clone(), 1).await.unwrap();
        assert!(
            DIRECTORY_CURSORS
                .lock()
                .unwrap()
                .contains_key(&lexical_root)
        );

        fs::remove_dir_all(&scan_root).unwrap();
        let _ = scan_directory_page(lexical_root.clone(), 1).await;

        let cursors = DIRECTORY_CURSORS.lock().unwrap();
        assert!(!cursors.contains_key(&lexical_root));
        assert!(!cursors.contains_key(&canonical_root));
    }

    #[tokio::test]
    async fn directory_iteration_error_releases_cursor_state() {
        let temp = tempfile::tempdir().unwrap();
        let dir = temp.path().join("scan");
        fs::create_dir_all(&dir).unwrap();
        fs::write(dir.join("first"), b"first").unwrap();
        fs::write(dir.join("second"), b"second").unwrap();

        scan_directory_page(dir.clone(), 1).await.unwrap();
        assert!(DIRECTORY_CURSORS.lock().unwrap().contains_key(&dir));

        let result = scan_directory_page_with(dir.clone(), 1, |_| {
            Err(io::Error::other("injected directory iteration failure"))
        })
        .await;
        assert!(result.as_ref().is_err_and(|error| {
            error
                .to_string()
                .contains("injected directory iteration failure")
        }));
        assert!(!DIRECTORY_CURSORS.lock().unwrap().contains_key(&dir));
    }

    #[test]
    fn file_move_copies_atomically_when_hard_links_are_unavailable() {
        let temp = tempfile::tempdir().unwrap();
        let source = temp.path().join("source");
        let target = temp.path().join("target");
        fs::write(&source, b"attachment").unwrap();

        let moved = move_file_without_replacing_with(&source, &target, |_, _| {
            Err(io::Error::new(
                io::ErrorKind::Unsupported,
                "injected hard-link failure",
            ))
        })
        .unwrap();

        assert_eq!(moved, FileMove::Moved);
        assert!(!source.exists());
        assert_eq!(fs::read(target).unwrap(), b"attachment");
    }

    #[test]
    fn file_move_copy_fallback_does_not_replace_target() {
        let temp = tempfile::tempdir().unwrap();
        let source = temp.path().join("source");
        let target = temp.path().join("target");
        fs::write(&source, b"source").unwrap();
        fs::write(&target, b"target").unwrap();

        let moved = move_file_without_replacing_with(&source, &target, |_, _| {
            Err(io::Error::new(
                io::ErrorKind::Unsupported,
                "injected hard-link failure",
            ))
        })
        .unwrap();

        assert_eq!(moved, FileMove::TargetExists);
        assert_eq!(fs::read(source).unwrap(), b"source");
        assert_eq!(fs::read(target).unwrap(), b"target");
    }

    #[tokio::test]
    async fn interrupted_atomic_create_object_is_reconciled_after_grace() {
        let temp = tempfile::tempdir().unwrap();
        let db_path = temp.path().join("test.sqlite");
        let pool = open_db(&db_path).await.unwrap();
        let mut conn = pool.acquire().await.unwrap();
        let blob_dir = temp.path().join("blobs");
        let hash = "cdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcd";
        let path = object_path(&blob_dir, hash).unwrap();
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        fs::write(&path, b"orphan").unwrap();
        let clock = TestClock::at("2030-07-01T00:00:00Z");
        let policy = LifecyclePolicy {
            grace: Duration::ZERO,
            ..LifecyclePolicy::default()
        };

        prune(&mut conn, &blob_dir, policy, true, &clock)
            .await
            .unwrap();
        assert!(!path.exists());
    }

    #[tokio::test]
    async fn orphan_traversal_obeys_limit_and_resumes_between_steps() {
        let temp = tempfile::tempdir().unwrap();
        let pool = open_db(&temp.path().join("test.sqlite")).await.unwrap();
        let mut conn = pool.acquire().await.unwrap();
        let blob_dir = temp.path().join("blobs");
        for digit in ['6', '7', '8', '9', 'a'] {
            let hash = digit.to_string().repeat(64);
            let path = object_path(&blob_dir, &hash).unwrap();
            fs::create_dir_all(path.parent().unwrap()).unwrap();
            fs::write(path, b"orphan").unwrap();
        }
        let clock = TestClock::at("2030-07-01T00:00:00Z");
        let policy = LifecyclePolicy {
            grace: Duration::ZERO,
            maintenance_limit: 2,
            ..LifecyclePolicy::default()
        };

        prune(&mut conn, &blob_dir, policy, true, &clock)
            .await
            .unwrap();
        let after_one = fs::read_dir(staging_dir(&blob_dir)).unwrap().count();
        assert!(after_one >= 3);

        for _ in 0..4 {
            prune(&mut conn, &blob_dir, policy, true, &clock)
                .await
                .unwrap();
        }
        assert_eq!(fs::read_dir(staging_dir(&blob_dir)).unwrap().count(), 0);
    }

    #[tokio::test]
    async fn interrupted_trash_move_restores_available_object() {
        let temp = tempfile::tempdir().unwrap();
        let db_path = temp.path().join("test.sqlite");
        let pool = open_db(&db_path).await.unwrap();
        let mut conn = pool.acquire().await.unwrap();
        let blob_dir = temp.path().join("blobs");
        let hash = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd";
        upsert_inventory_available(&mut conn, hash, 4, "image/png")
            .await
            .unwrap();
        let source = object_path(&blob_dir, hash).unwrap();
        let trash = trash_dir(&blob_dir).join(hash);
        fs::create_dir_all(source.parent().unwrap()).unwrap();
        fs::create_dir_all(trash.parent().unwrap()).unwrap();
        fs::write(&source, b"blob").unwrap();
        fs::rename(&source, &trash).unwrap();

        reconcile_trash(&mut conn, &blob_dir).await.unwrap();
        assert!(source.exists());
        assert!(!trash.exists());
    }
}