gunnar-server 3.0.0

//! **`Report.product`** (5.4, ONE monetize, G1 pull): what monetize used to dial a main
//! for, said by the main itself in its `Fleet.Report`.
//!
//! Monetize never calls into an estate. Its gunnar plugin used to ask the box over the
//! control plane for its tenants (`Repos.ListNamespaces`), their usage (`Usage.Read`),
//! what it can still serve (`Admin.Capacity`) and whether its paid door answers; under
//! ONE monetize the main reads the same answers from its OWN control plane, in process,
//! and sends them as JSON text of `monetize_product::pull::ProductReport`.
//!
//! The JSON is written by [`wire`], a mirror of the monetize-product types (serde's
//! default shapes: a newtype tenant id is its string, an enum is externally tagged).
//! The folds are the plugin's (`monetize/products/gunnar`): `usage_from_namespace_usage`,
//! `servable_from`, `growth_from` with `reconcile_growth`, `box_ceilings`, and the
//! `absorbs` table for when a bought disk helps.

use std::collections::BTreeMap;

use gunnar_grpc::ControlPlane;
use gunnar_proto::{CapacityVerdict, PrincipalKind, SetGrowth};

/// The monetize-product types `ProductReport` is made of, as JSON.
pub mod wire {
    use std::collections::BTreeMap;

    use serde::{Deserialize, Serialize};

    #[derive(Clone, PartialEq, Eq, Debug, Default, Serialize, Deserialize)]
    #[serde(default)]
    pub struct ProductReport {
        pub tenants: Vec<TenantOwner>,
        pub usage: BTreeMap<String, Usage>,
        pub servable: Option<Servable>,
        pub metered: Option<MeteredDoor>,
        pub room: BTreeMap<String, u64>,
        pub promised: BTreeMap<String, BTreeMap<String, u64>>,
        pub with_iron: Option<String>,
        pub names: Vec<String>,
        /// ★ UW-5: each account's room for "Add users" (monetize-product `pull::AccountRoom`).
        pub accounts: BTreeMap<String, AccountRoom>,
        /// ★ UW-10: each tenant's activity over the last [`super::ACTIVITY_DAYS`] days
        /// (monetize-product `Activity`), by tenant name.
        pub activity: BTreeMap<String, Activity>,
        pub measured_unix_ms: u64,
    }

    #[derive(Clone, PartialEq, Eq, Debug, Default, Serialize, Deserialize)]
    pub struct ActivityDay {
        pub day_unix_ms: u64,
        pub meters: BTreeMap<String, u64>,
    }

    #[derive(Clone, PartialEq, Eq, Debug, Default, Serialize, Deserialize)]
    pub struct Activity {
        pub days: Vec<ActivityDay>,
        pub retained_days: u64,
    }

    #[derive(Clone, PartialEq, Eq, Debug, Default, Serialize, Deserialize)]
    #[serde(default)]
    pub struct AccountRoom {
        pub unused_bytes: Option<u64>,
        pub room_bytes: u64,
        pub seats: u32,
        pub seats_open: u32,
    }

    #[derive(Clone, PartialEq, Eq, Debug, Serialize, Deserialize)]
    pub struct TenantOwner {
        pub tenant: String,
        pub person: bool,
        pub seller: bool,
    }

    #[derive(Clone, PartialEq, Eq, Debug, Default, Serialize, Deserialize)]
    pub struct Usage {
        pub meters: BTreeMap<String, u64>,
        pub measured_at_unix_ms: u64,
    }

    #[derive(Clone, PartialEq, Eq, Debug, Serialize, Deserialize)]
    pub enum Growth {
        Sealed(String),
        OperatorOnly(String),
        Unknown(String),
        AtRestart(String),
        AtRuntime(String),
    }

    #[derive(Clone, PartialEq, Eq, Debug, Serialize, Deserialize)]
    pub enum Servable {
        Measured {
            servable_bytes: u64,
            total_bytes: u64,
            growth: Growth,
            ceilings: BTreeMap<String, u64>,
        },
        Unmeasured(String),
    }

    #[derive(Clone, PartialEq, Eq, Debug, Serialize, Deserialize)]
    pub enum MeteredDoor {
        Enforcing(String),
        Absent(String),
        Unproven(String),
    }
}

use wire::{Growth, MeteredDoor, ProductReport, Servable, TenantOwner, Usage};

/// ★ UW-10: how many days of each tenant's activity the report carries (monetize-product
/// `pull::ACTIVITY_DAYS`).
pub const ACTIVITY_DAYS: u64 = 30;

/// The byte meters that live on `/data`: what `servable_bytes` is room for.
pub const DATA_METERS: [&str; 3] = ["pack_bytes", "lfs_bytes", "cache_bytes"];

/// RAM held per indexed object by an open store (the plugin's, measured: 382 B).
pub const RAM_BYTES_PER_OBJECT: u64 = 382;

const USAGE_PAGE: usize = 1000;

/// **The main's word about its product**, read from its own control plane.
/// `promised` is each tenant's enforced caps; `metered` is the paid door as this build
/// knows it. A part the plane cannot answer is said as such, never as a zero.
pub fn read(
    control: &dyn ControlPlane,
    promised: BTreeMap<String, BTreeMap<String, u64>>,
    metered: MeteredDoor,
    now_unix_ms: u64,
) -> ProductReport {
    let mut names = std::collections::BTreeSet::new();
    let tenants = match control.list_namespaces() {
        Ok(namespaces) => namespaces
            .into_iter()
            .inspect(|n| {
                names.insert(n.name.clone());
            })
            .filter(|n| n.owner_kind != PrincipalKind::Machine as i32)
            .map(|n| TenantOwner {
                person: n.owner_kind == PrincipalKind::User as i32,
                seller: n.owner_server_grant,
                tenant: n.name,
            })
            .collect(),
        Err(e) => {
            tracing::warn!(target: crate::repair::TARGET, "Report.product: the namespaces could not be listed: {e}");
            Vec::new()
        }
    };
    match control.list_principals() {
        Ok(principals) => names.extend(
            principals
                .into_iter()
                .flat_map(|p| [p.id, p.name])
                .filter(|n| !n.is_empty()),
        ),
        Err(e) => {
            tracing::warn!(target: crate::repair::TARGET, "Report.product: the principals could not be listed: {e}")
        }
    }

    let servable = servable_of(control);
    let (room, with_iron) = room_of(&servable);
    ProductReport {
        tenants,
        usage: usage_of(control, now_unix_ms),
        servable: Some(servable),
        metered: Some(metered),
        room,
        promised,
        with_iron,
        names: names.into_iter().collect(),
        accounts: BTreeMap::new(),
        activity: BTreeMap::new(),
        measured_unix_ms: now_unix_ms,
    }
}

/// ★ UW-10: **each tenant's kept activity** over the last [`ACTIVITY_DAYS`] days — the plane's
/// own `Usage.Activity`, meter for meter as the plugin names them (`activity_day`) — what
/// monetize's tenant page reads instead of dialling the box. A tenant the plane cannot answer
/// for is left out, never an empty series.
pub fn activity_of(
    control: &dyn ControlPlane,
    report: &ProductReport,
    now_unix_ms: u64,
) -> BTreeMap<String, wire::Activity> {
    let since = now_unix_ms.saturating_sub(ACTIVITY_DAYS * 86_400_000) as i64;
    report
        .tenants
        .iter()
        .filter_map(|t| match control.activity(&t.tenant, since, 0) {
            Ok((days, retained)) => Some((t.tenant.clone(), wire::Activity { days: days.iter().map(activity_day).collect(), retained_days: retained.max(0) as u64 })),
            Err(e) => {
                tracing::debug!(target: crate::repair::TARGET, "Report.product: {}'s activity: {e}", t.tenant);
                None
            }
        })
        .collect()
}

/// The plugin's `activity_day_from_pb`, meter for meter.
pub fn activity_day(d: &gunnar_proto::DayActivity) -> wire::ActivityDay {
    let mut meters: BTreeMap<String, u64> = [
        ("bytes_in", d.bytes_in),
        ("bytes_out", d.bytes_out),
        ("pushes", d.pushes),
        ("fetches", d.fetches),
        ("lfs_bytes_in", d.lfs_bytes_in),
        ("lfs_bytes_out", d.lfs_bytes_out),
        ("refused", d.refused),
        ("auth_failures", d.auth_failures),
        ("distinct_principals", d.distinct_principals),
        ("last_push_unix_ms", d.last_push_unix_ms.max(0) as u64),
        ("last_fetch_unix_ms", d.last_fetch_unix_ms.max(0) as u64),
    ]
    .into_iter()
    .map(|(k, v)| (k.to_string(), v))
    .collect();
    for (reason, count) in &d.refused_by_reason {
        meters.insert(format!("refused_{reason}"), *count);
    }
    wire::ActivityDay {
        day_unix_ms: d.day_unix_ms.max(0) as u64,
        meters,
    }
}

/// ★ UW-5: **each person's account's room for "Add users"** — what monetize answers "would
/// they fit" from: the room neither stored nor promised (the plane's own dry run, against the
/// account's enforced `pack_bytes` cap and what it stores), its seats, and the room one user
/// holds. An account the plane cannot answer for is left out, never guessed.
pub fn accounts_of(
    control: &dyn ControlPlane,
    stores: &crate::stores::ServerStores,
    report: &ProductReport,
) -> BTreeMap<String, wire::AccountRoom> {
    let room_bytes = gunnar_grpc::service::accounts::ROOM_PER_USER_BYTES;
    report
        .tenants
        .iter()
        .filter(|t| t.person)
        .filter_map(|t| {
            let cap = report.promised.get(&t.tenant).and_then(|c| c.get("pack_bytes")).copied().filter(|c| *c > 0);
            let stored = report.usage.get(&t.tenant).and_then(|u| u.meters.get("pack_bytes")).copied().unwrap_or(0);
            let ask = gunnar_grpc::EmployeeAsk { room_bytes, seats: 1, cap, stored, dry_run: true, ..Default::default() };
            let answer = control.enroll_employee(&t.tenant, &ask).map_err(|e| tracing::debug!(target: crate::repair::TARGET, "Report.product: {}'s room: {e}", t.tenant)).ok()?;
            let (seats, seats_open) = stores.auth().seats_of(&t.tenant).ok()?;
            Some((t.tenant.clone(), wire::AccountRoom { unused_bytes: answer.unused_bytes, room_bytes, seats, seats_open }))
        })
        .collect()
}

/// Every namespace's meters, paged off the usage ledger (`UsageBatch.ReadMany`'s read).
fn usage_of(control: &dyn ControlPlane, now_unix_ms: u64) -> BTreeMap<String, Usage> {
    let mut out = BTreeMap::new();
    let mut token = String::new();
    loop {
        let page = match control.usage_page(&token, USAGE_PAGE, "") {
            Ok(p) => p,
            Err(e) => {
                tracing::warn!(target: crate::repair::TARGET, "Report.product: usage could not be read: {e}");
                return out;
            }
        };
        for u in &page.usages {
            out.insert(u.namespace.clone(), usage_from(u, now_unix_ms));
        }
        if page.next_page_token.is_empty() || page.next_page_token == token {
            return out;
        }
        token = page.next_page_token;
    }
}

/// The plugin's `usage_from_namespace_usage`, meter for meter.
pub fn usage_from(u: &gunnar_proto::NamespaceUsage, fallback_now: u64) -> Usage {
    let meters = [
        ("pack_bytes", u.pack_bytes),
        ("cache_bytes", u.cache_bytes),
        ("cold_bytes", u.cold_bytes),
        ("lfs_bytes", u.lfs_bytes),
        ("tombstoned_bytes", u.tombstoned_bytes),
        ("repos", u.repos),
        ("objects", u.objects),
        (
            "open_store_ram_bytes",
            u.objects.saturating_mul(RAM_BYTES_PER_OBJECT),
        ),
        ("bytes_in_today", u.bytes_in_today),
        ("bytes_out_today", u.bytes_out_today),
        ("pushes_today", u.pushes_today),
        ("fetches_today", u.fetches_today),
        ("refused_today", u.refused_today),
        ("last_push_unix_ms", u.last_push_unix_ms.max(0) as u64),
        ("last_fetch_unix_ms", u.last_fetch_unix_ms.max(0) as u64),
    ]
    .into_iter()
    .map(|(k, v)| (k.to_owned(), v))
    .collect();
    let measured = if u.measured_at_unix_ms > 0 {
        u.measured_at_unix_ms as u64
    } else {
        fallback_now
    };
    Usage {
        meters,
        measured_at_unix_ms: measured,
    }
}

/// `Admin.Capacity` for the whole box, with the grow engine asked as the plugin did.
fn servable_of(control: &dyn ControlPlane) -> Servable {
    let mut answer = match control.capacity(&gunnar_proto::CapacityRequest::default()) {
        Ok(a) => a,
        Err(e) => {
            return Servable::Unmeasured(format!(
                "gunnar did not answer how much it can serve: {e}. THAT IS A MEASUREMENT THAT DID NOT HAPPEN, not a full disk"
            ))
        }
    };
    if answer.growth() == SetGrowth::GrowthSealed
        && control.grow_status().is_ok_and(|s| s.engine_present)
    {
        answer.growth = SetGrowth::GrowthAtRuntime as i32;
    }
    match answer.verdict() {
        CapacityVerdict::CapacityUnmeasured | CapacityVerdict::Unspecified => Servable::Unmeasured(format!(
            "gunnar COULD NOT MEASURE its data set, so it has not said how much it can serve: {}. An unmeasured box has not said it is full and has not said it is empty",
            answer.reason
        )),
        CapacityVerdict::CapacityAvailable | CapacityVerdict::CapacityExhausted => {
            let ceilings = [("cpu_millicores", answer.cpu_millicores_total), ("ram_bytes", answer.ram_bytes_total)]
                .into_iter()
                .filter_map(|(m, v)| v.filter(|v| *v > 0).map(|v| (m.to_owned(), v)))
                .collect();
            Servable::Measured { servable_bytes: answer.servable_bytes, total_bytes: answer.total_bytes, growth: growth_from(answer.growth()), ceilings }
        }
    }
}

/// gunnar's `growth`, each arm with the plugin's `remediation` sentence.
pub fn growth_from(growth: SetGrowth) -> Growth {
    match growth {
        SetGrowth::GrowthSealed => Growth::Sealed(
            "AND THIS SET CANNOT GROW WHILE GUNNAR RUNS: this gunnar build has no runtime data-set growth (its grow door reports no grow-runtime engine) and is PID 1 with no shell, so a disk bought and attached to this box would be BILLED AND INVISIBLE. Serving more is an OPERATOR MAINTENANCE OPERATION that enlarges the set's existing volumes in place (or a fresh install onto a larger set), not another disk: nothing here buys or attaches one".into(),
        ),
        SetGrowth::GrowthOperatorOnly => Growth::OperatorOnly(
            "gunnar will not grow the set itself, but this box has a shell beside it: an operator can extend the filesystem out of band and gunnar will measure the larger one on its next look".into(),
        ),
        SetGrowth::GrowthUnknown | SetGrowth::Unspecified => Growth::Unknown(
            "gunnar could not tell whether this set can ever grow, and UNKNOWN is not a promise that it can — nothing should be bought for this box until it can be measured".into(),
        ),
        SetGrowth::GrowthAtRestart => Growth::AtRestart(
            "this set GROWS AT ITS NEXT RESTART once every member volume has been enlarged to one equal size: still no hot-plug, so a disk attached to this box is billed and invisible — the way to serve more is the data-set growth (monetize's outside actor enlarges the members in place), never another disk".into(),
        ),
        SetGrowth::GrowthAtRuntime => Growth::AtRuntime(
            "this set GROWS IN PLACE AT RUNTIME: gunnar drains, unmounts and grows into enlarged member volumes while its control plane stays up — serve more with the Data set panel (Quote, then Grow: monetize's outside actor enlarges the members in place, gunnar grows into them). Only an EXTRA disk attached beside the set would be billed and invisible, so never buy another disk".into(),
        ),
    }
}

/// **The room a raise finds, and whether bought iron helps** — the plugin's `absorbs`
/// table said as numbers: every `/data` byte meter has `servable_bytes` of room (free
/// minus the disk floor, never the floor itself). An unmeasured box has NO room: nothing
/// is sold on a guess. Only an operator-only set turns a shortfall into a disk to buy.
pub fn room_of(servable: &Servable) -> (BTreeMap<String, u64>, Option<String>) {
    match servable {
        Servable::Measured {
            servable_bytes,
            growth,
            ..
        } => {
            let room = DATA_METERS
                .iter()
                .map(|m| ((*m).to_owned(), *servable_bytes))
                .collect();
            let iron = match growth {
                Growth::OperatorOnly(words) => Some(format!("A disk may be bought; {words}. Until an operator extends the filesystem onto it, the tenant's cap is not served")),
                _ => None,
            };
            (room, iron)
        }
        Servable::Unmeasured(_) => (
            DATA_METERS.iter().map(|m| ((*m).to_owned(), 0)).collect(),
            None,
        ),
    }
}