diff --git a/HOSTED.md b/HOSTED.md index cac6627..bdbae45 100644 --- a/HOSTED.md +++ b/HOSTED.md @@ -51,6 +51,15 @@ flowchart LR Launch with flat monthly tiers and task caps. Avoid pure usage billing until task duration and model-cost distribution are known. The concrete tier matrix, launch price proposal, and the enforcement mechanism behind every promised limit live in [docs/pricing.md](docs/pricing.md). +The adapter enforces purchased plans directly: GitHub Marketplace +`marketplace_purchase` webhooks record each account's plan, `installation` +events map installations to their account, and intake plus the worker apply +the tier's task caps and concurrency automatically. `[billing] require_plan` +turns the hosted deployment into a paid gate — installations without an +entitled plan (and no explicit `[[installations]]` entry) are recorded +`ignored:no_plan`. Explicit TOML limits always win, which is how Dedicated +contracts get custom numbers. + ## Buyer Promise > Your familiar on your GitHub: the one that already knows your code, your team, and how you ship. diff --git a/config/example.toml b/config/example.toml index a555cc3..df38e70 100644 --- a/config/example.toml +++ b/config/example.toml @@ -96,6 +96,18 @@ trigger_labels = ["coven:fix", "coven:review"] # Labels that trigger this famil # [installations.repos."your-org/quiet"] # labels = false # only the label lane off +# ── Billing entitlements (hosted, docs/pricing.md) ────────────────────────── +# Marketplace purchases map plans to the same limit knobs automatically: +# `marketplace_purchase` webhooks record the purchasing account's plan, and +# `installation` events map installations to their account. Explicit +# [installations.limits] values always win over plan defaults (Dedicated +# contracts, operator overrides). +# [billing] +# require_plan = false # true (hosted): installations with neither an +# # entitled plan (active/trial) nor an explicit +# # [[installations]] entry are recorded +# # ignored:no_plan — the monetization gate. + # ── Automatic review policy ───────────────────────────────────────────────── # Hosted PR review lanes (issue #10). Push/commit review is parsed today and # ships with headless contract v3. diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs index ee3f16f..eda9fa4 100644 --- a/crates/config/src/lib.rs +++ b/crates/config/src/lib.rs @@ -33,6 +33,106 @@ pub struct Config { /// fails closed for installations not listed. #[serde(default)] pub installations: Vec, + /// Hosted billing entitlement policy (docs/pricing.md). Absent section = + /// billing off: no plan is required and TOML limits are the only limits. + #[serde(default)] + pub billing: BillingConfig, +} + +/// Hosted billing entitlement policy: how purchased plans gate intake. +#[derive(Debug, Clone, Copy, Default, Deserialize, Serialize)] +pub struct BillingConfig { + /// When true, deliveries from installations that have neither an + /// entitled purchased plan (active or on trial) nor an explicit + /// `[[installations]]` entry are recorded `ignored:no_plan` — the hosted + /// monetization gate. Default false: self-hosted deployments never + /// require a plan. + #[serde(default)] + pub require_plan: bool, +} + +/// A hosted pricing tier (docs/pricing.md). Purchased plans map to the same +/// enforcement knobs as `[installations.limits]`; explicit TOML values always +/// win so operators can customize (e.g. Dedicated contracts). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PlanTier { + Starter, + Team, + /// Custom limits by contract — the tier itself imposes none; the + /// operator sets them per installation in TOML. + Dedicated, + /// A plan name we could not classify (e.g. a renamed Marketplace + /// listing). Treated as Starter — the most conservative paid tier — so + /// a rename never grants unlimited service or locks a paying customer + /// out. + Unknown, +} + +impl PlanTier { + /// Classify a marketplace plan name, e.g. "Hosted Team" → `Team`. + pub fn parse(name: &str) -> Self { + let name = name.to_ascii_lowercase(); + if name.contains("starter") { + Self::Starter + } else if name.contains("team") { + Self::Team + } else if name.contains("dedicated") { + Self::Dedicated + } else { + Self::Unknown + } + } + + /// Stable identifier for storage and audit records. + pub fn as_str(&self) -> &'static str { + match self { + Self::Starter => "starter", + Self::Team => "team", + Self::Dedicated => "dedicated", + Self::Unknown => "unknown", + } + } + + /// The tier's default limits (docs/pricing.md tier matrix). + pub fn limits(&self) -> InstallationLimits { + match self { + Self::Starter | Self::Unknown => InstallationLimits { + max_concurrent: Some(1), + max_tasks_per_day: Some(25), + }, + Self::Team => InstallationLimits { + max_concurrent: Some(4), + max_tasks_per_day: Some(150), + }, + Self::Dedicated => InstallationLimits::default(), + } + } +} + +impl std::str::FromStr for PlanTier { + type Err = std::convert::Infallible; + fn from_str(s: &str) -> Result { + Ok(match s { + "starter" => Self::Starter, + "team" => Self::Team, + "dedicated" => Self::Dedicated, + _ => Self::Unknown, + }) + } +} + +/// Effective limits for one installation: explicit TOML values win per field +/// (operator override, Dedicated contracts); a purchased plan supplies tier +/// defaults for the rest; neither = unlimited (the self-hosted default). +pub fn effective_limits( + explicit: InstallationLimits, + plan: Option, +) -> InstallationLimits { + let tier = plan.map(|p| p.limits()).unwrap_or_default(); + InstallationLimits { + max_concurrent: explicit.max_concurrent.or(tier.max_concurrent), + max_tasks_per_day: explicit.max_tasks_per_day.or(tier.max_tasks_per_day), + } } /// Routing and trigger policy for one GitHub App installation (issue #7). @@ -737,6 +837,12 @@ impl Config { .unwrap_or_default() } + /// Whether an installation has an explicit `[[installations]]` entry — + /// an operator-vouched tenant, exempt from `billing.require_plan`. + pub fn installation_listed(&self, installation_id: u64) -> bool { + self.installations.iter().any(|i| i.id == installation_id) + } + /// `installation id → max_concurrent` for every configured cap — the /// worker's claim filter (issue #15). pub fn concurrency_caps(&self) -> std::collections::HashMap { @@ -1488,6 +1594,7 @@ mod tests { gardener: GardenerConfig::default(), api: ApiConfig::default(), installations: vec![], + billing: BillingConfig::default(), } } @@ -2406,3 +2513,66 @@ mod tests { assert!(errs.contains(&"familiars[].bot_username")); } } + +#[cfg(test)] +mod plan_tier_tests { + //! Plan → limits mapping and precedence (docs/pricing.md). + use super::*; + + #[test] + fn marketplace_plan_names_classify_to_tiers() { + assert_eq!(PlanTier::parse("Hosted Starter"), PlanTier::Starter); + assert_eq!(PlanTier::parse("Hosted Team"), PlanTier::Team); + assert_eq!(PlanTier::parse("Hosted Dedicated"), PlanTier::Dedicated); + assert_eq!(PlanTier::parse("TEAM (annual)"), PlanTier::Team); + assert_eq!(PlanTier::parse("Mystery Plan"), PlanTier::Unknown); + } + + #[test] + fn tier_limits_match_the_pricing_matrix() { + let starter = PlanTier::Starter.limits(); + assert_eq!(starter.max_concurrent, Some(1)); + assert_eq!(starter.max_tasks_per_day, Some(25)); + + let team = PlanTier::Team.limits(); + assert_eq!(team.max_concurrent, Some(4)); + assert_eq!(team.max_tasks_per_day, Some(150)); + + // Dedicated is custom-by-contract: the tier imposes nothing. + let dedicated = PlanTier::Dedicated.limits(); + assert_eq!(dedicated.max_concurrent, None); + assert_eq!(dedicated.max_tasks_per_day, None); + } + + #[test] + fn unknown_plan_names_fail_safe_to_starter_limits() { + assert_eq!(PlanTier::Unknown.limits().max_tasks_per_day, Some(25)); + } + + #[test] + fn explicit_toml_limits_override_plan_defaults_per_field() { + let explicit = InstallationLimits { + max_concurrent: Some(8), + max_tasks_per_day: None, + }; + let effective = effective_limits(explicit, Some(PlanTier::Starter)); + // Operator's concurrency wins; the tier still supplies the daily cap. + assert_eq!(effective.max_concurrent, Some(8)); + assert_eq!(effective.max_tasks_per_day, Some(25)); + } + + #[test] + fn no_plan_and_no_toml_means_unlimited() { + let effective = effective_limits(InstallationLimits::default(), None); + assert_eq!(effective.max_concurrent, None); + assert_eq!(effective.max_tasks_per_day, None); + } + + #[test] + fn tier_ids_round_trip_through_storage_strings() { + for tier in [PlanTier::Starter, PlanTier::Team, PlanTier::Dedicated] { + assert_eq!(tier.as_str().parse::().unwrap(), tier); + } + assert_eq!("bogus".parse::().unwrap(), PlanTier::Unknown); + } +} diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index ab7c85f..d9ce17b 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -15,7 +15,7 @@ use std::path::Path; use std::sync::{Arc, Mutex}; /// Current schema version, stored in `PRAGMA user_version`. -const SCHEMA_VERSION: i32 = 5; +const SCHEMA_VERSION: i32 = 6; /// Handle to the durable store. Cheap to clone; all clones share one writer /// connection. @@ -480,6 +480,32 @@ pub struct UsageRow { pub runtime_secs: u64, } +/// The purchased plan for one account (billing entitlements). Written from +/// `marketplace_purchase` webhook events; resolved per installation through +/// `installation_accounts`. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct AccountPlan { + pub account_id: u64, + pub account_login: String, + /// The raw marketplace plan name, kept for audit. + pub plan_name: String, + /// Parsed tier id: `starter` | `team` | `dedicated` | `unknown`. + pub tier: String, + /// `active` | `trial` | `cancelled`. + pub state: String, + /// Where the plan record came from: `marketplace` | `manual`. + pub source: String, + pub updated_at: String, +} + +impl AccountPlan { + /// Whether this plan currently entitles service. + pub fn entitled(&self) -> bool { + matches!(self.state.as_str(), "active" | "trial") + } +} + /// One task-lifecycle audit event for the Cave dashboard (issue #18). #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)] #[serde(rename_all = "camelCase")] @@ -645,6 +671,213 @@ impl Store { .expect("store task panicked") } + /// Upsert the purchased plan for one account (billing entitlements). + pub async fn set_account_plan(&self, plan: AccountPlan) -> Result<()> { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().expect("store mutex poisoned"); + conn.execute( + "INSERT INTO account_plans + (account_id, account_login, plan_name, tier, state, source, updated_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) + ON CONFLICT(account_id) DO UPDATE SET + account_login = excluded.account_login, + plan_name = excluded.plan_name, + tier = excluded.tier, + state = excluded.state, + source = excluded.source, + updated_at = excluded.updated_at", + params![ + plan.account_id, + plan.account_login, + plan.plan_name, + plan.tier, + plan.state, + plan.source, + now_rfc3339(), + ], + )?; + Ok(()) + }) + .await + .expect("store task panicked") + } + + /// Records a `marketplace_purchase` delivery and applies its plan + /// transition in one transaction. "Delivery seen" and "plan applied" are + /// atomic: a storage failure can never mark the delivery as handled while + /// dropping the transition — a lost `cancelled` would leave an account + /// entitled forever, and a lost `purchased` would lock a paying customer + /// out with no redelivery able to repair either. + /// + /// `plan` = `None` records the delivery without a state change (pending + /// changes, unrecognized actions). Duplicates persist nothing further. + pub async fn record_delivery_with_plan( + &self, + delivery: Delivery, + reason: &str, + plan: Option, + ) -> Result { + let routing_label = format!("ignored:{reason}"); + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let mut conn = conn.lock().expect("store mutex poisoned"); + let now = now_rfc3339(); + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let inserted = tx.execute( + "INSERT INTO webhook_deliveries + (delivery_id, event, action, installation_id, repo, payload_hash, routing, received_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) + ON CONFLICT(delivery_id) DO NOTHING", + params![ + delivery.delivery_id, + delivery.event, + delivery.action, + delivery.installation_id, + delivery.repo, + delivery.payload_hash, + routing_label, + now, + ], + )?; + if inserted == 0 { + tx.commit()?; + return Ok(Recorded::Duplicate); + } + if let Some(plan) = plan { + tx.execute( + "INSERT INTO account_plans + (account_id, account_login, plan_name, tier, state, source, updated_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) + ON CONFLICT(account_id) DO UPDATE SET + account_login = excluded.account_login, + plan_name = excluded.plan_name, + tier = excluded.tier, + state = excluded.state, + source = excluded.source, + updated_at = excluded.updated_at", + params![ + plan.account_id, + plan.account_login, + plan.plan_name, + plan.tier, + plan.state, + plan.source, + now, + ], + )?; + } + tx.commit()?; + Ok(Recorded::New) + }) + .await + .expect("store task panicked") + } + + /// Upsert the account behind one installation (from `installation` + /// webhook events) so installation → plan can be resolved. + pub async fn record_installation_account( + &self, + installation_id: u64, + account_id: u64, + account_login: &str, + ) -> Result<()> { + let conn = self.conn.clone(); + let account_login = account_login.to_string(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().expect("store mutex poisoned"); + conn.execute( + "INSERT INTO installation_accounts + (installation_id, account_id, account_login, updated_at) + VALUES (?1, ?2, ?3, ?4) + ON CONFLICT(installation_id) DO UPDATE SET + account_id = excluded.account_id, + account_login = excluded.account_login, + updated_at = excluded.updated_at", + params![installation_id, account_id, account_login, now_rfc3339()], + )?; + Ok(()) + }) + .await + .expect("store task panicked") + } + + /// The purchased plan behind one installation, if its account is known + /// and has one. Returns cancelled plans too — callers check + /// [`AccountPlan::entitled`]. + pub async fn plan_for_installation( + &self, + installation_id: u64, + ) -> Result> { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().expect("store mutex poisoned"); + let plan = conn + .query_row( + "SELECT p.account_id, p.account_login, p.plan_name, + p.tier, p.state, p.source, p.updated_at + FROM account_plans p + JOIN installation_accounts ia ON ia.account_id = p.account_id + WHERE ia.installation_id = ?1", + params![installation_id], + |row| { + Ok(AccountPlan { + account_id: row.get(0)?, + account_login: row.get(1)?, + plan_name: row.get(2)?, + tier: row.get(3)?, + state: row.get(4)?, + source: row.get(5)?, + updated_at: row.get(6)?, + }) + }, + ) + .optional()?; + Ok(plan) + }) + .await + .expect("store task panicked") + } + + /// `installation id → tier` for every installation whose account holds an + /// entitled (active or trial) plan — the worker merges these into its + /// claim-time concurrency caps. + pub async fn entitled_installation_tiers(&self) -> Result> { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().expect("store mutex poisoned"); + let mut stmt = conn.prepare( + "SELECT ia.installation_id, p.tier + FROM installation_accounts ia + JOIN account_plans p ON p.account_id = ia.account_id + WHERE p.state IN ('active', 'trial')", + )?; + let rows = stmt + .query_map([], |row| Ok((row.get::<_, u64>(0)?, row.get::<_, String>(1)?)))? + .collect::, _>>()?; + Ok(rows) + }) + .await + .expect("store task panicked") + } + + /// Forget which account an installation belonged to (delete-on-uninstall). + /// The account's plan survives: other installations may share it, and + /// cancellation arrives separately via `marketplace_purchase`. + pub async fn delete_installation_account(&self, installation_id: u64) -> Result { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().expect("store mutex poisoned"); + let n = conn.execute( + "DELETE FROM installation_accounts WHERE installation_id = ?1", + params![installation_id], + )?; + Ok(n > 0) + }) + .await + .expect("store task panicked") + } + /// Usage metering rollup (issue #15): task counts and attempt runtime, /// grouped by installation, repository, and familiar. `scope` applies the /// same tenant boundary as the task list. @@ -1152,6 +1385,36 @@ fn migrate(conn: &Connection) -> Result<()> { ) .context("failed to apply schema v5")?; } + if version < 6 { + // v6: billing entitlements (docs/pricing.md). Marketplace purchases + // key on the purchasing *account*; tenancy keys on the *installation*. + // `account_plans` holds one plan per account; `installation_accounts` + // maps installations to their account so intake and the worker can + // resolve installation → plan. + conn.execute_batch( + r#" + CREATE TABLE account_plans ( + account_id INTEGER PRIMARY KEY, + account_login TEXT NOT NULL, + plan_name TEXT NOT NULL, + tier TEXT NOT NULL, + state TEXT NOT NULL, + source TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + + CREATE TABLE installation_accounts ( + installation_id INTEGER PRIMARY KEY, + account_id INTEGER NOT NULL, + account_login TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE INDEX installation_accounts_account + ON installation_accounts(account_id); + "#, + ) + .context("failed to apply schema v6")?; + } conn.pragma_update(None, "user_version", SCHEMA_VERSION)?; Ok(()) } @@ -2076,3 +2339,122 @@ mod metering_tests { assert_eq!(scoped[0].completed, 0); } } + +#[cfg(test)] +mod entitlement_tests { + //! Billing entitlements: account plans and installation resolution. + use super::*; + + fn plan(account_id: u64, tier: &str, state: &str) -> AccountPlan { + AccountPlan { + account_id, + account_login: "acme".into(), + plan_name: format!("Hosted {tier}"), + tier: tier.into(), + state: state.into(), + source: "marketplace".into(), + updated_at: String::new(), + } + } + + #[tokio::test] + async fn plan_resolves_through_the_installation_account_mapping() { + let store = Store::open_in_memory().unwrap(); + assert_eq!(store.plan_for_installation(7).await.unwrap(), None); + + store.record_installation_account(7, 42, "acme").await.unwrap(); + assert_eq!(store.plan_for_installation(7).await.unwrap(), None); + + store.set_account_plan(plan(42, "team", "active")).await.unwrap(); + let got = store.plan_for_installation(7).await.unwrap().expect("plan"); + assert_eq!(got.tier, "team"); + assert!(got.entitled()); + } + + #[tokio::test] + async fn upsert_replaces_the_plan_and_cancellation_removes_entitlement() { + let store = Store::open_in_memory().unwrap(); + store.record_installation_account(7, 42, "acme").await.unwrap(); + store.set_account_plan(plan(42, "starter", "trial")).await.unwrap(); + assert!(store.plan_for_installation(7).await.unwrap().unwrap().entitled()); + + store.set_account_plan(plan(42, "starter", "cancelled")).await.unwrap(); + let got = store.plan_for_installation(7).await.unwrap().expect("kept for audit"); + assert!(!got.entitled()); + } + + #[tokio::test] + async fn entitled_tiers_skip_cancelled_plans_and_unmapped_installations() { + let store = Store::open_in_memory().unwrap(); + store.record_installation_account(7, 42, "acme").await.unwrap(); + store.record_installation_account(8, 43, "beta").await.unwrap(); + store.set_account_plan(plan(42, "team", "active")).await.unwrap(); + store.set_account_plan(plan(43, "starter", "cancelled")).await.unwrap(); + + let tiers = store.entitled_installation_tiers().await.unwrap(); + assert_eq!(tiers.get(&7).map(String::as_str), Some("team")); + assert!(!tiers.contains_key(&8)); + } + + + #[tokio::test] + async fn delivery_and_plan_transition_are_atomic_and_dedupe_together() { + let store = Store::open_in_memory().unwrap(); + store.record_installation_account(7, 42, "acme").await.unwrap(); + let delivery = |id: &str| Delivery { + delivery_id: id.to_string(), + event: "marketplace_purchase".to_string(), + action: Some("purchased".to_string()), + installation_id: None, + repo: None, + payload_hash: "h".to_string(), + }; + + let first = store + .record_delivery_with_plan( + delivery("mp-1"), + "marketplace_purchase:purchased", + Some(plan(42, "team", "active")), + ) + .await + .unwrap(); + assert_eq!(first, Recorded::New); + assert!(store.plan_for_installation(7).await.unwrap().unwrap().entitled()); + + // A redelivery of the same id must not reapply state — a later, + // newer transition would otherwise be clobbered by a stale replay. + let replay = store + .record_delivery_with_plan( + delivery("mp-1"), + "marketplace_purchase:purchased", + Some(plan(42, "starter", "cancelled")), + ) + .await + .unwrap(); + assert_eq!(replay, Recorded::Duplicate); + assert_eq!(store.plan_for_installation(7).await.unwrap().unwrap().tier, "team"); + + // `None` records the delivery without touching the plan. + let noop = store + .record_delivery_with_plan(delivery("mp-2"), "marketplace_purchase:pending_change", None) + .await + .unwrap(); + assert_eq!(noop, Recorded::New); + assert_eq!(store.plan_for_installation(7).await.unwrap().unwrap().state, "active"); + } + + #[tokio::test] + async fn uninstall_forgets_the_mapping_but_keeps_the_account_plan() { + let store = Store::open_in_memory().unwrap(); + store.record_installation_account(7, 42, "acme").await.unwrap(); + store.set_account_plan(plan(42, "team", "active")).await.unwrap(); + + assert!(store.delete_installation_account(7).await.unwrap()); + assert!(!store.delete_installation_account(7).await.unwrap()); + assert_eq!(store.plan_for_installation(7).await.unwrap(), None); + + // Reinstall under the same account: the plan is still there. + store.record_installation_account(9, 42, "acme").await.unwrap(); + assert!(store.plan_for_installation(9).await.unwrap().unwrap().entitled()); + } +} diff --git a/crates/webhook/src/events.rs b/crates/webhook/src/events.rs index 87e6b31..51e088f 100644 --- a/crates/webhook/src/events.rs +++ b/crates/webhook/src/events.rs @@ -31,11 +31,37 @@ pub struct WebhookPayload { #[serde(default)] pub commits: Vec, pub sender: Option, + /// `marketplace_purchase` events: the purchase that changed. + pub marketplace_purchase: Option, } #[derive(Debug, Deserialize)] pub struct Installation { pub id: u64, + /// Present on `installation` lifecycle events; absent on the slim + /// `installation` object other events carry. + pub account: Option, +} + +/// A GitHub account (user or organization) — the unit Marketplace bills. +#[derive(Debug, Deserialize)] +pub struct Account { + pub id: u64, + pub login: String, +} + +/// `marketplace_purchase` payload: the purchase that changed (billing). +#[derive(Debug, Deserialize)] +pub struct MarketplacePurchase { + pub account: Account, + pub plan: Option, + #[serde(default)] + pub on_free_trial: bool, +} + +#[derive(Debug, Deserialize)] +pub struct MarketplacePlan { + pub name: String, } #[derive(Debug, Deserialize)] diff --git a/crates/webhook/src/routes.rs b/crates/webhook/src/routes.rs index bfc5235..6fba2e9 100644 --- a/crates/webhook/src/routes.rs +++ b/crates/webhook/src/routes.rs @@ -614,6 +614,30 @@ pub async fn handle_webhook( }; } + // Billing: `marketplace_purchase` deliveries carry plan changes for the + // purchasing *account*. Record the entitlement before acknowledging, and + // dedupe by delivery id like every other delivery. + if event_type == "marketplace_purchase" { + return handle_marketplace_purchase(&state, delivery, &payload).await; + } + + // `installation` lifecycle events name the account behind the + // installation — the join key that resolves a Marketplace purchase to + // this tenant. Keep the mapping current on every such event. + if event_type == "installation" && payload.action.as_deref() != Some("deleted") { + if let Some(inst) = &payload.installation { + if let Some(account) = &inst.account { + if let Err(e) = state + .store + .record_installation_account(inst.id, account.id, &account.login) + .await + { + error!("failed to record installation account: {e:#}"); + } + } + } + } + // Installation removed → purge that tenant's durable memory records // (issue #6, delete-on-uninstall). Idempotent via the delivery id. if event_type == "installation" && payload.action.as_deref() == Some("deleted") { @@ -645,6 +669,11 @@ pub async fn handle_webhook( 0 }); info!(installation_id = id, memory, tasks, "purged tenant data on uninstall"); + // Forget the account mapping too; the account's plan + // survives for its other installations. + if let Err(e) = state.store.delete_installation_account(id).await { + error!("failed to forget installation account on uninstall: {e:#}"); + } // Audit what was deleted (issue #12). let _ = state .store @@ -670,13 +699,61 @@ pub async fn handle_webhook( // delivery id must never dispatch twice (issue #2). match event_to_task(&state, event).await { Some(task) => { + // Entitlement resolution: explicit TOML limits win per field; + // otherwise a purchased plan supplies its tier defaults + // (docs/pricing.md). Neither = unlimited (self-hosted default). + let plan = match state.store.plan_for_installation(task.installation_id).await { + Ok(plan) => plan, + Err(e) => return storage_unavailable(e), + }; + let entitled = plan.as_ref().filter(|p| p.entitled()); + + // The hosted monetization gate: with `billing.require_plan`, an + // installation must hold an entitled plan or an explicit + // [[installations]] entry (operator-vouched, e.g. Dedicated). + if state.config.billing.require_plan + && entitled.is_none() + && !state.config.installation_listed(task.installation_id) + { + warn!( + installation_id = task.installation_id, + "no entitled plan — delivery ignored" + ); + if let Err(e) = state + .store + .record_delivery(delivery, Routing::Ignored("no_plan")) + .await + { + return storage_unavailable(e); + } + return ( + StatusCode::OK, + Json(json!({"ok": true, "ignored": "no_plan"})), + ) + .into_response(); + } + + let tier = entitled.map(|p| { + let tier: coven_github_config::PlanTier = + p.tier.parse().unwrap_or(coven_github_config::PlanTier::Unknown); + if tier == coven_github_config::PlanTier::Unknown { + warn!( + installation_id = task.installation_id, + plan = %p.plan_name, + "unclassified plan — applying Starter limits" + ); + } + tier + }); + let limits = coven_github_config::effective_limits( + state.config.limits_for(task.installation_id), + tier, + ); + // Daily task cap (issue #15): over-quota installations get their // delivery recorded as ignored — visible in the audit trail — and // GitHub still hears 200 (the delivery itself succeeded). - if let Some(cap) = state - .config - .limits_for(task.installation_id) - .max_tasks_per_day + if let Some(cap) = limits.max_tasks_per_day { let cutoff = (chrono::Utc::now() - chrono::Duration::hours(24)).to_rfc3339(); match state @@ -742,6 +819,105 @@ pub async fn handle_webhook( (StatusCode::OK, Json(json!({"ok": true}))).into_response() } +/// Handles a `marketplace_purchase` delivery: record the purchasing +/// account's plan (billing entitlement), audit the transition, and +/// acknowledge. Purchases key on the account; installations resolve to it +/// through the mapping kept by `installation` events. +async fn handle_marketplace_purchase( + state: &AppState, + delivery: Delivery, + payload: &WebhookPayload, +) -> axum::response::Response { + let action = payload.action.as_deref().unwrap_or(""); + let reason = format!("marketplace_purchase:{action}"); + + // Decide the transition first; the delivery record and the plan write + // then land in ONE store transaction, so a storage failure can never + // mark the delivery seen while dropping the transition (a redelivery + // would dedupe and the lost `cancelled`/`purchased` could never be + // recovered). + // + // Pending changes apply at the end of the billing cycle; GitHub sends a + // concrete `changed` event when they take effect. Nothing to do yet. + let transition = payload.marketplace_purchase.as_ref().and_then(|purchase| { + let plan_state = match action { + "purchased" | "changed" if purchase.on_free_trial => "trial", + "purchased" | "changed" => "active", + "cancelled" => "cancelled", + _ => return None, + }; + let plan_name = purchase + .plan + .as_ref() + .map(|p| p.name.clone()) + .unwrap_or_default(); + let tier = coven_github_config::PlanTier::parse(&plan_name); + Some((purchase, plan_state, plan_name, tier)) + }); + if payload.marketplace_purchase.is_none() { + warn!("marketplace_purchase delivery without a purchase object"); + } + + let record = transition + .as_ref() + .map(|(purchase, plan_state, plan_name, tier)| { + coven_github_store::AccountPlan { + account_id: purchase.account.id, + account_login: purchase.account.login.clone(), + plan_name: plan_name.clone(), + tier: tier.as_str().to_string(), + state: plan_state.to_string(), + source: "marketplace".to_string(), + updated_at: String::new(), + } + }); + + match state + .store + .record_delivery_with_plan(delivery, &reason, record) + .await + { + Ok(Recorded::New) => {} + Ok(Recorded::Duplicate) => { + return (StatusCode::OK, Json(json!({"ok": true, "duplicate": true}))) + .into_response(); + } + Err(e) => return storage_unavailable(e), + } + + let Some((purchase, plan_state, plan_name, tier)) = transition else { + info!(action, "marketplace_purchase acknowledged without state change"); + return (StatusCode::OK, Json(json!({"ok": true, "ignored": reason}))) + .into_response(); + }; + + info!( + account = %purchase.account.login, + plan = %plan_name, + tier = tier.as_str(), + state = plan_state, + "marketplace plan recorded" + ); + // Every plan transition lands in the audit trail (issue #12). Audit is + // observability, not entitlement state — failure logs but never 5xxs a + // delivery whose transition is already durable. + if let Err(e) = state + .store + .record_api_read( + "billing:marketplace", + &format!("account:{}", purchase.account.id), + action, + &format!("tier:{},state:{plan_state}", tier.as_str()), + ) + .await + { + error!("billing audit write failed: {e:#}"); + } + + (StatusCode::OK, Json(json!({"ok": true, "plan": tier.as_str(), "state": plan_state}))) + .into_response() +} + /// The durable store could not record the delivery: answer 5xx so GitHub /// retries later instead of treating unheld work as delivered (issue #2). fn storage_unavailable(e: anyhow::Error) -> axum::response::Response { @@ -1176,6 +1352,7 @@ mod tests { gardener: coven_github_config::GardenerConfig::default(), api: coven_github_config::ApiConfig::default(), installations: vec![], + billing: Default::default(), }), store: Store::open_in_memory().expect("in-memory store"), notify: std::sync::Arc::new(tokio::sync::Notify::new()), @@ -2818,3 +2995,241 @@ mod metering_route_tests { ); } } + +#[cfg(test)] +mod billing_route_tests { + //! Marketplace plan intake: entitlement recording, plan-derived limits, + //! and the `require_plan` monetization gate. + use super::tests::app_state; + use super::*; + use axum::extract::State; + use hmac::{Hmac, Mac}; + use sha2::Sha256 as HmacSha; + use std::sync::Arc; + + fn signed(event: &str, delivery_id: &str, body: &str) -> HeaderMap { + let mut mac = Hmac::::new_from_slice(b"secret").expect("hmac"); + mac.update(body.as_bytes()); + let sig = format!("sha256={}", hex::encode(mac.finalize().into_bytes())); + let mut headers = HeaderMap::new(); + headers.insert("x-github-event", event.parse().expect("header")); + headers.insert("x-hub-signature-256", sig.parse().expect("header")); + headers.insert("x-github-delivery", delivery_id.parse().expect("header")); + headers + } + + async fn deliver( + state: &AppState, + event: &str, + delivery_id: &str, + body: &str, + ) -> serde_json::Value { + let response = handle_webhook( + State(state.clone()), + signed(event, delivery_id, body), + Bytes::from(body.to_string()), + ) + .await + .into_response(); + assert_eq!(response.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .expect("body"); + serde_json::from_slice(&bytes).expect("json") + } + + fn purchase(action: &str, account_id: u64, plan: &str, on_free_trial: bool) -> String { + serde_json::json!({ + "action": action, + "marketplace_purchase": { + "account": { "id": account_id, "login": "acme", "type": "Organization" }, + "plan": { "name": plan }, + "on_free_trial": on_free_trial + } + }) + .to_string() + } + + fn installation_created(installation: u64, account_id: u64) -> String { + serde_json::json!({ + "action": "created", + "installation": { + "id": installation, + "account": { "id": account_id, "login": "acme" } + } + }) + .to_string() + } + + fn assigned(installation: u64, issue: u64) -> String { + serde_json::json!({ + "action": "assigned", + "issue": { "number": issue, "title": "t", "body": "b" }, + "assignee": { "login": "coven-cody[bot]" }, + "repository": { "name": "demo", "owner": { "login": "OpenCoven" } }, + "installation": { "id": installation } + }) + .to_string() + } + + fn require_plan_state() -> AppState { + let base = app_state(); + let mut config = (*base.config).clone(); + config.billing.require_plan = true; + AppState { + config: Arc::new(config), + ..base + } + } + + #[tokio::test] + async fn purchase_records_the_plan_and_resolves_through_installations() { + let state = app_state(); + let body = purchase("purchased", 42, "Hosted Team", false); + let response = deliver(&state, "marketplace_purchase", "mp-1", &body).await; + assert_eq!(response["plan"], "team"); + assert_eq!(response["state"], "active"); + + deliver( + &state, + "installation", + "in-1", + &installation_created(7, 42), + ) + .await; + let plan = state + .store + .plan_for_installation(7) + .await + .unwrap() + .expect("plan resolved"); + assert_eq!(plan.tier, "team"); + assert!(plan.entitled()); + } + + #[tokio::test] + async fn duplicate_purchase_deliveries_do_not_reapply() { + let state = app_state(); + let body = purchase("purchased", 42, "Hosted Team", false); + deliver(&state, "marketplace_purchase", "mp-1", &body).await; + let dup = deliver(&state, "marketplace_purchase", "mp-1", &body).await; + assert_eq!(dup["duplicate"], true); + } + + #[tokio::test] + async fn require_plan_gates_intake_until_a_purchase_lands() { + let state = require_plan_state(); + deliver( + &state, + "installation", + "in-1", + &installation_created(7, 42), + ) + .await; + + // No plan → the delivery is acknowledged but ignored, and audited. + let refused = deliver(&state, "issues", "dl-1", &assigned(7, 1)).await; + assert_eq!(refused["ignored"], "no_plan"); + assert_eq!( + state.store.delivery_routing("dl-1").await.unwrap().as_deref(), + Some("ignored:no_plan") + ); + + // Trial purchase → the same trigger now produces a task. + deliver( + &state, + "marketplace_purchase", + "mp-1", + &purchase("purchased", 42, "Hosted Starter", true), + ) + .await; + let accepted = deliver(&state, "issues", "dl-2", &assigned(7, 2)).await; + assert!(accepted.get("ignored").is_none()); + assert_eq!(state.store.task_states().await.unwrap().len(), 1); + } + + #[tokio::test] + async fn cancellation_revokes_entitlement() { + let state = require_plan_state(); + deliver( + &state, + "installation", + "in-1", + &installation_created(7, 42), + ) + .await; + deliver( + &state, + "marketplace_purchase", + "mp-1", + &purchase("purchased", 42, "Hosted Team", false), + ) + .await; + deliver( + &state, + "marketplace_purchase", + "mp-2", + &purchase("cancelled", 42, "Hosted Team", false), + ) + .await; + + let refused = deliver(&state, "issues", "dl-1", &assigned(7, 1)).await; + assert_eq!(refused["ignored"], "no_plan"); + } + + #[tokio::test] + async fn pending_changes_do_not_alter_the_plan() { + let state = app_state(); + deliver( + &state, + "marketplace_purchase", + "mp-1", + &purchase("purchased", 42, "Hosted Team", false), + ) + .await; + deliver( + &state, + "marketplace_purchase", + "mp-2", + &purchase("pending_change", 42, "Hosted Starter", false), + ) + .await; + deliver( + &state, + "installation", + "in-1", + &installation_created(7, 42), + ) + .await; + let plan = state.store.plan_for_installation(7).await.unwrap().unwrap(); + assert_eq!(plan.tier, "team"); + } + + #[tokio::test] + async fn uninstall_forgets_the_account_mapping() { + let state = app_state(); + deliver( + &state, + "marketplace_purchase", + "mp-1", + &purchase("purchased", 42, "Hosted Team", false), + ) + .await; + deliver( + &state, + "installation", + "in-1", + &installation_created(7, 42), + ) + .await; + assert!(state.store.plan_for_installation(7).await.unwrap().is_some()); + + let body = serde_json::json!({ + "action": "deleted", + "installation": { "id": 7, "account": { "id": 42, "login": "acme" } } + }) + .to_string(); + deliver(&state, "installation", "in-2", &body).await; + assert!(state.store.plan_for_installation(7).await.unwrap().is_none()); + } +} diff --git a/crates/worker/src/gardener_schedule.rs b/crates/worker/src/gardener_schedule.rs index ae13c7e..8a75120 100644 --- a/crates/worker/src/gardener_schedule.rs +++ b/crates/worker/src/gardener_schedule.rs @@ -251,6 +251,7 @@ mod tests { limits: InstallationLimits::default(), repos: HashMap::new(), }], + billing: Default::default(), }) } diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs index 2b31772..796a3e9 100644 --- a/crates/worker/src/lib.rs +++ b/crates/worker/src/lib.rs @@ -38,7 +38,10 @@ pub async fn run( ) { let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(config.worker.concurrency)); // Per-installation concurrency caps (issue #15), applied at claim time. - let caps = config.concurrency_caps(); + // Explicit TOML caps win; purchased plans supply tier defaults for the + // rest (billing entitlements), re-read each claim so plan changes apply + // without a restart. + let static_caps = config.concurrency_caps(); loop { // Hold capacity BEFORE claiming so a claimed task is never parked @@ -47,6 +50,23 @@ pub async fn run( Ok(permit) => permit, Err(_) => break, }; + let mut caps = static_caps.clone(); + match store.entitled_installation_tiers().await { + Ok(tiers) => { + for (installation, tier) in tiers { + if let std::collections::hash_map::Entry::Vacant(slot) = + caps.entry(installation) + { + let tier: coven_github_config::PlanTier = + tier.parse().unwrap_or(coven_github_config::PlanTier::Unknown); + if let Some(cap) = tier.limits().max_concurrent { + slot.insert(cap); + } + } + } + } + Err(e) => error!("failed to load plan concurrency caps: {e:#}"), + } match store.claim_next(&caps).await { Ok(Some(task)) => { let config = config.clone(); @@ -1915,6 +1935,7 @@ mod disposition_tests { gardener: coven_github_config::GardenerConfig::default(), api: coven_github_config::ApiConfig::default(), installations: vec![], + billing: Default::default(), } } @@ -2045,6 +2066,7 @@ mod process_tests { gardener: coven_github_config::GardenerConfig::default(), api: coven_github_config::ApiConfig::default(), installations: vec![], + billing: Default::default(), } } @@ -2317,6 +2339,7 @@ exit 0 gardener: coven_github_config::GardenerConfig::default(), api: coven_github_config::ApiConfig::default(), installations: vec![], + billing: Default::default(), }; let task = Task { id: "task-pub".to_string(), @@ -2552,6 +2575,7 @@ exit 0 gardener: coven_github_config::GardenerConfig::default(), api: coven_github_config::ApiConfig::default(), installations: vec![], + billing: Default::default(), }; let task = Task { id: "task-stale".to_string(), @@ -2696,6 +2720,7 @@ mod command_and_marker_tests { gardener: coven_github_config::GardenerConfig::default(), api: coven_github_config::ApiConfig::default(), installations: vec![], + billing: Default::default(), } } @@ -3797,6 +3822,7 @@ mod publication_gate_tests { gardener: coven_github_config::GardenerConfig::default(), api: coven_github_config::ApiConfig::default(), installations: vec![], + billing: Default::default(), }; let task = Task { id: "task-gates".to_string(), @@ -4010,6 +4036,7 @@ exit 0 gardener: coven_github_config::GardenerConfig::default(), api: coven_github_config::ApiConfig::default(), installations: vec![], + billing: Default::default(), } } @@ -4167,6 +4194,7 @@ mod cleanup_tests { gardener: coven_github_config::GardenerConfig::default(), api: coven_github_config::ApiConfig::default(), installations: vec![], + billing: Default::default(), }; let task = Task { id: "task-cleanup".to_string(), diff --git a/docs/pricing.md b/docs/pricing.md index a5a020e..b8a38dc 100644 --- a/docs/pricing.md +++ b/docs/pricing.md @@ -58,7 +58,8 @@ the product. | Promise | Mechanism | |---|---| -| Task caps / concurrency | `[installations.limits] max_tasks_per_day` (intake gate, audited as `ignored:quota_exceeded`) and `max_concurrent` (claim-time gate) — #15 | +| Plan purchase → entitlement | `marketplace_purchase` webhook → per-account plan record, resolved per installation; `[billing] require_plan` gates intake for plan-less installations (`ignored:no_plan`) | +| Task caps / concurrency | `[installations.limits] max_tasks_per_day` (intake gate, audited as `ignored:quota_exceeded`) and `max_concurrent` (claim-time gate) — #15; purchased plans supply tier defaults, explicit TOML wins | | Familiar count / routing | `[[installations]] familiars` allow-list + per-repo trigger policy — #7 | | Tenant data boundary | Fail-closed token-scoped task/usage/memory APIs, per-read audit — #3 | | Memory governance | `[memory]` opt-in, approval gates, retention_days, revoke + inspect — #6 |