gunnar-server 3.0.0

//! **`Usage.AdmitPlan`** — the admission plan for a namespace: usage over the
//! rolling window (the flow ledger's bytes in + out), the budget, and per kind
//! of work the class, the estimate and the plan's sentence. The SAME grant as
//! `Usage.Activity` (`Read` on the namespace).
//!
//! Without `admit-share` the answer is `pass-through` — no budget, every kind
//! Full — and the usage is still the ledger's; with it, the configured budget
//! decides the class exactly as the gate does.

use std::sync::Arc;

use gunnar_core::Transfer;
use gunnar_grpc::{Caller, EventBus, ServerConfig, ServiceContext, UsageService};
use gunnar_proto::usage_server::Usage as _;
use gunnar_proto::{AdmitPlanRequest, AdmitPlanResponse};
use gunnar_server::control_plane::CoreControlPlane;
use gunnar_server::stores::ServerStores;
use gunnar_ssh::Admission;

const DAY: i64 = gunnar_core::DAY_MS;

fn noon() -> i64 {
    gunnar_core::midnight_of(gunnar_core::day_of(gunnar_core::now_ms())) + DAY / 2
}

fn box_with(namespaces: &[&str]) -> (tempfile::TempDir, Arc<ServerStores>) {
    let dir = tempfile::tempdir().expect("tempdir");
    let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
    for ns in namespaces {
        stores.catalog().create_namespace(ns).expect("namespace");
        stores
            .auth()
            .add_principal(gunnar_core::Principal {
                id: (*ns).to_owned(),
                name: (*ns).to_owned(),
                kind: gunnar_core::PrincipalKind::User,
                created: 0,
            })
            .expect("principal");
    }
    (dir, stores)
}

/// The gate as `serve` builds it: the flow ledger, and (with the feature) a
/// usage share seeded from it.
fn gate(stores: &Arc<ServerStores>, limit: Option<u64>) -> Arc<Admission> {
    let gate = Admission::new(Some(8), Some(4)).with_flow(Arc::clone(stores.flow()));
    #[cfg(feature = "admit-share")]
    let gate = gate.with_policy(
        gunnar_ssh::edda_admit::Settings {
            limit,
            ..Default::default()
        }
        .build_seeded(Some(Admission::flow_seed(Arc::clone(stores.flow())))),
    );
    #[cfg(not(feature = "admit-share"))]
    let _ = limit;
    Arc::new(gate)
}

fn service(stores: &Arc<ServerStores>, gate: Arc<Admission>) -> UsageService {
    let plane: Arc<dyn gunnar_grpc::ControlPlane> =
        Arc::new(CoreControlPlane::new(Arc::clone(stores)).with_admission(gate));
    UsageService::new(ServiceContext::new(
        plane,
        Arc::new(EventBus::new(16)),
        ServerConfig::default(),
    ))
}

fn plan_as(
    service: &UsageService,
    principal: &str,
    namespace: &str,
) -> Result<AdmitPlanResponse, tonic::Status> {
    let mut request = tonic::Request::new(AdmitPlanRequest {
        namespace: namespace.to_owned(),
    });
    request.extensions_mut().insert(Caller {
        principal_id: principal.to_owned(),
        fingerprint: "SHA256:test".to_owned(),
        machine: false,
        authenticated: true,
    });
    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("runtime")
        .block_on(service.admit_plan(request))
        .map(tonic::Response::into_inner)
}

/// Yesterday 1 000 in (flushed), today 234 out (live): 1 234 over the window.
fn traffic(stores: &ServerStores) {
    let flow = stores.flow();
    flow.record(
        "alpha",
        noon() - DAY,
        &Transfer {
            bytes_in: 1_000,
            pushed: true,
            ..Transfer::default()
        },
    );
    flow.flush(noon() - DAY).expect("flush");
    flow.record(
        "alpha",
        noon(),
        &Transfer {
            bytes_out: 234,
            fetched: true,
            ..Transfer::default()
        },
    );
}

/// ★ **The plan is the namespace's own read**, like `Activity`: the owner
/// reads it, another tenant is refused, an empty name is an argument error
/// and an unknown one is not dressed up as an idle tenant.
#[test]
fn the_admission_plan_is_the_namespaces_own_read_grant() {
    let (_dir, stores) = box_with(&["alpha", "beta"]);
    traffic(&stores);
    let service = service(&stores, gate(&stores, Some(1_000_000)));

    let mine = plan_as(&service, "alpha", "alpha").expect("the owner reads her own");
    assert_eq!(mine.used_bytes, 1_234, "yesterday plus today, in plus out");
    assert_eq!(mine.window_days, 30);
    let kinds: Vec<&str> = mine.kinds.iter().map(|k| k.kind.as_str()).collect();
    assert_eq!(kinds, ["clone", "fetch", "push", "request"]);

    let theirs = plan_as(&service, "beta", "alpha").expect_err("beta must not read alpha's plan");
    assert_eq!(theirs.code(), tonic::Code::PermissionDenied, "{theirs:?}");
    assert_eq!(
        plan_as(&service, "alpha", "").unwrap_err().code(),
        tonic::Code::InvalidArgument
    );
    assert_eq!(
        plan_as(&service, "alpha", "gamma").unwrap_err().code(),
        tonic::Code::PermissionDenied
    );
}

/// ★ **Without the feature the plan is PassThrough**: no budget, every kind
/// Full, and the sentence says so — the truth about that build.
#[cfg(not(feature = "admit-share"))]
#[test]
fn without_the_share_every_kind_is_full_and_there_is_no_budget() {
    let (_dir, stores) = box_with(&["alpha"]);
    traffic(&stores);
    let plan = plan_as(&service(&stores, gate(&stores, None)), "alpha", "alpha").unwrap();
    assert_eq!(plan.policy, "pass-through");
    assert_eq!((plan.limit_bytes, plan.budget_left_bytes), (None, None));
    assert!(
        plan.kinds.iter().all(|k| k.admit_class == "Full"),
        "{plan:?}"
    );
    assert_eq!(
        plan.kinds[0].why,
        "no limit; 30d used 1.2 KB; this clone ~? (no estimate yet); Full"
    );
}

/// ★ **With the feature the budget decides**: within it every kind is Full
/// and the budget left is reported; past it every kind is Reduced — the same
/// answer the gate acts on.
#[cfg(feature = "admit-share")]
#[test]
fn with_the_share_the_budget_decides_the_class() {
    let (_dir, stores) = box_with(&["alpha"]);
    traffic(&stores);
    let within = plan_as(
        &service(&stores, gate(&stores, Some(10_000))),
        "alpha",
        "alpha",
    )
    .unwrap();
    assert_eq!(within.policy, "usage-share");
    assert_eq!(
        (within.limit_bytes, within.budget_left_bytes),
        (Some(10_000), Some(8_766))
    );
    assert!(
        within.kinds.iter().all(|k| k.admit_class == "Full"),
        "{within:?}"
    );
    assert_eq!(
        within.kinds[0].why,
        "10 KB/30d used 1.2 KB; this clone ~? (no estimate yet); Full"
    );

    let over = plan_as(
        &service(&stores, gate(&stores, Some(1_000))),
        "alpha",
        "alpha",
    )
    .unwrap();
    assert_eq!(over.budget_left_bytes, Some(0));
    assert!(
        over.kinds.iter().all(|k| k.admit_class == "Reduced"),
        "{over:?}"
    );
    assert!(
        over.kinds[0]
            .why
            .starts_with("1 KB/30d used 1.2 KB; this clone"),
        "{}",
        over.kinds[0].why
    );
}

/// ★ **The medium's token reaches the server's settings**: `GUNNAR_ADMIT_LIMIT`
/// (the `korp.env.` token `admit_limit` lands) is what `--admit-limit` reads;
/// `none` is PassThrough, a size is a budget, a typo refuses by name.
#[cfg(feature = "admit-share")]
#[test]
fn the_admit_flags_become_the_settings() {
    let s = gunnar_server::admit_settings(Some("none"), None).unwrap();
    assert!(s.is_pass_through());
    assert_eq!(s.build().policy(), "pass-through");
    let s = gunnar_server::admit_settings(Some("100GB"), Some(7)).unwrap();
    assert_eq!((s.limit, s.window_days), (Some(100_000_000_000), 7));
    assert_eq!(s.build().policy(), "usage-share");
    let e = gunnar_server::admit_settings(Some("5 parsecs"), None)
        .unwrap_err()
        .to_string();
    assert!(e.contains("--admit-limit"), "{e}");
    assert!(gunnar_server::admit_settings(None, Some(0)).is_err());
}