1import { SUBSCRIPTION_SCAN_BYTES } from "./accessHistory";2import { REVENUE_COUNTERS } from "./revenueCounters";3import { query, type QueryCtx } from "../_generated/server";4import { ConvexError, v, type Infer } from "convex/values";5import type { Doc, Id } from "../_generated/dataModel";6 7import {8 canReadRawStoreEvidence,9 resolveOperatorProject,10 resolveProjectByApiKeyFromDb,11 resolveProjectByIdForCurrentUserFromDb,12} from "../projects/helpers";13import {14 isValidSubscriptionUserId,15 MAX_SUBSCRIPTION_USER_ID_LENGTH,16 SUBSCRIPTION_ENVIRONMENT_SCOPES,17 type SubscriptionEnvironmentScope,18} from "./limits";19import {20 loadStatsPeriods,21 statsAreCurrent,22 statsContributionFor,23} from "./stats";24import { subscriptionEnvironment } from "./analyticsFacts";25import { ROLLUP_DEFINITIONS_VERSION } from "./revenueMetrics";26import { entitlementActive, isMisrecordedApplePass } from "./stateMachine";27import { selectMostRecentlyUpdatedSubscription } from "./selectLatest";28import {29 DEFAULT_REPORTING_CURRENCY,30 normalizeReportingCurrencyOrDefault,31} from "../utils/currency";32import { storeOf } from "../commerce/internal";33import { purchaseStoreValidator } from "../schema";34import {35 purchaseIssueCauseValidator,36 receiptEnvironmentValidator,37 subscriptionStateValidator,38} from "../utils/validation";39 40const subscriptionFields = {41 id: v.id("subscriptions"),42 productId: v.string(),43 platform: v.union(v.literal("IOS"), v.literal("Android")),44 state: subscriptionStateValidator,45 expiresAt: v.optional(v.number()),46 renewsAt: v.optional(v.number()),47 willRenew: v.optional(v.boolean()),48 cancellationReason: v.optional(v.string()),49 currency: v.optional(v.string()),50 priceAmountMicros: v.optional(v.number()),51 startedAt: v.number(),52 updatedAt: v.number(),53 // Legacy server responses preserve this required MAUI field; v2 omits it.54 purchaseToken: v.string(),55 originalTransactionId: v.optional(v.string()),56 userId: v.optional(v.string()),57};58const subscriptionShape = v.object(subscriptionFields);59const subscriptionV2Shape = v.object({60 id: v.id("subscriptions"),61 productId: v.string(),62 platform: v.union(v.literal("IOS"), v.literal("Android")),63 store: v.optional(purchaseStoreValidator),64 state: subscriptionStateValidator,65 expiresAt: v.optional(v.number()),66 renewsAt: v.optional(v.number()),67 willRenew: v.optional(v.boolean()),68 cancellationReason: v.optional(v.string()),69 currency: v.optional(v.string()),70 priceAmountMicros: v.optional(v.number()),71 startedAt: v.number(),72 updatedAt: v.number(),73 userId: v.optional(v.string()),74});75const subscriptionEvaluationRowShape = v.object({76 ...subscriptionFields,77 createdAt: v.number(),78});79type SubscriptionRow = Infer<typeof subscriptionShape>;80type SubscriptionV2Row = Infer<typeof subscriptionV2Shape>;81type SubscriptionEvaluationRow = Infer<typeof subscriptionEvaluationRowShape>;82export const MAX_USER_SUBSCRIPTION_ROWS = 200;83 84const environmentScopeValidator = v.union(85 ...SUBSCRIPTION_ENVIRONMENT_SCOPES.map((scope) => v.literal(scope)),86);87 88function subscriptionInEnvironment(89 sub: Doc<"subscriptions">,90 scope: SubscriptionEnvironmentScope | undefined,91): boolean {92 if (scope === undefined) return true;93 const environment = sub.environment ?? sub.lastEventSource?.environment;94 if (environment === undefined) return false;95 return (96 scope === "all" ||97 (scope === "production"98 ? environment === "Production"99 : environment !== "Production")100 );101}102 103function isActive(sub: Doc<"subscriptions">, now: number): boolean {104 return entitlementActive(sub, now);105}106 107function isEntitledState(sub: Doc<"subscriptions">): boolean {108 return sub.state === "Active" || sub.state === "InGracePeriod";109}110 111export function assertUserSubscriptionRowLimit(112 rows: Array<Doc<"subscriptions">>,113): void {114 if (rows.length <= MAX_USER_SUBSCRIPTION_ROWS) return;115 throw new ConvexError({116 code: "ENTITLEMENT_SNAPSHOT_TOO_LARGE",117 message:118 "This user has more than 200 subscription rows. Contact IAPKit support before retrying.",119 });120}121 122async function userSubscriptionRows(123 ctx: QueryCtx,124 projectId: Id<"projects">,125 userId: string,126): Promise<Array<Doc<"subscriptions">>> {127 const rows = await ctx.db128 .query("subscriptions")129 .withIndex("by_project_and_user_and_updated", (q) =>130 q.eq("projectId", projectId).eq("userId", userId),131 )132 .order("desc")133 .take(MAX_USER_SUBSCRIPTION_ROWS + 1);134 assertUserSubscriptionRowLimit(rows);135 return rows;136}137 138export function shapeSubscriptionEvaluationSnapshot(139 rows: Array<Doc<"subscriptions">>,140): {141 candidates: SubscriptionEvaluationRow[];142 fallback: SubscriptionEvaluationRow | null;143} {144 const fallback = selectMostRecentlyUpdatedSubscription(rows);145 return {146 // The snapshot omits lastEventId, so the legacy-pass guard runs here.147 candidates: rows148 .filter((row) => isEntitledState(row) && !isMisrecordedApplePass(row))149 .map(shapeSubscriptionEvaluationRow),150 fallback: fallback ? shapeSubscriptionEvaluationRow(fallback) : null,151 };152}153 154function shapeSubscriptionEvaluationRow(155 sub: Doc<"subscriptions">,156): SubscriptionEvaluationRow {157 return {158 ...shapeSubscriptionRow(sub),159 createdAt: sub._creationTime,160 };161}162 163export function shapeSubscriptionRow(164 sub: Doc<"subscriptions">,165): SubscriptionRow {166 const row: SubscriptionRow = {167 id: sub._id,168 productId: sub.productId,169 platform: sub.platform,170 state: sub.state,171 expiresAt: sub.expiresAt,172 renewsAt: sub.renewsAt,173 willRenew: sub.willRenew,174 cancellationReason: sub.cancellationReason,175 currency: sub.currency,176 priceAmountMicros: sub.priceAmountMicros,177 startedAt: sub.startedAt,178 updatedAt: sub.updatedAt,179 purchaseToken: sub.purchaseToken,180 userId: sub.userId,181 };182 183 if (sub.platform === "IOS") {184 return {185 ...row,186 originalTransactionId: sub.purchaseToken,187 };188 }189 190 return row;191}192 193export function shapeSubscriptionV2Row(194 sub: Doc<"subscriptions">,195): SubscriptionV2Row {196 return {197 id: sub._id,198 productId: sub.productId,199 platform: sub.platform,200 ...(sub.store ? { store: sub.store } : {}),201 state: sub.state,202 expiresAt: sub.expiresAt,203 renewsAt: sub.renewsAt,204 willRenew: sub.willRenew,205 cancellationReason: sub.cancellationReason,206 currency: sub.currency,207 priceAmountMicros: sub.priceAmountMicros,208 startedAt: sub.startedAt,209 updatedAt: sub.updatedAt,210 userId: sub.userId,211 };212}213 214async function projectByApiKey(215 ctx: QueryCtx,216 apiKey: string | undefined,217 requiredAccess: "client" | "admin" = "client",218): Promise<Doc<"projects"> | null> {219 if (!apiKey) return null;220 const resolved = await resolveProjectByApiKeyFromDb(221 ctx,222 apiKey,223 requiredAccess,224 );225 return resolved?.project ?? null;226}227 228async function requireAdminProjectByApiKey(229 ctx: QueryCtx,230 apiKey: string,231): Promise<Doc<"projects">> {232 const project = await projectByApiKey(ctx, apiKey, "admin");233 if (!project) {234 throw new ConvexError({235 code: "INVALID_API_KEY",236 message: "API key is invalid or inactive",237 });238 }239 return project;240}241 242async function projectByIdForCurrentUser(243 ctx: QueryCtx,244 projectId: Id<"projects"> | undefined,245): Promise<Doc<"projects"> | null> {246 if (!projectId) return null;247 const resolved = await resolveProjectByIdForCurrentUserFromDb(ctx, projectId);248 return resolved?.project ?? null;249}250 251async function projectForReadArgs(252 ctx: QueryCtx,253 args: {254 apiKey?: string;255 projectId?: Id<"projects">;256 },257 apiKeyAccess: "client" | "admin" = "client",258): Promise<Doc<"projects"> | null> {259 if (args.projectId) {260 return projectByIdForCurrentUser(ctx, args.projectId);261 }262 263 if (args.apiKey !== undefined) {264 return projectByApiKey(ctx, args.apiKey, apiKeyAccess);265 }266 267 throw new Error("apiKey or projectId is required.");268}269 270// A dashboard member reads the list without store tokens; secret API keys271// keep them for reconciliation.272async function listReadAccess(273 ctx: QueryCtx,274 args: { apiKey?: string; projectId?: Id<"projects"> },275): Promise<{ project: Doc<"projects">; storeTokens: boolean } | null> {276 if (args.projectId) {277 const access = await resolveProjectByIdForCurrentUserFromDb(278 ctx,279 args.projectId,280 );281 return access282 ? {283 project: access.project,284 storeTokens: canReadRawStoreEvidence(access.role),285 }286 : null;287 }288 const project = await projectForReadArgs(ctx, args, "admin");289 return project ? { project, storeTokens: true } : null;290}291 292// purchaseToken stays present but empty: the field is required (see above).293function withoutStoreToken(row: SubscriptionRow): SubscriptionRow {294 const { originalTransactionId: _originalTransactionId, ...rest } = row;295 return { ...rest, purchaseToken: "" };296}297 298export interface MrrCurrencyEntry {299 currency: string;300 mrrMicros: number;301}302 303/**304 * Selects the headline MRR entry for a project's reporting currency.305 *306 * `reportingCurrency` is normalized via `normalizeReportingCurrencyOrDefault`,307 * so unknown or invalid input falls back to `DEFAULT_REPORTING_CURRENCY`.308 * If no entry matches the normalized currency, returned `mrrMicros` is `0`.309 * Returned `excludedMrrByCurrency` contains every non-reporting-currency entry.310 *311 * @param entries Per-currency MRR rows already summed for the project.312 * @param reportingCurrency Project-configured reporting currency, raw or normalized.313 * @returns The normalized `currency`, selected `mrrMicros`, and excluded rows.314 */315export function selectReportingMrr(316 entries: MrrCurrencyEntry[],317 reportingCurrency: string | null | undefined,318): {319 currency: string;320 mrrMicros: number;321 excludedMrrByCurrency: MrrCurrencyEntry[];322} {323 const normalizedReportingCurrency =324 normalizeReportingCurrencyOrDefault(reportingCurrency);325 const reportingEntry = entries.find(326 (entry) => entry.currency === normalizedReportingCurrency,327 );328 329 return {330 currency: normalizedReportingCurrency,331 mrrMicros: reportingEntry?.mrrMicros ?? 0,332 excludedMrrByCurrency: entries.filter(333 (entry) => entry.currency !== normalizedReportingCurrency,334 ),335 };336}337 338// Clock-free snapshot for the Fly HTTP layer, so Convex can cache it and339// invalidate it on row changes. Of the historical rows only the latest is kept,340// as the status fallback; Fly drops candidates whose expiresAt has passed using341// its own clock.342export const subscriptionEvaluationSnapshot = query({343 args: {344 apiKey: v.string(),345 userId: v.string(),346 },347 returns: v.object({348 projectId: v.union(v.id("projects"), v.null()),349 candidates: v.array(subscriptionEvaluationRowShape),350 fallback: v.union(subscriptionEvaluationRowShape, v.null()),351 }),352 handler: async (ctx, args) => {353 const project = await requireAdminProjectByApiKey(ctx, args.apiKey);354 355 const rows = await userSubscriptionRows(ctx, project._id, args.userId);356 return {357 projectId: project._id,358 ...shapeSubscriptionEvaluationSnapshot(rows),359 };360 },361});362 363// `/v2` account reads are server-to-server. Convex repeats the admin check and364// shapes the response without store credentials before it crosses the HTTP365// boundary.366export const subscriptionStatusV2 = query({367 args: {368 apiKey: v.string(),369 userId: v.string(),370 now: v.number(),371 environment: v.optional(environmentScopeValidator),372 },373 returns: v.object({374 active: v.boolean(),375 subscription: v.union(subscriptionV2Shape, v.null()),376 environmentScope: v.optional(environmentScopeValidator),377 }),378 handler: async (ctx, args) => {379 const project = await requireAdminProjectByApiKey(ctx, args.apiKey);380 381 const subs = (382 await userSubscriptionRows(ctx, project._id, args.userId)383 ).filter((sub) => subscriptionInEnvironment(sub, args.environment));384 const activeSubs = subs.filter((candidate) =>385 isActive(candidate, args.now),386 );387 const selected = selectMostRecentlyUpdatedSubscription(388 activeSubs.length > 0 ? activeSubs : subs,389 );390 391 return {392 active: activeSubs.length > 0,393 subscription: selected ? shapeSubscriptionV2Row(selected) : null,394 ...(args.environment === undefined395 ? {}396 : { environmentScope: args.environment }),397 };398 },399});400 401export const entitlementsV2 = query({402 args: {403 apiKey: v.string(),404 userId: v.string(),405 now: v.number(),406 environment: v.optional(environmentScopeValidator),407 },408 returns: v.object({409 userId: v.string(),410 productIds: v.array(v.string()),411 subscriptions: v.array(subscriptionV2Shape),412 environmentScope: v.optional(environmentScopeValidator),413 }),414 handler: async (ctx, args) => {415 const project = await requireAdminProjectByApiKey(ctx, args.apiKey);416 417 const all = await userSubscriptionRows(ctx, project._id, args.userId);418 const active = all.filter(419 (sub) =>420 subscriptionInEnvironment(sub, args.environment) &&421 isActive(sub, args.now),422 );423 return {424 userId: args.userId,425 productIds: Array.from(new Set(active.map((sub) => sub.productId))),426 subscriptions: active.map(shapeSubscriptionV2Row),427 ...(args.environment === undefined428 ? {}429 : { environmentScope: args.environment }),430 };431 },432});433 434// Authoritative server-credential gate with no side effect and no data. The435// commerce bindPurchase handler calls this before it parses store evidence, so436// an unknown or under-scoped key is rejected (UNAUTHORIZED / FORBIDDEN) before437// it can learn which stores bind or which evidence a store requires.438export const assertServerAccess = query({439 args: { apiKey: v.string() },440 returns: v.object({ ok: v.boolean() }),441 handler: async (ctx, args) => {442 await requireAdminProjectByApiKey(ctx, args.apiKey);443 return { ok: true };444 },445});446 447export const userErasureStatusV2 = query({448 args: {449 apiKey: v.string(),450 // The id arrives from a URL path; a malformed one must read as "no such451 // job" (404), not fail argument validation into a 500.452 jobId: v.string(),453 },454 returns: v.union(455 v.null(),456 v.object({457 jobId: v.id("subscriptionUserErasureJobs"),458 status: v.union(459 v.literal("queued"),460 v.literal("running"),461 v.literal("completed"),462 ),463 subscriptionsErased: v.number(),464 commerceEventsErased: v.number(),465 createdAt: v.number(),466 updatedAt: v.number(),467 completedAt: v.optional(v.number()),468 }),469 ),470 handler: async (ctx, args) => {471 const project = await requireAdminProjectByApiKey(ctx, args.apiKey);472 const jobId = ctx.db.normalizeId("subscriptionUserErasureJobs", args.jobId);473 if (!jobId) return null;474 const job = await ctx.db.get(jobId);475 if (!job || job.projectId !== project._id) return null;476 return {477 jobId: job._id,478 status: job.status,479 subscriptionsErased: job.subscriptionsErased,480 commerceEventsErased: job.commerceEventsErased,481 createdAt: job.createdAt,482 updatedAt: job.updatedAt,483 completedAt: job.completedAt,484 };485 },486});487 488// Match onesub's `/onesub/status?userId=` — returns the most-recently-489// updated active subscription when the user is entitled, otherwise the490// most-recently-updated subscription overall, plus one `active` boolean491// for simple gating.492export const subscriptionStatus = query({493 args: { apiKey: v.string(), userId: v.string() },494 returns: v.object({495 active: v.boolean(),496 subscription: v.union(subscriptionShape, v.null()),497 }),498 handler: async (ctx, args) => {499 const project = await requireAdminProjectByApiKey(ctx, args.apiKey);500 501 const subs = await userSubscriptionRows(ctx, project._id, args.userId);502 503 const now = Date.now();504 const activeSubs = subs.filter((candidate) => isActive(candidate, now));505 const sub = selectMostRecentlyUpdatedSubscription(506 activeSubs.length > 0 ? activeSubs : subs,507 );508 if (!sub) return { active: false, subscription: null };509 510 return {511 active: activeSubs.length > 0,512 subscription: shapeSubscriptionRow(sub),513 };514 },515});516 517// Match onesub's entitlement evaluation — every productId the user518// currently has rights to. Aggregates across all subscription rows so519// a user with multiple offers (resub, family share, cross-grade) sees520// the union.521export const entitlements = query({522 args: { apiKey: v.string(), userId: v.string() },523 returns: v.object({524 userId: v.string(),525 productIds: v.array(v.string()),526 subscriptions: v.array(subscriptionShape),527 }),528 handler: async (ctx, args) => {529 const project = await requireAdminProjectByApiKey(ctx, args.apiKey);530 531 const all = await userSubscriptionRows(ctx, project._id, args.userId);532 533 const now = Date.now();534 const active = all.filter((sub) => isActive(sub, now));535 return {536 userId: args.userId,537 productIds: Array.from(new Set(active.map((sub) => sub.productId))),538 subscriptions: active.map(shapeSubscriptionRow),539 };540 },541});542 543// Filtered list for the dashboard's subscriptions page. Mirrors544// onesub's `SubscriptionStore.listFiltered` API.545export const listSubscriptions = query({546 args: {547 apiKey: v.optional(v.string()),548 projectId: v.optional(v.id("projects")),549 state: v.optional(subscriptionStateValidator),550 productId: v.optional(v.string()),551 userId: v.optional(v.string()),552 limit: v.optional(v.number()),553 },554 returns: v.object({555 items: v.array(subscriptionShape),556 total: v.number(),557 }),558 handler: async (ctx, args) => {559 const access = await listReadAccess(ctx, args);560 if (!access) return { items: [], total: 0 };561 const { project } = access;562 const shape = access.storeTokens563 ? shapeSubscriptionRow564 : (sub: Doc<"subscriptions">) =>565 withoutStoreToken(shapeSubscriptionRow(sub));566 567 const limit = Math.min(Math.max(args.limit ?? 50, 1), 200);568 569 // A user has few subscriptions, so load them all and filter by state and570 // productId in memory.571 if (args.userId) {572 const userRows = await ctx.db573 .query("subscriptions")574 .withIndex("by_project_and_user", (q) =>575 q.eq("projectId", project._id).eq("userId", args.userId),576 )577 .order("desc")578 .collect();579 const filtered = userRows.filter((sub) => {580 if (args.state && sub.state !== args.state) return false;581 if (args.productId && sub.productId !== args.productId) return false;582 return true;583 });584 return {585 items: filtered.slice(0, limit).map(shape),586 total: filtered.length,587 };588 }589 590 // Every filter combination has an index (state + productId is a composite),591 // so take() never cuts off rows a post-filter would need.592 let rows: Array<Doc<"subscriptions">>;593 if (args.state && args.productId) {594 rows = await ctx.db595 .query("subscriptions")596 .withIndex("by_project_and_state_and_product", (q) =>597 q598 .eq("projectId", project._id)599 .eq("state", args.state!)600 .eq("productId", args.productId!),601 )602 .order("desc")603 .take(limit);604 } else if (args.state) {605 rows = await ctx.db606 .query("subscriptions")607 .withIndex("by_project_and_state", (q) =>608 q.eq("projectId", project._id).eq("state", args.state!),609 )610 .order("desc")611 .take(limit);612 } else if (args.productId) {613 rows = await ctx.db614 .query("subscriptions")615 .withIndex("by_project_and_product", (q) =>616 q.eq("projectId", project._id).eq("productId", args.productId!),617 )618 .order("desc")619 .take(limit);620 } else {621 rows = await ctx.db622 .query("subscriptions")623 .withIndex("by_project_and_updated", (q) =>624 q.eq("projectId", project._id),625 )626 .order("desc")627 .take(limit);628 }629 630 // `total` counts the rows loaded, not every match; a true count would need631 // an unbounded scan.632 return {633 items: rows.slice(0, limit).map(shape),634 total: rows.length,635 };636 },637});638 639// Admin reads fail on an unknown key rather than answer with an empty project.640async function projectForAdminRead(641 ctx: QueryCtx,642 args: { apiKey?: string; projectId?: Id<"projects"> },643): Promise<Doc<"projects"> | null> {644 if (!args.projectId && args.apiKey !== undefined) {645 return requireAdminProjectByApiKey(ctx, args.apiKey);646 }647 return projectForReadArgs(ctx, args, "admin");648}649 650// Current caches avoid a subscription scan. Legacy or expired caches use a651// bounded fallback; an incomplete result is explicit, never a current total.652export const metricsSummary = query({653 args: {654 apiKey: v.optional(v.string()),655 projectId: v.optional(v.id("projects")),656 refreshAt: v.optional(v.number()),657 },658 returns: v.object({659 environment: v.literal("Production"),660 stats: v.object({661 state: v.union(662 v.literal("current"),663 v.literal("rebuilding"),664 v.literal("incomplete"),665 ),666 asOf: v.number(),667 validUntil: v.union(v.number(), v.null()),668 }),669 recentCountsTruncated: v.boolean(),670 activeSubs: v.number(),671 inGracePeriod: v.number(),672 inBillingRetry: v.number(),673 refunded30d: v.number(),674 canceled30d: v.number(),675 // Monthly MRR in the reporting currency only; the name predates676 // per-currency MRR and is kept for dashboard and MCP consumers.677 mrrMicros: v.number(),678 currency: v.optional(v.string()),679 reportingCurrency: v.string(),680 // Per-currency MRR, each normalized to monthly by the product's681 // billingPeriod.682 mrrByCurrency: v.array(683 v.object({ currency: v.string(), mrrMicros: v.number() }),684 ),685 excludedMrrByCurrency: v.array(686 v.object({ currency: v.string(), mrrMicros: v.number() }),687 ),688 }),689 handler: async (ctx, args) => {690 const project = await projectForAdminRead(ctx, args);691 if (!project) {692 return {693 environment: "Production" as const,694 stats: {695 state: "incomplete" as const,696 asOf: Date.now(),697 validUntil: null,698 },699 recentCountsTruncated: false,700 activeSubs: 0,701 inGracePeriod: 0,702 inBillingRetry: 0,703 refunded30d: 0,704 canceled30d: 0,705 mrrMicros: 0,706 currency: DEFAULT_REPORTING_CURRENCY,707 reportingCurrency: DEFAULT_REPORTING_CURRENCY,708 mrrByCurrency: [],709 excludedMrrByCurrency: [],710 };711 }712 const now = Date.now();713 const cutoff = now - 30 * 24 * 60 * 60 * 1000;714 715 // Live state counters + MRR — read out of the incrementally716 // maintained `subscriptionStats` table.717 const statsRows = await ctx.db718 .query("subscriptionStats")719 .withIndex("by_project", (q) => q.eq("projectId", project._id))720 .collect();721 722 const status = await ctx.db723 .query("revenueMetricsRunStatus")724 .withIndex("by_project", (q) => q.eq("projectId", project._id))725 .unique();726 const cacheCurrent = statsRows.length > 0 && statsAreCurrent(status, now);727 // Reactive pagination retains a journal, so its indexed range stays fixed.728 const result = await ctx.db729 .query("subscriptions")730 .withIndex("by_project_and_updated", (q) =>731 q.eq("projectId", project._id),732 )733 .order("desc")734 .paginate({735 cursor: null,736 numItems: 10_000,737 maximumRowsRead: 10_000,738 maximumBytesRead: SUBSCRIPTION_SCAN_BYTES,739 });740 const pageComplete = result.isDone && result.pageStatus !== "SplitRequired";741 const recentlyChanged = result.page.filter(742 (sub) => sub.updatedAt >= cutoff,743 );744 const recentCountsTruncated =745 result.pageStatus === "SplitRequired" ||746 (!result.isDone && (result.page.at(-1)?.updatedAt ?? now) >= cutoff);747 let validUntil: number | null = cacheCurrent748 ? (status?.statsValidUntil ?? null)749 : null;750 let statsState: "current" | "rebuilding" | "incomplete" = "current";751 let activeSubs = 0;752 let inGracePeriod = 0;753 let inBillingRetry = 0;754 const mrrAccumulators = new Map<string, number>();755 756 if (cacheCurrent) {757 for (const row of statsRows) {758 activeSubs += row.activeSubs;759 inGracePeriod += row.inGracePeriod;760 inBillingRetry += row.inBillingRetry;761 if (row.currency && row.mrrMicros > 0) {762 mrrAccumulators.set(763 row.currency,764 (mrrAccumulators.get(row.currency) ?? 0) + row.mrrMicros,765 );766 }767 }768 } else {769 const periods = await loadStatsPeriods(ctx, project._id, result.page);770 if (!pageComplete || periods === null) {771 const rebuilding = await ctx.db772 .query("subscriptionStatsRebuilds")773 .withIndex("by_project", (q) => q.eq("projectId", project._id))774 .unique();775 statsState = rebuilding ? "rebuilding" : "incomplete";776 }777 for (const sub of result.page) {778 const contribution = statsContributionFor(779 sub,780 periods?.get(`${sub.platform}:${sub.productId}`),781 now,782 );783 if (contribution?.expiresAt !== undefined)784 validUntil =785 validUntil === null786 ? contribution.expiresAt787 : Math.min(validUntil, contribution.expiresAt);788 if (contribution?.state === "Active") {789 activeSubs += 1;790 if (contribution.mrrMicros > 0) {791 mrrAccumulators.set(792 contribution.currency,793 (mrrAccumulators.get(contribution.currency) ?? 0) +794 contribution.mrrMicros,795 );796 }797 } else if (contribution?.state === "InGracePeriod") {798 inGracePeriod += 1;799 } else if (contribution?.state === "InBillingRetry") {800 inBillingRetry += 1;801 }802 }803 }804 805 let refunded30d = 0;806 let canceled30d = 0;807 const CANCELED_STATES = new Set([808 "Active",809 "InGracePeriod",810 "InBillingRetry",811 "Expired",812 ]);813 for (const sub of recentlyChanged) {814 if (subscriptionEnvironment(sub) !== "Production") continue;815 if (sub.state === "Refunded") {816 refunded30d += 1;817 }818 if (819 sub.willRenew === false &&820 sub.cancellationReason === "UserCanceled" &&821 CANCELED_STATES.has(sub.state)822 ) {823 canceled30d += 1;824 }825 }826 827 // Sorted for stable rendering. Only the reporting currency feeds the828 // headline; the others stay in `excludedMrrByCurrency` instead of being829 // summed.830 const sorted = Array.from(mrrAccumulators.entries()).sort(831 ([a, av], [b, bv]) => (bv !== av ? bv - av : a.localeCompare(b)),832 );833 const mrrByCurrency = sorted.map(([currency, mrrMicros]) => ({834 currency,835 mrrMicros,836 }));837 const reportingMrr = selectReportingMrr(838 mrrByCurrency,839 project.reportingCurrency,840 );841 842 return {843 environment: "Production" as const,844 stats: {845 state: statsState,846 asOf: cacheCurrent ? status!.statsUpdatedAt! : now,847 validUntil,848 },849 recentCountsTruncated,850 activeSubs,851 inGracePeriod,852 inBillingRetry,853 refunded30d,854 canceled30d,855 mrrMicros: reportingMrr.mrrMicros,856 currency: reportingMrr.currency,857 reportingCurrency: reportingMrr.currency,858 mrrByCurrency,859 excludedMrrByCurrency: reportingMrr.excludedMrrByCurrency,860 };861 },862});863 864// Daily metrics for the Analytics dashboard from the `revenueMetricsDaily`865// rollups, so a render never scans `webhookEvents`. `fromDay` and `toDay` are866// inclusive UTC YYYY-MM-DD, compared as strings on the index. Returns one entry867// per (day, currency, productId, platform) of one environment, production by868// default; rows in different currencies need an FX rate to add. Definitions869// are in `revenueMetrics.ts`; the optional counters are absent on rows written870// before `definitionsFromDay`, and the trial split before `trialSplitFromDay`.871const platformValidator = v.union(v.literal("IOS"), v.literal("Android"));872 873// not_collected: no store notification has reached the environment.874// processing: notifications exist but the window is not rolled up yet.875// zero: the window was rolled up and nothing happened.876// ready: the window has rows.877const analyticsDataStateValidator = v.union(878 v.literal("not_collected"),879 v.literal("processing"),880 v.literal("zero"),881 v.literal("ready"),882);883 884const revenueCohortShape = v.object({885 month: v.string(),886 currency: v.string(),887 members: v.number(),888 revenueMicros: v.number(),889 refundedMicros: v.number(),890 // Gross revenue per member observed through `observedThrough`; never projected.891 realizedLtvMicros: v.number(),892 retained: v.array(v.number()),893 retentionObserved: v.optional(v.array(v.number())),894 trialStarts: v.number(),895 trialConversions: v.number(),896 trialsPending: v.number(),897 trialsUnobserved: v.optional(v.number()),898 observedThrough: v.string(),899});900 901const revenueMetricsResponse = v.object({902 days: v.array(903 v.object({904 day: v.string(),905 currency: v.string(),906 productId: v.string(),907 platform: platformValidator,908 activeSubs: v.number(),909 activeHistoryComplete: v.optional(v.boolean()),910 activeTrials: v.optional(v.number()),911 newSubs: v.number(),912 trialStarts: v.optional(v.number()),913 trialConversions: v.optional(v.number()),914 directPaidStarts: v.optional(v.number()),915 renewals: v.number(),916 cancellations: v.number(),917 expirations: v.optional(v.number()),918 refunds: v.number(),919 revenueMicros: v.number(),920 refundedMicros: v.optional(v.number()),921 }),922 ),923 // Every value in the window, for the dashboard's filter dropdowns.924 currencies: v.array(v.string()),925 productIds: v.array(v.string()),926 platforms: v.array(platformValidator),927 // The scan hit REVENUE_SCAN_CAP; the dashboard shows a banner.928 truncated: v.boolean(),929 environment: receiptEnvironmentValidator,930 dataState: analyticsDataStateValidator,931 lastRunAt: v.optional(v.number()),932 // Days from here on follow the current definitions; earlier rows mix933 // sandbox into production and have no trial split.934 definitionsFromDay: v.optional(v.string()),935 // Days from here on separate free trials, trial conversions, unpaid trial936 // expirations, refund reversals and Apple resubscribes. Earlier days omit937 // the counters that need them; unset when every day has them.938 trialSplitFromDay: v.optional(v.string()),939 // Notifications that arrived after their day's rollup and are waiting for940 // a catch-up recompute.941 pendingLateEvents: v.number(),942 oldestPendingLateDay: v.optional(v.string()),943 cohorts: v.optional(v.array(revenueCohortShape)),944 cohortsComputedAt: v.optional(v.number()),945});946 947// `days` returns every row, and a Convex array holds at most 8,192 items: about948// 92 days of 30 SKUs × 3 currencies on one platform. Hitting it sets949// `truncated`.950const REVENUE_SCAN_CAP = 8_192;951 952export const getRevenueMetrics = query({953 args: {954 apiKey: v.optional(v.string()),955 projectId: v.optional(v.id("projects")),956 fromDay: v.string(),957 toDay: v.string(),958 // No product, currency or platform args: the dashboard filters client-side959 // and fills its dropdowns from the unfiltered rows. Add a separate query if960 // another caller needs narrowing.961 environment: v.optional(receiptEnvironmentValidator),962 includeCohorts: v.optional(v.boolean()),963 },964 returns: revenueMetricsResponse,965 handler: async (ctx, args): Promise<Infer<typeof revenueMetricsResponse>> => {966 const project = await projectForAdminRead(ctx, args);967 const environment = args.environment ?? "Production";968 if (!project) {969 return {970 days: [],971 currencies: [],972 productIds: [],973 platforms: [],974 truncated: false,975 environment,976 dataState: "not_collected" as const,977 pendingLateEvents: 0,978 };979 }980 981 // Cap at the dashboard's longest preset plus the day before it (`RANGES`982 // in analytics.tsx; change both together) so a client cannot load every983 // rollup row.984 const MAX_RANGE_DAYS = 92;985 if (args.fromDay > args.toDay) {986 throw new Error(987 `getRevenueMetrics: fromDay (${args.fromDay}) is after toDay (${args.toDay}).`,988 );989 }990 const fromMs = Date.parse(`${args.fromDay}T00:00:00.000Z`);991 const toMs = Date.parse(`${args.toDay}T00:00:00.000Z`);992 if (Number.isNaN(fromMs) || Number.isNaN(toMs)) {993 throw new Error(994 `getRevenueMetrics: invalid ISO date(s) fromDay=${args.fromDay} toDay=${args.toDay}.`,995 );996 }997 const spanDays = Math.round((toMs - fromMs) / 86_400_000) + 1;998 if (spanDays > MAX_RANGE_DAYS) {999 throw new Error(1000 `getRevenueMetrics: span of ${spanDays} days exceeds MAX_RANGE_DAYS=${MAX_RANGE_DAYS}.`,1001 );1002 }1003 1004 // Rows from before the environment split have none and count as1005 // production.1006 const environmentKeys =1007 environment === "Production"1008 ? (["Production", undefined] as const)1009 : (["Sandbox"] as const);1010 // Newest first, so a full window drops its oldest days, and legacy rows,1011 // the oldest of all, go before any current one.1012 const rows: Doc<"revenueMetricsDaily">[] = [];1013 for (const key of environmentKeys) {1014 rows.push(1015 ...(await ctx.db1016 .query("revenueMetricsDaily")1017 .withIndex("by_project_and_environment_and_day", (q) =>1018 q1019 .eq("projectId", project._id)1020 .eq("environment", key)1021 .gte("day", args.fromDay)1022 .lte("day", args.toDay),1023 )1024 .order("desc")1025 .take(REVENUE_SCAN_CAP + 1 - rows.length)),1026 );1027 }1028 const truncated = rows.length > REVENUE_SCAN_CAP;1029 if (truncated) {1030 // The day the cap cut into is partial, which would skew its totals and1031 // churn base, so it goes whole unless it is the only day left.1032 const cutDay = rows[REVENUE_SCAN_CAP].day;1033 const keep = rows.findIndex((row) => row.day <= cutDay);1034 rows.splice(keep > 0 ? keep : REVENUE_SCAN_CAP);1035 }1036 // Return main's day order.1037 rows.sort(1038 (a, b) =>1039 a.day.localeCompare(b.day) ||1040 a.currency.localeCompare(b.currency) ||1041 a._creationTime - b._creationTime,1042 );1043 if (truncated) {1044 console.warn(1045 `[getRevenueMetrics] revenueMetricsDaily scan hit REVENUE_SCAN_CAP=${REVENUE_SCAN_CAP} for project=${project._id} environment=${environment} range=${args.fromDay}..${args.toDay}; the oldest days are missing.`,1046 );1047 }1048 1049 const currencies = new Set<string>();1050 const productIds = new Set<string>();1051 const platforms = new Set<"IOS" | "Android">();1052 for (const row of rows) {1053 if (row.currency) currencies.add(row.currency);1054 productIds.add(row.productId);1055 platforms.add(row.platform);1056 }1057 1058 const status = await ctx.db1059 .query("revenueMetricsRunStatus")1060 .withIndex("by_project", (q) => q.eq("projectId", project._id))1061 .unique();1062 const dataState = await revenueDataState(ctx, {1063 projectId: project._id,1064 environment,1065 hasRows: rows.some(1066 (row) =>1067 row.activeHistoryUnknown !== 0 ||1068 REVENUE_COUNTERS.some((key) => (row[key] ?? 0) !== 0),1069 ),1070 toDay: args.toDay,1071 status,1072 });1073 const cohorts = args.includeCohorts1074 ? await revenueCohorts(ctx, project._id, environment)1075 : undefined;1076 const trialSplitFromDay = status?.trialSplitFromDay;1077 1078 return {1079 days: rows.map((row) => ({1080 day: row.day,1081 currency: row.currency,1082 productId: row.productId,1083 platform: row.platform,1084 activeSubs: row.activeSubs,1085 activeHistoryComplete: row.activeHistoryUnknown === 0,1086 newSubs: row.newSubs,1087 renewals: row.renewals,1088 cancellations: row.cancellations,1089 refunds: row.refunds,1090 revenueMicros: row.revenueMicros,1091 refundedMicros: row.refundedMicros,1092 // These need the free-trial flags notifications carry from1093 // `trialSplitFromDay` on, so earlier days omit them.1094 ...(trialSplitFromDay === undefined || row.day >= trialSplitFromDay1095 ? {1096 activeTrials: row.activeTrials,1097 trialStarts: row.trialStarts,1098 trialConversions: row.trialConversions,1099 directPaidStarts: row.directPaidStarts,1100 expirations: row.expirations,1101 }1102 : {}),1103 })),1104 currencies: Array.from(currencies).sort(),1105 productIds: Array.from(productIds).sort(),1106 platforms: Array.from(platforms).sort(),1107 truncated,1108 environment,1109 dataState,1110 lastRunAt: status?.lastRunAt,1111 definitionsFromDay: status?.definitionsFromDay,1112 trialSplitFromDay,1113 pendingLateEvents: status?.pendingLateEvents ?? 0,1114 oldestPendingLateDay: status?.oldestPendingLateDay,1115 ...(cohorts1116 ? { cohorts, cohortsComputedAt: status?.cohortsComputedAt }1117 : {}),1118 };1119 },1120});1121 1122// Rows alone never mean collected: a project that only calls verify still gets1123// activeSubs rows, but its event counters would read as real zeros.1124async function revenueDataState(1125 ctx: QueryCtx,1126 args: {1127 projectId: Id<"projects">;1128 environment: Infer<typeof receiptEnvironmentValidator>;1129 hasRows: boolean;1130 toDay: string;1131 status: Doc<"revenueMetricsRunStatus"> | null;1132 },1133): Promise<Infer<typeof analyticsDataStateValidator>> {1134 const status = args.status;1135 if (1136 status?.lastRunAt === undefined ||1137 status.definitionsVersion !== ROLLUP_DEFINITIONS_VERSION1138 ) {1139 // No finished run has looked for store notifications yet. Oldest rows1140 // only, so new notifications never invalidate this query.1141 const anyEvent = await ctx.db1142 .query("webhookEvents")1143 .withIndex("by_project", (q) => q.eq("projectId", args.projectId))1144 .first();1145 return anyEvent ? "processing" : "not_collected";1146 }1147 if (status.lastNotificationAt?.[args.environment] === undefined) {1148 return "not_collected";1149 }1150 if (args.hasRows) return "ready";1151 if (args.toDay > new Date(status.lastRunAt).toISOString().slice(0, 10)) {1152 return "processing";1153 }1154 return "zero";1155}1156 1157async function revenueCohorts(1158 ctx: QueryCtx,1159 projectId: Id<"projects">,1160 environment: Infer<typeof receiptEnvironmentValidator>,1161): Promise<Array<Infer<typeof revenueCohortShape>>> {1162 // Rebuilt as a whole each day: at most COHORT_MONTHS × 2 × currencies rows.1163 const rows = await ctx.db1164 .query("revenueCohortsMonthly")1165 .withIndex("by_project", (q) => q.eq("projectId", projectId))1166 .collect();1167 return rows1168 .filter((row) => row.environment === environment)1169 .sort((a, b) =>1170 a.month !== b.month1171 ? a.month.localeCompare(b.month)1172 : a.currency.localeCompare(b.currency),1173 )1174 .map((row) => ({1175 month: row.month,1176 currency: row.currency,1177 members: row.members,1178 revenueMicros: row.revenueMicros,1179 refundedMicros: row.refundedMicros,1180 realizedLtvMicros:1181 row.members > 0 ? Math.round(row.revenueMicros / row.members) : 0,1182 retained: row.retained,1183 retentionObserved: row.retentionObserved,1184 trialStarts: row.trialStarts,1185 trialConversions: row.trialConversions,1186 trialsPending: row.trialsPending,1187 trialsUnobserved: row.trialsUnobserved,1188 observedThrough: row.observedThrough,1189 }));1190}1191 1192const TIMELINE_ROWS_PER_KIND = 50;1193 1194const timelineItemValidator = v.union(1195 v.object({1196 kind: v.literal("purchase"),1197 id: v.id("purchases"),1198 at: v.number(),1199 store: purchaseStoreValidator,1200 productId: v.optional(v.string()),1201 state: v.string(),1202 isValid: v.boolean(),1203 }),1204 v.object({1205 kind: v.literal("subscription"),1206 id: v.id("subscriptions"),1207 at: v.number(),1208 store: purchaseStoreValidator,1209 productId: v.string(),1210 state: subscriptionStateValidator,1211 expiresAt: v.optional(v.number()),1212 willRenew: v.optional(v.boolean()),1213 }),1214 v.object({1215 kind: v.literal("event"),1216 id: v.id("commerceEvents"),1217 at: v.number(),1218 store: purchaseStoreValidator,1219 eventType: v.string(),1220 productId: v.optional(v.string()),1221 }),1222 v.object({1223 kind: v.literal("issue"),1224 id: v.id("purchaseIssues"),1225 at: v.number(),1226 store: purchaseStoreValidator,1227 cause: purchaseIssueCauseValidator,1228 productId: v.optional(v.string()),1229 count: v.number(),1230 }),1231);1232export type TimelineItem = Infer<typeof timelineItemValidator>;1233 1234/**1235 * One app user's purchases, subscriptions, commerce events and purchase1236 * issues, newest first. Each kind is capped at 50 rows; `truncated` says a cap1237 * was reached.1238 */1239export const userTimeline = query({1240 args: {1241 projectId: v.optional(v.id("projects")),1242 apiKey: v.optional(v.string()),1243 appUserId: v.string(),1244 },1245 returns: v.object({1246 appUserId: v.string(),1247 items: v.array(timelineItemValidator),1248 truncated: v.boolean(),1249 }),1250 handler: async (ctx, args) => {1251 const project = await resolveOperatorProject(ctx, args, "read");1252 if (!isValidSubscriptionUserId(args.appUserId)) {1253 throw new ConvexError({1254 code: "INVALID_INPUT",1255 message: `appUserId must be 1 to ${MAX_SUBSCRIPTION_USER_ID_LENGTH} characters`,1256 });1257 }1258 const take = TIMELINE_ROWS_PER_KIND + 1;1259 const [updatedPurchases, legacyPurchases, subscriptions, events, issues] =1260 await Promise.all([1261 ctx.db1262 .query("purchases")1263 .withIndex("by_project_and_app_user_and_updated", (q) =>1264 q1265 .eq("projectId", project._id)1266 .eq("appUserId", args.appUserId)1267 .gt("updatedAt", undefined),1268 )1269 .order("desc")1270 .take(take),1271 ctx.db1272 .query("purchases")1273 .withIndex("by_project_and_app_user_and_updated", (q) =>1274 q1275 .eq("projectId", project._id)1276 .eq("appUserId", args.appUserId)1277 .eq("updatedAt", undefined),1278 )1279 .order("desc")1280 .take(take),1281 ctx.db1282 .query("subscriptions")1283 .withIndex("by_project_and_user_and_updated", (q) =>1284 q.eq("projectId", project._id).eq("userId", args.appUserId),1285 )1286 .order("desc")1287 .take(take),1288 ctx.db1289 .query("commerceEvents")1290 .withIndex("by_project_and_user_and_occurred", (q) =>1291 q.eq("projectId", project._id).eq("userId", args.appUserId),1292 )1293 .order("desc")1294 .take(take),1295 ctx.db1296 .query("purchaseIssues")1297 .withIndex("by_project_and_appUserId_and_lastSeenAt", (q) =>1298 q.eq("projectId", project._id).eq("appUserId", args.appUserId),1299 )1300 .order("desc")1301 .take(take),1302 ]);1303 const purchases = [...updatedPurchases, ...legacyPurchases]1304 .sort(1305 (a, b) =>1306 (b.updatedAt ?? b._creationTime) - (a.updatedAt ?? a._creationTime),1307 )1308 .slice(0, take);1309 const items: TimelineItem[] = [1310 ...purchases.slice(0, TIMELINE_ROWS_PER_KIND).map(1311 (purchase): TimelineItem => ({1312 kind: "purchase",1313 id: purchase._id,1314 at: purchase.updatedAt ?? purchase._creationTime,1315 store: purchase.store,1316 ...(purchase.productId ? { productId: purchase.productId } : {}),1317 state: purchase.state,1318 isValid: purchase.isValid ?? false,1319 }),1320 ),1321 ...subscriptions.slice(0, TIMELINE_ROWS_PER_KIND).map(1322 (subscription): TimelineItem => ({1323 kind: "subscription",1324 id: subscription._id,1325 at: subscription.updatedAt,1326 store: storeOf(subscription),1327 productId: subscription.productId,1328 state: subscription.state,1329 ...(subscription.expiresAt !== undefined1330 ? { expiresAt: subscription.expiresAt }1331 : {}),1332 ...(subscription.willRenew !== undefined1333 ? { willRenew: subscription.willRenew }1334 : {}),1335 }),1336 ),1337 ...events.slice(0, TIMELINE_ROWS_PER_KIND).map(1338 (event): TimelineItem => ({1339 kind: "event",1340 id: event._id,1341 at: event.occurredAt,1342 store: event.store,1343 eventType: event.eventType,1344 ...(event.productId ? { productId: event.productId } : {}),1345 }),1346 ),1347 ...issues.slice(0, TIMELINE_ROWS_PER_KIND).map(1348 (issue): TimelineItem => ({1349 kind: "issue",1350 id: issue._id,1351 at: issue.lastSeenAt,1352 store: issue.store,1353 cause: issue.cause,1354 ...(issue.productId ? { productId: issue.productId } : {}),1355 count: issue.count,1356 }),1357 ),1358 ].sort((a, b) => b.at - a.at);1359 return {1360 appUserId: args.appUserId,1361 items,1362 truncated: [purchases, subscriptions, events, issues].some(1363 (rows) => rows.length > TIMELINE_ROWS_PER_KIND,1364 ),1365 };1366 },1367});1368