//! The engine's resume store: interrupted blob uploads, on disk.
//!
//! Implements [`BlobResumeStore`] over the app's own SQLite database, so a
//! process killed mid-upload finds the session again on restart. The tables are
//! engine-owned bookkeeping like the changelog and the conflict stash: local
//! only, absent from every sync manifest, never pushed.
//!
//! Nothing here is secret. A nonce rides in the clear at the head of the sealed
//! chunk it belongs to, an ETag is an S3 identifier, and the plaintext digests
//! are of content the app already holds in the file being uploaded. Losing the
//! whole table costs a restart from zero and nothing else, which is what the
//! [`best_effort`](crate::client::resume) contract on the trait is about.
//!
//! See [`crate::client::resume`] for why the nonces have to be here at all.

use std::sync::Arc;

use rusqlite::{OptionalExtension, params};

use super::db::DbSource;
use crate::client::resume::{
    BlobResumeStore, ResumeChunk, ResumePart, ResumeRecord, ResumeSession,
};
use crate::crypto::BLOB_NONCE_LEN;
use crate::error::{Result, SyncKitError};

/// DDL for the resume tables.
///
/// Applied from [`configure_connection`](super::db::configure_connection)
/// rather than from `SyncSchema::migration_sql`, because an app that snapshotted
/// the generated migration into a versioned file would never see a table added
/// later. These are pure engine bookkeeping with no app-visible shape, so
/// creating them on connection open is both safe and the only way to guarantee
/// they exist wherever the engine runs.
pub(crate) const RESUME_DDL: &str = "\
-- An in-flight multipart blob upload, so a killed process resumes it.
CREATE TABLE IF NOT EXISTS sync_blob_resume (
    hash       TEXT PRIMARY KEY NOT NULL,
    upload_id  TEXT NOT NULL,
    part_size  INTEGER NOT NULL,
    part_count INTEGER NOT NULL,
    size_bytes INTEGER NOT NULL,
    created_at INTEGER NOT NULL
) WITHOUT ROWID;

-- One completed part, with the ETag S3 needs to assemble the object.
CREATE TABLE IF NOT EXISTS sync_blob_resume_part (
    hash        TEXT NOT NULL,
    part_number INTEGER NOT NULL,
    etag        TEXT NOT NULL,
    PRIMARY KEY (hash, part_number),
    FOREIGN KEY (hash) REFERENCES sync_blob_resume(hash) ON DELETE CASCADE
) WITHOUT ROWID;

-- The nonce each sealed chunk was sealed with, so the chunk spanning a part
-- boundary can be reproduced byte for byte. `plain_sha` is checked before the
-- nonce is re-used: sealing different plaintext under a used nonce would break
-- the cipher outright, so the resume path must be able to prove the file has
-- not changed underneath it.
CREATE TABLE IF NOT EXISTS sync_blob_resume_chunk (
    hash        TEXT NOT NULL,
    chunk_index INTEGER NOT NULL,
    nonce       BLOB NOT NULL,
    plain_sha   BLOB NOT NULL,
    PRIMARY KEY (hash, chunk_index),
    FOREIGN KEY (hash) REFERENCES sync_blob_resume(hash) ON DELETE CASCADE
) WITHOUT ROWID;
";

/// A [`BlobResumeStore`] over the engine's database.
///
/// Opens a connection per call rather than holding one: the call rate is one
/// per completed multipart part, which is megabytes of transfer apart, and a
/// long-lived second writer on the app's file would be a worse trade than the
/// open.
pub struct SqliteResumeStore {
    db: DbSource,
}

impl SqliteResumeStore {
    /// A resume store backed by `db`.
    pub fn new(db: DbSource) -> Self {
        Self { db }
    }

    /// A resume store as the client wants it.
    pub fn shared(db: DbSource) -> Arc<dyn BlobResumeStore> {
        Arc::new(Self::new(db))
    }

    fn conn(&self) -> Result<rusqlite::Connection> {
        let conn = self.db.open()?;
        // The app's own connections are on the same file. A blob pass runs
        // alongside whatever the app is doing, so wait rather than fail on a
        // held write lock; every statement here is short.
        conn.busy_timeout(std::time::Duration::from_secs(5))?;
        Ok(conn)
    }
}

/// Read a fixed-width blob column, rejecting a wrong-length value rather than
/// padding or truncating it into something that would seal wrongly.
fn fixed<const N: usize>(bytes: &[u8], what: &str) -> Result<[u8; N]> {
    <[u8; N]>::try_from(bytes)
        .map_err(|_| SyncKitError::Database(format!("{what} is {} bytes, want {N}", bytes.len())))
}

impl BlobResumeStore for SqliteResumeStore {
    fn load(&self, hash: &str) -> Result<Option<ResumeRecord>> {
        let conn = self.conn()?;
        let Some((upload_id, part_size, part_count, size_bytes, created_at)) = conn
            .query_row(
                "SELECT upload_id, part_size, part_count, size_bytes, created_at
                   FROM sync_blob_resume WHERE hash = ?1",
                params![hash],
                |r| {
                    Ok((
                        r.get::<_, String>(0)?,
                        r.get::<_, i64>(1)?,
                        r.get::<_, i64>(2)?,
                        r.get::<_, i64>(3)?,
                        r.get::<_, i64>(4)?,
                    ))
                },
            )
            .optional()?
        else {
            return Ok(None);
        };

        let mut parts_stmt = conn.prepare(
            "SELECT part_number, etag FROM sync_blob_resume_part
              WHERE hash = ?1 ORDER BY part_number",
        )?;
        let parts = parts_stmt
            .query_map(params![hash], |r| {
                Ok(ResumePart {
                    part_number: r.get::<_, i64>(0)? as u32,
                    etag: r.get(1)?,
                })
            })?
            .collect::<rusqlite::Result<Vec<_>>>()?;

        let mut chunks_stmt = conn.prepare(
            "SELECT chunk_index, nonce, plain_sha FROM sync_blob_resume_chunk
              WHERE hash = ?1 ORDER BY chunk_index",
        )?;
        let chunks = chunks_stmt
            .query_map(params![hash], |r| {
                Ok((
                    r.get::<_, i64>(0)? as u32,
                    r.get::<_, Vec<u8>>(1)?,
                    r.get::<_, Vec<u8>>(2)?,
                ))
            })?
            .collect::<rusqlite::Result<Vec<_>>>()?
            .into_iter()
            .map(|(index, nonce, plain_sha)| {
                Ok(ResumeChunk {
                    index,
                    nonce: fixed::<BLOB_NONCE_LEN>(&nonce, "resume nonce")?,
                    plain_sha: fixed::<32>(&plain_sha, "resume plaintext digest")?,
                })
            })
            .collect::<Result<Vec<_>>>()?;

        Ok(Some(ResumeRecord {
            session: ResumeSession {
                upload_id,
                part_size: part_size as u64,
                part_count: part_count as u32,
                size_bytes: size_bytes as u64,
            },
            age_secs: (chrono::Utc::now().timestamp() - created_at).max(0),
            parts,
            chunks,
        }))
    }

    fn begin(&self, hash: &str, session: &ResumeSession) -> Result<()> {
        let mut conn = self.conn()?;
        let tx = conn.transaction()?;
        // Replace rather than merge: a new session means the parts and nonces
        // recorded against the old one describe an upload that no longer exists.
        tx.execute(
            "DELETE FROM sync_blob_resume WHERE hash = ?1",
            params![hash],
        )?;
        tx.execute(
            "INSERT INTO sync_blob_resume
               (hash, upload_id, part_size, part_count, size_bytes, created_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
            params![
                hash,
                session.upload_id,
                session.part_size as i64,
                i64::from(session.part_count),
                session.size_bytes as i64,
                chrono::Utc::now().timestamp(),
            ],
        )?;
        tx.commit()?;
        Ok(())
    }

    fn record_part(&self, hash: &str, part: &ResumePart, chunks: &[ResumeChunk]) -> Result<()> {
        let mut conn = self.conn()?;
        let tx = conn.transaction()?;
        // If the session row is gone the record was cleared under us; the
        // foreign keys would reject these anyway, so say so rather than write
        // orphans.
        let live: bool = tx
            .query_row(
                "SELECT 1 FROM sync_blob_resume WHERE hash = ?1",
                params![hash],
                |_| Ok(true),
            )
            .optional()?
            .unwrap_or(false);
        if !live {
            return Ok(());
        }
        for chunk in chunks {
            // REPLACE: a second attempt re-seals its chunks with fresh nonces,
            // and the newest is the one that describes what is at S3.
            tx.execute(
                "INSERT OR REPLACE INTO sync_blob_resume_chunk
                   (hash, chunk_index, nonce, plain_sha) VALUES (?1, ?2, ?3, ?4)",
                params![
                    hash,
                    i64::from(chunk.index),
                    chunk.nonce.as_slice(),
                    chunk.plain_sha.as_slice()
                ],
            )?;
        }
        tx.execute(
            "INSERT OR REPLACE INTO sync_blob_resume_part (hash, part_number, etag)
             VALUES (?1, ?2, ?3)",
            params![hash, i64::from(part.part_number), part.etag],
        )?;
        tx.commit()?;
        Ok(())
    }

    fn clear(&self, hash: &str) -> Result<()> {
        // ON DELETE CASCADE takes the parts and chunks; `foreign_keys` is ON for
        // every engine connection (see `configure_connection`).
        self.conn()?.execute(
            "DELETE FROM sync_blob_resume WHERE hash = ?1",
            params![hash],
        )?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn store() -> SqliteResumeStore {
        use std::sync::atomic::{AtomicU64, Ordering};
        static N: AtomicU64 = AtomicU64::new(0);
        let mut p = std::env::temp_dir();
        p.push(format!(
            "synckit_resume_{}_{}",
            std::process::id(),
            N.fetch_add(1, Ordering::Relaxed)
        ));
        std::fs::create_dir_all(&p).unwrap();
        let store = SqliteResumeStore::new(DbSource::path(p.join("app.db")));
        // Opening is what applies the DDL.
        drop(store.conn().unwrap());
        store
    }

    fn session() -> ResumeSession {
        ResumeSession {
            upload_id: "upload-1".into(),
            part_size: 5 * 1024 * 1024,
            part_count: 3,
            size_bytes: 11 * 1024 * 1024,
        }
    }

    #[test]
    fn a_session_round_trips_with_its_parts_and_chunks() {
        let store = store();
        store.begin("aa", &session()).unwrap();
        store
            .record_part(
                "aa",
                &ResumePart {
                    part_number: 1,
                    etag: "\"etag-1\"".into(),
                },
                &[ResumeChunk {
                    index: 0,
                    nonce: [7u8; BLOB_NONCE_LEN],
                    plain_sha: [9u8; 32],
                }],
            )
            .unwrap();

        let record = store.load("aa").unwrap().unwrap();
        assert_eq!(record.session.upload_id, "upload-1");
        assert_eq!(record.session.part_count, 3);
        assert_eq!(record.first_missing_part(), 2);
        assert_eq!(record.parts[0].etag, "\"etag-1\"");
        assert_eq!(record.chunk(0).unwrap().nonce, [7u8; BLOB_NONCE_LEN]);
        assert!(record.age_secs >= 0 && record.age_secs < 60);
    }

    #[test]
    fn beginning_again_discards_the_old_session_entirely() {
        let store = store();
        store.begin("aa", &session()).unwrap();
        store
            .record_part(
                "aa",
                &ResumePart {
                    part_number: 1,
                    etag: "old".into(),
                },
                &[ResumeChunk {
                    index: 0,
                    nonce: [1u8; BLOB_NONCE_LEN],
                    plain_sha: [1u8; 32],
                }],
            )
            .unwrap();

        let mut next = session();
        next.upload_id = "upload-2".into();
        store.begin("aa", &next).unwrap();

        let record = store.load("aa").unwrap().unwrap();
        assert_eq!(record.session.upload_id, "upload-2");
        // Parts and nonces belong to the dead session; keeping them would
        // resume a session S3 no longer has.
        assert!(record.parts.is_empty());
        assert!(record.chunks.is_empty());
    }

    #[test]
    fn clearing_takes_the_children_with_it() {
        let store = store();
        store.begin("aa", &session()).unwrap();
        store
            .record_part(
                "aa",
                &ResumePart {
                    part_number: 1,
                    etag: "e".into(),
                },
                &[ResumeChunk {
                    index: 0,
                    nonce: [1u8; BLOB_NONCE_LEN],
                    plain_sha: [1u8; 32],
                }],
            )
            .unwrap();
        store.clear("aa").unwrap();
        assert!(store.load("aa").unwrap().is_none());

        let conn = store.conn().unwrap();
        let parts: i64 = conn
            .query_row("SELECT count(*) FROM sync_blob_resume_part", [], |r| {
                r.get(0)
            })
            .unwrap();
        let chunks: i64 = conn
            .query_row("SELECT count(*) FROM sync_blob_resume_chunk", [], |r| {
                r.get(0)
            })
            .unwrap();
        assert_eq!((parts, chunks), (0, 0));
    }

    #[test]
    fn recording_against_a_cleared_session_writes_nothing() {
        let store = store();
        store
            .record_part(
                "gone",
                &ResumePart {
                    part_number: 1,
                    etag: "e".into(),
                },
                &[],
            )
            .unwrap();
        assert!(store.load("gone").unwrap().is_none());
    }

    #[test]
    fn a_wrong_length_nonce_is_rejected_rather_than_reshaped() {
        let store = store();
        store.begin("aa", &session()).unwrap();
        store
            .conn()
            .unwrap()
            .execute(
                "INSERT INTO sync_blob_resume_chunk (hash, chunk_index, nonce, plain_sha)
                 VALUES ('aa', 0, X'0102', ?1)",
                params![[0u8; 32].as_slice()],
            )
            .unwrap();
        let err = store.load("aa").unwrap_err();
        assert!(err.to_string().contains("resume nonce"), "{err}");
    }
}