monetize-product 0.3.2

//! **The product as monetize serves it** (5.4, G1 pull; lane MONETIZE-FIX step 3c): every
//! answer the server loops ask of a product — its tenants and who they are, their usage and
//! activity, what the box can serve, the paid door, the room a raise finds, a grant of users,
//! a purge — read from the newest report of the estate's main ([`ProductReport`]), and every
//! change queued as a signed order the main pulls. Monetize never calls into an estate.
//!
//! [`Pulled`] is concrete: there is no per-product host adapter. A product's own words are its
//! facts ([`ProductFacts`], from its guest); the box's are its report. [`PullLane`] is the
//! guardian's store as a pulled product reads and queues on it (`monetize-guardian`'s `Lane`
//! in production).

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

use crate::pull::{ProductReport, PurgeAnswer};
use crate::{
    Absorb, Activity, EntitlementFact, MeteredDoor, ProductError, ProductFacts, Purged, Servable,
    TenantId, TenantOwner, TenantUsage, Usage, UsagePage, UsagePageReq, UsersAnswer, UsersAsk,
};

/// How long [`Pulled::add_users`] waits for the main's ack of a grant it just queued: the
/// main long-polls its orders, runs one at once and reports straight after, so a live estate
/// answers within a few seconds; past this the grant stays queued and is answered on the
/// next ask (the paid path asks again on every read).
pub const USERS_ACK_WAIT: std::time::Duration = std::time::Duration::from_secs(20);

/// Where a product's facts come from, asked each time (a guest replaced live answers anew).
pub type FactsSource = Arc<dyn Fn() -> ProductFacts + Send + Sync>;

/// **One product's estate over the pull channel**, as [`Pulled`] reads and orders it: the
/// estate's main's reports, the fleet's other estates of the product, and the signed orders
/// queued for them (marked, so a repeat answers from the first one's ack).
pub trait PullLane: Send + Sync {
    /// The product's own estate.
    fn instance(&self) -> &str;
    /// The estate whose main serves `tenant` and the tenant's name there.
    fn serving(&self, tenant: &TenantId) -> Result<(String, String), String>;
    /// `instance`'s main's newest report and when monetize heard it.
    fn report_of(&self, instance: &str) -> Result<(ProductReport, u64), String>;
    /// The same, refused when the main has been silent past the guardian's bound.
    fn fresh_of(&self, instance: &str) -> Result<ProductReport, String>;
    /// The product's other estates on the fleet.
    fn sold_estates(&self) -> Vec<String>;
    /// Queue the fact for the main that serves its tenant (newest per tenant wins).
    fn queue_entitlement(&self, fact: &EntitlementFact) -> Result<(), String>;
    /// Whether the grant under `reference` is queued and not yet acked.
    fn users_queued(&self, tenant: &TenantId, reference: &str) -> bool;
    /// Queue a grant of users once per `reference`; `Some` once the main acked it.
    fn add_users(
        &self,
        tenant: &TenantId,
        ask: &UsersAsk,
        reference: &str,
    ) -> Result<Option<Result<UsersAnswer, String>>, String>;
    /// Queue ONE `grow_room` order for the main that serves `tenant`.
    fn queue_room(&self, tenant: &TenantId, need_bytes: u64, why: &str) -> Result<(), String>;
    /// Queue the purge once; `Some` once the main says how it went.
    fn purge(
        &self,
        tenant: &TenantId,
        reason: &str,
    ) -> Result<Option<Result<PurgeAnswer, String>>, String>;
    /// The own estate's newest report.
    fn report(&self) -> Result<(ProductReport, u64), String> {
        self.report_of(self.instance())
    }
    /// The own estate's report, fresh.
    fn fresh(&self) -> Result<ProductReport, String> {
        self.fresh_of(self.instance())
    }
}

/// **A product, served** (see the module): its facts and the lane of its estate.
#[derive(Clone)]
pub struct Pulled {
    facts: FactsSource,
    lane: Arc<dyn PullLane>,
    /// How long `add_users` waits for its ack ([`USERS_ACK_WAIT`]).
    ack_wait: std::time::Duration,
}

impl std::fmt::Debug for Pulled {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Pulled")
            .field("instance", &self.lane.instance())
            .finish()
    }
}

impl Pulled {
    /// The product whose facts `facts` answers, served from `lane`.
    pub fn new(facts: FactsSource, lane: Arc<dyn PullLane>) -> Self {
        Self {
            facts,
            lane,
            ack_wait: USERS_ACK_WAIT,
        }
    }

    /// [`Self::new`] waiting `wait` for a grant's ack (tests: none).
    pub fn with_ack_wait(mut self, wait: std::time::Duration) -> Self {
        self.ack_wait = wait;
        self
    }

    /// The estate whose main serves the product's own tenants.
    pub fn instance(&self) -> &str {
        self.lane.instance()
    }

    fn report(&self) -> Result<(ProductReport, u64), String> {
        self.lane.report()
    }

    fn fresh(&self) -> Result<ProductReport, String> {
        self.lane.fresh()
    }

    /// The own estate's report (required: the product is unavailable without it), then each
    /// sold estate's that has reported, its tenants keyed `@<estate>/<ns>`. A sold estate that
    /// has not called home is left out; monetize never dials it.
    fn every_report(&self) -> Result<Vec<ProductReport>, ProductError> {
        let (own, _) = self.lane.report().map_err(ProductError::Unavailable)?;
        let mut all = vec![own];
        for estate in self.lane.sold_estates() {
            let Ok((mut p, _)) = self.lane.report_of(&estate) else {
                continue;
            };
            for o in &mut p.tenants {
                o.tenant = TenantId::on_estate(&estate, &o.tenant.0);
            }
            p.usage = std::mem::take(&mut p.usage)
                .into_iter()
                .map(|(t, u)| (TenantId::on_estate(&estate, &t).0, u))
                .collect();
            all.push(p);
        }
        Ok(all)
    }

    pub fn facts(&self) -> crate::ProductFacts {
        (self.facts)()
    }
    pub fn list_tenants(&self) -> Result<Vec<TenantId>, ProductError> {
        Ok(self
            .every_report()?
            .into_iter()
            .flat_map(|p| p.tenants)
            .map(|o| o.tenant)
            .collect())
    }
    pub fn read_usage(&self, tenant: &TenantId) -> Result<Usage, ProductError> {
        let (instance, ns) = self.lane.serving(tenant).map_err(ProductError::Refused)?;
        let (mut p, _) = self
            .lane
            .report_of(&instance)
            .map_err(ProductError::Unavailable)?;
        p.usage.remove(&ns).ok_or_else(|| {
            ProductError::Refused(format!(
                "{instance}: the main's report has no usage for tenant {}",
                tenant.0
            ))
        })
    }
    pub fn read_usage_page(&self, req: &UsagePageReq) -> Result<Option<UsagePage>, ProductError> {
        let _ = req;
        let all = self.every_report()?;
        let as_of_unix_ms = all
            .iter()
            .map(|p| p.measured_unix_ms)
            .min()
            .unwrap_or_default();
        let rows = all
            .into_iter()
            .flat_map(|p| p.usage)
            .map(|(t, usage)| TenantUsage {
                tenant: TenantId(t),
                usage,
            })
            .collect();
        Ok(Some(UsagePage {
            rows,
            as_of_unix_ms,
            ..Default::default()
        }))
    }
    pub fn tenant_owners(&self) -> Result<Option<Vec<TenantOwner>>, ProductError> {
        Ok(Some(
            self.every_report()?
                .into_iter()
                .flat_map(|p| p.tenants)
                .collect(),
        ))
    }
    /// ★ UW-10: the days of the asked window from the serving main's report, with how far
    /// back the box remembers. A report that carries none for the tenant is unavailable,
    /// never an empty series (that would read as a silent tenant).
    pub fn read_activity(
        &self,
        tenant: &TenantId,
        since_unix_ms: u64,
        until_unix_ms: u64,
    ) -> Result<Activity, ProductError> {
        let (instance, ns) = self.lane.serving(tenant).map_err(ProductError::Refused)?;
        let (mut p, _) = self
            .lane
            .report_of(&instance)
            .map_err(ProductError::Unavailable)?;
        let mut a = p.activity.remove(&ns).ok_or_else(|| {
            ProductError::Unavailable(format!(
                "{instance}: the main's report carries no activity for tenant {}",
                tenant.0
            ))
        })?;
        let until = if until_unix_ms == 0 {
            u64::MAX
        } else {
            until_unix_ms
        };
        a.days
            .retain(|d| d.day_unix_ms >= since_unix_ms && d.day_unix_ms <= until);
        Ok(a)
    }
    pub fn push_entitlement(&self, fact: &EntitlementFact) -> Result<(), ProductError> {
        self.lane
            .queue_entitlement(fact)
            .map_err(ProductError::Unavailable)
    }
    /// A raise to `caps` needs `caps − promised` of the reported room of the main that serves
    /// the tenant (a sold estate's own for `@<estate>/<ns>`), per meter the report names room
    /// for. Short: bought iron helps only when the report says how.
    pub fn can_absorb(
        &self,
        tenant: &TenantId,
        caps: &BTreeMap<String, u64>,
    ) -> Result<Absorb, String> {
        let (instance, ns) = self.lane.serving(tenant)?;
        let p = self.lane.fresh_of(&instance)?;
        let held = p.promised.get(&ns);
        let short: Vec<String> = caps
            .iter()
            .filter_map(|(meter, target)| {
                let room = *p.room.get(meter)?;
                let need =
                    target.saturating_sub(held.and_then(|h| h.get(meter)).copied().unwrap_or(0));
                (need > room).then(|| {
                    format!("{meter} needs {need} beyond the {room} the main has room for")
                })
            })
            .collect();
        if short.is_empty() {
            return Ok(Absorb::OnHand);
        }
        match p.with_iron {
            Some(note) => Ok(Absorb::WithIron { note }),
            None => Err(format!("{instance}: {}", short.join("; "))),
        }
    }
    pub fn servable(&self) -> Servable {
        match self.fresh() {
            Ok(p) => p.servable.unwrap_or_else(|| {
                Servable::Unmeasured(format!(
                    "{}: its main's report does not say what it can serve",
                    self.instance()
                ))
            }),
            Err(w) => Servable::Unmeasured(w),
        }
    }
    /// ST-11: another estate of the product answers from its OWN main's fresh report.
    pub fn servable_on(&self, estate: &str) -> Servable {
        if estate == self.instance() {
            return self.servable();
        }
        match self.lane.fresh_of(estate) {
            Ok(p) => p.servable.unwrap_or_else(|| {
                Servable::Unmeasured(format!(
                    "{estate}: its main's report does not say what it can serve"
                ))
            }),
            Err(w) => Servable::Unmeasured(w),
        }
    }
    pub fn metered_door(&self) -> MeteredDoor {
        match self.report() {
            Ok((p, _)) => p.metered.unwrap_or_else(|| {
                MeteredDoor::Unproven(format!(
                    "{}: its main's report does not say",
                    self.instance()
                ))
            }),
            Err(w) => MeteredDoor::Unproven(w),
        }
    }
    /// ★ UW-5: from the serving main's fresh report — the account's room, its seats and the
    /// room one user holds — and its names (a held name is refused before money moves). The
    /// grant re-checks in the box; a refusal there is refunded.
    pub fn users_fit(
        &self,
        tenant: &TenantId,
        ask: &UsersAsk,
    ) -> Result<UsersAnswer, ProductError> {
        let (instance, ns) = self.lane.serving(tenant).map_err(ProductError::Refused)?;
        let p = self
            .lane
            .fresh_of(&instance)
            .map_err(ProductError::Unavailable)?;
        let name = ask.name.trim();
        if ask.count.max(1) == 1 && !name.is_empty() && p.names.iter().any(|n| n == name) {
            return Err(ProductError::Refused(format!(
                "{name} is already a name on {instance}: pick another"
            )));
        }
        let room = p.accounts.get(&ns).ok_or_else(|| {
            ProductError::Unavailable(format!(
                "{instance}: the main's report says nothing of {}'s room yet",
                tenant.0
            ))
        })?;
        let room_bytes = if ask.room_bytes > 0 {
            ask.room_bytes
        } else {
            room.room_bytes
        };
        let need_bytes = room_bytes.saturating_mul(u64::from(ask.count.max(1)));
        let shortfall_bytes = room
            .unused_bytes
            .map(|u| need_bytes.saturating_sub(u))
            .unwrap_or(0);
        Ok(UsersAnswer {
            principal: String::new(),
            room_bytes,
            need_bytes,
            unused_bytes: room.unused_bytes,
            shortfall_bytes,
            seats: room.seats,
            seats_open: room.seats_open,
            unchanged: true,
            invite_secret: String::new(),
        })
    }
    /// ★ UW-5: ONE signed `add_users` order per `order`, answered from the main's ack — waited
    /// for a short while (a live main runs it at once); still waiting, `Unavailable` (queued),
    /// and the next ask reads the ack.
    pub fn add_users(
        &self,
        tenant: &TenantId,
        ask: &UsersAsk,
        order: &str,
    ) -> Result<UsersAnswer, ProductError> {
        // Waited for only by the call that queues it: a read of a grant queued before (the
        // poller, a status page) answers at once and never holds a thread.
        let waits = !self.lane.users_queued(tenant, order);
        let until = std::time::Instant::now() + self.ack_wait;
        loop {
            match self
                .lane
                .add_users(tenant, ask, order)
                .map_err(ProductError::Unavailable)?
            {
                Some(Ok(answer)) => return Ok(answer),
                Some(Err(words)) => return Err(ProductError::Refused(words)),
                None if !waits || std::time::Instant::now() >= until => {
                    return Err(ProductError::Unavailable(format!(
                        "{order}: queued for {}'s main to grant on its next call",
                        self.instance()
                    )))
                }
                None => std::thread::sleep(std::time::Duration::from_millis(250)),
            }
        }
    }
    /// ★ UW-4: the serving main says its set grows at runtime.
    /// ★ The customer never sees a cap (owner): past the room of a main whose set is not
    /// SEALED — it grows at runtime or at a restart, an operator grows it, or it cannot tell —
    /// the order is queued and the main told (`grow_room`); a sealed set refuses, as before.
    pub fn queues_room(&self) -> bool {
        matches!(self.fresh().ok().and_then(|p| p.servable), Some(Servable::Measured { growth, .. }) if !matches!(growth, crate::Growth::Sealed(_)))
    }
    pub fn queue_room(
        &self,
        tenant: &TenantId,
        need_bytes: u64,
        why: &str,
    ) -> Result<(), ProductError> {
        self.lane
            .queue_room(tenant, need_bytes, why)
            .map_err(ProductError::Unavailable)
    }
    /// ★ UW-11: ONE signed `purge_tenant` order at a time, asked again (naming the job) until
    /// the main says it settled. Waiting or running is `still_locked` (nonpay asks again next
    /// tick); a FAILED job is refused, so nothing of the tenant's is released.
    pub fn purge_tenant(&self, tenant: &TenantId, reason: &str) -> Result<Purged, ProductError> {
        let locked = |detail: String| Purged {
            tenant: tenant.clone(),
            detail,
            bytes_reclaimed: None,
            still_locked: true,
            failures: Vec::new(),
        };
        match self.lane.purge(tenant, reason).map_err(ProductError::Unavailable)? {
            None => Ok(locked(format!("queued for {}'s main to purge on its next call; the account stays locked", self.instance()))),
            Some(Err(words)) => Err(ProductError::Refused(words)),
            Some(Ok(a)) if a.failed => Err(ProductError::Refused(format!(
                "{}'s purge of {} did not finish ({}): {}{}",
                self.instance(),
                tenant.0,
                a.detail,
                if a.error.is_empty() { "no reason given" } else { &a.error },
                if a.failures.is_empty() { String::new() } else { format!("; still on the disk: {}", a.failures.join("; ")) }
            ))),
            Some(Ok(a)) if a.running => Ok(locked(format!("{}; the main's job {} is still running — the account stays LOCKED until it settles", a.detail, a.job))),
            Some(Ok(a)) => Ok(Purged { tenant: tenant.clone(), detail: a.detail, bytes_reclaimed: a.bytes_reclaimed, still_locked: a.still_locked, failures: a.failures }),
        }
    }
}

/// **A product placed on no estate**: `[products.<id>]` names no `instance`, so no main's report
/// answers for it; every read and order is refused by name.
pub struct Unplaced(pub &'static str);

impl Unplaced {
    fn no(&self) -> String {
        format!("[products.{}] names no instance: no estate's main reports for it, and monetize never dials one", self.0)
    }
}

impl PullLane for Unplaced {
    fn instance(&self) -> &str {
        ""
    }
    fn serving(&self, _tenant: &TenantId) -> Result<(String, String), String> {
        Err(self.no())
    }
    fn report_of(&self, _instance: &str) -> Result<(ProductReport, u64), String> {
        Err(self.no())
    }
    fn fresh_of(&self, _instance: &str) -> Result<ProductReport, String> {
        Err(self.no())
    }
    fn sold_estates(&self) -> Vec<String> {
        Vec::new()
    }
    fn queue_entitlement(&self, _fact: &EntitlementFact) -> Result<(), String> {
        Err(self.no())
    }
    fn users_queued(&self, _tenant: &TenantId, _reference: &str) -> bool {
        false
    }
    fn add_users(
        &self,
        _tenant: &TenantId,
        _ask: &UsersAsk,
        _reference: &str,
    ) -> Result<Option<Result<UsersAnswer, String>>, String> {
        Err(self.no())
    }
    fn queue_room(&self, _tenant: &TenantId, _need_bytes: u64, _why: &str) -> Result<(), String> {
        Err(self.no())
    }
    fn purge(
        &self,
        _tenant: &TenantId,
        _reason: &str,
    ) -> Result<Option<Result<PurgeAnswer, String>>, String> {
        Err(self.no())
    }
}

/// **One estate's main held in memory** (tests, and a host that keeps no guardian store): its
/// newest report, the orders queued for it, and the main's answers to the ones that answer.
/// Every report is fresh. What [`Pulled`] reads and queues on, as on a real main.
#[derive(Default)]
pub struct HeldMain {
    instance: String,
    held: std::sync::Mutex<Held>,
}

/// What a [`HeldMain`] holds.
#[derive(Default, Clone)]
pub struct Held {
    pub report: Option<ProductReport>,
    /// The entitlement facts queued, oldest first.
    pub entitlements: Vec<EntitlementFact>,
    /// The grants of users queued, oldest first.
    pub users: Vec<crate::pull::AddUsers>,
    /// The newest `grow_room` order, unacked.
    pub room: Option<crate::pull::GrowRoom>,
    /// The purges queued.
    pub purges: Vec<crate::pull::PurgeTenant>,
    /// The main's answer to a grant of users, once it ran one.
    pub users_answer: Option<Result<UsersAnswer, String>>,
    /// The main's answer to a purge, once it ran one.
    pub purge_answer: Option<Result<PurgeAnswer, String>>,
}

impl HeldMain {
    /// The main of `instance`, which has not reported yet.
    pub fn new(instance: impl Into<String>) -> Arc<HeldMain> {
        Arc::new(HeldMain {
            instance: instance.into(),
            ..Default::default()
        })
    }
    /// **The main reports**: `edit` changes its newest report (an empty one at first).
    pub fn report(&self, edit: impl FnOnce(&mut ProductReport)) {
        edit(
            self.lock()
                .report
                .get_or_insert_with(ProductReport::default),
        );
    }
    /// **The main changes what it holds or answers** (its answers to grants and purges).
    pub fn hold(&self, edit: impl FnOnce(&mut Held)) {
        edit(&mut self.lock());
    }
    /// What it holds now.
    pub fn held(&self) -> Held {
        self.lock().clone()
    }
    /// The entitlement facts queued, oldest first.
    pub fn entitlements(&self) -> Vec<EntitlementFact> {
        self.lock().entitlements.clone()
    }
    fn lock(&self) -> std::sync::MutexGuard<'_, Held> {
        self.held.lock().unwrap_or_else(|e| e.into_inner())
    }
    /// A product whose facts `facts` answers, served from this main (no ack wait).
    pub fn serve(self: &Arc<Self>, facts: ProductFacts) -> Pulled {
        Pulled::new(
            Arc::new(move || facts.clone()),
            Arc::clone(self) as Arc<dyn PullLane>,
        )
        .with_ack_wait(std::time::Duration::ZERO)
    }
}

impl PullLane for HeldMain {
    fn instance(&self) -> &str {
        &self.instance
    }
    fn serving(&self, tenant: &TenantId) -> Result<(String, String), String> {
        match tenant.estate() {
            Some((estate, ns)) if estate != self.instance => {
                Err(format!("{estate}: no estate of this product holds {ns}"))
            }
            Some((_, ns)) => Ok((self.instance.clone(), ns.to_owned())),
            None => Ok((self.instance.clone(), tenant.0.clone())),
        }
    }
    fn report_of(&self, instance: &str) -> Result<(ProductReport, u64), String> {
        if instance != self.instance {
            return Err(format!("{instance}: not this estate"));
        }
        self.lock().report.clone().map(|r| (r, 1)).ok_or_else(|| {
            format!("{instance}: its main has sent no report yet; monetize never dials it")
        })
    }
    fn fresh_of(&self, instance: &str) -> Result<ProductReport, String> {
        self.report_of(instance).map(|(r, _)| r)
    }
    fn sold_estates(&self) -> Vec<String> {
        Vec::new()
    }
    fn queue_entitlement(&self, fact: &EntitlementFact) -> Result<(), String> {
        self.lock().entitlements.push(fact.clone());
        Ok(())
    }
    fn users_queued(&self, _tenant: &TenantId, reference: &str) -> bool {
        let h = self.lock();
        h.users_answer.is_none() && h.users.iter().any(|u| u.order == reference)
    }
    fn add_users(
        &self,
        tenant: &TenantId,
        ask: &UsersAsk,
        reference: &str,
    ) -> Result<Option<Result<UsersAnswer, String>>, String> {
        let ns = self.serving(tenant)?.1;
        let mut h = self.lock();
        if !h.users.iter().any(|u| u.order == reference) {
            h.users.push(crate::pull::AddUsers {
                tenant: ns,
                ask: ask.clone(),
                order: reference.to_owned(),
            });
        }
        Ok(h.users_answer.clone())
    }
    fn queue_room(&self, _tenant: &TenantId, need_bytes: u64, why: &str) -> Result<(), String> {
        self.lock().room = Some(crate::pull::GrowRoom {
            need_bytes,
            reason: why.to_owned(),
        });
        Ok(())
    }
    fn purge(
        &self,
        tenant: &TenantId,
        reason: &str,
    ) -> Result<Option<Result<PurgeAnswer, String>>, String> {
        let ns = self.serving(tenant)?.1;
        let mut h = self.lock();
        if !h.purges.iter().any(|p| p.tenant == ns) {
            h.purges.push(crate::pull::PurgeTenant {
                tenant: ns,
                reason: reason.to_owned(),
                ..Default::default()
            });
        }
        Ok(h.purge_answer.clone())
    }
}