//! V2 binding envelope around the existing immutable registration/serving journal.
//! V1 receipt and recovery codecs remain unchanged for existing Cloud originals.
use crate::{
    app_master_admission::{self, Binding},
    identity::NodeIdentity,
    media_registration::{
        self as io, AuthorizedSelection, InstallationPin, Intent, Limits, Receipt, Retirement,
    },
};
use anyhow::{Context, Result};
use ed25519_dalek::Signer;
use serde::{Deserialize, Serialize};
use std::{
    fs::File,
    path::Path,
    time::{Duration, Instant},
};

pub(crate) const APPROVAL: &str = "archipelago.media-master.approval.v2";
pub(crate) const RESOLUTION: &str = "archipelago.media-master.resolution.v2";
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub(crate) struct Operation {
    pub intent: Intent,
    pub binding: Binding,
    pub payment_methods: Vec<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub(crate) struct BoundReceipt {
    pub version: u8,
    pub binding: Binding,
    pub registration: Receipt,
    pub signature: String,
}
impl BoundReceipt {
    pub(crate) fn preimage(&self) -> Result<Vec<u8>> {
        Ok(serde_json::to_vec(&serde_json::json!([
            "archipelago.indeehub.app-master.receipt.v2",
            self.binding.canonical()?,
            String::from_utf8(self.registration.preimage()?)?,
            self.registration.signature
        ]))?)
    }
    fn validate(&self, identity: &NodeIdentity, op: &Operation) -> Result<()> {
        let r = &self.registration;
        let i = &op.intent;
        anyhow::ensure!(
            self.version == 2
                && self.binding == op.binding
                && r.version == 1
                && r.request_id == i.request_id
                && r.nonce == i.nonce
                && r.project_id == i.project_id
                && r.producer == i.producer
                && r.node_did == i.node_did
                && r.app_audience == i.app_audience
                && r.price_sats == i.price_sats
                && r.viewing_seconds == i.viewing_seconds
                && r.expires_at == i.expires_at
                && r.sha256 == op.binding.sha256
                && r.size_bytes == op.binding.size_bytes
                && r.payment_methods == op.payment_methods
                && NodeIdentity::verify(&identity.pubkey_hex(), &r.preimage()?, &r.signature)?
                && NodeIdentity::verify(
                    &identity.pubkey_hex(),
                    &self.preimage()?,
                    &self.signature
                )?,
            "Original master receipt changed"
        );
        Ok(())
    }
}
fn checked_binding(op: &Operation, pin: &InstallationPin) -> Result<()> {
    op.binding.validate(pin)?;
    let i = &op.intent;
    let b = &op.binding;
    let id = uuid::Uuid::parse_str(&i.request_id)?;
    anyhow::ensure!(
        id.to_string() == i.request_id
            && id.get_version_num() == 4
            && i.version == 1
            && matches!(b.version, 2 | 3)
            && i.app_audience == pin.app_audience
            && i.node_did == pin.node_did
            && b.app_audience == i.app_audience
            && b.node_did == i.node_did
            && b.producer == i.producer
            && b.asset.project_id == i.project_id
            && !op.payment_methods.is_empty()
            && op.payment_methods.len() <= 3
            && op
                .payment_methods
                .iter()
                .all(|method| matches!(method.as_str(), "cashu" | "lightning" | "onchain"))
            && op.payment_methods.windows(2).all(|pair| pair[0] < pair[1]),
        "Master approval binding changed"
    );
    Ok(())
}
fn checked(op: &Operation, pin: &InstallationPin) -> Result<()> {
    checked_binding(op, pin)?;
    anyhow::ensure!(
        !op.payment_methods.iter().any(|m| m == "onchain") || op.intent.price_sats >= 546,
        "On-chain receiving requires at least 546 sats"
    );
    Ok(())
}
fn operation_dir(data: &Path, op: &Operation, limits: &Limits<'_>) -> Result<File> {
    let root = io::open_directory(data)?;
    let store = io::private_directory(&root, "app-master-registrations")?;
    let dir = io::private_directory(&store, &op.intent.request_id)?;
    io::lock_operation(&dir, limits, Instant::now() + Duration::from_secs(30))?;
    if let Some(saved) = io::read_record::<Operation>(&dir, "operation.json")? {
        anyhow::ensure!(
            saved == *op,
            "Original master, price or receiving methods changed; return to review"
        );
    } else {
        io::save_record(&dir, "operation.json", op)?;
    }
    Ok(dir)
}
fn complete(
    dir: &File,
    identity: &NodeIdentity,
    op: &Operation,
    registration: Receipt,
) -> Result<BoundReceipt> {
    let mut receipt = BoundReceipt {
        version: 2,
        binding: op.binding.clone(),
        registration,
        signature: String::new(),
    };
    receipt.signature = hex::encode(identity.signing_key().sign(&receipt.preimage()?).to_bytes());
    receipt.validate(identity, op)?;
    if let Some(saved) = io::read_record::<BoundReceipt>(dir, "receipt.json")? {
        anyhow::ensure!(saved == receipt, "Original master receipt changed");
    } else {
        io::save_record(dir, "receipt.json", &receipt)?;
    }
    dir.sync_all()?;
    Ok(receipt)
}
pub(crate) async fn prepare(
    data: &Path,
    identity: &NodeIdentity,
    pin: &InstallationPin,
    op: &Operation,
    now: u64,
    limits: &Limits<'_>,
) -> Result<BoundReceipt> {
    checked(op, pin)?;
    let dir = operation_dir(data, op, limits)?;
    if let Some(receipt) = io::read_record::<BoundReceipt>(&dir, "receipt.json")? {
        receipt.validate(identity, op)?;
        let recovered = crate::registered_media::resolve_registration(
            data,
            identity,
            &op.intent,
            &op.intent.producer,
            now,
            limits,
        )?;
        anyhow::ensure!(
            recovered["phase"] == "completed"
                && recovered["receipt"] == serde_json::to_value(&receipt.registration)?,
            "Durable master media changed"
        );
        return Ok(receipt);
    }
    // Recover a crash after durable v1 serving commit, before the v2 envelope.
    let recovered = crate::registered_media::resolve_registration(
        data,
        identity,
        &op.intent,
        &op.intent.producer,
        now,
        limits,
    )?;
    if recovered["phase"] == "completed" {
        return complete(
            &dir,
            identity,
            op,
            serde_json::from_value(recovered["receipt"].clone())?,
        );
    }
    anyhow::ensure!(
        now < op.intent.expires_at && recovered["phase"] != "retired",
        "Original preparation expired; recover or retire it"
    );
    // Revalidate at explicit confirmation, even if the browser review is fresh.
    let fresh = app_master_admission::resolve(
        data,
        identity,
        pin,
        &op.binding.owner_id,
        &op.binding.producer,
        &op.binding.asset,
        now,
        false,
    )
    .await?;
    anyhow::ensure!(
        fresh == op.binding,
        "Selected master changed; return to review"
    );
    let source =
        app_master_admission::cache_source(data, identity, pin, &op.binding, now, limits).await?;
    let current = u64::try_from(chrono::Utc::now().timestamp())?;
    let fresh = app_master_admission::resolve(
        data,
        identity,
        pin,
        &op.binding.owner_id,
        &op.binding.producer,
        &op.binding.asset,
        current,
        false,
    )
    .await?;
    anyhow::ensure!(
        fresh == op.binding && current < op.intent.expires_at,
        "Master changed or preparation expired; return to review"
    );
    let selection = AuthorizedSelection {
        relative_path: source.file_name().context("Private source missing")?.into(),
        payment_methods: op.payment_methods.clone(),
    };
    let registration = crate::registered_media::register_approved_selection(
        data,
        source.parent().context("Private cache missing")?,
        identity,
        &crate::registered_media::ApprovedSelection {
            authenticated_producer: &op.intent.producer,
            authenticated_project: &op.intent.project_id,
            intent: &op.intent,
            selection: &selection,
        },
        current,
        limits,
        |_| Ok(()),
    )?;
    complete(&dir, identity, op, registration)
}
pub(crate) fn resolve(
    data: &Path,
    identity: &NodeIdentity,
    pin: &InstallationPin,
    op: &Operation,
    now: u64,
    limits: &Limits<'_>,
) -> Result<serde_json::Value> {
    let invalid_onchain_dust =
        op.intent.price_sats < 546 && op.payment_methods.iter().any(|method| method == "onchain");
    if invalid_onchain_dust {
        // A previously signed, unconsumed underpriced operation cannot be
        // prepared, but must be retireable after its exact intent expires.
        // This path never resolves a source, creates a receipt or publishes.
        checked_binding(op, pin)?;
        anyhow::ensure!(
            now >= op.intent.expires_at,
            "Original preparation is still active"
        );
        let dir = operation_dir(data, op, limits)?;
        anyhow::ensure!(
            io::read_record::<BoundReceipt>(&dir, "receipt.json")?.is_none(),
            "Invalid payment methods cannot complete a master"
        );
        let retirement = io::retire_unconsumed(data, identity, pin, &op.intent, now, limits)?;
        return signed_retirement(&dir, identity, op, retirement);
    }
    checked(op, pin)?;
    let dir = operation_dir(data, op, limits)?;
    if let Some(receipt) = io::read_record::<BoundReceipt>(&dir, "receipt.json")? {
        receipt.validate(identity, op)?;
        let recovered = crate::registered_media::resolve_registration(
            data,
            identity,
            &op.intent,
            &op.intent.producer,
            now,
            limits,
        )?;
        anyhow::ensure!(
            recovered["phase"] == "completed"
                && recovered["receipt"] == serde_json::to_value(&receipt.registration)?,
            "Durable master media changed"
        );
        return Ok(serde_json::json!({"phase":"completed","receipt":receipt}));
    }
    let outcome = crate::registered_media::resolve_registration(
        data,
        identity,
        &op.intent,
        &op.intent.producer,
        now,
        limits,
    )?;
    if outcome["phase"] == "completed" {
        let receipt = complete(
            &dir,
            identity,
            op,
            serde_json::from_value(outcome["receipt"].clone())?,
        )?;
        return Ok(serde_json::json!({"phase":"completed","receipt":receipt}));
    }
    if outcome["phase"] == "retired" {
        let retirement: Retirement = serde_json::from_value(outcome["retirement"].clone())?;
        return signed_retirement(&dir, identity, op, retirement);
    }
    Ok(outcome)
}

fn signed_retirement(
    dir: &File,
    identity: &NodeIdentity,
    op: &Operation,
    retirement: Retirement,
) -> Result<serde_json::Value> {
    let preimage = serde_json::to_vec(&serde_json::json!([
        "archipelago.indeehub.app-master.retirement.v2",
        op.binding.canonical()?,
        String::from_utf8(retirement.preimage()?)?,
        retirement.signature
    ]))?;
    let signed = serde_json::json!({"version":2,"binding":op.binding,"retirement":retirement,
        "signature":hex::encode(identity.signing_key().sign(&preimage).to_bytes())});
    if let Some(saved) = io::read_record::<serde_json::Value>(dir, "retirement.json")? {
        anyhow::ensure!(saved == signed, "Original retirement changed");
    } else {
        io::save_record(dir, "retirement.json", &signed)?;
    }
    Ok(serde_json::json!({"phase":"retired","retirement":signed}))
}

#[cfg(test)]
mod tests {
    use super::*;
    use sha2::{Digest, Sha256};
    use std::{
        os::unix::fs::PermissionsExt,
        sync::{atomic::AtomicBool, Arc},
    };
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    async fn fixture(
        corrupt: bool,
        current: bool,
    ) -> (
        tempfile::TempDir,
        NodeIdentity,
        InstallationPin,
        Operation,
        tokio::task::JoinHandle<()>,
    ) {
        let root = tempfile::tempdir().unwrap();
        let identity = NodeIdentity::load_or_create(&root.path().join("identity"))
            .await
            .unwrap();
        let installed = crate::container::registration_pin::ensure_for_installation(
            root.path(),
            "indeedhub-api",
            &identity,
        )
        .unwrap();
        let pin = InstallationPin {
            node_did: installed.node_did,
            app_audience: installed.app_audience,
        };
        let uuid = uuid::Uuid::new_v4().to_string();
        let now = chrono::Utc::now().timestamp() as u64;
        let mut binding:Binding=serde_json::from_value(serde_json::json!({"version":2,"appAudience":pin.app_audience,"nodeDid":pin.node_did,
            "producer":"aa".repeat(32),"ownerId":"owner","asset":{"projectId":"project","backendScopeId":uuid,"contentId":"content",
                "file":"projects/project/file/master.mp4","revision":{"configuration":uuid,"authority":uuid}},
            "storage":{"version":1,"backendScopeId":uuid,"storageProfileId":uuid,"storageGeneration":uuid,"bucket":"private"},
            "source":{"uploadIdentity":uuid,"etag":"\"original\"","size":4,"versionId":null},"uploadCommandId":uuid,"uploadReceiptId":uuid,
            "fileDigest":format!("sha256-chunks-v1:{}","bb".repeat(32)),"sha256":hex::encode(Sha256::digest(b"test")),"sizeBytes":"4"})).unwrap();
        if current {
            binding.version = 3;
            binding.storage.version = 2;
            binding.storage.storage_profile_id = "installed-private-s3-current-v1".into();
            binding.storage.storage_generation = "ee".repeat(32);
            binding.source.upload_identity = None;
            binding.upload_command_id = None;
            binding.upload_receipt_id = None;
            binding.current_verification = Some(crate::app_master_admission::CurrentVerification {
                kind: "owner-current-source-v1".into(),
                verification_id: uuid::Uuid::new_v4().to_string(),
                confirmation_id: uuid::Uuid::new_v4().to_string(),
                verified_at: now,
            });
        }
        let op = Operation {
            intent: Intent {
                version: 1,
                request_id: uuid::Uuid::new_v4().to_string(),
                nonce: "cc".repeat(32),
                app_audience: pin.app_audience.clone(),
                node_did: pin.node_did.clone(),
                producer: binding.producer.clone(),
                project_id: "project".into(),
                price_sats: 15,
                viewing_seconds: 3600,
                created_at: now,
                expires_at: now + 600,
            },
            binding: binding.clone(),
            payment_methods: vec!["cashu".into()],
        };
        let key = ed25519_dalek::SigningKey::from_bytes(&[9u8; 32]);
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let config = serde_json::json!({"version":1,"node_did":pin.node_did,"app_audience":pin.app_audience,
            "base_url":format!("http://{}/",listener.local_addr().unwrap()),"service_public_key":hex::encode(key.verifying_key().to_bytes())});
        let path = root.path().join("app-master-admission.json");
        std::fs::write(&path, serde_json::to_vec(&config).unwrap()).unwrap();
        std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)).unwrap();
        let task = tokio::spawn(async move {
            loop {
                let (mut socket, _) = listener.accept().await.unwrap();
                let mut bytes = Vec::new();
                let (header_end, length) = loop {
                    let mut part = [0u8; 1024];
                    let n = socket.read(&mut part).await.unwrap();
                    if n == 0 {
                        return;
                    }
                    bytes.extend_from_slice(&part[..n]);
                    if let Some(index) = bytes.windows(4).position(|v| v == b"\r\n\r\n") {
                        let text = String::from_utf8_lossy(&bytes[..index]);
                        let length = text
                            .lines()
                            .find_map(|line| {
                                line.to_ascii_lowercase()
                                    .strip_prefix("content-length:")
                                    .map(|v| v.trim().parse::<usize>().unwrap())
                            })
                            .unwrap();
                        break (index + 4, length);
                    }
                };
                while bytes.len() < header_end + length {
                    let mut part = [0u8; 1024];
                    let n = socket.read(&mut part).await.unwrap();
                    bytes.extend_from_slice(&part[..n]);
                }
                let r: serde_json::Value =
                    serde_json::from_slice(&bytes[header_end..header_end + length]).unwrap();
                let body = if r["action"] == "source" {
                    if corrupt {
                        b"evil".to_vec()
                    } else {
                        b"test".to_vec()
                    }
                } else {
                    let a = &r["asset"];
                    let preimage = serde_json::to_string(&serde_json::json!([
                        "archipelago.indeehub.app-master.request.v2",
                        2,
                        r["action"],
                        r["issuedAt"],
                        r["nonce"],
                        binding.app_audience,
                        binding.node_did,
                        r["ownerId"],
                        r["producer"],
                        a["projectId"],
                        a["backendScopeId"],
                        a["contentId"],
                        a["file"],
                        a["revision"]["configuration"],
                        a["revision"]["authority"],
                        r["bindingHash"],
                        r["proofVersions"]
                    ]))
                    .unwrap();
                    let request_hash = hex::encode(Sha256::digest(
                        serde_json::to_vec(&serde_json::json!([preimage, r["signature"]])).unwrap(),
                    ));
                    let at = chrono::Utc::now().timestamp() as u64;
                    let proof = serde_json::to_vec(&serde_json::json!([
                        "archipelago.indeehub.app-master.response.v2",
                        request_hash,
                        at,
                        binding.canonical().unwrap()
                    ]))
                    .unwrap();
                    serde_json::to_vec(&serde_json::json!({"version":2,"requestHash":request_hash,"issuedAt":at,"binding":binding,"signature":hex::encode(key.sign(&proof).to_bytes())})).unwrap()
                };
                let header = format!(
                    "HTTP/1.1 201 Created\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
                    body.len()
                );
                socket.write_all(header.as_bytes()).await.unwrap();
                socket.write_all(&body).await.unwrap();
            }
        });
        (root, identity, pin, op, task)
    }
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn receiving_methods_are_canonical_and_pending_selection_is_immutable() {
        let (root, _identity, pin, mut op, task) = fixture(false, false).await;
        op.payment_methods = vec!["lightning".into()];
        checked(&op, &pin).unwrap();
        let cancelled = AtomicBool::new(false);
        let limits = Limits {
            max_bytes: 1024,
            cancelled: &cancelled,
        };
        drop(operation_dir(root.path(), &op, &limits).unwrap());
        op.payment_methods = vec!["cashu".into(), "lightning".into()];
        checked(&op, &pin).unwrap();
        assert!(operation_dir(root.path(), &op, &limits).is_err());
        for methods in [
            vec![],
            vec!["onchain"],
            vec!["cashu", "cashu"],
            vec!["lightning", "cashu"],
        ] {
            op.payment_methods = methods.into_iter().map(str::to_owned).collect();
            assert!(checked(&op, &pin).is_err());
        }
        task.abort();
    }
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn expired_underpriced_onchain_master_can_only_retire_if_unconsumed() {
        let (root, identity, pin, mut op, task) = fixture(false, false).await;
        op.intent.price_sats = 2;
        op.payment_methods = vec!["cashu".into(), "lightning".into(), "onchain".into()];
        let cancelled = AtomicBool::new(false);
        let limits = Limits {
            max_bytes: 1024,
            cancelled: &cancelled,
        };
        assert!(checked_binding(&op, &pin).is_ok());
        assert!(checked(&op, &pin).is_err());
        assert!(prepare(
            root.path(),
            &identity,
            &pin,
            &op,
            op.intent.created_at,
            &limits
        )
        .await
        .is_err());
        assert!(resolve(
            root.path(),
            &identity,
            &pin,
            &op,
            op.intent.created_at,
            &limits
        )
        .is_err());
        let retired = resolve(
            root.path(),
            &identity,
            &pin,
            &op,
            op.intent.expires_at + 1,
            &limits,
        )
        .unwrap();
        assert_eq!(retired["phase"], "retired");
        assert_eq!(
            retired["retirement"]["binding"],
            serde_json::to_value(&op.binding).unwrap()
        );
        assert_eq!(
            retired,
            resolve(
                root.path(),
                &identity,
                &pin,
                &op,
                op.intent.expires_at + 2,
                &limits
            )
            .unwrap()
        );
        let dir = root
            .path()
            .join("app-master-registrations")
            .join(&op.intent.request_id);
        assert!(!dir.join("receipt.json").exists());
        assert!(dir.join("retirement.json").is_file());
        let media = root
            .path()
            .join("media-registration")
            .join(&op.intent.request_id);
        assert!(!media.join("operation.json").exists());
        assert!(!media.join("media").exists());
        let mut changed = op.clone();
        changed.payment_methods = vec!["cashu".into(), "onchain".into()];
        assert!(resolve(
            root.path(),
            &identity,
            &pin,
            &changed,
            changed.intent.expires_at + 2,
            &limits
        )
        .is_err());
        task.abort();

        let (root, identity, pin, mut op, task) = fixture(false, false).await;
        op.intent.price_sats = 2;
        op.payment_methods = vec!["cashu".into(), "lightning".into(), "onchain".into()];
        let media = root
            .path()
            .join("media-registration")
            .join(&op.intent.request_id);
        std::fs::create_dir_all(&media).unwrap();
        std::fs::write(media.join("media"), b"retained partial media").unwrap();
        assert!(resolve(
            root.path(),
            &identity,
            &pin,
            &op,
            op.intent.expires_at + 1,
            &limits
        )
        .is_err());
        assert!(!media.join("retirement.json").exists());
        task.abort();
    }
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn exact_source_prepares_durable_bound_receipt_reopens_and_rejects_substitution() {
        let (root, identity, pin, op, task) = fixture(false, false).await;
        let cancelled = AtomicBool::new(false);
        let limits = Limits {
            max_bytes: 1024,
            cancelled: &cancelled,
        };
        let now = op.intent.created_at;
        let receipt = prepare(root.path(), &identity, &pin, &op, now, &limits)
            .await
            .unwrap();
        assert_eq!(receipt.registration.sha256, op.binding.sha256);
        task.abort(); // Recovery must work without app/source availability, after expiry.
        let recovered = resolve(root.path(), &identity, &pin, &op, now + 1000, &limits).unwrap();
        assert_eq!(
            recovered["receipt"],
            serde_json::to_value(&receipt).unwrap()
        );
        for changed in ["asset", "price", "source"] {
            let mut other = op.clone();
            match changed {
                "asset" => other.binding.asset.content_id = "other".into(),
                "price" => other.intent.price_sats += 1,
                _ => other.binding.source.etag = "\"other\"".into(),
            };
            assert!(resolve(root.path(), &identity, &pin, &other, now + 1000, &limits).is_err());
        }
        let media = root
            .path()
            .join("media-registration")
            .join(&op.intent.request_id)
            .join("media");
        std::fs::set_permissions(&media, std::fs::Permissions::from_mode(0o600)).unwrap();
        std::fs::write(&media, b"evil").unwrap();
        std::fs::set_permissions(&media, std::fs::Permissions::from_mode(0o400)).unwrap();
        assert!(resolve(root.path(), &identity, &pin, &op, now + 1000, &limits).is_err());
    }
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn corrupt_source_never_issues_receipt_and_expired_original_retires() {
        let (root, identity, pin, op, task) = fixture(true, false).await;
        let cancelled = AtomicBool::new(false);
        let limits = Limits {
            max_bytes: 1024,
            cancelled: &cancelled,
        };
        assert!(prepare(
            root.path(),
            &identity,
            &pin,
            &op,
            op.intent.created_at,
            &limits
        )
        .await
        .is_err());
        task.abort();
        assert!(!root
            .path()
            .join("app-master-registrations")
            .join(&op.intent.request_id)
            .join("receipt.json")
            .exists());
        let retired = resolve(
            root.path(),
            &identity,
            &pin,
            &op,
            op.intent.expires_at + 1,
            &limits,
        )
        .unwrap();
        assert_eq!(retired["phase"], "retired");
        assert_eq!(
            retired["retirement"]["binding"],
            serde_json::to_value(&op.binding).unwrap()
        );
        assert_eq!(
            retired,
            resolve(
                root.path(),
                &identity,
                &pin,
                &op,
                op.intent.expires_at + 50,
                &limits
            )
            .unwrap()
        );
    }
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn current_owner_verification_prepares_and_recovers_without_claiming_original_upload_history(
    ) {
        let (root, identity, pin, op, task) = fixture(false, true).await;
        let cancelled = AtomicBool::new(false);
        let limits = Limits {
            max_bytes: 1024,
            cancelled: &cancelled,
        };
        let receipt = prepare(
            root.path(),
            &identity,
            &pin,
            &op,
            op.intent.created_at,
            &limits,
        )
        .await
        .unwrap();
        task.abort();
        assert_eq!(receipt.binding.version, 3);
        assert!(receipt.binding.upload_receipt_id.is_none());
        assert!(String::from_utf8(receipt.preimage().unwrap())
            .unwrap()
            .contains("owner-current-source-v1"));
        assert_eq!(
            resolve(
                root.path(),
                &identity,
                &pin,
                &op,
                op.intent.expires_at + 10,
                &limits
            )
            .unwrap()["receipt"],
            serde_json::to_value(receipt).unwrap()
        );
        let mut false_history = op.clone();
        false_history.binding.upload_receipt_id = Some(uuid::Uuid::new_v4().to_string());
        assert!(resolve(
            root.path(),
            &identity,
            &pin,
            &false_history,
            op.intent.expires_at + 10,
            &limits
        )
        .is_err());
        let mut changed = op.clone();
        changed
            .binding
            .current_verification
            .as_mut()
            .unwrap()
            .verification_id = uuid::Uuid::new_v4().to_string();
        assert!(resolve(
            root.path(),
            &identity,
            &pin,
            &changed,
            op.intent.expires_at + 10,
            &limits
        )
        .is_err());
    }
}