//! Group key rotation: the membership batch the client builds, and reading a
//! pull that spans GCK generations.

/// Rotation is the removal primitive: the batch an admin posts becomes the new
/// membership, so the server drops anyone absent from it and re-keys in the same
/// transaction. These tests pin the batch the client builds, because everything
/// the server can enforce depends on the client getting that batch right.
mod membership_batch {
    use crate::common::*;
    use synckit_client::{
        GroupId, IdentityKeypair, IdentityPublicKey, generate_group_key, open_gck_grant,
        seal_gck_to_member,
    };
    use wiremock::matchers::path_regex;

    /// A member we control both halves of, so a grant sealed to them can be
    /// opened and checked rather than merely counted.
    struct Member {
        user_id: UserId,
        keypair: IdentityKeypair,
    }

    impl Member {
        fn new() -> Self {
            Self {
                user_id: UserId::new(Uuid::new_v4()),
                keypair: IdentityKeypair::generate(),
            }
        }

        fn pubkey_json(&self) -> serde_json::Value {
            json!({
                "user_id": self.user_id,
                "pubkey": self.keypair.public_key().to_base64(),
            })
        }
    }

    /// The client whose master key seeds the admin identity, plus that identity.
    fn admin_client(kit: &MockKit) -> (SyncKitClient, IdentityKeypair) {
        let (client, master) = kit.keyed();
        let identity = IdentityKeypair::from_master_key(&master);
        (client, identity)
    }

    /// Mount the two reads a rotation makes: the admin's own grant (for the
    /// current generation) and the member pubkey list (the re-seal inputs).
    async fn mount_reads(
        kit: &MockKit,
        group_id: GroupId,
        gck: &[u8; 32],
        admin: &IdentityKeypair,
        admin_id: UserId,
        version: i32,
        members: &[&Member],
    ) {
        let sealed = seal_gck_to_member(gck, &admin.public_key(), &group_id.to_string(), version)
            .expect("seal admin grant");
        kit.get(&format!("/api/v1/sync/groups/{group_id}/grant"))
            .json(json!({
                "sealed_gck": sealed,
                "gck_version": version,
            }))
            .await;

        let mut pubkeys = vec![json!({
            "user_id": admin_id,
            "pubkey": admin.public_key().to_base64(),
        })];
        pubkeys.extend(members.iter().map(|m| m.pubkey_json()));
        kit.get(&format!("/api/v1/sync/groups/{group_id}/pubkeys"))
            .json(pubkeys)
            .await;
    }

    async fn mount_rotate(kit: &MockKit) {
        kit.matching("POST", path_regex(r"^/api/v1/sync/groups/[^/]+/rotate$"))
            .code(204)
            .empty()
            .await;
    }

    /// The body the client POSTed to `/rotate`.
    async fn posted_batch(kit: &MockKit, group_id: GroupId) -> serde_json::Value {
        kit.body(&format!("/api/v1/sync/groups/{group_id}/rotate"))
            .await
    }

    #[tokio::test]
    async fn removing_a_member_rekeys_and_reseals_to_everyone_who_stays() {
        let kit = MockKit::start().await;
        let (client, admin_identity) = admin_client(&kit);
        let (admin_id, _) = test_ids();
        let group_id = GroupId::new(Uuid::new_v4());
        let old_gck = generate_group_key();

        let bob = Member::new();
        let carol = Member::new();
        mount_reads(
            &kit,
            group_id,
            &old_gck,
            &admin_identity,
            admin_id,
            7,
            &[&bob, &carol],
        )
        .await;
        mount_rotate(&kit).await;

        client
            .remove_member(group_id, carol.user_id)
            .await
            .expect("remove member");

        let batch = posted_batch(&kit, group_id).await;
        assert_eq!(
            batch["gck_version"], 8,
            "the generation must advance past the one our grant reports"
        );

        let grants = batch["grants"].as_array().expect("grants array");
        assert_eq!(grants.len(), 2, "admin and bob, not carol: {grants:?}");
        let recipients: Vec<&str> = grants
            .iter()
            .map(|g| g["user_id"].as_str().expect("user_id"))
            .collect();
        assert!(recipients.contains(&admin_id.to_string().as_str()));
        assert!(recipients.contains(&bob.user_id.to_string().as_str()));
        assert!(
            !recipients.contains(&carol.user_id.to_string().as_str()),
            "the removed member must not be re-granted"
        );

        // The grants are real seals of one new key, not placeholders: Bob's opens,
        // and what comes out is neither the old GCK nor something private to the
        // admin's copy.
        let bobs = grants
            .iter()
            .find(|g| g["user_id"].as_str() == Some(&bob.user_id.to_string()))
            .expect("bob's grant");
        let new_gck = open_gck_grant(
            bobs["sealed_gck"].as_str().expect("sealed_gck"),
            &bob.keypair,
            &group_id.to_string(),
            8,
        )
        .expect("bob opens his grant");
        assert_ne!(new_gck, old_gck, "rotation must mint a fresh key");

        let admins = grants
            .iter()
            .find(|g| g["user_id"].as_str() == Some(&admin_id.to_string()))
            .expect("admin's grant");
        let admin_copy = open_gck_grant(
            admins["sealed_gck"].as_str().expect("sealed_gck"),
            &admin_identity,
            &group_id.to_string(),
            8,
        )
        .expect("admin opens their own grant");
        assert_eq!(
            admin_copy, new_gck,
            "every member must be sealed the same new key"
        );
    }

    #[tokio::test]
    async fn a_grant_cannot_be_opened_by_the_member_it_was_not_sealed_to() {
        let kit = MockKit::start().await;
        let (client, admin_identity) = admin_client(&kit);
        let (admin_id, _) = test_ids();
        let group_id = GroupId::new(Uuid::new_v4());

        let bob = Member::new();
        let carol = Member::new();
        mount_reads(
            &kit,
            group_id,
            &generate_group_key(),
            &admin_identity,
            admin_id,
            1,
            &[&bob, &carol],
        )
        .await;
        mount_rotate(&kit).await;

        client
            .rotate_group_key(group_id, &[])
            .await
            .expect("rotate without removing anyone");

        let batch = posted_batch(&kit, group_id).await;
        let bobs = batch["grants"]
            .as_array()
            .expect("grants")
            .iter()
            .find(|g| g["user_id"].as_str() == Some(&bob.user_id.to_string()))
            .expect("bob's grant")["sealed_gck"]
            .as_str()
            .expect("sealed_gck")
            .to_string();

        assert!(
            open_gck_grant(&bobs, &carol.keypair, &group_id.to_string(), 2).is_err(),
            "a grant sealed to bob must not open under carol's key"
        );
    }

    #[tokio::test]
    async fn an_empty_removal_set_rekeys_without_dropping_anyone() {
        let kit = MockKit::start().await;
        let (client, admin_identity) = admin_client(&kit);
        let (admin_id, _) = test_ids();
        let group_id = GroupId::new(Uuid::new_v4());

        let bob = Member::new();
        mount_reads(
            &kit,
            group_id,
            &generate_group_key(),
            &admin_identity,
            admin_id,
            3,
            &[&bob],
        )
        .await;
        mount_rotate(&kit).await;

        client
            .rotate_group_key(group_id, &[])
            .await
            .expect("rekey after a suspected compromise");

        let batch = posted_batch(&kit, group_id).await;
        assert_eq!(batch["gck_version"], 4);
        assert_eq!(
            batch["grants"].as_array().expect("grants").len(),
            2,
            "a bare re-key keeps the whole membership"
        );
    }

    #[tokio::test]
    async fn a_member_pubkey_the_client_cannot_parse_aborts_the_rotation() {
        let kit = MockKit::start().await;
        let (client, admin_identity) = admin_client(&kit);
        let (admin_id, _) = test_ids();
        let group_id = GroupId::new(Uuid::new_v4());

        let sealed = seal_gck_to_member(
            &generate_group_key(),
            &admin_identity.public_key(),
            &group_id.to_string(),
            1,
        )
        .expect("seal admin grant");
        kit.get(&format!("/api/v1/sync/groups/{group_id}/grant"))
            .json(json!({
                "sealed_gck": sealed,
                "gck_version": 1,
            }))
            .await;
        kit.get(&format!("/api/v1/sync/groups/{group_id}/pubkeys"))
            .json(json!([
                { "user_id": admin_id, "pubkey": admin_identity.public_key().to_base64() },
                { "user_id": Uuid::new_v4(), "pubkey": "not-a-key" },
            ]))
            .await;
        mount_rotate(&kit).await;

        client
            .rotate_group_key(group_id, &[])
            .await
            .expect_err("an unreadable member key must not produce a partial rotation");

        assert_eq!(
            kit.hits(&format!("/api/v1/sync/groups/{group_id}/rotate"))
                .await,
            0,
            "nothing may be posted when the batch could not be built in full"
        );
    }

    /// `IdentityPublicKey` round-trips through the wire form the pubkey list uses.
    /// If this ever stops holding, every rotation silently degrades to the error
    /// path above.
    #[test]
    fn member_pubkeys_round_trip_through_base64() {
        let identity = IdentityKeypair::generate();
        let encoded = identity.public_key().to_base64();
        let decoded = IdentityPublicKey::from_base64(&encoded).expect("round-trip");
        assert_eq!(decoded.as_bytes(), identity.public_key().as_bytes());
    }
}

/// The client half of history-survives-rotation: one pull can span GCK
/// generations, and each entry is opened under the key it was sealed with.
mod generations {
    use crate::common::*;
    use synckit_client::{
        ChangeEntry, GroupId, IdentityKeypair, generate_group_key, seal_gck_to_member,
    };
    use wiremock::matchers::{path_regex, query_param};

    fn change(row: &str, title: &str) -> ChangeEntry {
        ChangeEntry {
            table: "tasks".to_string(),
            op: ChangeOp::Insert,
            row_id: row.to_string(),
            timestamp: Utc::now(),
            hlc: Hlc::zero(DeviceId::nil()),
            data: Some(json!({ "title": title })),
            extra: serde_json::Map::default(),
        }
    }

    /// Push one change under `gck` and return the ciphertext the client produced,
    /// so it can be served straight back in a pull. Going through the real push
    /// path keeps the fixture honest: no test-local reimplementation of the AAD
    /// binding to drift from the one the client uses.
    async fn sealed_entry(
        client: &SyncKitClient,
        kit: &MockKit,
        group_id: GroupId,
        gck: &[u8; 32],
        device: DeviceId,
        row: &str,
        title: &str,
    ) -> serde_json::Value {
        let push_path = format!("/api/v1/sync/groups/{group_id}/push");
        let before = kit.hits(&push_path).await;
        client
            .group_push(group_id, gck, device, vec![change(row, title)])
            .await
            .expect("group push");
        let pushes = kit.requests_to(&push_path).await;
        assert!(pushes.len() > before, "a push was sent");
        let body: serde_json::Value =
            serde_json::from_slice(&pushes[before].body).expect("push body");
        body["changes"][0].clone()
    }

    #[tokio::test]
    async fn a_pull_spanning_two_generations_opens_each_under_its_own_key() {
        let kit = MockKit::start().await;
        let (client, master) = kit.keyed();
        let identity = IdentityKeypair::from_master_key(&master);

        let group_id = GroupId::new(Uuid::new_v4());
        let device = DeviceId::new(Uuid::new_v4());
        let group_ref = group_id.to_string();
        let gck_v1 = generate_group_key();
        let gck_v2 = generate_group_key();

        kit.matching("POST", path_regex(r"^/api/v1/sync/groups/[^/]+/push$"))
            .json(json!({ "cursor": 1 }))
            .await;

        let old = sealed_entry(
            &client,
            &kit,
            group_id,
            &gck_v1,
            device,
            "r-old",
            "before rotation",
        )
        .await;
        let new = sealed_entry(
            &client,
            &kit,
            group_id,
            &gck_v2,
            device,
            "r-new",
            "after rotation",
        )
        .await;

        // Each generation's grant is fetched by version. Serving only these two
        // means a client that ignored the per-entry version and asked for one key
        // would still get an answer, and then fail to decrypt half the batch.
        for (version, gck) in [(1, &gck_v1), (2, &gck_v2)] {
            let sealed =
                seal_gck_to_member(gck, &identity.public_key(), &group_ref, version).expect("seal");
            kit.get(&format!("/api/v1/sync/groups/{group_id}/grant"))
                .and(query_param("version", version.to_string()))
                .json(json!({
                    "sealed_gck": sealed,
                    "gck_version": version,
                }))
                .await;
        }

        kit.matching("POST", path_regex(r"^/api/v1/sync/groups/[^/]+/pull$"))
            .json(json!({
                "changes": [
                    {
                        "seq": 1,
                        "device_id": device,
                        "table": old["table"],
                        "op": old["op"],
                        "row_id": old["row_id"],
                        "timestamp": old["timestamp"],
                        "data": old["data"],
                        "gck_version": 1,
                    },
                    {
                        "seq": 2,
                        "device_id": device,
                        "table": new["table"],
                        "op": new["op"],
                        "row_id": new["row_id"],
                        "timestamp": new["timestamp"],
                        "data": new["data"],
                        "gck_version": 2,
                    },
                ],
                "cursor": 2,
                "has_more": false,
            }))
            .await;

        let (changes, cursor, has_more) = client
            .group_pull_rich(group_id, 2, device, 0)
            .await
            .expect("a pull spanning generations must succeed");

        assert_eq!(cursor, 2);
        assert!(!has_more);
        assert_eq!(changes.len(), 2);
        assert_eq!(
            changes[0].entry.data.as_ref().expect("old plaintext"),
            &json!({ "title": "before rotation" }),
            "the pre-rotation entry must open under generation 1"
        );
        assert_eq!(
            changes[1].entry.data.as_ref().expect("new plaintext"),
            &json!({ "title": "after rotation" }),
            "the post-rotation entry must open under generation 2"
        );
    }

    /// An entry with no generation is what a server predating per-generation
    /// grants returns. The caller's current generation is the fallback, so an old
    /// server keeps working rather than failing every pull.
    #[tokio::test]
    async fn an_entry_without_a_generation_falls_back_to_the_current_one() {
        let kit = MockKit::start().await;
        let (client, master) = kit.keyed();
        let identity = IdentityKeypair::from_master_key(&master);

        let group_id = GroupId::new(Uuid::new_v4());
        let device = DeviceId::new(Uuid::new_v4());
        let gck = generate_group_key();

        kit.matching("POST", path_regex(r"^/api/v1/sync/groups/[^/]+/push$"))
            .json(json!({ "cursor": 1 }))
            .await;
        let entry = sealed_entry(
            &client,
            &kit,
            group_id,
            &gck,
            device,
            "r-legacy",
            "legacy row",
        )
        .await;

        let sealed = seal_gck_to_member(&gck, &identity.public_key(), &group_id.to_string(), 5)
            .expect("seal");
        kit.get(&format!("/api/v1/sync/groups/{group_id}/grant"))
            .json(json!({
                "sealed_gck": sealed,
                "gck_version": 5,
            }))
            .await;

        kit.matching("POST", path_regex(r"^/api/v1/sync/groups/[^/]+/pull$"))
            .json(json!({
                "changes": [{
                    "seq": 1,
                    "device_id": device,
                    "table": entry["table"],
                    "op": entry["op"],
                    "row_id": entry["row_id"],
                    "timestamp": entry["timestamp"],
                    "data": entry["data"],
                }],
                "cursor": 1,
                "has_more": false,
            }))
            .await;

        let (changes, _, _) = client
            .group_pull_rich(group_id, 5, device, 0)
            .await
            .expect("a generation-less entry must not fail the pull");
        assert_eq!(
            changes[0].entry.data.as_ref().expect("plaintext"),
            &json!({ "title": "legacy row" })
        );
    }
}