gunnar-server 3.0.0

//! **`restate_entitlements`** (5.4, ONE monetize, owner 13:35Z): after a recharge,
//! monetize's guardian queues the estate's REAL entitlement snapshot to the main —
//! every tenant's fact from the ONE ledger. The main verifies the snapshot (envelope
//! and every fact) under its monetize key, refuses the whole order on any failure,
//! acts only while it serves, applies each fact through the `Entitlement.Set` path
//! (newer wins, the same fact is a no-op) and deletes NOTHING.
//!
//! Driven through `gunnar_repair::exec::execute`, the one dispatch a pulled order
//! takes, on the server's own [`ServerEstate`].
#![cfg(all(feature = "replication", feature = "metered"))]

use std::collections::BTreeMap;
use std::sync::Arc;

use ed25519_dalek::{Signer, SigningKey};
use gunnar_common::monetize_key::MonetizePubkey;
use gunnar_core::ssh_key::PrivateKey;
use gunnar_core::{Principal, PrincipalKind};
use gunnar_repair::exec::execute;
use gunnar_repair::{Order, OrderAck};
use gunnar_server::entitlement::{EntitlementSlot, MeteredConfig};
use gunnar_server::repair::{Entitlements, ServerEstate};
use gunnar_server::stores::ServerStores;
use monetize_embed::signing::Snapshot;
use monetize_embed::{EntitlementFact, State, TenantId};

const T: u64 = 1_789_000_000_000;

struct Rig {
    _dir: tempfile::TempDir,
    stores: Arc<ServerStores>,
    slot: Arc<EntitlementSlot>,
    key: SigningKey,
    forger: SigningKey,
}

impl Rig {
    /// A main that serves: one principal, so it is not a fresh box.
    fn serving() -> Rig {
        let r = Rig::fresh();
        r.stores
            .auth()
            .add_principal(Principal {
                id: "alice".into(),
                name: "alice".into(),
                kind: PrincipalKind::User,
                created: 0,
            })
            .expect("principal");
        r
    }

    /// A re-laid main nobody restored: it does not serve.
    fn fresh() -> Rig {
        let dir = tempfile::tempdir().expect("tempdir");
        let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
        let key = SigningKey::from_bytes(&[7u8; 32]);
        let pubkey = dir.path().join("monetize.pub");
        std::fs::write(&pubkey, key.verifying_key().to_bytes()).expect("key");
        let config = MeteredConfig::load(
            &MonetizePubkey::Path(pubkey),
            Some("https://gunnar.rs/billing/{tenant}"),
        )
        .expect("config");
        let slot = EntitlementSlot::open(Arc::clone(&stores), &config).expect("slot");
        Rig {
            _dir: dir,
            stores,
            slot,
            key,
            forger: SigningKey::from_bytes(&[9u8; 32]),
        }
    }

    fn fact(&self, by: &SigningKey, ns: &str, state: State, issued: u64) -> EntitlementFact {
        let mut f = EntitlementFact {
            tenant: TenantId(ns.into()),
            plan: "10gb".into(),
            state,
            paid_until_unix_ms: None,
            caps: BTreeMap::new(),
            source: format!("payment:{ns}:{issued}"),
            signature: vec![],
            issued_unix_ms: Some(issued),
            issued_signature: vec![],
        };
        f.signature = by
            .sign(&monetize_embed::signing::fact_message(&f))
            .to_bytes()
            .to_vec();
        f.issued_signature = by
            .sign(&monetize_embed::signing::fact_message_issued(&f))
            .to_bytes()
            .to_vec();
        f
    }

    fn snapshot(
        &self,
        by: &SigningKey,
        issued: u64,
        facts: Vec<EntitlementFact>,
    ) -> serde_json::Value {
        let signature = by
            .sign(&monetize_embed::signing::snapshot_message(issued, &facts))
            .to_bytes()
            .to_vec();
        serde_json::to_value(Snapshot {
            issued_unix_ms: issued,
            facts,
            signature,
        })
        .expect("json")
    }

    fn estate(&self) -> ServerEstate {
        let pem = gunnar_core::mint_ed25519().expect("mint");
        let key = PrivateKey::from_openssh(std::str::from_utf8(&pem).unwrap()).expect("parse");
        ServerEstate {
            key,
            own: "127.0.0.1:1".into(),
            entitlements: Entitlements::of(Arc::clone(&self.stores))
                .with_slot(Some(Arc::clone(&self.slot))),
        }
    }

    async fn restate(&self, seq: u64, snapshot: serde_json::Value) -> OrderAck {
        execute(
            &self.estate(),
            seq,
            &Order::RestateEntitlements {
                snapshot,
                case: Some("case-1".into()),
            },
            &|| false,
        )
        .await
    }

    fn issued(&self, ns: &str) -> Option<u64> {
        self.slot
            .get(ns)
            .expect("get")
            .and_then(|f| f.issued_unix_ms)
    }
}

#[tokio::test]
async fn a_verified_snapshot_is_restated_newer_wins_and_nothing_is_deleted() {
    let r = Rig::serving();
    let first = r.snapshot(
        &r.key,
        T,
        vec![
            r.fact(&r.key, "held", State::Paid, T + 10),
            r.fact(&r.key, "same", State::Paid, T + 3),
            r.fact(&r.key, "kept", State::Paid, T + 1),
        ],
    );
    let a = r.restate(1, first).await;
    assert!(a.ok, "{a:?}");
    assert_eq!(
        a.detail,
        "restated 3 entitlement facts (0 already held as new or newer)"
    );

    let same = r.fact(&r.key, "same", State::Paid, T + 3);
    let second = r.snapshot(
        &r.key,
        T + 20,
        vec![
            r.fact(&r.key, "new", State::Paid, T + 11),
            r.fact(&r.key, "held", State::Suspended, T + 5),
            same,
        ],
    );
    let a = r.restate(2, second).await;
    assert!(a.ok, "{a:?}");
    assert_eq!(
        a.detail,
        "restated 1 entitlement facts (2 already held as new or newer)"
    );
    assert_eq!(
        r.issued("new"),
        Some(T + 11),
        "a tenant the box did not hold is restated"
    );
    assert_eq!(
        r.issued("held"),
        Some(T + 10),
        "an OLDER fact does not move a tenant back"
    );
    assert_eq!(r.issued("same"), Some(T + 3));
    assert_eq!(
        r.issued("kept"),
        Some(T + 1),
        "a tenant the snapshot leaves out is NOT deleted"
    );
}

#[tokio::test]
async fn a_forged_envelope_or_a_forged_fact_refuses_the_whole_order() {
    let r = Rig::serving();
    let forged_envelope = r.snapshot(&r.forger, T, vec![r.fact(&r.key, "a", State::Paid, T + 1)]);
    let a = r.restate(1, forged_envelope).await;
    assert!(!a.ok, "{a:?}");
    assert!(
        a.detail.starts_with("bad_order:") && a.detail.contains("does not verify"),
        "{}",
        a.detail
    );

    let one_forged_fact = r.snapshot(
        &r.key,
        T,
        vec![
            r.fact(&r.key, "a", State::Paid, T + 1),
            r.fact(&r.forger, "b", State::Paid, T + 1),
        ],
    );
    let a = r.restate(2, one_forged_fact).await;
    assert!(!a.ok, "{a:?}");
    assert!(a.detail.contains("does not verify"), "{}", a.detail);
    assert_eq!(
        r.issued("a"),
        None,
        "the good fact beside a forged one is NOT applied: the whole order is refused"
    );

    let a = r.restate(3, serde_json::json!({"facts": "nope"})).await;
    assert!(
        !a.ok && a.detail.starts_with("bad_order:") && a.detail.contains("snapshot"),
        "{}",
        a.detail
    );
}

#[tokio::test]
async fn an_empty_snapshot_is_restated_as_nothing() {
    let r = Rig::serving();
    let a = r.restate(1, r.snapshot(&r.key, T, vec![])).await;
    assert!(a.ok, "{a:?}");
    assert_eq!(
        a.detail,
        "restated 0 entitlement facts (0 already held as new or newer)"
    );
}

#[tokio::test]
async fn a_box_that_does_not_serve_restates_nothing_and_says_so() {
    let r = Rig::fresh();
    let a = r
        .restate(
            1,
            r.snapshot(&r.key, T, vec![r.fact(&r.key, "a", State::Paid, T + 1)]),
        )
        .await;
    assert!(!a.ok, "{a:?}");
    assert!(
        a.detail.starts_with("not_serving:") && a.detail.contains("serving=no"),
        "{}",
        a.detail
    );
    assert_eq!(r.issued("a"), None);
}

#[tokio::test]
async fn a_box_without_entitlements_refuses_a_snapshot_with_facts_by_name() {
    let r = Rig::serving();
    let estate = ServerEstate {
        entitlements: Entitlements::of(Arc::clone(&r.stores)),
        ..r.estate()
    };
    let snap = r.snapshot(&r.key, T, vec![r.fact(&r.key, "a", State::Paid, T + 1)]);
    let a = execute(
        &estate,
        1,
        &Order::RestateEntitlements {
            snapshot: snap,
            case: None,
        },
        &|| false,
    )
    .await;
    assert!(!a.ok, "{a:?}");
    assert!(
        a.detail.starts_with("bad_order:") && a.detail.contains("1 entitlement facts"),
        "{}",
        a.detail
    );
    let a = execute(
        &estate,
        2,
        &Order::RestateEntitlements {
            snapshot: r.snapshot(&r.key, T, vec![]),
            case: None,
        },
        &|| false,
    )
    .await;
    assert!(
        a.ok,
        "an empty snapshot asks nothing of a box without entitlements: {a:?}"
    );
}

/// **Order `entitlement`** (ONE monetize, G1-pull #1): what `Entitlement.Set` took when
/// monetize still dialled in now arrives as a pulled order, the signed fact its body.
/// The same path as a restatement: verified under the box's monetize key, newer wins.
#[tokio::test]
async fn an_entitlement_order_applies_a_newer_fact_holds_an_older_one_and_refuses_a_forgery() {
    let r = Rig::serving();
    let entitle = |seq: u64, fact: EntitlementFact| {
        let estate = r.estate();
        async move {
            execute(
                &estate,
                seq,
                &Order::Entitlement {
                    fact: serde_json::to_value(fact).expect("json"),
                },
                &|| false,
            )
            .await
        }
    };
    let a = entitle(1, r.fact(&r.key, "acme", State::Paid, T + 5)).await;
    assert!(a.ok, "{a:?}");
    assert!(
        a.detail.contains("acme") && a.detail.contains("Paid"),
        "{}",
        a.detail
    );
    assert_eq!(r.issued("acme"), Some(T + 5));

    let a = entitle(2, r.fact(&r.key, "acme", State::Suspended, T + 1)).await;
    assert!(a.ok, "an older fact is not news, and not a failure: {a:?}");
    assert!(a.detail.contains("already holds"), "{}", a.detail);
    assert_eq!(r.issued("acme"), Some(T + 5));

    let a = entitle(3, r.fact(&r.forger, "acme", State::Suspended, T + 9)).await;
    assert!(
        !a.ok && a.detail.starts_with("bad_order:") && a.detail.contains("does not verify"),
        "{a:?}"
    );
    assert_eq!(r.issued("acme"), Some(T + 5));

    let a = entitle(4, r.fact(&r.key, "acme", State::Suspended, T + 9)).await;
    assert!(a.ok, "{a:?}");
    assert_eq!(r.issued("acme"), Some(T + 9));

    let a = execute(
        &r.estate(),
        5,
        &Order::Entitlement {
            fact: serde_json::json!({"tenant": 1}),
        },
        &|| false,
    )
    .await;
    assert!(!a.ok && a.detail.starts_with("bad_order:"), "{a:?}");

    let fresh = Rig::fresh();
    let a = execute(
        &fresh.estate(),
        1,
        &Order::Entitlement {
            fact: serde_json::to_value(fresh.fact(&fresh.key, "acme", State::Paid, T)).unwrap(),
        },
        &|| false,
    )
    .await;
    assert!(!a.ok && a.detail.starts_with("not_serving:"), "{a:?}");
}