//! Blob metadata persistence; opaque bytes remain in `BlobStore`.

use cid::Cid;
use rusqlite::{Transaction, params};

use crate::{
    domain::blob::{Blob, referenced_blob_cids, writable_blob_refs},
    error::{Error, ErrorContext as _, Result},
};

use super::{Database, now, query, require_account_active};

impl Database {
    pub(crate) fn is_public_blob(&self, did: &str, cid: &Cid) -> Result<bool> {
        self.query_one(
            "SELECT EXISTS(SELECT 1 FROM record_blobs WHERE did = ?1 AND cid = ?2)",
            params![did, cid.to_string()],
            |row| Ok(row.get(0)?),
        )
    }

    /// Reserve metadata before publishing blob bytes.
    ///
    /// # Errors
    ///
    /// Returns an error when metadata cannot be written, conflicts with an
    /// earlier upload, or deletion already owns this content-addressed key.
    pub fn begin_blob_upload(&self, did: &str, blob: &Blob) -> Result<()> {
        let connection = self.connection()?;
        let changed = connection.execute(
            "INSERT INTO blobs (did, cid, media_type, size, created_at)
             SELECT ?1, ?2, ?3, ?4, ?5
             WHERE (SELECT status FROM accounts WHERE did = ?1) = 'active'
             ON CONFLICT(did, cid) DO UPDATE SET
                created_at = excluded.created_at
             WHERE blobs.delete_started_at IS NULL
               AND blobs.media_type = excluded.media_type AND blobs.size = excluded.size",
            params![
                did,
                blob.cid.to_string(),
                blob.media_type,
                i64::try_from(blob.size).unwrap_or(i64::MAX),
                now(),
            ],
        )?;
        if changed == 0 {
            require_account_active(&connection, did)?;
            return Err(Error::Conflict(
                "blob metadata differs from its earlier upload or deletion is in progress"
                    .to_owned(),
            ));
        }
        Ok(())
    }

    /// Mark a reserved blob as visible after its bytes are durable.
    ///
    /// # Errors
    ///
    /// Returns an error when the reservation disappeared or entered deletion.
    pub fn finish_blob_upload(&self, did: &str, cid: &Cid) -> Result<()> {
        let connection = self.connection()?;
        let changed = connection.execute(
            "UPDATE blobs SET available_at = COALESCE(available_at, ?3)
             WHERE did = ?1 AND cid = ?2 AND delete_started_at IS NULL
               AND (SELECT status FROM accounts WHERE did = ?1) = 'active'",
            params![did, cid.to_string(), now()],
        )?;
        if changed == 0 {
            require_account_active(&connection, did)?;
            return Err(Error::Internal(format!(
                "blob upload reservation disappeared for {did}/{cid}"
            )));
        }
        Ok(())
    }

    /// Find available metadata for a content-addressed blob, or `None` when absent.
    ///
    /// # Errors
    ///
    /// Returns an error for failed lookup or invalid stored metadata.
    pub fn find_blob(&self, did: &str, cid: &Cid) -> Result<Option<Blob>> {
        self.query_optional(
            "SELECT media_type, size FROM blobs
                 WHERE did = ?1 AND cid = ?2
                   AND delete_started_at IS NULL AND available_at IS NOT NULL",
            params![did, cid.to_string()],
            |row| decode_blob(*cid, row.get(0)?, row.get(1)?),
        )
    }

    pub(crate) fn validate_blob_refs(&self, did: &str, value: &serde_json::Value) -> Result<()> {
        for referenced in writable_blob_refs(value).map_err(Error::InvalidRequest)? {
            let stored = self.find_blob(did, &referenced.cid)?.ok_or_else(|| {
                Error::InvalidRequest(format!(
                    "record references blob {} before it is uploaded",
                    referenced.cid
                ))
            })?;
            if referenced.media_type != stored.media_type || referenced.size != stored.size {
                return Err(Error::InvalidRequest(format!(
                    "record blob {} metadata does not match the uploaded blob",
                    referenced.cid
                )));
            }
        }
        Ok(())
    }

    pub(crate) fn all_blobs(&self) -> Result<Vec<(String, Blob)>> {
        self.query(
            "SELECT did, cid, media_type, size FROM blobs
             WHERE delete_started_at IS NULL AND available_at IS NOT NULL
             ORDER BY did, cid",
            [],
            |row| {
                let (did, cid, media_type, size) = <(_, String, _, _)>::try_from(row)?;
                Ok((did, decode_blob(parse_stored_cid(&cid)?, media_type, size)?))
            },
        )
    }

    pub(crate) fn claim_orphan_blobs(
        &self,
        created_before: i64,
        limit: usize,
    ) -> Result<Vec<(String, Cid)>> {
        let candidates = self.transaction(|transaction| {
            let candidates = query(
                transaction,
                "SELECT blobs.did, blobs.cid
                 FROM blobs
                 WHERE blobs.delete_started_at IS NOT NULL
                    OR (blobs.created_at <= ?1
                        AND NOT EXISTS (
                            SELECT 1 FROM record_blobs
                            WHERE record_blobs.did = blobs.did
                              AND record_blobs.cid = blobs.cid
                        )
                        AND NOT EXISTS (
                            SELECT 1 FROM space_record_blobs
                            WHERE space_record_blobs.did = blobs.did
                              AND space_record_blobs.cid = blobs.cid
                        ))
                 ORDER BY blobs.delete_started_at IS NULL, blobs.created_at,
                          blobs.did, blobs.cid
                 LIMIT ?2",
                params![created_before, i64::try_from(limit).unwrap_or(i64::MAX)],
                |row| Ok(<(String, String)>::try_from(row)?),
            )?;
            for (did, cid) in &candidates {
                transaction.execute(
                    "UPDATE blobs SET delete_started_at = COALESCE(delete_started_at, ?3)
                 WHERE did = ?1 AND cid = ?2",
                    params![did, cid, now()],
                )?;
            }
            Ok(candidates)
        })?;
        candidates
            .into_iter()
            .map(|(did, cid)| Ok((did, parse_stored_cid(&cid)?)))
            .collect()
    }

    pub(crate) fn finish_blob_deletion(&self, did: &str, cid: &Cid) -> Result<()> {
        let changed = self.connection()?.execute(
            "DELETE FROM blobs
             WHERE did = ?1 AND cid = ?2 AND delete_started_at IS NOT NULL",
            params![did, cid.to_string()],
        )?;
        if changed == 0 {
            return Err(Error::Internal(format!(
                "blob deletion claim disappeared for {did}/{cid}"
            )));
        }
        Ok(())
    }

    pub(crate) fn referenced_blob_cids(
        &self,
        did: &str,
        since: Option<&str>,
        cursor: Option<&str>,
        limit: usize,
    ) -> Result<Vec<Cid>> {
        self.query(
            "SELECT DISTINCT record_blobs.cid
             FROM record_blobs
             WHERE record_blobs.did = ?1
               AND (?2 IS NULL OR record_blobs.cid > ?2)
               AND (
                 ?3 IS NULL OR EXISTS (
                   SELECT 1 FROM repo_ops
                   WHERE repo_ops.did = record_blobs.did
                     AND repo_ops.collection = record_blobs.collection
                     AND repo_ops.rkey = record_blobs.rkey
                     AND repo_ops.rev > ?3
                 )
               )
             ORDER BY record_blobs.cid
             LIMIT ?4",
            params![did, cursor, since, i64::try_from(limit).unwrap_or(i64::MAX)],
            |row| parse_stored_cid(&row.get::<_, String>(0)?),
        )
    }

    pub(crate) fn missing_blob_references(
        &self,
        did: &str,
        cursor: Option<&str>,
        limit: usize,
    ) -> Result<Vec<(Cid, String)>> {
        self.query(
            "SELECT record_blobs.cid,
                    MIN(record_blobs.collection || '/' || record_blobs.rkey)
             FROM record_blobs
             LEFT JOIN blobs
               ON blobs.did = record_blobs.did AND blobs.cid = record_blobs.cid
              AND blobs.delete_started_at IS NULL AND blobs.available_at IS NOT NULL
             WHERE record_blobs.did = ?1
               AND blobs.cid IS NULL
               AND (?2 IS NULL OR record_blobs.cid > ?2)
             GROUP BY record_blobs.cid
             ORDER BY record_blobs.cid
             LIMIT ?3",
            params![did, cursor, i64::try_from(limit).unwrap_or(i64::MAX)],
            |row| {
                let (cid, path) = <(String, String)>::try_from(row)?;
                Ok((parse_stored_cid(&cid)?, format!("at://{did}/{path}")))
            },
        )
    }
}

fn parse_stored_cid(value: &str) -> Result<Cid> {
    value
        .parse()
        .context(Error::Internal, "stored blob CID is invalid")
}

pub(crate) fn active_blob_ref_cids(
    transaction: &Transaction<'_>,
    did: &str,
    value: &serde_json::Value,
) -> Result<Vec<String>> {
    let cids = referenced_blob_cids(value).into_iter().collect::<Vec<_>>();
    for cid in &cids {
        let exists = transaction.query_row(
            "SELECT EXISTS(
                SELECT 1 FROM blobs
                WHERE did = ?1 AND cid = ?2
                  AND delete_started_at IS NULL AND available_at IS NOT NULL
             )",
            params![did, cid],
            |row| row.get::<_, bool>(0),
        )?;
        if !exists {
            return Err(Error::InvalidRequest(format!(
                "record references blob {cid} before it is uploaded"
            )));
        }
    }
    Ok(cids)
}

fn decode_blob(cid: Cid, media_type: String, size: i64) -> Result<Blob> {
    Ok(Blob {
        cid,
        media_type,
        size: u64::try_from(size)
            .map_err(|_| Error::Internal("stored blob size is negative".to_owned()))?,
    })
}

#[cfg(test)]
mod tests {
    use std::str::FromStr as _;

    use super::*;

    const DID: &str = "did:web:alice.test";

    fn previous_blob_lookup(database: &Database, cid: &Cid) -> Result<Blob> {
        database
            .query_optional(
                "SELECT media_type, size FROM blobs
                 WHERE did = ?1 AND cid = ?2
                   AND delete_started_at IS NULL AND available_at IS NOT NULL",
                params![DID, cid.to_string()],
                |row| decode_blob(*cid, row.get(0)?, row.get(1)?),
            )?
            .ok_or(Error::NotFound)
    }

    fn assert_optional_blob_parity(database: &Database, reference: &Blob) {
        let previous = previous_blob_lookup(database, &reference.cid);
        let actual = database
            .find_blob(DID, &reference.cid)
            .and_then(|blob| blob.ok_or(Error::NotFound));
        assert_eq!(format!("{actual:?}"), format!("{previous:?}"));
        let expected = previous
            .map_err(|error| {
                if matches!(error, Error::NotFound) {
                    Error::InvalidRequest(format!(
                        "record references blob {} before it is uploaded",
                        reference.cid
                    ))
                } else {
                    error
                }
            })
            .and_then(|stored| {
                if reference.media_type != stored.media_type || reference.size != stored.size {
                    return Err(Error::InvalidRequest(format!(
                        "record blob {} metadata does not match the uploaded blob",
                        reference.cid
                    )));
                }
                Ok(())
            });
        let value = serde_json::json!({
            "$type": "app.test.record",
            "image": {"$type": "blob", "ref": {"$link": reference.cid.to_string()}, "mimeType": reference.media_type, "size": reference.size},
        });
        let actual = database.validate_blob_refs(DID, &value);
        assert_eq!(format!("{actual:?}"), format!("{expected:?}"));
    }

    #[test]
    fn optional_blob_lookup_preserves_visibility_and_reference_validation_errors() {
        use rusqlite::types::Value as SqlValue;

        let database = Database::open_in_memory().unwrap();
        let reference = Blob {
            cid: cpds_protocol::sha256_cid(0x55, b"blob"),
            media_type: "image/png".to_owned(),
            size: 3,
        };
        assert!(database.find_blob(DID, &reference.cid).unwrap().is_none());
        assert_optional_blob_parity(&database, &reference);
        database.connection().unwrap().execute_batch(
            "CREATE TEMP TABLE blobs (did, cid, media_type, size, available_at, delete_started_at)",
        ).unwrap();
        for media_type in [
            SqlValue::Text("image/png".to_owned()),
            SqlValue::Text("text/plain".to_owned()),
            SqlValue::Text("not a MIME type".to_owned()),
            SqlValue::Null,
            SqlValue::Integer(1),
            SqlValue::Blob(vec![0]),
        ] {
            for size in [
                SqlValue::Integer(3),
                SqlValue::Integer(0),
                SqlValue::Integer(-1),
                SqlValue::Integer(i64::MAX),
                SqlValue::Real(3.5),
                SqlValue::Text("3".to_owned()),
                SqlValue::Null,
                SqlValue::Blob(vec![0]),
            ] {
                for available_at in [None, Some(1)] {
                    for delete_started_at in [None, Some(1)] {
                        {
                            let connection = database.connection().unwrap();
                            connection.execute("DELETE FROM temp.blobs", []).unwrap();
                            connection
                                .execute(
                                    "INSERT INTO temp.blobs VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                                    params![
                                        DID,
                                        reference.cid.to_string(),
                                        media_type,
                                        size,
                                        available_at,
                                        delete_started_at
                                    ],
                                )
                                .unwrap();
                        }
                        assert_optional_blob_parity(&database, &reference);
                    }
                }
            }
        }
        database
            .connection()
            .unwrap()
            .execute_batch("DROP TABLE temp.blobs; DROP TABLE main.blobs")
            .unwrap();
        assert_optional_blob_parity(&database, &reference);
        assert!(matches!(
            database.find_blob(DID, &reference.cid),
            Err(Error::Database(_))
        ));
    }

    #[test]
    fn stored_blob_cid_context_preserves_values_and_exact_errors() {
        let cid = cpds_protocol::sha256_cid(0x55, b"blob");
        let v0 = Cid::new_v0(*cid.hash()).unwrap().to_string();
        let valid = cid.to_string();
        for value in [
            valid.as_str(),
            v0.as_str(),
            "",
            "invalid-cid",
            "b",
            "Qm",
            "é",
            " bafkreia",
        ] {
            let old = Cid::from_str(value)
                .map_err(|error| Error::Internal(format!("stored blob CID is invalid: {error}")));
            let current = parse_stored_cid(value);
            assert_eq!(format!("{current:?}"), format!("{old:?}"), "{value:?}");
        }
    }

    #[test]
    fn repeated_uploads_cannot_change_content_addressed_metadata() {
        let database = Database::open_in_memory().unwrap();
        database
            .ensure_account(DID, "alice.test", "password")
            .unwrap();
        let blob = Blob {
            cid: Cid::from_str("bafkreibme22gw2h7y2h7tg2fhqotaqjucnbc24deqo72b6mkl2egezxhvy")
                .unwrap(),
            media_type: "image/png".to_owned(),
            size: 3,
        };
        database.begin_blob_upload(DID, &blob).unwrap();
        for available in [false, true] {
            if available {
                database.finish_blob_upload(DID, &blob.cid).unwrap();
            }
            database.begin_blob_upload(DID, &blob).unwrap();
            for inconsistent in [
                Blob {
                    media_type: "text/plain".to_owned(),
                    ..blob.clone()
                },
                Blob {
                    size: 4,
                    ..blob.clone()
                },
            ] {
                assert!(matches!(
                    database.begin_blob_upload(DID, &inconsistent),
                    Err(Error::Conflict(_))
                ));
            }
        }
        assert_eq!(database.find_blob(DID, &blob.cid).unwrap().unwrap(), blob);
    }

    #[test]
    fn deactivation_prevents_blob_reservation_and_publication() {
        let database = Database::open_in_memory().unwrap();
        database
            .ensure_account(DID, "alice.test", "password")
            .unwrap();
        let blob = Blob {
            cid: Cid::from_str("bafkreibme22gw2h7y2h7tg2fhqotaqjucnbc24deqo72b6mkl2egezxhvy")
                .unwrap(),
            media_type: "image/png".to_owned(),
            size: 3,
        };

        database.begin_blob_upload(DID, &blob).unwrap();
        database
            .connection()
            .unwrap()
            .execute(
                "UPDATE accounts SET status = 'deactivated' WHERE did = ?1",
                [DID],
            )
            .unwrap();
        assert!(matches!(
            database.finish_blob_upload(DID, &blob.cid),
            Err(Error::Xrpc {
                name: "AccountDeactivated",
                ..
            })
        ));
        assert!(matches!(
            database
                .find_blob(DID, &blob.cid)
                .and_then(|blob| blob.ok_or(Error::NotFound)),
            Err(Error::NotFound)
        ));

        let other = Blob {
            cid: Cid::from_str("bafkreid7zfv2egb7qol7vr43m6xju2wwhf4l75uz2e6fmh5f3av7p7ddca")
                .unwrap(),
            ..blob
        };
        assert!(matches!(
            database.begin_blob_upload(DID, &other),
            Err(Error::Xrpc {
                name: "AccountDeactivated",
                ..
            })
        ));
    }
}