Skip to content
Sign up

flagon-io/g1t

public
0

Commit

Billing's fifteen minutes: spend alerts first, each step on its own, this month's users read once from an index (billing 0050)

syntaqxcommitted Parenta8cc493Browse files
3 files+100−400/3 viewed
+4−0
1+-- This month's users, read every fifteen minutes for autopay and the limit
2+-- warnings: usage entries from a date on, by kind and time, and the
3+-- workspace in the index so the read never touches the ledger itself.
4+CREATE INDEX IF NOT EXISTS ledger_usage_by_time ON ledger (kind, created_at, workspace);
+50−19
11541154 }
11551155 }
11561156
1157−#[event(scheduled)]
1158−async fn scheduled(event: ScheduledEvent, env: Env, _ctx: ScheduleContext) {
1159− let Ok(billing) = Billing::from_env(&env) else {
1160− return;
1161− };
1162− let keeper = keeper::Keeper::from_env(&env);
1163− if let Err(error) = billing.settle_runs(&keeper).await {
1157+/// The fifteen-minute steps, each on its own: one failing never skips the
1158+/// rest.
1159+async fn tick(billing: &Billing, env: &Env, keeper: &keeper::Keeper) {
1160+ if let Err(error) = billing.settle_runs(keeper).await {
11641161 worker::console_error!("settling runs failed: {error}");
11651162 }
11661163 if let Err(error) = billing.settle_own_runs().await {
11701167 match billing.replay_events().await {
11711168 Ok(done) => worker::console_log!("stripe replay: {done}"),
11721169 Err(error) => worker::console_error!("replaying Stripe events failed: {error}"),
1173− }
1174− if let Err(error) = billing.autopay().await {
1175− worker::console_error!("paying at the limit failed: {error}");
11761170 }
1171+ // This month's users, read once for autopay and the warnings.
1172+ let users = if billing.stripe.is_some() {
1173+ match billing.month_users().await {
1174+ Ok(users) => Some(users),
1175+ Err(error) => {
1176+ worker::console_error!("reading this month's users failed: {error}");
1177+ None
1178+ }
1179+ }
1180+ } else {
1181+ None
1182+ };
1183+ if let Some(users) = &users
1184+ && let Err(error) = billing.autopay(users).await {
1185+ worker::console_error!("paying at the limit failed: {error}");
1186+ }
11771187 // AI credit below a workspace's auto-reload threshold, reloaded (ai.rs).
11781188 match billing.reload_ai_credit().await {
11791189 Ok(0) => {}
11911201 if let Err(error) = billing.invoice_enterprises().await {
11921202 worker::console_error!("invoicing enterprises failed: {error}");
11931203 }
1204+ if let Some(users) = &users
1205+ && let Ok(identity) = env.service("IDENTITY")
1206+ && let Err(error) = billing.warn_limits(&identity, users).await {
1207+ worker::console_error!("warning owners failed: {error}");
1208+ }
1209+}
1210+
1211+/// Every fifteen minutes: runs settled, Stripe replayed, cards charged at
1212+/// the limit, months closed, owners warned. Once a day, at
1213+/// [`keeper::DAILY`], in an invocation of its own: the slower checks
1214+/// against Stripe and Cloudflare, and measuring.
1215+///
1216+/// Each step logs its own failure and the next still runs. The spend
1217+/// breaker and budget alerts go first: a step that runs out of CPU ends
1218+/// the invocation, and must never take the alerts with it.
1219+#[event(scheduled)]
1220+async fn scheduled(event: ScheduledEvent, env: Env, _ctx: ScheduleContext) {
1221+ let Ok(billing) = Billing::from_env(&env) else {
1222+ return;
1223+ };
11941224 // Comped budgets' alerts, and a tripped breaker staff were not told of.
11951225 if let Err(error) = billing.watch_spend().await {
11961226 worker::console_error!("watching g1t's own spend failed: {error}");
11971227 }
1198− if let Ok(identity) = env.service("IDENTITY")
1199− && let Err(error) = billing.warn_limits(&identity).await {
1200− worker::console_error!("warning owners failed: {error}");
1201− }
1228+ let keeper = keeper::Keeper::from_env(&env);
1229+ let daily = event.cron() == keeper::DAILY;
1230+ if !daily {
1231+ tick(&billing, &env, &keeper).await;
1232+ }
12021233 // Once a day: Stripe's endpoint kept listening to billing's events and
12031234 // enabled, and saved cards and plans not read in a while read again.
1204− if event.cron() == keeper::DAILY {
1235+ if daily {
12051236 match billing.keep_endpoint("billing").await {
12061237 Ok(done) => worker::console_log!("stripe endpoint: {done}"),
12071238 Err(error) => worker::console_error!("keeping Stripe's endpoint failed: {error}"),
12131244 }
12141245 // Once a day, and at once if the costs were never checked: check every
12151246 // cost against what Cloudflare billed.
1216− if (event.cron() == keeper::DAILY || billing.never_checked().await.unwrap_or(false))
1247+ if (daily || billing.never_checked().await.unwrap_or(false))
12171248 && let Err(error) = billing.reconcile(&keeper).await {
12181249 worker::console_error!("checking costs against Cloudflare failed: {error}");
12191250 }
12201251 // Once a day: credit from g1t past its expiry stops counting
12211252 // (grants.rs), before the day is reconciled.
1222− if event.cron() == keeper::DAILY {
1253+ if daily {
12231254 match billing.expire_credits().await {
12241255 Ok(closed) => worker::console_log!("credits expired: {closed}"),
12251256 Err(error) => worker::console_error!("expiring credits failed: {error}"),
12281259 // Once a day: what Cloudflare charged, reconciled against what g1t
12291260 // counted and charged; prices whose day has come; margin alerts
12301261 // (margin.rs). After the keeper, so its proposals are in.
1231− if event.cron() == keeper::DAILY {
1262+ if daily {
12321263 match billing.costs_daily(&env, &keeper).await {
12331264 Ok(run) => worker::console_log!("costs: {} lines, {} days, {} proposals, {} alerts", run.lines, run.days, run.proposals, run.alerts),
12341265 Err(error) => worker::console_error!("reconciling costs failed: {error}"),
12371268 // Once a day: what each workspace's private repositories hold, its git
12381269 // operations, Deployments plans from before the g1t plan set to end,
12391270 // and old reservations cleared.
1240− if event.cron() == keeper::DAILY {
1271+ if daily {
12411272 if let Err(error) = billing.measure_packages().await {
12421273 worker::console_error!("measuring package storage failed: {error}");
12431274 }
+46−21
3737 //! Usage counts at what it cost g1t or what it is charged, whichever is
3838 //! more. Test-mode payments are not money, so they do not raise trust.
3939
40+use std::collections::BTreeSet;
41+
4042 use futures_util::future::{try_join, try_join5, try_join_all};
4143 use g1t_contracts::billing::{
4244 BillingAccount, CheckLimitArgs, Limit, LimitArgs, LimitState, NotePendingArgs, PlanKind, SetBudgetArgs, SetSpendLimitArgs, Trust,
714716 Ok(true)
715717 }
716718
719+ /// The workspaces with usage on the ledger this month: autopay's and
720+ /// the limit warnings' candidates, read once a tick for both. Served by
721+ /// `ledger_usage_by_time` (migration 0050), not a scan of the ledger.
722+ pub(crate) async fn month_users(&self) -> Result<BTreeSet<String>> {
723+ #[derive(Deserialize)]
724+ struct User {
725+ workspace: String,
726+ }
727+ let month_start = format!("{}-01", &rfc3339(now_ms())[..7]);
728+ Ok(self
729+ .db
730+ .prepare("SELECT DISTINCT workspace FROM ledger WHERE kind = 'usage' AND created_at >= ?")
731+ .bind(&[month_start.into()])?
732+ .all()
733+ .await?
734+ .results::<User>()?
735+ .into_iter()
736+ .map(|user| user.workspace)
737+ .collect())
738+ }
739+
717740 /// Charges the saved card of each workspace nearing its limit, for what
718741 /// it owes, so that a workspace that pays never has its work stopped.
719742 /// Only with live payments: test-mode payments are not money and lower
720743 /// nothing. Not for a workspace's own spend limit, which means stop, nor
721744 /// for enterprises, which are invoiced. A charge at the limit always
722745 /// goes through, whatever the minimum charge.
723− pub(crate) async fn autopay(&self) -> Result<()> {
746+ ///
747+ /// `users` are the workspaces with usage this month ([`Self::month_users`]).
748+ pub(crate) async fn autopay(&self, users: &BTreeSet<String>) -> Result<()> {
724749 if !self.stripe.as_ref().is_some_and(crate::stripe::Stripe::live) {
725750 return Ok(());
726751 }
728753 struct Candidate {
729754 workspace: String,
730755 }
731− let month_start = format!("{}-01", &rfc3339(now_ms())[..7]);
732756 // With a card, and not already declined: a declined card waits for
733757 // the owners, rather than being tried again every few minutes.
734758 let candidates = self
735759 .db
736760 .prepare(
737− "SELECT DISTINCT ledger.workspace AS workspace
738− FROM ledger JOIN accounts ON accounts.workspace = ledger.workspace
739− LEFT JOIN limits ON limits.workspace = ledger.workspace
740− WHERE ledger.kind = 'usage' AND ledger.created_at >= ? AND accounts.customer_id IS NOT NULL
741− AND limits.autopay_failed_at IS NULL",
761+ "SELECT accounts.workspace AS workspace
762+ FROM accounts LEFT JOIN limits ON limits.workspace = accounts.workspace
763+ WHERE accounts.customer_id IS NOT NULL AND limits.autopay_failed_at IS NULL",
742764 )
743− .bind(&[month_start.as_str().into()])?
744765 .all()
745766 .await?
746− .results::<Candidate>()?;
767+ .results::<Candidate>()?
768+ .into_iter()
769+ .filter(|candidate| users.contains(&candidate.workspace));
747770 for candidate in candidates {
748771 let limit = self.limit_of(&candidate.workspace).await?;
749772 // Near g1t's ceiling on what is unpaid; the spend limit is the
850873 /// plan's included usage, its spend limit and g1t's ceiling, once each a
851874 /// month; when its card was declined; and when a spend spike paused it.
852875 /// The same alerts show in the app (`entitlements`).
853− pub(crate) async fn warn_limits(&self, identity: &worker::Fetcher) -> Result<()> {
876+ ///
877+ /// `users` are the workspaces with usage this month ([`Self::month_users`]).
878+ pub(crate) async fn warn_limits(&self, identity: &worker::Fetcher, users: &BTreeSet<String>) -> Result<()> {
854879 if self.stripe.is_none() {
855880 return Ok(());
856881 }
860885 struct Candidate {
861886 workspace: String,
862887 }
863− let candidates = self
864− .db
865− .prepare(
866− "SELECT DISTINCT workspace FROM ledger WHERE kind = 'usage' AND created_at >= ?1
867− UNION SELECT workspace FROM limits WHERE autopay_failed_at IS NOT NULL",
868− )
869− .bind(&[format!("{month}-01").into()])?
870− .all()
871− .await?
872− .results::<Candidate>()?;
888+ let mut candidates = users.clone();
889+ candidates.extend(
890+ self.db
891+ .prepare("SELECT workspace FROM limits WHERE autopay_failed_at IS NOT NULL")
892+ .all()
893+ .await?
894+ .results::<Candidate>()?
895+ .into_iter()
896+ .map(|candidate| candidate.workspace),
897+ );
873898 #[derive(Deserialize)]
874899 struct Told {
875900 autopay_failed_at: Option<String>,
880905 meter: String,
881906 level: Option<i64>,
882907 }
883− for Candidate { workspace } in candidates {
908+ for workspace in candidates {
884909 let told = self
885910 .db
886911 .prepare("SELECT autopay_failed_at, declined_told_at FROM limits WHERE workspace = ?")