1import { foldAccessHistory } from "./accessHistory";2import {3 internalMutation,4 internalQuery,5 type MutationCtx,6 type QueryCtx,7} from "../_generated/server";8import { ConvexError, v, type Infer } from "convex/values";9import type { Doc, Id } from "../_generated/dataModel";10import { internal } from "../_generated/api";11 12import { HarmonizedPurchaseState } from "../purchases/purchaseState";13import type { ReceiptEnvironment } from "../purchases/shared";14import {15 bindAmazonPurchaseOwner,16 subscriptionPurchaseRemoteId,17} from "../purchases/binding";18import {19 applySubscriptionTransition,20 entitlementActive,21 type CurrentSubscription,22 type SubscriptionEventInput,23} from "./stateMachine";24import { applyStatsTransition, statsContributionFor } from "./stats";25import { assertProjectWritable } from "../projects/writable";26import { isValidSubscriptionUserId } from "./limits";27import { isUserErasureRequested } from "./erasure";28import { emitCommerceEvent, storeOf } from "../commerce/internal";29import {30 mergeAnalyticsFacts,31 reconcilePaidStart,32 recordSubscriptionAnalyticsFacts,33} from "./analyticsFacts";34import {35 receiptEnvironmentValidator,36 type bindReasonValidator,37} from "../utils/validation";38 39export const USER_ERASURE_BATCH_SIZE = 100;40export const USER_ERASURE_JOB_RETENTION_MS = 7 * 24 * 60 * 60 * 1_000;41const USER_ERASURE_STALE_MS = 10 * 60 * 1_000;42const USER_ERASURE_PRUNER_BATCH_SIZE = 100;43 44const subscriptionPlatformValidator = v.union(45 v.literal("IOS"),46 v.literal("Android"),47);48 49type RawEventInput = Pick<50 Doc<"webhookEvents">,51 | "type"52 | "productId"53 | "subscriptionState"54 | "expiresAt"55 | "renewsAt"56 | "willRenew"57 | "cancellationReason"58 | "currency"59 | "priceAmountMicros"60 | "platform"61 | "effectiveImmediately"62> & { purchaseToken: string };63type SubscriptionState = Doc<"subscriptions">["state"];64type SubscriptionCancellationReason = NonNullable<65 Doc<"subscriptions">["cancellationReason"]66>;67type SubscriptionPlatform = Doc<"subscriptions">["platform"];68 69export type VerifiedSubscriptionInput = {70 platform: SubscriptionPlatform;71 productId: string;72 purchaseState: HarmonizedPurchaseState;73 subscriptionState?: string;74 expiresAt?: number;75 renewsAt?: number;76 willRenew?: boolean;77 currency?: string;78 priceAmountMicros?: number;79 revocationReasonIOS?: number;80};81 82export type VerifiedSubscriptionSnapshot = {83 productId: string;84 state: SubscriptionState;85 expiresAt?: number;86 renewsAt?: number;87 willRenew?: boolean;88 cancellationReason?: SubscriptionCancellationReason;89 clearCancellationReason?: boolean;90 currency?: string;91 priceAmountMicros?: number;92};93 94type ExistingSubscriptionSnapshotFields = Pick<95 Doc<"subscriptions">,96 | "expiresAt"97 | "renewsAt"98 | "willRenew"99 | "cancellationReason"100 | "currency"101 | "priceAmountMicros"102>;103 104type RecordVerifiedSubscriptionArgs = {105 projectId: Id<"projects">;106 platform: SubscriptionPlatform;107 purchaseToken: string;108 productId: string;109 purchaseState: string;110 subscriptionState?: string;111 expiresAt?: number;112 renewsAt?: number;113 willRenew?: boolean;114 currency?: string;115 priceAmountMicros?: number;116 revocationReasonIOS?: number;117 verifiedPurchaseDate?: number;118 verifiedSignedDate?: number;119 evidenceAt?: number;120 environment?: ReceiptEnvironment;121};122 123type BindSubscriptionToUserArgs = {124 projectId: Id<"projects">;125 purchaseToken: string;126 userId: string;127};128 129export interface PersistSubscriptionSnapshotArgs {130 projectId: Id<"projects">;131 platform: SubscriptionPlatform;132 purchaseToken: string;133 existing: Doc<"subscriptions"> | null;134 store?: Doc<"subscriptions">["store"];135 next: NonNullable<CurrentSubscription>;136 now: number;137 verifiedPurchaseDate?: number;138 verifiedSignedDate?: number;139 environment?: ReceiptEnvironment;140 resyncedAt?: number;141 verifiedEvidenceAt?: number;142 lastEvent?: Pick<143 Doc<"webhookEvents">,144 | "_id"145 | "_creationTime"146 | "store"147 | "type"148 | "occurredAt"149 | "sourceNotificationId"150 | "environment"151 | "environmentDefaulted"152 | "productId"153 | "applicationId"154 | "transactionId"155 | "originalTransactionId"156 | "currency"157 | "priceAmountMicros"158 >;159}160 161interface ApplySubscriptionEventArgs {162 projectId: Id<"projects">;163 eventId: Id<"webhookEvents">;164}165 166interface ApplySubscriptionEventResult {167 transition: string | null;168 active: boolean;169 subscriptionId?: Id<"subscriptions">;170}171 172// Applies a webhook event to `subscriptions`. The event's appliedAt marker173// commits in the same transaction as the subscription and stats writes, so a174// crash before this mutation can be repaired and a processed event is never175// replayed after a newer lastEventId.176export const applySubscriptionEvent = internalMutation({177 args: {178 projectId: v.id("projects"),179 eventId: v.id("webhookEvents"),180 },181 returns: v.object({182 transition: v.union(v.string(), v.null()),183 active: v.boolean(),184 subscriptionId: v.optional(v.id("subscriptions")),185 }),186 handler: async (ctx, args) => applySubscriptionEventHandler(ctx, args),187});188 189// Analytics facts follow the lifecycle transition and only on the event's190// first application, so a redelivery never counts its money twice.191export async function applySubscriptionEventHandler(192 ctx: MutationCtx,193 args: ApplySubscriptionEventArgs,194): Promise<ApplySubscriptionEventResult> {195 const event = await ctx.db.get(args.eventId);196 const firstApplication = event !== null && event.appliedAt === undefined;197 const { analyticsBefore, ...result } = await applyLifecycleEvent(ctx, args);198 if (firstApplication && result.subscriptionId) {199 await recordSubscriptionAnalyticsFacts(200 ctx,201 result.subscriptionId,202 event,203 analyticsBefore,204 );205 }206 return result;207}208 209async function applyLifecycleEvent(210 ctx: MutationCtx,211 args: ApplySubscriptionEventArgs,212): Promise<ApplySubscriptionEventResult & { analyticsBefore?: number | null }> {213 await assertProjectWritable(ctx, args.projectId);214 const storedEvent = await ctx.db.get(args.eventId);215 if (!storedEvent || storedEvent.projectId !== args.projectId) {216 throw new Error("Webhook event not found for project");217 }218 219 const now = Date.now();220 if (storedEvent.productKind === "one_time") {221 if (storedEvent.appliedAt === undefined) {222 await ctx.db.patch(storedEvent._id, { appliedAt: now });223 }224 return { transition: null, active: false };225 }226 if (!storedEvent.purchaseToken) {227 if (storedEvent.appliedAt === undefined) {228 await ctx.db.patch(storedEvent._id, { appliedAt: now });229 }230 return { transition: null, active: false };231 }232 233 const supersedingResolution = await findSupersedingSubscription(234 ctx,235 args.projectId,236 storedEvent.purchaseToken,237 );238 const existingByCurrentToken = supersedingResolution.aliased239 ? null240 : await findSubscriptionByToken(241 ctx,242 args.projectId,243 storedEvent.purchaseToken,244 );245 if (supersedingResolution.aliased) {246 const supersedingSubscription = supersedingResolution.subscription;247 const aliasedTransition =248 storedEvent.type === "PurchaseRefunded"249 ? ("Refunded" as const)250 : storedEvent.type === "SubscriptionRevoked"251 ? ("Revoked" as const)252 : null;253 const active = supersedingSubscription254 ? isActive(supersedingSubscription, now)255 : false;256 const firstApplication = storedEvent.appliedAt === undefined;257 if (firstApplication) {258 if (supersedingSubscription && aliasedTransition) {259 const predecessorProductId =260 storedEvent.productId ?? supersedingResolution.productId;261 await emitCommerceEvent(ctx, {262 projectId: args.projectId,263 transition: aliasedTransition,264 active: false,265 previouslyActive: false,266 sourceEvent: {267 ...storedEvent,268 productId: predecessorProductId,269 },270 ...(predecessorProductId271 ? {272 subscription: {273 state: aliasedTransition,274 productId: predecessorProductId,275 ...(supersedingSubscription.userId276 ? { userId: supersedingSubscription.userId }277 : {}),278 },279 }280 : {}),281 });282 }283 await ctx.db.patch(storedEvent._id, { appliedAt: now });284 }285 return {286 analyticsBefore: supersedingResolution.analyticsReplacementAt ?? null,287 transition: firstApplication ? aliasedTransition : null,288 active,289 ...(supersedingSubscription290 ? { subscriptionId: supersedingSubscription._id }291 : {}),292 };293 }294 const linkedResolution = storedEvent.linkedPurchaseToken295 ? await findSupersedingSubscription(296 ctx,297 args.projectId,298 storedEvent.linkedPurchaseToken,299 )300 : ({ aliased: false } as const);301 const linkedExact =302 storedEvent.linkedPurchaseToken && !linkedResolution.aliased303 ? await findSubscriptionByToken(304 ctx,305 args.projectId,306 storedEvent.linkedPurchaseToken,307 )308 : null;309 if (linkedResolution.aliased && !linkedResolution.subscription) {310 if (storedEvent.appliedAt === undefined) {311 await ctx.db.patch(storedEvent._id, { appliedAt: now });312 }313 return { transition: null, active: false };314 }315 const linkedExisting =316 linkedExact ??317 (linkedResolution.aliased ? linkedResolution.subscription : null);318 let existing = preferredReplacementSnapshot(319 existingByCurrentToken,320 linkedExisting,321 );322 const priorStoreSnapshot = existingByCurrentToken?.lastEventId323 ? existingByCurrentToken324 : linkedExisting?.lastEventId325 ? linkedExisting326 : null;327 if (storedEvent.appliedAt !== undefined) {328 return {329 transition: null,330 active: existing ? isActive(existing, now) : false,331 ...(existing ? { subscriptionId: existing._id } : {}),332 };333 }334 335 // A linked-token event can arrive after both tokens already have rows. Keep336 // current-token state, merge predecessor identity/history, and remove the337 // duplicate contribution before applying same-token ordering.338 if (339 existingByCurrentToken &&340 linkedExisting &&341 existingByCurrentToken._id !== linkedExisting._id342 ) {343 // Two accounts on one subscription is a conflict whether or not either row344 // carries an erasure marker; a marker never decides which of them wins.345 if (346 existingByCurrentToken.userId &&347 linkedExisting.userId &&348 existingByCurrentToken.userId !== linkedExisting.userId349 ) {350 throw new ConvexError({351 code: "SUBSCRIPTION_USER_CONFLICT",352 message: "Linked Google purchase tokens belong to different users.",353 });354 }355 // The marker records that the row's own owner was erased. A live binding on356 // either row is a later, valid association, so it survives the merge and357 // the marker does not carry.358 const boundUserId =359 existingByCurrentToken.userId ?? linkedExisting.userId ?? undefined;360 const accountErased =361 !boundUserId &&362 (existingByCurrentToken.accountErased === true ||363 linkedExisting.accountErased === true);364 const survivor = preferredReplacementSnapshot(365 existingByCurrentToken,366 linkedExisting,367 )!;368 const removed =369 survivor._id === existingByCurrentToken._id370 ? linkedExisting371 : existingByCurrentToken;372 const removedPeriod = await fetchBillingPeriod(373 ctx,374 args.projectId,375 removed.platform,376 removed.productId,377 );378 await ctx.db.delete(removed._id);379 await applyStatsTransition(380 ctx,381 args.projectId,382 statsContributionFor(removed, removedPeriod, now),383 null,384 removed,385 );386 const previousAnalytics = [survivor.analytics, removed.analytics];387 existing = {388 ...survivor,389 purchaseToken: storedEvent.purchaseToken,390 accountErased: accountErased || undefined,391 userId: boundUserId,392 startedAt: Math.min(survivor.startedAt, removed.startedAt),393 analytics: mergeAnalyticsFacts(survivor.analytics, removed.analytics, {394 at:395 storedEvent.type === "SubscriptionStarted"396 ? storedEvent.occurredAt397 : undefined,398 now,399 }),400 };401 await ctx.db.patch(survivor._id, {402 purchaseToken: existing.purchaseToken,403 accountErased: existing.accountErased,404 userId: existing.userId,405 startedAt: existing.startedAt,406 analytics: existing.analytics,407 updatedAt: now,408 });409 await reconcilePaidStart(ctx, previousAnalytics, existing.analytics);410 await recordSubscriptionTokenAlias(ctx, {411 projectId: args.projectId,412 purchaseToken: removed.purchaseToken,413 successorPurchaseToken: existing.purchaseToken,414 predecessorProductId: removed.productId,415 analyticsReplacementAt:416 storedEvent.type === "SubscriptionStarted"417 ? storedEvent.occurredAt418 : undefined,419 now,420 });421 }422 423 if (424 existing &&425 !existingByCurrentToken &&426 existing.purchaseToken !== storedEvent.purchaseToken427 ) {428 existing = {429 ...existing,430 analytics: mergeAnalyticsFacts(undefined, existing.analytics, {431 at:432 storedEvent.type === "SubscriptionStarted"433 ? storedEvent.occurredAt434 : undefined,435 now,436 }),437 };438 }439 440 const previouslyActive =441 reportedEntitlement(existingByCurrentToken) ??442 reportedEntitlement(linkedExisting) ??443 false;444 const noOpResult = (): ApplySubscriptionEventResult => ({445 transition: null,446 active: existing ? isActive(existing, now) : false,447 ...(existing ? { subscriptionId: existing._id } : {}),448 });449 450 // Store timestamps only order events for the same purchase token. A linked451 // predecessor can expire after its replacement became active.452 const orderingExisting = existingByCurrentToken;453 454 // For events from before appliedAt existed: the current last event counts as455 // applied, and an older one is marked handled without rolling state back.456 // Millisecond store timestamps can tie, so ingestion order breaks ties. An457 // unapplied newest event falls through and repairs the action/mutation gap.458 if (459 orderingExisting?.lastEventSourceNotificationId ===460 storedEvent.sourceNotificationId461 ) {462 await ctx.db.patch(storedEvent._id, { appliedAt: now });463 return noOpResult();464 }465 if (orderingExisting?.lastEventOccurredAt !== undefined) {466 const stale =467 orderingExisting.lastEventOccurredAt > storedEvent.occurredAt ||468 (orderingExisting.lastEventOccurredAt === storedEvent.occurredAt &&469 (orderingExisting.lastEventCreationTime ?? 0) >470 storedEvent._creationTime);471 if (stale) {472 await ctx.db.patch(storedEvent._id, { appliedAt: now });473 return noOpResult();474 }475 } else if (orderingExisting?.lastEventId) {476 if (orderingExisting.lastEventId === args.eventId) {477 await ctx.db.patch(storedEvent._id, { appliedAt: now });478 return noOpResult();479 }480 const lastEvent = await ctx.db.get(orderingExisting.lastEventId);481 if (482 lastEvent?.projectId === args.projectId &&483 lastEvent.purchaseToken === storedEvent.purchaseToken &&484 lastEvent.platform === storedEvent.platform &&485 (lastEvent.occurredAt > storedEvent.occurredAt ||486 (lastEvent.occurredAt === storedEvent.occurredAt &&487 lastEvent._creationTime > storedEvent._creationTime))488 ) {489 await ctx.db.patch(storedEvent._id, { appliedAt: now });490 return noOpResult();491 }492 }493 494 // Google subscription bundles can report ITEMS_CHANGED without identifying495 // which line item changed. Keep the raw event for operations, but never496 // overwrite the singular canonical product with an arbitrary bundle item.497 // Linked replacements still move the canonical row after the ordering guard.498 if (499 storedEvent.type === "SubscriptionProductChanged" &&500 !storedEvent.productId501 ) {502 if (existing) {503 const verifiedReplacementProductChanged =504 existingByCurrentToken?.lastEventId === undefined &&505 priorStoreSnapshot !== null &&506 priorStoreSnapshot.productId !== existing.productId;507 const subscriptionId = await persistSubscriptionSnapshot(ctx, {508 projectId: args.projectId,509 platform: existing.platform,510 purchaseToken: storedEvent.purchaseToken,511 existing,512 next: {513 state: existing.state,514 productId: existing.productId,515 expiresAt: existing.expiresAt,516 renewsAt: existing.renewsAt,517 willRenew: existing.willRenew,518 cancellationReason: existing.cancellationReason,519 currency: existing.currency,520 priceAmountMicros: existing.priceAmountMicros,521 },522 now,523 lastEvent: storedEvent,524 });525 await recordSubscriptionTokenAlias(ctx, {526 projectId: args.projectId,527 purchaseToken: existing.purchaseToken,528 successorPurchaseToken: storedEvent.purchaseToken,529 predecessorProductId: existing.productId,530 now,531 });532 await recordSubscriptionTokenAlias(ctx, {533 projectId: args.projectId,534 purchaseToken: storedEvent.linkedPurchaseToken,535 successorPurchaseToken: storedEvent.purchaseToken,536 predecessorProductId:537 priorStoreSnapshot?.productId ?? existing.productId,538 now,539 });540 await ctx.db.patch(storedEvent._id, { appliedAt: now });541 const active = isActive(existing, now);542 if (verifiedReplacementProductChanged) {543 await emitCommerceEvent(ctx, {544 projectId: args.projectId,545 transition: "ProductChanged",546 active,547 previouslyActive: active,548 sourceEvent: storedEvent,549 subscriptionId,550 previousProductId: priorStoreSnapshot.productId,551 subscription: {552 state: existing.state,553 productId: existing.productId,554 ...(existing.expiresAt !== undefined555 ? { expiresAt: existing.expiresAt }556 : {}),557 ...(existing.renewsAt !== undefined558 ? { renewsAt: existing.renewsAt }559 : {}),560 ...(existing.willRenew !== undefined561 ? { willRenew: existing.willRenew }562 : {}),563 ...(existing.cancellationReason564 ? { cancellationReason: existing.cancellationReason }565 : {}),566 ...(existing.userId ? { userId: existing.userId } : {}),567 },568 });569 }570 return {571 transition: verifiedReplacementProductChanged ? "ProductChanged" : null,572 active,573 subscriptionId,574 };575 }576 await ctx.db.patch(storedEvent._id, { appliedAt: now });577 return noOpResult();578 }579 580 const current: CurrentSubscription = existing581 ? {582 state: existing.state,583 productId: existing.productId,584 expiresAt: existing.expiresAt,585 renewsAt: existing.renewsAt,586 willRenew: existing.willRenew,587 cancellationReason: existing.cancellationReason,588 currency: existing.currency,589 priceAmountMicros: existing.priceAmountMicros,590 }591 : null;592 const event: RawEventInput = {593 type: storedEvent.type,594 productId: storedEvent.productId,595 subscriptionState: storedEvent.subscriptionState,596 expiresAt: storedEvent.expiresAt,597 renewsAt: storedEvent.renewsAt,598 willRenew: storedEvent.willRenew,599 cancellationReason: storedEvent.cancellationReason,600 currency: storedEvent.currency,601 priceAmountMicros: storedEvent.priceAmountMicros,602 platform: storedEvent.platform,603 effectiveImmediately: storedEvent.effectiveImmediately,604 purchaseToken: storedEvent.purchaseToken,605 };606 const transition = applySubscriptionTransition(607 current,608 coerceEventInput(event),609 );610 const linkedStartedOnActivePredecessor =611 storedEvent.type === "SubscriptionStarted" &&612 storedEvent.linkedPurchaseToken !== undefined &&613 priorStoreSnapshot !== null &&614 isActive(priorStoreSnapshot, priorStoreSnapshot.updatedAt);615 const effectiveTransition = linkedStartedOnActivePredecessor616 ? transition.next?.productId !== priorStoreSnapshot.productId617 ? "ProductChanged"618 : priorStoreSnapshot.willRenew === false &&619 transition.next?.willRenew === true620 ? "Uncanceled"621 : transition.next?.expiresAt !== undefined &&622 priorStoreSnapshot.expiresAt !== undefined &&623 transition.next.expiresAt > priorStoreSnapshot.expiresAt624 ? "Renewed"625 : null626 : transition.transition;627 const firstStoreEventAfterVerification =628 storedEvent.type === "SubscriptionStarted" &&629 existing !== null &&630 existing.lastEventId === undefined &&631 priorStoreSnapshot === null;632 // Computed before persistSubscriptionSnapshot stamps lastEvent*: a row633 // bootstrapped by receipt verification has no store history, so a price634 // change or deferral as its first store event has no baseline. The mapping635 // vectors pin these to no event.636 const priceOrDeferralWithoutBaseline =637 existing !== null &&638 existing.lastEventId === undefined &&639 priorStoreSnapshot === null &&640 (effectiveTransition === "PriceChanged" ||641 effectiveTransition === "Deferred");642 const previousProductId =643 priorStoreSnapshot?.productId ?? existing?.productId;644 645 if (!transition.next) {646 await ctx.db.patch(storedEvent._id, { appliedAt: now });647 return {648 transition: transition.transition ?? null,649 active: false,650 ...(existing ? { subscriptionId: existing._id } : {}),651 };652 }653 // A re-sync already read the store at or after this event, so its state654 // stands. The event is reported only when it names that state (SPEC 9.1).655 const next =656 current &&657 orderingExisting?.resyncedAt !== undefined &&658 storedEvent.occurredAt <= orderingExisting.resyncedAt659 ? current660 : transition.next;661 662 const subscriptionId = await persistSubscriptionSnapshot(ctx, {663 projectId: args.projectId,664 platform: event.platform,665 purchaseToken: event.purchaseToken,666 existing,667 next,668 now,669 lastEvent: storedEvent,670 });671 await recordSubscriptionTokenAlias(ctx, {672 projectId: args.projectId,673 purchaseToken: existing?.purchaseToken,674 successorPurchaseToken: storedEvent.purchaseToken,675 predecessorProductId: existing?.productId,676 analyticsReplacementAt:677 storedEvent.type === "SubscriptionStarted"678 ? storedEvent.occurredAt679 : undefined,680 now,681 });682 await recordSubscriptionTokenAlias(ctx, {683 projectId: args.projectId,684 purchaseToken: storedEvent.linkedPurchaseToken,685 successorPurchaseToken: storedEvent.purchaseToken,686 predecessorProductId: priorStoreSnapshot?.productId ?? existing?.productId,687 analyticsReplacementAt:688 storedEvent.type === "SubscriptionStarted"689 ? storedEvent.occurredAt690 : undefined,691 now,692 });693 await ctx.db.patch(storedEvent._id, { appliedAt: now });694 const active = entitlementActive(next, now);695 const commerceTransition =696 next.state !== transition.next.state697 ? null698 : linkedStartedOnActivePredecessor699 ? effectiveTransition700 : firstStoreEventAfterVerification701 ? "Started"702 : priceOrDeferralWithoutBaseline703 ? null704 : effectiveTransition;705 await emitCommerceEvent(ctx, {706 projectId: args.projectId,707 transition: commerceTransition ?? null,708 active,709 previouslyActive,710 sourceEvent: storedEvent,711 subscriptionId,712 ...(previousProductId && previousProductId !== next.productId713 ? { previousProductId }714 : {}),715 subscription: {716 state: next.state,717 productId: next.productId,718 ...(next.expiresAt !== undefined ? { expiresAt: next.expiresAt } : {}),719 ...(next.renewsAt !== undefined ? { renewsAt: next.renewsAt } : {}),720 ...(next.willRenew !== undefined ? { willRenew: next.willRenew } : {}),721 ...(next.cancellationReason722 ? { cancellationReason: next.cancellationReason }723 : {}),724 ...(existing?.userId ? { userId: existing.userId } : {}),725 },726 });727 728 return {729 transition: effectiveTransition ?? null,730 active,731 subscriptionId,732 };733}734 735export function buildVerifiedSubscriptionSnapshot(736 input: VerifiedSubscriptionInput,737): VerifiedSubscriptionSnapshot | null {738 if (input.productId.length === 0 || input.productId === "unknown") {739 return null;740 }741 if (input.purchaseState === HarmonizedPurchaseState.INAUTHENTIC) return null;742 743 const base: Pick<744 VerifiedSubscriptionSnapshot,745 | "productId"746 | "expiresAt"747 | "renewsAt"748 | "willRenew"749 | "currency"750 | "priceAmountMicros"751 > = {752 productId: input.productId,753 expiresAt: input.expiresAt,754 renewsAt: input.renewsAt,755 willRenew: input.willRenew,756 currency: input.currency,757 priceAmountMicros: input.priceAmountMicros,758 };759 760 if (input.platform === "IOS") {761 switch (input.purchaseState) {762 case HarmonizedPurchaseState.ENTITLED:763 return {764 ...base,765 state: "Active",766 cancellationReason: undefined,767 clearCancellationReason: true,768 };769 case HarmonizedPurchaseState.EXPIRED:770 return {771 ...base,772 state: "Expired",773 willRenew: false,774 };775 case HarmonizedPurchaseState.CANCELED:776 return {777 ...base,778 state: "Revoked",779 willRenew: false,780 // Apple's revocationReason field covers a transaction "refunded781 // ... or revoked from family sharing": value 1 is unambiguously a782 // refund, while 0 also covers Family Sharing loss. Mirroring the783 // webhook REVOKE policy, only the unambiguous value asserts money784 // moved back; webhook REFUND notifications carry the rest.785 ...(input.revocationReasonIOS === 1786 ? { cancellationReason: "Refunded" as const }787 : {}),788 };789 case HarmonizedPurchaseState.PENDING_ACKNOWLEDGMENT:790 case HarmonizedPurchaseState.PENDING:791 case HarmonizedPurchaseState.UNKNOWN:792 return {793 ...base,794 state: "Unknown",795 };796 case HarmonizedPurchaseState.READY_TO_CONSUME:797 case HarmonizedPurchaseState.CONSUMED:798 return null;799 }800 }801 802 switch (input.purchaseState) {803 case HarmonizedPurchaseState.UNKNOWN:804 return {805 ...base,806 state: "Unknown",807 };808 case HarmonizedPurchaseState.EXPIRED:809 return {810 ...base,811 state: "Expired",812 willRenew: false,813 };814 case HarmonizedPurchaseState.READY_TO_CONSUME:815 case HarmonizedPurchaseState.CONSUMED:816 return null;817 case HarmonizedPurchaseState.ENTITLED:818 case HarmonizedPurchaseState.CANCELED:819 break;820 }821 822 switch (input.subscriptionState?.toUpperCase()) {823 case "SUBSCRIPTION_STATE_ACTIVE":824 return {825 ...base,826 state: "Active",827 willRenew: input.willRenew ?? true,828 cancellationReason: undefined,829 clearCancellationReason: true,830 };831 case "SUBSCRIPTION_STATE_CANCELED":832 return {833 ...base,834 state: "Active",835 willRenew: false,836 };837 case "SUBSCRIPTION_STATE_IN_GRACE_PERIOD":838 return {839 ...base,840 state: "InGracePeriod",841 willRenew: input.willRenew ?? true,842 cancellationReason: undefined,843 clearCancellationReason: true,844 };845 case "SUBSCRIPTION_STATE_ON_HOLD":846 return {847 ...base,848 state: "InBillingRetry",849 cancellationReason: "BillingError",850 };851 case "SUBSCRIPTION_STATE_PAUSED":852 return {853 ...base,854 state: "Paused",855 willRenew: false,856 };857 case "SUBSCRIPTION_STATE_EXPIRED":858 return {859 ...base,860 state: "Expired",861 willRenew: false,862 };863 case "SUBSCRIPTION_STATE_PENDING":864 return {865 ...base,866 state: "Unknown",867 };868 }869 870 return {871 ...base,872 state:873 input.purchaseState === HarmonizedPurchaseState.ENTITLED ||874 input.purchaseState === HarmonizedPurchaseState.PENDING_ACKNOWLEDGMENT875 ? "Active"876 : "Unknown",877 };878}879 880export function mergeVerifiedSubscriptionSnapshot(881 existing: ExistingSubscriptionSnapshotFields | null,882 snapshot: VerifiedSubscriptionSnapshot,883): VerifiedSubscriptionSnapshot {884 if (!existing) return snapshot;885 886 return {887 ...snapshot,888 expiresAt: snapshot.expiresAt ?? existing.expiresAt,889 renewsAt:890 snapshot.willRenew === false891 ? undefined892 : (snapshot.renewsAt ?? existing.renewsAt),893 willRenew: snapshot.willRenew ?? existing.willRenew,894 cancellationReason:895 snapshot.cancellationReason !== undefined ||896 snapshot.clearCancellationReason897 ? snapshot.cancellationReason898 : existing.cancellationReason,899 currency: snapshot.currency ?? existing.currency,900 priceAmountMicros: snapshot.priceAmountMicros ?? existing.priceAmountMicros,901 };902}903 904// Receipt verification is the synchronous bootstrap path for905// subscriptions. Webhooks keep lifecycle state fresh later, but a906// successful verify must be enough for SDK clients to bind the just-907// purchased token to their app user.908export const recordVerifiedSubscription = internalMutation({909 args: {910 projectId: v.id("projects"),911 platform: subscriptionPlatformValidator,912 purchaseToken: v.string(),913 productId: v.string(),914 purchaseState: v.string(),915 subscriptionState: v.optional(v.string()),916 expiresAt: v.optional(v.number()),917 renewsAt: v.optional(v.number()),918 willRenew: v.optional(v.boolean()),919 currency: v.optional(v.string()),920 priceAmountMicros: v.optional(v.number()),921 revocationReasonIOS: v.optional(v.number()),922 verifiedPurchaseDate: v.optional(v.number()),923 verifiedSignedDate: v.optional(v.number()),924 // The newest fact the evidence reports; see predatesResync.925 evidenceAt: v.optional(v.number()),926 environment: v.optional(receiptEnvironmentValidator),927 },928 returns: v.union(v.id("subscriptions"), v.null()),929 handler: async (ctx, args) => recordVerifiedSubscriptionHandler(ctx, args),930});931 932export async function recordVerifiedSubscriptionHandler(933 ctx: MutationCtx,934 args: RecordVerifiedSubscriptionArgs,935): Promise<Id<"subscriptions"> | null> {936 await assertProjectWritable(ctx, args.projectId);937 const snapshot = buildVerifiedSubscriptionSnapshot({938 platform: args.platform,939 productId: args.productId,940 purchaseState: normalizeHarmonizedPurchaseState(args.purchaseState),941 subscriptionState: args.subscriptionState,942 expiresAt: args.expiresAt,943 renewsAt: args.renewsAt,944 willRenew: args.willRenew,945 currency: args.currency,946 priceAmountMicros: args.priceAmountMicros,947 revocationReasonIOS: args.revocationReasonIOS,948 });949 if (!snapshot) return null;950 951 const supersedingResolution = await findSupersedingSubscription(952 ctx,953 args.projectId,954 args.purchaseToken,955 );956 if (supersedingResolution.aliased) {957 return supersedingResolution.subscription?._id ?? null;958 }959 const existing = await findSubscriptionByToken(960 ctx,961 args.projectId,962 args.purchaseToken,963 );964 // Verification bootstraps a token before server notifications arrive. Once965 // a store event governs the row, a client can replay an older but still-valid966 // transaction; without comparable ordering metadata that snapshot must not967 // roll canonical product, state, or expiry backward.968 if (existing?.lastEventId) return existing._id;969 const now = Date.now();970 if (existing && isOlderVerifiedSnapshot(existing, args)) return existing._id;971 972 const next = mergeVerifiedSubscriptionSnapshot(existing, snapshot);973 if (existing && predatesResync(existing, args)) {974 // Evidence the re-sync may already cover, yet it disagrees: an Apple975 // refund reversal keeps its purchase date. Ask the store, never guess.976 if (disputesResync(existing, next.state, now)) {977 await ctx.db.patch(existing._id, { resyncDisputedAt: now });978 await ctx.scheduler.runAfter(979 0,980 internal.subscriptions.action.resyncDisputedSubscription,981 { projectId: args.projectId, subscriptionId: existing._id },982 );983 }984 return existing._id;985 }986 return persistSubscriptionSnapshot(ctx, {987 projectId: args.projectId,988 platform: args.platform,989 purchaseToken: args.purchaseToken,990 existing,991 next,992 now,993 verifiedPurchaseDate: args.verifiedPurchaseDate,994 verifiedSignedDate: args.verifiedSignedDate,995 verifiedEvidenceAt: args.evidenceAt,996 environment: args.environment,997 });998}999 1000// Google snapshots follow store-read time; Apple transactions follow purchase1001// and signing dates because an upgrade may end sooner and a refund keeps its date.1002function isOlderVerifiedSnapshot(1003 stored: Pick<1004 Doc<"subscriptions">,1005 "verifiedPurchaseDate" | "verifiedSignedDate" | "verifiedEvidenceAt"1006 >,1007 incoming: Pick<1008 RecordVerifiedSubscriptionArgs,1009 "verifiedPurchaseDate" | "verifiedSignedDate" | "platform" | "evidenceAt"1010 >,1011): boolean {1012 if (incoming.platform === "Android") {1013 return (1014 stored.verifiedEvidenceAt !== undefined &&1015 incoming.evidenceAt !== undefined &&1016 incoming.evidenceAt < stored.verifiedEvidenceAt1017 );1018 }1019 if (1020 stored.verifiedPurchaseDate === undefined ||1021 incoming.verifiedPurchaseDate === undefined1022 ) {1023 return false;1024 }1025 if (incoming.verifiedPurchaseDate !== stored.verifiedPurchaseDate) {1026 return incoming.verifiedPurchaseDate < stored.verifiedPurchaseDate;1027 }1028 if (stored.verifiedSignedDate === undefined) return false;1029 // A same-transaction verify without a signing date cannot prove it is newer.1030 return (1031 incoming.verifiedSignedDate === undefined ||1032 incoming.verifiedSignedDate < stored.verifiedSignedDate1033 );1034}1035 1036// A re-sync read the store's current state at `resyncedAt`, so a verify only1037// replaces it with evidence of something later: Apple's transaction purchase1038// or revocation date (a replay is re-signed, so its signing date proves1039// nothing), or for Google, which reports current state, when Play was read.1040function predatesResync(1041 stored: Pick<Doc<"subscriptions">, "resyncedAt">,1042 incoming: Pick<RecordVerifiedSubscriptionArgs, "evidenceAt">,1043): boolean {1044 return (1045 stored.resyncedAt !== undefined &&1046 (incoming.evidenceAt === undefined ||1047 incoming.evidenceAt <= stored.resyncedAt)1048 );1049}1050 1051export const RESYNC_DISPUTE_COOLDOWN_MS = 60 * 60 * 1_000;1052 1053// A transaction cannot show billing retry or grace: there its period simply1054// ended, so Expired evidence agrees with both. An Apple refund verifies as1055// Revoked, which the re-sync stores as Refunded. A real dispute re-reads the1056// store at most once per cooldown, however often the app verifies.1057function disputesResync(1058 stored: Pick<Doc<"subscriptions">, "state" | "resyncDisputedAt">,1059 verifiedState: SubscriptionState,1060 now: number,1061): boolean {1062 if (verifiedState === stored.state) return false;1063 if (1064 verifiedState === "Expired" &&1065 (stored.state === "InBillingRetry" || stored.state === "InGracePeriod")1066 ) {1067 return false;1068 }1069 if (verifiedState === "Revoked" && stored.state === "Refunded") return false;1070 return (stored.resyncDisputedAt ?? 0) + RESYNC_DISPUTE_COOLDOWN_MS <= now;1071}1072 1073/**1074 * The entitlement gate a row's owner was last told, or undefined when nothing1075 * was. A store event reports the row's state (a late bind repeats its grant),1076 * read at the row's own `updatedAt` since today's clock may already have1077 * closed it. Without one, only a re-sync that reached the owner told anything,1078 * and a verify since then may have moved the row silently.1079 */1080export function reportedEntitlement(1081 row: Doc<"subscriptions"> | null,1082): boolean | undefined {1083 if (!row) return undefined;1084 if (row.lastEventId === undefined) return row.resyncReportedActive;1085 return isActive(row, row.updatedAt);1086}1087 1088export async function persistSubscriptionSnapshot(1089 ctx: MutationCtx,1090 args: PersistSubscriptionSnapshotArgs,1091): Promise<Id<"subscriptions">> {1092 // Stats deltas must compare the old row against the old catalog entry1093 // and the new row against the new one; otherwise product/platform changes1094 // subtract the wrong MRR bucket.1095 const afterBillingPeriod =1096 args.store === "amazon"1097 ? undefined1098 : await fetchBillingPeriod(1099 ctx,1100 args.projectId,1101 args.platform,1102 args.next.productId,1103 );1104 const beforeBillingPeriod =1105 args.existing &&1106 args.existing.store !== "amazon" &&1107 (args.existing.productId !== args.next.productId ||1108 args.existing.platform !== args.platform)1109 ? await fetchBillingPeriod(1110 ctx,1111 args.projectId,1112 args.existing.platform,1113 args.existing.productId,1114 )1115 : afterBillingPeriod;1116 const beforeContribution = args.existing1117 ? statsContributionFor(args.existing, beforeBillingPeriod, args.now)1118 : null;1119 // An event Play sent without an environment keeps the row's known one.1120 const environment = args.lastEvent?.environmentDefaulted1121 ? (args.existing?.environment ??1122 args.existing?.lastEventSource?.environment)1123 : (args.lastEvent?.environment ?? args.environment);1124 1125 const resyncAdvanced =1126 args.resyncedAt !== undefined &&1127 args.resyncedAt > (args.existing?.resyncedAt ?? -Infinity);1128 const evidenceAdvanced =1129 args.verifiedEvidenceAt !== undefined &&1130 args.verifiedEvidenceAt > (args.existing?.verifiedEvidenceAt ?? -Infinity);1131 // An expired Apple receipt still proves its dated purchase-to-expiry period.1132 const expiredApplePeriod =1133 !resyncAdvanced &&1134 args.platform === "IOS" &&1135 args.next.state === "Expired" &&1136 args.verifiedPurchaseDate !== undefined &&1137 args.verifiedEvidenceAt === args.verifiedPurchaseDate &&1138 args.next.expiresAt !== undefined;1139 // A newer signed receipt can show a reversal without dating its access change.1140 const undatedAppleChange =1141 !resyncAdvanced &&1142 !evidenceAdvanced &&1143 args.platform === "IOS" &&1144 args.existing !== null &&1145 args.verifiedSignedDate !== undefined &&1146 args.verifiedSignedDate > (args.existing.verifiedSignedDate ?? -Infinity) &&1147 !(1148 expiredApplePeriod &&1149 args.existing.state === "Active" &&1150 args.existing.verifiedPurchaseDate === args.verifiedPurchaseDate &&1151 args.existing.expiresAt === args.next.expiresAt1152 ) &&1153 (args.next.state !== args.existing.state ||1154 args.next.expiresAt !== args.existing.expiresAt);1155 const observed = resyncAdvanced || evidenceAdvanced || undatedAppleChange;1156 // A delayed commit or receipt replay is not new store evidence.1157 const observedAt = undatedAppleChange1158 ? args.verifiedSignedDate!1159 : Math.max(1160 args.resyncedAt ?? -Infinity,1161 args.verifiedEvidenceAt ?? -Infinity,1162 );1163 const analytics = observed1164 ? {1165 ...(args.existing?.analytics ?? { paidMicros: 0, refundedMicros: 0 }),1166 accessHistory: foldAccessHistory(1167 args.existing?.analytics?.accessHistory,1168 [1169 {1170 effectiveAt: observedAt,1171 sourceAt:1172 args.resyncedAt ?? args.verifiedSignedDate ?? observedAt,1173 sourceCreatedAt: args.now,1174 ...(!undatedAppleChange && args.next.state !== "Unknown"1175 ? {1176 active:1177 args.next.state === "Active" ||1178 args.next.state === "InGracePeriod" ||1179 expiredApplePeriod,1180 }1181 : {}),1182 ...(args.next.expiresAt !== undefined1183 ? { expiresAt: args.next.expiresAt }1184 : {}),1185 },1186 ],1187 args.now,1188 ),1189 }1190 : args.existing?.analytics;1191 1192 const row = {1193 ...(analytics ? { analytics } : {}),1194 purchaseToken: args.purchaseToken,1195 productKind: "subscription" as const,1196 ...(args.existing?.userId ? { userId: args.existing.userId } : {}),1197 productId: args.next.productId,1198 platform: args.platform,1199 ...(args.store ? { store: args.store } : {}),1200 state: args.next.state,1201 expiresAt: args.next.expiresAt,1202 renewsAt: args.next.renewsAt,1203 willRenew: args.next.willRenew,1204 cancellationReason: args.next.cancellationReason,1205 currency: args.next.currency,1206 priceAmountMicros: args.next.priceAmountMicros,1207 ...(environment ? { environment } : {}),1208 updatedAt: args.now,1209 ...(args.resyncedAt !== undefined ? { resyncedAt: args.resyncedAt } : {}),1210 ...(args.verifiedEvidenceAt !== undefined1211 ? { verifiedEvidenceAt: args.verifiedEvidenceAt }1212 : {}),1213 // The pair is written together so a signing date never outlives its purchase date.1214 ...(args.verifiedPurchaseDate !== undefined1215 ? {1216 verifiedPurchaseDate: args.verifiedPurchaseDate,1217 verifiedSignedDate: args.verifiedSignedDate,1218 }1219 : {}),1220 ...(args.lastEvent1221 ? {1222 lastEventId: args.lastEvent._id,1223 lastEventOccurredAt: args.lastEvent.occurredAt,1224 lastEventCreationTime: args.lastEvent._creationTime,1225 lastEventSourceNotificationId: args.lastEvent.sourceNotificationId,1226 lastEventSource: {1227 type: args.lastEvent.type,1228 store: args.lastEvent.store,1229 environment: args.lastEvent.environment,1230 productId: args.lastEvent.productId,1231 applicationId: args.lastEvent.applicationId,1232 transactionId: args.lastEvent.transactionId,1233 originalTransactionId: args.lastEvent.originalTransactionId,1234 currency: args.lastEvent.currency,1235 priceAmountMicros: args.lastEvent.priceAmountMicros,1236 },1237 }1238 : {}),1239 };1240 1241 const subscriptionId = args.existing1242 ? args.existing._id1243 : await ctx.db.insert("subscriptions", {1244 projectId: args.projectId,1245 startedAt: args.now,1246 ...row,1247 });1248 1249 if (args.existing) {1250 await ctx.db.patch(args.existing._id, row);1251 }1252 1253 const updatedRow = (await ctx.db.get(subscriptionId))!;1254 const afterContribution = statsContributionFor(1255 updatedRow,1256 afterBillingPeriod,1257 args.now,1258 );1259 await applyStatsTransition(1260 ctx,1261 args.projectId,1262 beforeContribution,1263 afterContribution,1264 updatedRow,1265 );1266 1267 return subscriptionId;1268}1269 1270export function findSubscriptionByToken(1271 ctx: Pick<QueryCtx, "db">,1272 projectId: Id<"projects">,1273 purchaseToken: string,1274): Promise<Doc<"subscriptions"> | null> {1275 return ctx.db1276 .query("subscriptions")1277 .withIndex("by_project_and_token", (q) =>1278 q.eq("projectId", projectId).eq("purchaseToken", purchaseToken),1279 )1280 .unique();1281}1282 1283const MAX_SUBSCRIPTION_TOKEN_ALIAS_HOPS = 64;1284 1285type SupersedingSubscriptionResolution =1286 | { aliased: false }1287 | {1288 aliased: true;1289 subscription: Doc<"subscriptions"> | null;1290 productId?: string;1291 analyticsReplacementAt?: number;1292 };1293 1294// Follows a token Google replaced to the row that holds it now.1295export async function findSupersedingSubscription(1296 ctx: Pick<QueryCtx, "db">,1297 projectId: Id<"projects">,1298 purchaseToken: string,1299): Promise<SupersedingSubscriptionResolution> {1300 let token = purchaseToken;1301 let productId: string | undefined;1302 let analyticsReplacementAt: number | undefined;1303 const seen = new Set<string>([token]);1304 for (let hop = 0; hop <= MAX_SUBSCRIPTION_TOKEN_ALIAS_HOPS; hop += 1) {1305 const alias = await ctx.db1306 .query("subscriptionTokenAliases")1307 .withIndex("by_project_and_token", (q) =>1308 q.eq("projectId", projectId).eq("purchaseToken", token),1309 )1310 .unique();1311 if (!alias) {1312 return hop === 01313 ? { aliased: false }1314 : {1315 aliased: true,1316 subscription: await findSubscriptionByToken(ctx, projectId, token),1317 productId,1318 analyticsReplacementAt,1319 };1320 }1321 if (hop === 0) {1322 productId = alias.predecessorProductId;1323 analyticsReplacementAt = alias.analyticsReplacementAt;1324 }1325 if (hop === MAX_SUBSCRIPTION_TOKEN_ALIAS_HOPS) {1326 return {1327 aliased: true,1328 subscription: null,1329 productId,1330 analyticsReplacementAt,1331 };1332 }1333 if (seen.has(alias.successorPurchaseToken)) {1334 return {1335 aliased: true,1336 subscription: null,1337 productId,1338 analyticsReplacementAt,1339 };1340 }1341 token = alias.successorPurchaseToken;1342 seen.add(token);1343 }1344 return {1345 aliased: true,1346 subscription: null,1347 productId,1348 analyticsReplacementAt,1349 };1350}1351 1352async function recordSubscriptionTokenAlias(1353 ctx: MutationCtx,1354 args: {1355 projectId: Id<"projects">;1356 purchaseToken?: string;1357 successorPurchaseToken: string;1358 predecessorProductId?: string;1359 analyticsReplacementAt?: number;1360 now: number;1361 },1362): Promise<void> {1363 if (1364 !args.purchaseToken ||1365 args.purchaseToken === args.successorPurchaseToken1366 ) {1367 return;1368 }1369 const existing = await ctx.db1370 .query("subscriptionTokenAliases")1371 .withIndex("by_project_and_token", (q) =>1372 q1373 .eq("projectId", args.projectId)1374 .eq("purchaseToken", args.purchaseToken as string),1375 )1376 .unique();1377 if (existing) {1378 if (1379 existing.successorPurchaseToken !== args.successorPurchaseToken ||1380 (existing.analyticsReplacementAt === undefined &&1381 args.analyticsReplacementAt !== undefined)1382 ) {1383 await ctx.db.patch(existing._id, {1384 successorPurchaseToken: args.successorPurchaseToken,1385 analyticsReplacementAt:1386 existing.successorPurchaseToken === args.successorPurchaseToken1387 ? (existing.analyticsReplacementAt ?? args.analyticsReplacementAt)1388 : args.analyticsReplacementAt,1389 predecessorProductId:1390 existing.predecessorProductId ?? args.predecessorProductId,1391 updatedAt: args.now,1392 });1393 }1394 return;1395 }1396 await ctx.db.insert("subscriptionTokenAliases", {1397 projectId: args.projectId,1398 purchaseToken: args.purchaseToken,1399 successorPurchaseToken: args.successorPurchaseToken,1400 predecessorProductId: args.predecessorProductId,1401 analyticsReplacementAt: args.analyticsReplacementAt,1402 createdAt: args.now,1403 updatedAt: args.now,1404 });1405}1406 1407function preferredReplacementSnapshot(1408 current: Doc<"subscriptions"> | null,1409 linked: Doc<"subscriptions"> | null,1410): Doc<"subscriptions"> | null {1411 if (!current) return linked;1412 // Store timestamps from predecessor and replacement tokens are not a total1413 // order: the predecessor can expire after the replacement becomes active.1414 // Current-token state therefore wins even when it came from verification;1415 // user identity and start history are merged separately during reconciliation.1416 return current;1417}1418 1419export const getSourceProductIdByToken = internalQuery({1420 args: {1421 projectId: v.id("projects"),1422 purchaseToken: v.string(),1423 },1424 returns: v.union(v.string(), v.null()),1425 handler: async (ctx, args) => getSourceProductIdByTokenHandler(ctx, args),1426});1427 1428export async function getSourceProductIdByTokenHandler(1429 ctx: Pick<QueryCtx, "db">,1430 args: { projectId: Id<"projects">; purchaseToken: string },1431): Promise<string | null> {1432 const resolution = await findSupersedingSubscription(1433 ctx,1434 args.projectId,1435 args.purchaseToken,1436 );1437 if (resolution.aliased) return resolution.productId ?? null;1438 const subscription = await findSubscriptionByToken(1439 ctx,1440 args.projectId,1441 args.purchaseToken,1442 );1443 return subscription?.productId ?? null;1444}1445 1446export const getCurrentProductIdByToken = internalQuery({1447 args: {1448 projectId: v.id("projects"),1449 purchaseToken: v.string(),1450 },1451 returns: v.union(v.string(), v.null()),1452 handler: async (ctx, args) => getCurrentProductIdByTokenHandler(ctx, args),1453});1454 1455export async function getCurrentProductIdByTokenHandler(1456 ctx: Pick<QueryCtx, "db">,1457 args: { projectId: Id<"projects">; purchaseToken: string },1458): Promise<string | null> {1459 const resolution = await findSupersedingSubscription(1460 ctx,1461 args.projectId,1462 args.purchaseToken,1463 );1464 if (resolution.aliased) return resolution.subscription?.productId ?? null;1465 const subscription = await findSubscriptionByToken(1466 ctx,1467 args.projectId,1468 args.purchaseToken,1469 );1470 return subscription?.productId ?? null;1471}1472 1473// Billing period for the exact (platform, productId): one SKU can have a1474// different period per store. Undefined when untracked or unset;1475// monthlyMicrosForSub then assumes P1M.1476async function fetchBillingPeriod(1477 ctx: MutationCtx,1478 projectId: Id<"projects">,1479 platform: SubscriptionPlatform,1480 productId: string,1481): Promise<string | undefined> {1482 const product = await ctx.db1483 .query("products")1484 .withIndex("by_project_and_platform_and_product", (q) =>1485 q1486 .eq("projectId", projectId)1487 .eq("platform", platform)1488 .eq("productId", productId),1489 )1490 .unique();1491 return product?.billingPeriod ?? undefined;1492}1493 1494function coerceEventInput(raw: RawEventInput): SubscriptionEventInput {1495 const appleRenewalPreference =1496 raw.platform === "IOS" &&1497 raw.type === "SubscriptionProductChanged" &&1498 raw.effectiveImmediately !== true;1499 return {1500 type: raw.type,1501 // Apple's renewal preference is the product for the next billing period;1502 // the current transaction remains active until then. Keep the target on1503 // the source event, but do not replace the canonical subscription early.1504 productId: appleRenewalPreference ? undefined : raw.productId,1505 subscriptionState: raw.subscriptionState,1506 expiresAt: raw.expiresAt,1507 renewsAt: raw.renewsAt,1508 willRenew: raw.willRenew,1509 cancellationReason: raw.cancellationReason,1510 currency: appleRenewalPreference ? undefined : raw.currency,1511 priceAmountMicros: appleRenewalPreference1512 ? undefined1513 : raw.priceAmountMicros,1514 };1515}1516 1517function isActive(1518 sub: Doc<"subscriptions">,1519 now: number = Date.now(),1520): boolean {1521 return entitlementActive(sub, now);1522}1523 1524function normalizeHarmonizedPurchaseState(1525 state: string,1526): HarmonizedPurchaseState {1527 const normalized = state.trim().toUpperCase().replace(/-/g, "_");1528 if (normalized in HarmonizedPurchaseState) {1529 return HarmonizedPurchaseState[1530 normalized as keyof typeof HarmonizedPurchaseState1531 ];1532 }1533 return HarmonizedPurchaseState.UNKNOWN;1534}1535 1536// Bind a subscription to a userId. Called by the SDK after a successful1537// receipt validation when the host app knows which user owns the receipt.1538export const bindSubscriptionToUser = internalMutation({1539 args: {1540 projectId: v.id("projects"),1541 purchaseToken: v.string(),1542 userId: v.string(),1543 },1544 returns: v.union(v.id("subscriptions"), v.null()),1545 handler: async (ctx, args) => bindSubscriptionToUserHandler(ctx, args),1546});1547 1548/**1549 * The retained source a re-emitted entitlement event is attributed to. The1550 * stored row is preferred; a pruned one is reconstructed from the snapshot the1551 * subscription keeps. Price is stripped: a webhook already reported it, and1552 * repeating it would put the same money on another event.1553 */1554async function retainedSourceEventFor(1555 ctx: MutationCtx,1556 projectId: Id<"projects">,1557 sub: Doc<"subscriptions">,1558) {1559 if (!sub.lastEventId) return null;1560 const retained = await ctx.db.get(sub.lastEventId);1561 const source =1562 retained?.projectId === projectId1563 ? retained1564 : sub.lastEventSource &&1565 sub.lastEventOccurredAt !== undefined &&1566 sub.lastEventSourceNotificationId1567 ? {1568 _id: sub.lastEventId,1569 type: sub.lastEventSource.type,1570 platform: sub.platform,1571 store: sub.lastEventSource.store ?? sub.store,1572 environment: sub.lastEventSource.environment,1573 productId: sub.lastEventSource.productId,1574 applicationId: sub.lastEventSource.applicationId,1575 transactionId: sub.lastEventSource.transactionId,1576 originalTransactionId: sub.lastEventSource.originalTransactionId,1577 sourceNotificationId: sub.lastEventSourceNotificationId,1578 occurredAt: sub.lastEventOccurredAt,1579 }1580 : null;1581 if (!source) return null;1582 return {1583 ...source,1584 currency: undefined,1585 priceAmountMicros: undefined,1586 amountProvenance: undefined,1587 };1588}1589 1590/**1591 * Moves an existing binding. `bindSubscriptionToUserHandler` refuses this on1592 * purpose — a caller holding a purchase token has not proved it owns the1593 * purchase — so correcting a wrong binding needs an operator-authorized path.1594 * Callers must gate this on a secret key.1595 */1596export async function rebindSubscriptionToUserHandler(1597 ctx: MutationCtx,1598 args: BindSubscriptionToUserArgs,1599): Promise<{ subscriptionId: Id<"subscriptions">; notified: boolean } | null> {1600 if (!isValidSubscriptionUserId(args.userId)) {1601 throw new ConvexError("userId must be nonblank and at most 256 characters");1602 }1603 const project = await assertProjectWritable(ctx, args.projectId);1604 if (await isUserErasureRequested(ctx, project, args.userId)) return null;1605 // Resolve exactly as `bind` does. An operator correcting a wrong binding1606 // usually has the token the customer reported, which on Play may be the one1607 // a replacement superseded.1608 const supersedingResolution = await findSupersedingSubscription(1609 ctx,1610 args.projectId,1611 args.purchaseToken,1612 );1613 const sub = supersedingResolution.aliased1614 ? supersedingResolution.subscription1615 : await findSubscriptionByToken(ctx, args.projectId, args.purchaseToken);1616 if (!sub) return null;1617 if (1618 sub.userId &&1619 sub.userId !== args.userId &&1620 (await isUserErasureRequested(ctx, project, sub.userId))1621 )1622 return null;1623 if (1624 (await bindAmazonPurchaseOwner(ctx, project, sub, args.userId, true)) !==1625 "linked"1626 )1627 return null;1628 if (sub.userId === args.userId) {1629 return { subscriptionId: sub._id, notified: true };1630 }1631 const previousUserId = sub.userId;1632 const now = Date.now();1633 1634 // A consumer that gates access on commerce events has no other way to learn1635 // the purchase moved: without these it keeps the wrong user entitled and1636 // never entitles the real one. Decide before writing, so the caller is never1637 // told a rebind succeeded when the notification half of it could not.1638 // The old owner loses what it was told, which a silent verify may have1639 // already changed; the new owner gains what the row grants now.1640 const entitled = isActive({ ...sub, userId: args.userId }, now);1641 const revokeFromPrevious =1642 previousUserId !== undefined && reportedEntitlement(sub) === true;1643 const sourceEvent =1644 entitled || revokeFromPrevious1645 ? await retainedSourceEventFor(ctx, args.projectId, sub)1646 : null;1647 await ctx.db.patch(sub._id, {1648 userId: args.userId,1649 accountErased: undefined,1650 resyncReportedActive: undefined,1651 updatedAt: now,1652 });1653 if (!entitled && !revokeFromPrevious) {1654 return { subscriptionId: sub._id, notified: true };1655 }1656 if (!sourceEvent) {1657 // Every retained trace of the originating notification is gone, so no1658 // event can be attributed. The binding moved; the operator must reconcile1659 // the developer backend by hand.1660 console.warn(1661 "[subscriptions/rebind] moved a binding without emitting entitlement events",1662 { projectId: args.projectId, subscriptionId: sub._id },1663 );1664 return { subscriptionId: sub._id, notified: false };1665 }1666 const snapshot = {1667 state: sub.state,1668 productId: sub.productId,1669 ...(sub.expiresAt !== undefined ? { expiresAt: sub.expiresAt } : {}),1670 ...(sub.renewsAt !== undefined ? { renewsAt: sub.renewsAt } : {}),1671 ...(sub.willRenew !== undefined ? { willRenew: sub.willRenew } : {}),1672 ...(sub.cancellationReason1673 ? { cancellationReason: sub.cancellationReason }1674 : {}),1675 };1676 for (const [userId, active, previouslyActive] of [1677 ...(revokeFromPrevious && previousUserId1678 ? ([[previousUserId, false, true]] as const)1679 : ([] as const)),1680 ...(entitled ? ([[args.userId, true, false]] as const) : ([] as const)),1681 ]) {1682 await emitCommerceEvent(ctx, {1683 projectId: args.projectId,1684 transition: null,1685 active,1686 previouslyActive,1687 sourceEvent,1688 subscriptionId: sub._id,1689 subscription: { ...snapshot, userId },1690 });1691 }1692 return { subscriptionId: sub._id, notified: true };1693}1694 1695export async function bindSubscriptionToUserHandler(1696 ctx: MutationCtx,1697 args: BindSubscriptionToUserArgs,1698): Promise<Id<"subscriptions"> | null> {1699 return (1700 (await bindSubscriptionToUserOutcome(ctx, args)).subscriptionId ?? null1701 );1702}1703 1704/** Why a bind did or did not link; see bindReasonValidator. */1705export type BindOutcome = {1706 reason: Infer<typeof bindReasonValidator>;1707 subscriptionId?: Id<"subscriptions">;1708 // The purchase the bind resolved to, for the issue log.1709 store?: Doc<"purchases">["store"];1710 productId?: string;1711 remoteId?: string;1712 limitReached?: boolean;1713};1714 1715export async function bindSubscriptionToUserOutcome(1716 ctx: MutationCtx,1717 args: BindSubscriptionToUserArgs,1718): Promise<BindOutcome> {1719 if (!isValidSubscriptionUserId(args.userId)) {1720 throw new ConvexError("userId must be nonblank and at most 256 characters");1721 }1722 const project = await assertProjectWritable(ctx, args.projectId);1723 if (await isUserErasureRequested(ctx, project, args.userId))1724 return { reason: "not_bindable" };1725 const supersedingResolution = await findSupersedingSubscription(1726 ctx,1727 args.projectId,1728 args.purchaseToken,1729 );1730 const sub = supersedingResolution.aliased1731 ? supersedingResolution.subscription1732 : await findSubscriptionByToken(ctx, args.projectId, args.purchaseToken);1733 if (!sub) return { reason: "not_found" };1734 const target: Pick<BindOutcome, "store" | "productId" | "remoteId"> = {1735 store: storeOf(sub),1736 productId: sub.productId,1737 remoteId: subscriptionPurchaseRemoteId(sub),1738 };1739 if (sub.userId && sub.userId !== args.userId) {1740 // Publishable callers get the same answer as for an unknown token: a1741 // distinct one would tell any holder of the app-embedded key whether a1742 // token exists and whether someone owns it.1743 console.warn(1744 "[subscriptions/bind] rejected a bind for an already-bound subscription",1745 { projectId: args.projectId, subscriptionId: sub._id },1746 );1747 return { reason: "bound_to_other_user", ...target };1748 }1749 const amazonOwner = await bindAmazonPurchaseOwner(1750 ctx,1751 project,1752 sub,1753 args.userId,1754 );1755 if (amazonOwner !== "linked")1756 return {1757 reason: "not_bindable",1758 ...(amazonOwner === "limit" ? { limitReached: true } : {}),1759 ...target,1760 };1761 if (sub.userId === args.userId)1762 return { reason: "already_bound", subscriptionId: sub._id, ...target };1763 const now = Date.now();1764 // Erasure only unlinked the previous owner; this is a new association, and1765 // the pending-erasure gate above still refuses a user who is being erased.1766 await ctx.db.patch(sub._id, {1767 userId: args.userId,1768 accountErased: undefined,1769 updatedAt: now,1770 });1771 1772 // If the store webhook won the race against verify/bind, its initial event1773 // had no account identity. Emit one correlated entitlement grant after the1774 // binding so a developer backend never has to recover a purchase token from1775 // the public payload. The normal verify -> bind -> webhook path has no1776 // lastEventId yet and emits its grant when that first webhook arrives.1777 if (sub.lastEventId && isActive(sub, now)) {1778 const retainedSource = await ctx.db.get(sub.lastEventId);1779 const sourceEvent =1780 retainedSource?.projectId === args.projectId1781 ? retainedSource1782 : sub.lastEventSource &&1783 sub.lastEventOccurredAt !== undefined &&1784 sub.lastEventSourceNotificationId1785 ? {1786 _id: sub.lastEventId,1787 type: sub.lastEventSource.type,1788 platform: sub.platform,1789 store: sub.lastEventSource.store ?? sub.store,1790 environment: sub.lastEventSource.environment,1791 productId: sub.lastEventSource.productId,1792 applicationId: sub.lastEventSource.applicationId,1793 transactionId: sub.lastEventSource.transactionId,1794 originalTransactionId: sub.lastEventSource.originalTransactionId,1795 currency: sub.lastEventSource.currency,1796 priceAmountMicros: sub.lastEventSource.priceAmountMicros,1797 sourceNotificationId: sub.lastEventSourceNotificationId,1798 occurredAt: sub.lastEventOccurredAt,1799 }1800 : null;1801 if (sourceEvent) {1802 // This only fires after a webhook already emitted the notification's1803 // amount, so repeating it would put the same money on a second event.1804 const entitlementSource = {1805 ...sourceEvent,1806 currency: undefined,1807 priceAmountMicros: undefined,1808 amountProvenance: undefined,1809 };1810 await emitCommerceEvent(ctx, {1811 projectId: args.projectId,1812 transition: null,1813 active: true,1814 previouslyActive: false,1815 sourceEvent: entitlementSource,1816 subscriptionId: sub._id,1817 subscription: {1818 state: sub.state,1819 productId: sub.productId,1820 ...(sub.expiresAt !== undefined ? { expiresAt: sub.expiresAt } : {}),1821 ...(sub.renewsAt !== undefined ? { renewsAt: sub.renewsAt } : {}),1822 ...(sub.willRenew !== undefined ? { willRenew: sub.willRenew } : {}),1823 ...(sub.cancellationReason1824 ? { cancellationReason: sub.cancellationReason }1825 : {}),1826 userId: args.userId,1827 },1828 });1829 }1830 }1831 return { reason: "bound", subscriptionId: sub._id, ...target };1832}1833 1834export async function drainSubscriptionUserErasurePage(1835 ctx: MutationCtx,1836 jobId: Id<"subscriptionUserErasureJobs">,1837): Promise<{1838 done: boolean;1839 subscriptionsErased: number;1840 commerceEventsErased: number;1841}> {1842 const job = await ctx.db.get(jobId);1843 if (!job || job.status === "completed" || !job.userId) {1844 return {1845 done: true,1846 subscriptionsErased: job?.subscriptionsErased ?? 0,1847 commerceEventsErased: job?.commerceEventsErased ?? 0,1848 };1849 }1850 1851 const now = Date.now();1852 await ctx.db.patch(jobId, { status: "running", updatedAt: now });1853 const [subscriptions, commerceEvents, purchases, issues] = await Promise.all([1854 ctx.db1855 .query("subscriptions")1856 .withIndex("by_project_and_user", (q) =>1857 q.eq("projectId", job.projectId).eq("userId", job.userId),1858 )1859 .take(USER_ERASURE_BATCH_SIZE),1860 ctx.db1861 .query("commerceEvents")1862 .withIndex("by_project_and_user", (q) =>1863 q.eq("projectId", job.projectId).eq("userId", job.userId),1864 )1865 .take(USER_ERASURE_BATCH_SIZE),1866 ctx.db1867 .query("purchases")1868 .withIndex("by_project_and_app_user", (q) =>1869 q.eq("projectId", job.projectId).eq("appUserId", job.userId),1870 )1871 .take(USER_ERASURE_BATCH_SIZE),1872 ctx.db1873 .query("purchaseIssues")1874 .withIndex("by_project_and_appUserId_and_lastSeenAt", (q) =>1875 q.eq("projectId", job.projectId).eq("appUserId", job.userId),1876 )1877 .take(USER_ERASURE_BATCH_SIZE),1878 ]);1879 1880 for (const subscription of subscriptions) {1881 await ctx.db.patch(subscription._id, {1882 userId: undefined,1883 accountErased: true,1884 resyncReportedActive: undefined,1885 });1886 }1887 for (const purchase of purchases) {1888 await ctx.db.patch(purchase._id, {1889 appUserId: undefined,1890 accountErased: true,1891 });1892 }1893 for (const issue of issues) {1894 await ctx.db.patch(issue._id, { appUserId: undefined });1895 }1896 let commerceEventsErasedThisPage = 0;1897 let waitingForClaimedDelivery = false;1898 for (const event of commerceEvents) {1899 const deliveries = await ctx.db1900 .query("outboundDeliveries")1901 .withIndex("by_event", (q) => q.eq("eventId", event._id))1902 .collect();1903 if (deliveries.some((delivery) => delivery.status === "delivering")) {1904 waitingForClaimedDelivery = true;1905 continue;1906 }1907 if (event.eventType.startsWith("entitlement.")) {1908 for (const delivery of deliveries) await ctx.db.delete(delivery._id);1909 await ctx.db.delete(event._id);1910 } else {1911 await ctx.db.patch(event._id, { userId: undefined });1912 }1913 commerceEventsErasedThisPage += 1;1914 }1915 1916 const subscriptionsErased = job.subscriptionsErased + subscriptions.length;1917 const commerceEventsErased =1918 job.commerceEventsErased + commerceEventsErasedThisPage;1919 const done =1920 !waitingForClaimedDelivery &&1921 subscriptions.length < USER_ERASURE_BATCH_SIZE &&1922 purchases.length < USER_ERASURE_BATCH_SIZE &&1923 issues.length < USER_ERASURE_BATCH_SIZE &&1924 commerceEvents.length < USER_ERASURE_BATCH_SIZE;1925 1926 if (done) {1927 await ctx.db.patch(jobId, {1928 userId: undefined,1929 status: "completed",1930 subscriptionsErased,1931 commerceEventsErased,1932 updatedAt: now,1933 completedAt: now,1934 });1935 } else {1936 await ctx.db.patch(jobId, {1937 subscriptionsErased,1938 commerceEventsErased,1939 updatedAt: now,1940 });1941 await ctx.scheduler.runAfter(1942 waitingForClaimedDelivery ? 1_000 : 0,1943 internal.subscriptions.internal.drainSubscriptionUserErasureJob,1944 { jobId },1945 );1946 }1947 1948 return { done, subscriptionsErased, commerceEventsErased };1949}1950 1951export const drainSubscriptionUserErasureJob = internalMutation({1952 args: { jobId: v.id("subscriptionUserErasureJobs") },1953 handler: async (ctx, args) =>1954 drainSubscriptionUserErasurePage(ctx, args.jobId),1955});1956 1957export const resumeSubscriptionUserErasureJobs = internalMutation({1958 args: {},1959 returns: v.object({ scheduled: v.number() }),1960 handler: async (ctx) => {1961 const staleBefore = Date.now() - USER_ERASURE_STALE_MS;1962 const [queued, stale] = await Promise.all([1963 ctx.db1964 .query("subscriptionUserErasureJobs")1965 .withIndex("by_status_and_updated", (q) => q.eq("status", "queued"))1966 .take(10),1967 ctx.db1968 .query("subscriptionUserErasureJobs")1969 .withIndex("by_status_and_updated", (q) =>1970 q.eq("status", "running").lt("updatedAt", staleBefore),1971 )1972 .take(10),1973 ]);1974 const jobs = [...queued, ...stale].slice(0, 10);1975 for (const job of jobs) {1976 await ctx.scheduler.runAfter(1977 0,1978 internal.subscriptions.internal.drainSubscriptionUserErasureJob,1979 { jobId: job._id },1980 );1981 }1982 return { scheduled: jobs.length };1983 },1984});1985 1986export async function pruneCompletedSubscriptionUserErasureJobsHandler(1987 ctx: MutationCtx,1988): Promise<{ pruned: number }> {1989 const completedBefore = Date.now() - USER_ERASURE_JOB_RETENTION_MS;1990 const completed = await ctx.db1991 .query("subscriptionUserErasureJobs")1992 .withIndex("by_status_and_updated", (q) =>1993 q.eq("status", "completed").lt("updatedAt", completedBefore),1994 )1995 .take(USER_ERASURE_PRUNER_BATCH_SIZE);1996 1997 for (const job of completed) {1998 await ctx.db.delete(job._id);1999 }2000 if (completed.length === USER_ERASURE_PRUNER_BATCH_SIZE) {2001 await ctx.scheduler.runAfter(2002 0,2003 internal.subscriptions.internal.pruneCompletedSubscriptionUserErasureJobs,2004 {},2005 );2006 }2007 return { pruned: completed.length };2008}2009 2010export const pruneCompletedSubscriptionUserErasureJobs = internalMutation({2011 args: {},2012 returns: v.object({ pruned: v.number() }),2013 handler: async (ctx) => pruneCompletedSubscriptionUserErasureJobsHandler(ctx),2014});2015