//! Content-addressed blob synchronization.
//!
//! Blob sync is a separate pass the scheduler interleaves *after* the metadata
//! pull, and it is non-fatal: a failed blob never aborts a sync cycle. An app
//! declares its blob policy via [`BlobPolicy`], which rows own a blob, where it
//! lives on disk, and how to reflect local-vs-cloud presence, and the engine
//! owns the invariant machinery: dedup-aware upload, integrity-verified download,
//! and atomic writes.
//!
//! Like [`super::sync`], the client is abstracted behind a trait
//! ([`BlobTransport`]) so the engine tests end-to-end against an in-memory store,
//! and `SyncKitClient` satisfies it in production.

use std::future::Future;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use rusqlite::Connection;
use sha2::{Digest, Sha256};

use super::db::DbSource;
use crate::client::{BlobUploadOutcome, SyncKitClient};
use crate::error::{Result, SyncKitError};

/// A content-addressed blob: its SHA-256 hash (the address), file extension, and
/// plaintext size in bytes.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BlobRef {
    /// SHA-256 content address (the blob's identity).
    pub hash: String,
    /// File extension, without the leading dot.
    pub ext: String,
    /// Plaintext size in bytes.
    pub size: u64,
}

/// The blob operations the engine needs from a server. A seam for testing and
/// the reference a future non-Rust SDK mirrors; [`SyncKitClient`] forwards to its
/// inherent `blob_*` methods.
pub trait BlobTransport: Sync {
    /// Upload the file at `path` under the content address `hash`, or report
    /// that the server already holds it. Takes a path rather than bytes so the
    /// implementation can stream: the engine must never be the reason a blob is
    /// read whole into memory.
    fn upload_file(
        &self,
        hash: &str,
        path: &Path,
    ) -> impl Future<Output = Result<BlobUploadOutcome>> + Send;
    /// Confirm an uploaded blob so the server commits it at `hash`.
    fn confirm(&self, hash: &str, size: u64) -> impl Future<Output = Result<()>> + Send;
    /// Presigned download URL for the blob at `hash`.
    fn download_url(&self, hash: &str) -> impl Future<Output = Result<String>> + Send;
    /// Download the blob at `hash` from `url`, returning its ciphertext bytes.
    fn download(&self, hash: &str, url: &str) -> impl Future<Output = Result<Vec<u8>>> + Send;
}

// The trait declares `-> impl Future + Send`; implementing with `async fn` is
// equivalent (the Send bound is enforced by the trait) and reads cleaner.
impl BlobTransport for SyncKitClient {
    async fn upload_file(&self, hash: &str, path: &Path) -> Result<BlobUploadOutcome> {
        SyncKitClient::blob_upload_streaming(self, hash, path).await
    }
    async fn confirm(&self, hash: &str, size: u64) -> Result<()> {
        SyncKitClient::blob_confirm(self, hash, size as i64).await
    }
    async fn download_url(&self, hash: &str) -> Result<String> {
        SyncKitClient::blob_download_url(self, hash).await
    }
    async fn download(&self, hash: &str, url: &str) -> Result<Vec<u8>> {
        SyncKitClient::blob_download(self, hash, url).await
    }
}

/// Which rows own blobs, where they live, and how to reflect their presence.
///
/// The three query/path methods are required; the presence hooks default to
/// no-ops for an app that tracks presence purely by disk existence (GoingsOn). An
/// app with an explicit presence flag (audiofiles' `cloud_only`) overrides them.
pub trait BlobPolicy: Send + Sync {
    /// Rows whose bytes should be uploaded. May return the full candidate set,
    /// the engine skips any whose [`local_path`](Self::local_path) is absent.
    fn pending_uploads(&self, conn: &Connection) -> Result<Vec<BlobRef>>;
    /// Rows whose bytes are wanted. The engine skips any already present on disk.
    fn pending_downloads(&self, conn: &Connection) -> Result<Vec<BlobRef>>;
    /// The on-disk location for a blob.
    fn local_path(&self, blob: &BlobRef) -> PathBuf;
    /// Reconcile presence flags before the passes (e.g. mark missing files cloud-only).
    fn reconcile(&self, _conn: &Connection) -> Result<()> {
        Ok(())
    }
    /// Reflect that a blob's bytes are now on the server.
    fn on_uploaded(&self, _conn: &Connection, _blob: &BlobRef) -> Result<()> {
        Ok(())
    }
    /// Reflect that a blob's bytes are now on local disk.
    fn on_downloaded(&self, _conn: &Connection, _blob: &BlobRef) -> Result<()> {
        Ok(())
    }
}

/// Result of a blob sync pass.
#[derive(Debug, Default, PartialEq, Eq)]
pub struct BlobOutcome {
    /// Number of blobs uploaded this pass.
    pub uploaded: u64,
    /// Number of blobs downloaded this pass.
    pub downloaded: u64,
}

type Policy = Arc<dyn BlobPolicy>;

fn join_err(e: &tokio::task::JoinError) -> SyncKitError {
    SyncKitError::Internal(format!("blocking task failed: {e}"))
}

fn io_err(context: &str, e: &std::io::Error) -> SyncKitError {
    SyncKitError::Internal(format!("blob {context}: {e}"))
}

/// First 8 chars of a hash for logs, panic-safe on short/multibyte input.
fn short(hash: &str) -> &str {
    hash.get(..8).unwrap_or(hash)
}

/// Run a policy database method on a blocking thread with a fresh connection.
async fn on_conn<F, R>(db: &DbSource, policy: &Policy, f: F) -> Result<R>
where
    F: FnOnce(&dyn BlobPolicy, &Connection) -> Result<R> + Send + 'static,
    R: Send + 'static,
{
    let db = db.clone();
    let policy = policy.clone();
    tokio::task::spawn_blocking(move || f(policy.as_ref(), &db.open()?))
        .await
        .map_err(|e| join_err(&e))?
}

/// The full blob pass: reconcile, then upload local, then download missing.
/// Non-fatal by construction, per-blob failures are logged and skipped.
pub async fn sync_blobs<T: BlobTransport>(
    db: &DbSource,
    client: &T,
    policy: &Policy,
) -> Result<BlobOutcome> {
    on_conn(db, policy, |p, c| p.reconcile(c)).await?;
    let uploaded = upload_blobs(db, client, policy).await?;
    let downloaded = download_blobs(db, client, policy).await?;
    Ok(BlobOutcome {
        uploaded,
        downloaded,
    })
}

/// Upload every pending local blob the server doesn't already have.
pub async fn upload_blobs<T: BlobTransport>(
    db: &DbSource,
    client: &T,
    policy: &Policy,
) -> Result<u64> {
    let pending = on_conn(db, policy, |p, c| p.pending_uploads(c)).await?;
    let mut uploaded = 0u64;
    for blob in pending {
        match upload_one(db, client, policy, &blob).await {
            Ok(true) => uploaded += 1,
            Ok(false) => {}
            Err(e) => tracing::warn!(
                hash = short(&blob.hash),
                "blob upload failed (non-fatal): {e}"
            ),
        }
    }
    Ok(uploaded)
}

async fn upload_one<T: BlobTransport>(
    db: &DbSource,
    client: &T,
    policy: &Policy,
    blob: &BlobRef,
) -> Result<bool> {
    let path = policy.local_path(blob);
    if !tokio::fs::try_exists(&path).await.unwrap_or(false) {
        return Ok(false); // no local file to upload
    }
    // The transport streams from the path, so a blob is never read whole into
    // memory here regardless of size, and it runs its own dedup check before
    // touching the file.
    if client.upload_file(&blob.hash, &path).await? == BlobUploadOutcome::AlreadyPresent {
        // Content-addressed dedup: the server says it holds this hash, so the
        // bytes stay home. Two consequences worth knowing, both bounded:
        //   - The claim is unverified. A server that answers `true` wrongly
        //     withholds the blob from other devices; it cannot substitute one,
        //     because `download_one` re-hashes before committing the file.
        //   - `on_uploaded` is deliberately not fired: nothing was uploaded,
        //     and the confirm call that pairs with it never happened. A policy
        //     using that hook to clear a presence flag will keep re-offering
        //     the blob on every pass, costing one round trip each time.
        //     Harmless for the no-op default (the only implementors today), but
        //     a policy with an explicit flag needs its own way to settle the
        //     dedup case.
        return Ok(true);
    }
    client.confirm(&blob.hash, blob.size).await?;

    let blob = blob.clone();
    on_conn(db, policy, move |p, c| p.on_uploaded(c, &blob)).await?;
    Ok(true)
}

/// Download every wanted blob not already on local disk.
pub async fn download_blobs<T: BlobTransport>(
    db: &DbSource,
    client: &T,
    policy: &Policy,
) -> Result<u64> {
    let pending = on_conn(db, policy, |p, c| p.pending_downloads(c)).await?;
    let mut downloaded = 0u64;
    for blob in pending {
        let path = policy.local_path(&blob);
        if tokio::fs::try_exists(&path).await.unwrap_or(false) {
            continue; // already have it locally
        }
        match download_one(db, client, policy, &blob).await {
            Ok(true) => downloaded += 1,
            Ok(false) => {}
            Err(e) => tracing::warn!(
                hash = short(&blob.hash),
                "blob download failed (non-fatal): {e}"
            ),
        }
    }
    Ok(downloaded)
}

/// Download a single blob by reference, unconditionally (the GUI "download now"
/// path). Verifies the content hash before committing the file.
pub async fn download_one<T: BlobTransport>(
    db: &DbSource,
    client: &T,
    policy: &Policy,
    blob: &BlobRef,
) -> Result<bool> {
    let path = policy.local_path(blob);
    let url = client.download_url(&blob.hash).await?;
    let data = client.download(&blob.hash, &url).await?;

    // Verify the plaintext hashes to its address before it lands under that name.
    // The SDK's blob_download also re-verifies; this guards the store regardless
    // of transport (and is what the in-memory tests exercise).
    let actual = hex::encode(Sha256::digest(&data));
    if actual != blob.hash {
        return Err(SyncKitError::IntegrityFailed {
            expected: blob.hash.clone(),
            actual,
        });
    }

    if let Some(parent) = path.parent() {
        tokio::fs::create_dir_all(parent)
            .await
            .map_err(|e| io_err("mkdir", &e))?;
    }
    // Atomic: write to a sibling temp then rename, so a crash never leaves a
    // half-written file under the content-addressed name.
    let tmp = {
        let mut s = path.clone().into_os_string();
        s.push(".downloading");
        PathBuf::from(s)
    };
    tokio::fs::write(&tmp, &data)
        .await
        .map_err(|e| io_err("write", &e))?;
    tokio::fs::rename(&tmp, &path)
        .await
        .map_err(|e| io_err("rename", &e))?;

    let blob = blob.clone();
    on_conn(db, policy, move |p, c| p.on_downloaded(c, &blob)).await?;
    Ok(true)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashMap;
    use std::sync::Mutex;

    /// In-memory content-addressed blob store standing in for the server.
    #[derive(Clone, Default)]
    struct FakeBlobs {
        store: Arc<Mutex<HashMap<String, Vec<u8>>>>,
    }
    impl FakeBlobs {
        fn put_raw(&self, hash: &str, bytes: Vec<u8>) {
            self.store.lock().unwrap().insert(hash.into(), bytes);
        }
        fn has(&self, hash: &str) -> bool {
            self.store.lock().unwrap().contains_key(hash)
        }
    }
    impl BlobTransport for FakeBlobs {
        async fn upload_file(&self, hash: &str, path: &Path) -> Result<BlobUploadOutcome> {
            if self.has(hash) {
                return Ok(BlobUploadOutcome::AlreadyPresent);
            }
            let data = std::fs::read(path).map_err(|e| io_err("read", &e))?;
            self.store.lock().unwrap().insert(hash.to_string(), data);
            Ok(BlobUploadOutcome::Uploaded)
        }
        async fn confirm(&self, _hash: &str, _size: u64) -> Result<()> {
            Ok(())
        }
        async fn download_url(&self, hash: &str) -> Result<String> {
            Ok(format!("fake://get/{hash}"))
        }
        async fn download(&self, hash: &str, _url: &str) -> Result<Vec<u8>> {
            self.store
                .lock()
                .unwrap()
                .get(hash)
                .cloned()
                .ok_or_else(|| SyncKitError::Internal("blob not on server".into()))
        }
    }

    /// A policy backed by a `blobmeta(hash, ext, size, cloud_only)` table.
    struct TestPolicy {
        dir: PathBuf,
    }
    impl BlobPolicy for TestPolicy {
        fn pending_uploads(&self, conn: &Connection) -> Result<Vec<BlobRef>> {
            rows(
                conn,
                "SELECT hash, ext, size FROM blobmeta WHERE cloud_only = 0",
            )
        }
        fn pending_downloads(&self, conn: &Connection) -> Result<Vec<BlobRef>> {
            rows(
                conn,
                "SELECT hash, ext, size FROM blobmeta WHERE cloud_only = 1",
            )
        }
        fn local_path(&self, blob: &BlobRef) -> PathBuf {
            self.dir.join(format!("{}.{}", blob.hash, blob.ext))
        }
        fn reconcile(&self, conn: &Connection) -> Result<()> {
            // Flip cloud_only=0 rows whose file is missing to cloud_only=1.
            let missing = rows(
                conn,
                "SELECT hash, ext, size FROM blobmeta WHERE cloud_only = 0",
            )?
            .into_iter()
            .filter(|b| !self.local_path(b).exists())
            .collect::<Vec<_>>();
            for b in missing {
                conn.execute(
                    "UPDATE blobmeta SET cloud_only = 1 WHERE hash = ?1",
                    [&b.hash],
                )?;
            }
            Ok(())
        }
        fn on_downloaded(&self, conn: &Connection, blob: &BlobRef) -> Result<()> {
            conn.execute(
                "UPDATE blobmeta SET cloud_only = 0 WHERE hash = ?1",
                [&blob.hash],
            )?;
            Ok(())
        }
    }

    fn rows(conn: &Connection, sql: &str) -> Result<Vec<BlobRef>> {
        let mut stmt = conn.prepare(sql)?;
        let out = stmt
            .query_map([], |r| {
                Ok(BlobRef {
                    hash: r.get(0)?,
                    ext: r.get(1)?,
                    size: r.get::<_, i64>(2)? as u64,
                })
            })?
            .collect::<rusqlite::Result<Vec<_>>>()?;
        Ok(out)
    }

    fn hash_of(content: &[u8]) -> String {
        hex::encode(Sha256::digest(content))
    }

    fn tempdir() -> PathBuf {
        use std::sync::atomic::{AtomicU64, Ordering};
        static N: AtomicU64 = AtomicU64::new(0);
        let mut p = std::env::temp_dir();
        p.push(format!(
            "synckit_b6_{}_{}",
            std::process::id(),
            N.fetch_add(1, Ordering::Relaxed)
        ));
        std::fs::create_dir_all(&p).unwrap();
        p
    }

    fn setup(dir: &std::path::Path) -> (DbSource, Policy) {
        let db_path = dir.join("meta.db");
        let db = DbSource::path(&db_path);
        db.open()
            .unwrap()
            .execute_batch("CREATE TABLE blobmeta (hash TEXT PRIMARY KEY, ext TEXT, size INT, cloud_only INT);")
            .unwrap();
        let content = dir.join("content");
        std::fs::create_dir_all(&content).unwrap();
        let policy: Policy = Arc::new(TestPolicy { dir: content });
        (db, policy)
    }

    fn seed_meta(db: &DbSource, hash: &str, ext: &str, size: u64, cloud_only: i64) {
        db.open()
            .unwrap()
            .execute(
                "INSERT INTO blobmeta (hash, ext, size, cloud_only) VALUES (?1, ?2, ?3, ?4)",
                (hash, ext, size as i64, cloud_only),
            )
            .unwrap();
    }

    #[tokio::test]
    async fn upload_sends_local_blob_and_dedups() {
        let dir = tempdir();
        let (db, policy) = setup(&dir);
        let server = FakeBlobs::default();
        let content = b"blob-one".to_vec();
        let h = hash_of(&content);
        seed_meta(&db, &h, "wav", content.len() as u64, 0);
        std::fs::write(
            policy.local_path(&BlobRef {
                hash: h.clone(),
                ext: "wav".into(),
                size: 0,
            }),
            &content,
        )
        .unwrap();

        assert_eq!(upload_blobs(&db, &server, &policy).await.unwrap(), 1);
        assert!(server.has(&h));
        // Second run: server already_exists → deduped, no error, still counted handled.
        assert_eq!(upload_blobs(&db, &server, &policy).await.unwrap(), 1);
    }

    #[tokio::test]
    async fn upload_skips_when_local_file_absent() {
        let dir = tempdir();
        let (db, policy) = setup(&dir);
        let server = FakeBlobs::default();
        seed_meta(&db, &hash_of(b"nope"), "wav", 4, 0); // no file on disk
        assert_eq!(upload_blobs(&db, &server, &policy).await.unwrap(), 0);
    }

    #[tokio::test]
    async fn download_verifies_writes_and_clears_cloud_only() {
        let dir = tempdir();
        let (db, policy) = setup(&dir);
        let server = FakeBlobs::default();
        let content = b"downloaded-bytes".to_vec();
        let h = hash_of(&content);
        server.put_raw(&h, content.clone());
        seed_meta(&db, &h, "flac", content.len() as u64, 1);

        assert_eq!(download_blobs(&db, &server, &policy).await.unwrap(), 1);
        let path = policy.local_path(&BlobRef {
            hash: h.clone(),
            ext: "flac".into(),
            size: 0,
        });
        assert_eq!(std::fs::read(&path).unwrap(), content);
        // on_downloaded cleared cloud_only.
        let co: i64 = db
            .open()
            .unwrap()
            .query_row("SELECT cloud_only FROM blobmeta WHERE hash=?1", [&h], |r| {
                r.get(0)
            })
            .unwrap();
        assert_eq!(co, 0);
        // No temp file left behind.
        assert!(
            !path
                .with_file_name(format!("{h}.flac.downloading"))
                .exists()
        );
    }

    #[tokio::test]
    async fn download_rejects_corrupt_bytes_and_writes_nothing() {
        let dir = tempdir();
        let (db, policy) = setup(&dir);
        let server = FakeBlobs::default();
        let h = hash_of(b"the-real-content");
        server.put_raw(&h, b"WRONG BYTES".to_vec()); // hash(bytes) != h
        seed_meta(&db, &h, "wav", 16, 1);

        // Non-fatal: download_blobs logs and counts 0; file not written.
        assert_eq!(download_blobs(&db, &server, &policy).await.unwrap(), 0);
        let path = policy.local_path(&BlobRef {
            hash: h.clone(),
            ext: "wav".into(),
            size: 0,
        });
        assert!(!path.exists());
        // And download_one surfaces the integrity error directly.
        let err = download_one(
            &db,
            &server,
            &policy,
            &BlobRef {
                hash: h,
                ext: "wav".into(),
                size: 16,
            },
        )
        .await;
        assert!(matches!(err, Err(SyncKitError::IntegrityFailed { .. })));
    }

    #[tokio::test]
    async fn download_skips_when_already_present() {
        let dir = tempdir();
        let (db, policy) = setup(&dir);
        let server = FakeBlobs::default();
        let content = b"already-here".to_vec();
        let h = hash_of(&content);
        seed_meta(&db, &h, "wav", content.len() as u64, 1);
        let path = policy.local_path(&BlobRef {
            hash: h.clone(),
            ext: "wav".into(),
            size: 0,
        });
        std::fs::write(&path, &content).unwrap();
        // Server does NOT have it; if the engine tried to download it would fail,
        // but it must skip because the file is present.
        assert_eq!(download_blobs(&db, &server, &policy).await.unwrap(), 0);
    }

    #[tokio::test]
    async fn reconcile_flags_missing_file_cloud_only() {
        let dir = tempdir();
        let (db, policy) = setup(&dir);
        let server = FakeBlobs::default();
        // A row claims local presence (cloud_only=0) but no file exists.
        let h = hash_of(b"ghost");
        seed_meta(&db, &h, "wav", 5, 0);
        sync_blobs(&db, &server, &policy).await.unwrap();
        let co: i64 = db
            .open()
            .unwrap()
            .query_row("SELECT cloud_only FROM blobmeta WHERE hash=?1", [&h], |r| {
                r.get(0)
            })
            .unwrap();
        assert_eq!(co, 1, "reconcile flips a missing-file row to cloud_only");
    }
}