1import {2 internalMutation,3 internalQuery,4 type MutationCtx,5 type QueryCtx,6} from "../_generated/server";7import { ConvexError, v } from "convex/values";8import type { Doc, Id } from "../_generated/dataModel";9import { internal } from "../_generated/api";10 11import { HarmonizedPurchaseState } from "../purchases/purchaseState";12import {13 applySubscriptionTransition,14 entitlementActive,15 type CurrentSubscription,16 type SubscriptionEventInput,17} from "./stateMachine";18import { applyStatsTransition, statsContributionFor } from "./stats";19import { assertProjectWritable } from "../projects/writable";20import { isValidSubscriptionUserId } from "./limits";21import { isUserErasureRequested } from "./erasure";22 23export const USER_ERASURE_BATCH_SIZE = 100;24export const USER_ERASURE_JOB_RETENTION_MS = 7 * 24 * 60 * 60 * 1_000;25const USER_ERASURE_STALE_MS = 10 * 60 * 1_000;26const USER_ERASURE_PRUNER_BATCH_SIZE = 100;27 28const subscriptionPlatformValidator = v.union(29 v.literal("IOS"),30 v.literal("Android"),31);32 33type RawEventInput = Pick<34 Doc<"webhookEvents">,35 | "type"36 | "productId"37 | "subscriptionState"38 | "expiresAt"39 | "renewsAt"40 | "willRenew"41 | "cancellationReason"42 | "currency"43 | "priceAmountMicros"44 | "platform"45 | "effectiveImmediately"46> & { purchaseToken: string };47type SubscriptionState = Doc<"subscriptions">["state"];48type SubscriptionCancellationReason = NonNullable<49 Doc<"subscriptions">["cancellationReason"]50>;51type SubscriptionPlatform = Doc<"subscriptions">["platform"];52 53export type VerifiedSubscriptionInput = {54 platform: SubscriptionPlatform;55 productId: string;56 purchaseState: HarmonizedPurchaseState;57 subscriptionState?: string;58 expiresAt?: number;59 renewsAt?: number;60 willRenew?: boolean;61 currency?: string;62 priceAmountMicros?: number;63 revocationReasonIOS?: number;64};65 66export type VerifiedSubscriptionSnapshot = {67 productId: string;68 state: SubscriptionState;69 expiresAt?: number;70 renewsAt?: number;71 willRenew?: boolean;72 cancellationReason?: SubscriptionCancellationReason;73 clearCancellationReason?: boolean;74 currency?: string;75 priceAmountMicros?: number;76};77 78type ExistingSubscriptionSnapshotFields = Pick<79 Doc<"subscriptions">,80 | "expiresAt"81 | "renewsAt"82 | "willRenew"83 | "cancellationReason"84 | "currency"85 | "priceAmountMicros"86>;87 88type RecordVerifiedSubscriptionArgs = {89 projectId: Id<"projects">;90 platform: SubscriptionPlatform;91 purchaseToken: string;92 productId: string;93 purchaseState: string;94 subscriptionState?: string;95 expiresAt?: number;96 renewsAt?: number;97 willRenew?: boolean;98 currency?: string;99 priceAmountMicros?: number;100 revocationReasonIOS?: number;101};102 103type BindSubscriptionToUserArgs = {104 projectId: Id<"projects">;105 purchaseToken: string;106 userId: string;107};108 109interface PersistSubscriptionSnapshotArgs {110 projectId: Id<"projects">;111 platform: SubscriptionPlatform;112 purchaseToken: string;113 existing: Doc<"subscriptions"> | null;114 next: NonNullable<CurrentSubscription>;115 now: number;116 lastEvent?: Pick<117 Doc<"webhookEvents">,118 | "_id"119 | "_creationTime"120 | "type"121 | "occurredAt"122 | "sourceNotificationId"123 | "environment"124 | "productId"125 | "applicationId"126 | "transactionId"127 | "originalTransactionId"128 | "currency"129 | "priceAmountMicros"130 >;131}132 133interface ApplySubscriptionEventArgs {134 projectId: Id<"projects">;135 eventId: Id<"webhookEvents">;136}137 138interface ApplySubscriptionEventResult {139 transition: string | null;140 active: boolean;141 subscriptionId?: Id<"subscriptions">;142}143 144// Apply a webhook event to the canonical `subscriptions` table. The event's145// durable appliedAt marker is committed in the same Convex transaction as146// the subscription and stats writes, so both retry gaps are safe: a crash147// before this mutation can be repaired, while any event this mutation already148// processed can never be replayed after a newer lastEventId replaces it.149import { emitCommerceEvent } from "../commerce/internal";150 151export const applySubscriptionEvent = internalMutation({152 args: {153 projectId: v.id("projects"),154 eventId: v.id("webhookEvents"),155 },156 returns: v.object({157 transition: v.union(v.string(), v.null()),158 active: v.boolean(),159 subscriptionId: v.optional(v.id("subscriptions")),160 }),161 handler: async (ctx, args) => applySubscriptionEventHandler(ctx, args),162});163 164export async function applySubscriptionEventHandler(165 ctx: MutationCtx,166 args: ApplySubscriptionEventArgs,167): Promise<ApplySubscriptionEventResult> {168 const project = await assertProjectWritable(ctx, args.projectId);169 const storedEvent = await ctx.db.get(args.eventId);170 if (!storedEvent || storedEvent.projectId !== args.projectId) {171 throw new Error("Webhook event not found for project");172 }173 174 const now = Date.now();175 if (storedEvent.productKind === "one_time") {176 if (storedEvent.appliedAt === undefined) {177 await ctx.db.patch(storedEvent._id, { appliedAt: now });178 }179 return { transition: null, active: false };180 }181 if (!storedEvent.purchaseToken) {182 if (storedEvent.appliedAt === undefined) {183 await ctx.db.patch(storedEvent._id, { appliedAt: now });184 }185 return { transition: null, active: false };186 }187 188 const supersedingResolution = await findSupersedingSubscription(189 ctx,190 args.projectId,191 storedEvent.purchaseToken,192 );193 const existingByCurrentToken = supersedingResolution.aliased194 ? null195 : await findSubscriptionByToken(196 ctx,197 args.projectId,198 storedEvent.purchaseToken,199 );200 if (supersedingResolution.aliased) {201 const supersedingSubscription = supersedingResolution.subscription;202 const aliasedTransition =203 storedEvent.type === "PurchaseRefunded"204 ? ("Refunded" as const)205 : storedEvent.type === "SubscriptionRevoked"206 ? ("Revoked" as const)207 : null;208 const active = supersedingSubscription209 ? isActive(supersedingSubscription, now)210 : false;211 const firstApplication = storedEvent.appliedAt === undefined;212 if (firstApplication) {213 if (supersedingSubscription && aliasedTransition) {214 const predecessorProductId =215 storedEvent.productId ?? supersedingResolution.productId;216 await emitCommerceEvent(ctx, {217 projectId: args.projectId,218 transition: aliasedTransition,219 active: false,220 previouslyActive: false,221 sourceEvent: {222 ...storedEvent,223 productId: predecessorProductId,224 },225 ...(predecessorProductId226 ? {227 subscription: {228 state: aliasedTransition,229 productId: predecessorProductId,230 ...(supersedingSubscription.userId231 ? { userId: supersedingSubscription.userId }232 : {}),233 },234 }235 : {}),236 });237 }238 await ctx.db.patch(storedEvent._id, { appliedAt: now });239 }240 return {241 transition: firstApplication ? aliasedTransition : null,242 active,243 ...(supersedingSubscription244 ? { subscriptionId: supersedingSubscription._id }245 : {}),246 };247 }248 const linkedResolution = storedEvent.linkedPurchaseToken249 ? await findSupersedingSubscription(250 ctx,251 args.projectId,252 storedEvent.linkedPurchaseToken,253 )254 : ({ aliased: false } as const);255 const linkedExact =256 storedEvent.linkedPurchaseToken && !linkedResolution.aliased257 ? await findSubscriptionByToken(258 ctx,259 args.projectId,260 storedEvent.linkedPurchaseToken,261 )262 : null;263 if (linkedResolution.aliased && !linkedResolution.subscription) {264 if (storedEvent.appliedAt === undefined) {265 await ctx.db.patch(storedEvent._id, { appliedAt: now });266 }267 return { transition: null, active: false };268 }269 const linkedExisting =270 linkedExact ??271 (linkedResolution.aliased ? linkedResolution.subscription : null);272 let existing = preferredReplacementSnapshot(273 existingByCurrentToken,274 linkedExisting,275 );276 const priorStoreSnapshot = existingByCurrentToken?.lastEventId277 ? existingByCurrentToken278 : linkedExisting?.lastEventId279 ? linkedExisting280 : null;281 if (storedEvent.appliedAt !== undefined) {282 return {283 transition: null,284 active: existing ? isActive(existing, now) : false,285 ...(existing ? { subscriptionId: existing._id } : {}),286 };287 }288 289 // A linked-token event can arrive after both tokens already have rows. Keep290 // current-token state, merge predecessor identity/history, and remove the291 // duplicate contribution before applying same-token ordering.292 if (293 existingByCurrentToken &&294 linkedExisting &&295 existingByCurrentToken._id !== linkedExisting._id296 ) {297 // Two accounts on one subscription is a conflict whether or not either row298 // carries an erasure marker; a marker never decides which of them wins.299 if (300 existingByCurrentToken.userId &&301 linkedExisting.userId &&302 existingByCurrentToken.userId !== linkedExisting.userId303 ) {304 throw new ConvexError({305 code: "SUBSCRIPTION_USER_CONFLICT",306 message: "Linked Google purchase tokens belong to different users.",307 });308 }309 // The marker records that the row's own owner was erased. A live binding on310 // either row is a later, valid association, so it survives the merge and311 // the marker does not carry.312 const boundUserId =313 existingByCurrentToken.userId ?? linkedExisting.userId ?? undefined;314 const accountErased =315 !boundUserId &&316 (existingByCurrentToken.accountErased === true ||317 linkedExisting.accountErased === true);318 const survivor = preferredReplacementSnapshot(319 existingByCurrentToken,320 linkedExisting,321 )!;322 const removed =323 survivor._id === existingByCurrentToken._id324 ? linkedExisting325 : existingByCurrentToken;326 const removedPeriod = await fetchBillingPeriod(327 ctx,328 args.projectId,329 removed.platform,330 removed.productId,331 );332 await applyStatsTransition(333 ctx,334 args.projectId,335 statsContributionFor(removed, removedPeriod, now),336 null,337 );338 await ctx.db.delete(removed._id);339 existing = {340 ...survivor,341 purchaseToken: storedEvent.purchaseToken,342 accountErased: accountErased || undefined,343 userId: boundUserId,344 startedAt: Math.min(survivor.startedAt, removed.startedAt),345 };346 await ctx.db.patch(survivor._id, {347 purchaseToken: existing.purchaseToken,348 accountErased: existing.accountErased,349 userId: existing.userId,350 startedAt: existing.startedAt,351 updatedAt: now,352 });353 await recordSubscriptionTokenAlias(ctx, {354 projectId: args.projectId,355 purchaseToken: removed.purchaseToken,356 successorPurchaseToken: existing.purchaseToken,357 predecessorProductId: removed.productId,358 now,359 });360 }361 362 // Compare with the last persisted gate; today's clock may already have expired it.363 const previouslyActive = priorStoreSnapshot364 ? isActive(priorStoreSnapshot, priorStoreSnapshot.updatedAt)365 : false;366 const noOpResult = (): ApplySubscriptionEventResult => ({367 transition: null,368 active: existing ? isActive(existing, now) : false,369 ...(existing ? { subscriptionId: existing._id } : {}),370 });371 372 // Store timestamps only order events for the same purchase token. A linked373 // predecessor can expire after its replacement became active.374 const orderingExisting = existingByCurrentToken;375 376 // Rollout compatibility for events written before appliedAt existed. The377 // current last event proves itself applied; an event older than the current378 // last event must be marked handled without being allowed to roll state379 // backwards. Store timestamps are only millisecond-precision, so ingestion380 // order breaks ties between distinct same-timestamp events. A recorded-but-381 // unapplied newest event still falls through and repairs the original382 // action/mutation gap.383 if (384 orderingExisting?.lastEventSourceNotificationId ===385 storedEvent.sourceNotificationId386 ) {387 await ctx.db.patch(storedEvent._id, { appliedAt: now });388 return noOpResult();389 }390 if (orderingExisting?.lastEventOccurredAt !== undefined) {391 const stale =392 orderingExisting.lastEventOccurredAt > storedEvent.occurredAt ||393 (orderingExisting.lastEventOccurredAt === storedEvent.occurredAt &&394 (orderingExisting.lastEventCreationTime ?? 0) >395 storedEvent._creationTime);396 if (stale) {397 await ctx.db.patch(storedEvent._id, { appliedAt: now });398 return noOpResult();399 }400 } else if (orderingExisting?.lastEventId) {401 if (orderingExisting.lastEventId === args.eventId) {402 await ctx.db.patch(storedEvent._id, { appliedAt: now });403 return noOpResult();404 }405 const lastEvent = await ctx.db.get(orderingExisting.lastEventId);406 if (407 lastEvent?.projectId === args.projectId &&408 lastEvent.purchaseToken === storedEvent.purchaseToken &&409 lastEvent.platform === storedEvent.platform &&410 (lastEvent.occurredAt > storedEvent.occurredAt ||411 (lastEvent.occurredAt === storedEvent.occurredAt &&412 lastEvent._creationTime > storedEvent._creationTime))413 ) {414 await ctx.db.patch(storedEvent._id, { appliedAt: now });415 return noOpResult();416 }417 }418 419 // Google subscription bundles can report ITEMS_CHANGED without identifying420 // which line item changed. Keep the raw event for operations, but never421 // overwrite the singular canonical product with an arbitrary bundle item.422 // Linked replacements still move the canonical row after the ordering guard.423 if (424 storedEvent.type === "SubscriptionProductChanged" &&425 !storedEvent.productId426 ) {427 if (existing) {428 const verifiedReplacementProductChanged =429 existingByCurrentToken?.lastEventId === undefined &&430 priorStoreSnapshot !== null &&431 priorStoreSnapshot.productId !== existing.productId;432 const subscriptionId = await persistSubscriptionSnapshot(ctx, {433 projectId: args.projectId,434 platform: existing.platform,435 purchaseToken: storedEvent.purchaseToken,436 existing,437 next: {438 state: existing.state,439 productId: existing.productId,440 expiresAt: existing.expiresAt,441 renewsAt: existing.renewsAt,442 willRenew: existing.willRenew,443 cancellationReason: existing.cancellationReason,444 currency: existing.currency,445 priceAmountMicros: existing.priceAmountMicros,446 },447 now,448 lastEvent: storedEvent,449 });450 await recordSubscriptionTokenAlias(ctx, {451 projectId: args.projectId,452 purchaseToken: existing.purchaseToken,453 successorPurchaseToken: storedEvent.purchaseToken,454 predecessorProductId: existing.productId,455 now,456 });457 await recordSubscriptionTokenAlias(ctx, {458 projectId: args.projectId,459 purchaseToken: storedEvent.linkedPurchaseToken,460 successorPurchaseToken: storedEvent.purchaseToken,461 predecessorProductId:462 priorStoreSnapshot?.productId ?? existing.productId,463 now,464 });465 await ctx.db.patch(storedEvent._id, { appliedAt: now });466 const active = isActive(existing, now);467 if (verifiedReplacementProductChanged) {468 await emitCommerceEvent(ctx, {469 projectId: args.projectId,470 transition: "ProductChanged",471 active,472 previouslyActive: active,473 sourceEvent: storedEvent,474 subscriptionId,475 previousProductId: priorStoreSnapshot.productId,476 subscription: {477 state: existing.state,478 productId: existing.productId,479 ...(existing.expiresAt !== undefined480 ? { expiresAt: existing.expiresAt }481 : {}),482 ...(existing.renewsAt !== undefined483 ? { renewsAt: existing.renewsAt }484 : {}),485 ...(existing.willRenew !== undefined486 ? { willRenew: existing.willRenew }487 : {}),488 ...(existing.cancellationReason489 ? { cancellationReason: existing.cancellationReason }490 : {}),491 ...(existing.userId ? { userId: existing.userId } : {}),492 },493 });494 }495 return {496 transition: verifiedReplacementProductChanged ? "ProductChanged" : null,497 active,498 subscriptionId,499 };500 }501 await ctx.db.patch(storedEvent._id, { appliedAt: now });502 return noOpResult();503 }504 505 const current: CurrentSubscription = existing506 ? {507 state: existing.state,508 productId: existing.productId,509 expiresAt: existing.expiresAt,510 renewsAt: existing.renewsAt,511 willRenew: existing.willRenew,512 cancellationReason: existing.cancellationReason,513 currency: existing.currency,514 priceAmountMicros: existing.priceAmountMicros,515 }516 : null;517 const event: RawEventInput = {518 type: storedEvent.type,519 productId: storedEvent.productId,520 subscriptionState: storedEvent.subscriptionState,521 expiresAt: storedEvent.expiresAt,522 renewsAt: storedEvent.renewsAt,523 willRenew: storedEvent.willRenew,524 cancellationReason: storedEvent.cancellationReason,525 currency: storedEvent.currency,526 priceAmountMicros: storedEvent.priceAmountMicros,527 platform: storedEvent.platform,528 effectiveImmediately: storedEvent.effectiveImmediately,529 purchaseToken: storedEvent.purchaseToken,530 };531 const transition = applySubscriptionTransition(532 current,533 coerceEventInput(event),534 );535 const linkedStartedOnActivePredecessor =536 storedEvent.type === "SubscriptionStarted" &&537 storedEvent.linkedPurchaseToken !== undefined &&538 priorStoreSnapshot !== null &&539 previouslyActive;540 const effectiveTransition = linkedStartedOnActivePredecessor541 ? transition.next?.productId !== priorStoreSnapshot.productId542 ? "ProductChanged"543 : priorStoreSnapshot.willRenew === false &&544 transition.next?.willRenew === true545 ? "Uncanceled"546 : transition.next?.expiresAt !== undefined &&547 priorStoreSnapshot.expiresAt !== undefined &&548 transition.next.expiresAt > priorStoreSnapshot.expiresAt549 ? "Renewed"550 : null551 : transition.transition;552 const firstStoreEventAfterVerification =553 storedEvent.type === "SubscriptionStarted" &&554 existing !== null &&555 existing.lastEventId === undefined &&556 priorStoreSnapshot === null;557 // Computed before persistSubscriptionSnapshot stamps lastEvent* onto the558 // row: a record bootstrapped by receipt verification has no store history,559 // so a price change or deferral arriving as its first store event has no560 // baseline to describe. The mapping vectors pin these to no event.561 const priceOrDeferralWithoutBaseline =562 existing !== null &&563 existing.lastEventId === undefined &&564 priorStoreSnapshot === null &&565 (effectiveTransition === "PriceChanged" ||566 effectiveTransition === "Deferred");567 const previousProductId =568 priorStoreSnapshot?.productId ?? existing?.productId;569 570 if (!transition.next) {571 await ctx.db.patch(storedEvent._id, { appliedAt: now });572 return {573 transition: transition.transition ?? null,574 active: false,575 ...(existing ? { subscriptionId: existing._id } : {}),576 };577 }578 579 const subscriptionId = await persistSubscriptionSnapshot(ctx, {580 projectId: args.projectId,581 platform: event.platform,582 purchaseToken: event.purchaseToken,583 existing,584 next: transition.next,585 now,586 lastEvent: storedEvent,587 });588 await recordSubscriptionTokenAlias(ctx, {589 projectId: args.projectId,590 purchaseToken: existing?.purchaseToken,591 successorPurchaseToken: storedEvent.purchaseToken,592 predecessorProductId: existing?.productId,593 now,594 });595 await recordSubscriptionTokenAlias(ctx, {596 projectId: args.projectId,597 purchaseToken: storedEvent.linkedPurchaseToken,598 successorPurchaseToken: storedEvent.purchaseToken,599 predecessorProductId: priorStoreSnapshot?.productId ?? existing?.productId,600 now,601 });602 await ctx.db.patch(storedEvent._id, { appliedAt: now });603 const active = entitlementActive(transition.next, now);604 // A record bootstrapped by receipt verification has no store history, so a605 // price change or deferral arriving as its first store event has no606 // baseline to describe. The mapping vectors pin these to no event.607 const commerceTransition = linkedStartedOnActivePredecessor608 ? effectiveTransition609 : firstStoreEventAfterVerification610 ? "Started"611 : priceOrDeferralWithoutBaseline612 ? null613 : effectiveTransition;614 await emitCommerceEvent(ctx, {615 projectId: args.projectId,616 transition: commerceTransition ?? null,617 active,618 previouslyActive,619 sourceEvent: storedEvent,620 subscriptionId,621 ...(previousProductId && previousProductId !== transition.next.productId622 ? { previousProductId }623 : {}),624 subscription: {625 state: transition.next.state,626 productId: transition.next.productId,627 ...(transition.next.expiresAt !== undefined628 ? { expiresAt: transition.next.expiresAt }629 : {}),630 ...(transition.next.renewsAt !== undefined631 ? { renewsAt: transition.next.renewsAt }632 : {}),633 ...(transition.next.willRenew !== undefined634 ? { willRenew: transition.next.willRenew }635 : {}),636 ...(transition.next.cancellationReason637 ? { cancellationReason: transition.next.cancellationReason }638 : {}),639 ...(existing?.userId ? { userId: existing.userId } : {}),640 },641 });642 643 return {644 transition: effectiveTransition ?? null,645 active,646 subscriptionId,647 };648}649 650export function buildVerifiedSubscriptionSnapshot(651 input: VerifiedSubscriptionInput,652): VerifiedSubscriptionSnapshot | null {653 if (input.productId.length === 0 || input.productId === "unknown") {654 return null;655 }656 if (input.purchaseState === HarmonizedPurchaseState.INAUTHENTIC) return null;657 658 const base: Pick<659 VerifiedSubscriptionSnapshot,660 | "productId"661 | "expiresAt"662 | "renewsAt"663 | "willRenew"664 | "currency"665 | "priceAmountMicros"666 > = {667 productId: input.productId,668 expiresAt: input.expiresAt,669 renewsAt: input.renewsAt,670 willRenew: input.willRenew,671 currency: input.currency,672 priceAmountMicros: input.priceAmountMicros,673 };674 675 if (input.platform === "IOS") {676 switch (input.purchaseState) {677 case HarmonizedPurchaseState.ENTITLED:678 return {679 ...base,680 state: "Active",681 cancellationReason: undefined,682 clearCancellationReason: true,683 };684 case HarmonizedPurchaseState.EXPIRED:685 return {686 ...base,687 state: "Expired",688 willRenew: false,689 };690 case HarmonizedPurchaseState.CANCELED:691 return {692 ...base,693 state: "Revoked",694 willRenew: false,695 // Apple's revocationReason field covers a transaction "refunded696 // ... or revoked from family sharing": value 1 is unambiguously a697 // refund, while 0 also covers Family Sharing loss. Mirroring the698 // webhook REVOKE policy, only the unambiguous value asserts money699 // moved back; webhook REFUND notifications carry the rest.700 ...(input.revocationReasonIOS === 1701 ? { cancellationReason: "Refunded" as const }702 : {}),703 };704 case HarmonizedPurchaseState.PENDING_ACKNOWLEDGMENT:705 case HarmonizedPurchaseState.PENDING:706 case HarmonizedPurchaseState.UNKNOWN:707 return {708 ...base,709 state: "Unknown",710 };711 case HarmonizedPurchaseState.READY_TO_CONSUME:712 case HarmonizedPurchaseState.CONSUMED:713 return null;714 }715 }716 717 switch (input.purchaseState) {718 case HarmonizedPurchaseState.UNKNOWN:719 return {720 ...base,721 state: "Unknown",722 };723 case HarmonizedPurchaseState.EXPIRED:724 return {725 ...base,726 state: "Expired",727 willRenew: false,728 };729 case HarmonizedPurchaseState.READY_TO_CONSUME:730 case HarmonizedPurchaseState.CONSUMED:731 return null;732 case HarmonizedPurchaseState.ENTITLED:733 case HarmonizedPurchaseState.CANCELED:734 break;735 }736 737 switch (input.subscriptionState?.toUpperCase()) {738 case "SUBSCRIPTION_STATE_ACTIVE":739 return {740 ...base,741 state: "Active",742 willRenew: input.willRenew ?? true,743 cancellationReason: undefined,744 clearCancellationReason: true,745 };746 case "SUBSCRIPTION_STATE_CANCELED":747 return {748 ...base,749 state: "Active",750 willRenew: false,751 };752 case "SUBSCRIPTION_STATE_IN_GRACE_PERIOD":753 return {754 ...base,755 state: "InGracePeriod",756 willRenew: input.willRenew ?? true,757 cancellationReason: undefined,758 clearCancellationReason: true,759 };760 case "SUBSCRIPTION_STATE_ON_HOLD":761 return {762 ...base,763 state: "InBillingRetry",764 cancellationReason: "BillingError",765 };766 case "SUBSCRIPTION_STATE_PAUSED":767 return {768 ...base,769 state: "Paused",770 willRenew: false,771 };772 case "SUBSCRIPTION_STATE_EXPIRED":773 return {774 ...base,775 state: "Expired",776 willRenew: false,777 };778 case "SUBSCRIPTION_STATE_PENDING":779 return {780 ...base,781 state: "Unknown",782 };783 }784 785 return {786 ...base,787 state:788 input.purchaseState === HarmonizedPurchaseState.ENTITLED ||789 input.purchaseState === HarmonizedPurchaseState.PENDING_ACKNOWLEDGMENT790 ? "Active"791 : "Unknown",792 };793}794 795export function mergeVerifiedSubscriptionSnapshot(796 existing: ExistingSubscriptionSnapshotFields | null,797 snapshot: VerifiedSubscriptionSnapshot,798): VerifiedSubscriptionSnapshot {799 if (!existing) return snapshot;800 801 return {802 ...snapshot,803 expiresAt: snapshot.expiresAt ?? existing.expiresAt,804 renewsAt:805 snapshot.willRenew === false806 ? undefined807 : (snapshot.renewsAt ?? existing.renewsAt),808 willRenew: snapshot.willRenew ?? existing.willRenew,809 cancellationReason:810 snapshot.cancellationReason !== undefined ||811 snapshot.clearCancellationReason812 ? snapshot.cancellationReason813 : existing.cancellationReason,814 currency: snapshot.currency ?? existing.currency,815 priceAmountMicros: snapshot.priceAmountMicros ?? existing.priceAmountMicros,816 };817}818 819// Receipt verification is the synchronous bootstrap path for820// subscriptions. Webhooks keep lifecycle state fresh later, but a821// successful verify must be enough for SDK clients to bind the just-822// purchased token to their app user.823export const recordVerifiedSubscription = internalMutation({824 args: {825 projectId: v.id("projects"),826 platform: subscriptionPlatformValidator,827 purchaseToken: v.string(),828 productId: v.string(),829 purchaseState: v.string(),830 subscriptionState: v.optional(v.string()),831 expiresAt: v.optional(v.number()),832 renewsAt: v.optional(v.number()),833 willRenew: v.optional(v.boolean()),834 currency: v.optional(v.string()),835 priceAmountMicros: v.optional(v.number()),836 revocationReasonIOS: v.optional(v.number()),837 },838 returns: v.union(v.id("subscriptions"), v.null()),839 handler: async (ctx, args) => recordVerifiedSubscriptionHandler(ctx, args),840});841 842export async function recordVerifiedSubscriptionHandler(843 ctx: MutationCtx,844 args: RecordVerifiedSubscriptionArgs,845): Promise<Id<"subscriptions"> | null> {846 await assertProjectWritable(ctx, args.projectId);847 const snapshot = buildVerifiedSubscriptionSnapshot({848 platform: args.platform,849 productId: args.productId,850 purchaseState: normalizeHarmonizedPurchaseState(args.purchaseState),851 subscriptionState: args.subscriptionState,852 expiresAt: args.expiresAt,853 renewsAt: args.renewsAt,854 willRenew: args.willRenew,855 currency: args.currency,856 priceAmountMicros: args.priceAmountMicros,857 revocationReasonIOS: args.revocationReasonIOS,858 });859 if (!snapshot) return null;860 861 const supersedingResolution = await findSupersedingSubscription(862 ctx,863 args.projectId,864 args.purchaseToken,865 );866 if (supersedingResolution.aliased) {867 return supersedingResolution.subscription?._id ?? null;868 }869 const existing = await findSubscriptionByToken(870 ctx,871 args.projectId,872 args.purchaseToken,873 );874 // Verification bootstraps a token before server notifications arrive. Once875 // a store event governs the row, a client can replay an older but still-valid876 // transaction; without comparable ordering metadata that snapshot must not877 // roll canonical product, state, or expiry backward.878 if (existing?.lastEventId) return existing._id;879 const now = Date.now();880 881 const next = mergeVerifiedSubscriptionSnapshot(existing, snapshot);882 return persistSubscriptionSnapshot(ctx, {883 projectId: args.projectId,884 platform: args.platform,885 purchaseToken: args.purchaseToken,886 existing,887 next,888 now,889 });890}891 892async function persistSubscriptionSnapshot(893 ctx: MutationCtx,894 args: PersistSubscriptionSnapshotArgs,895): Promise<Id<"subscriptions">> {896 // Stats deltas must compare the old row against the old catalog entry897 // and the new row against the new one; otherwise product/platform changes898 // subtract the wrong MRR bucket.899 const afterBillingPeriod = await fetchBillingPeriod(900 ctx,901 args.projectId,902 args.platform,903 args.next.productId,904 );905 const beforeBillingPeriod =906 args.existing &&907 (args.existing.productId !== args.next.productId ||908 args.existing.platform !== args.platform)909 ? await fetchBillingPeriod(910 ctx,911 args.projectId,912 args.existing.platform,913 args.existing.productId,914 )915 : afterBillingPeriod;916 const beforeContribution = args.existing917 ? statsContributionFor(args.existing, beforeBillingPeriod, args.now)918 : null;919 920 const row = {921 purchaseToken: args.purchaseToken,922 productKind: "subscription" as const,923 ...(args.existing?.userId ? { userId: args.existing.userId } : {}),924 productId: args.next.productId,925 platform: args.platform,926 state: args.next.state,927 expiresAt: args.next.expiresAt,928 renewsAt: args.next.renewsAt,929 willRenew: args.next.willRenew,930 cancellationReason: args.next.cancellationReason,931 currency: args.next.currency,932 priceAmountMicros: args.next.priceAmountMicros,933 updatedAt: args.now,934 ...(args.lastEvent935 ? {936 lastEventId: args.lastEvent._id,937 lastEventOccurredAt: args.lastEvent.occurredAt,938 lastEventCreationTime: args.lastEvent._creationTime,939 lastEventSourceNotificationId: args.lastEvent.sourceNotificationId,940 lastEventSource: {941 type: args.lastEvent.type,942 environment: args.lastEvent.environment,943 productId: args.lastEvent.productId,944 applicationId: args.lastEvent.applicationId,945 transactionId: args.lastEvent.transactionId,946 originalTransactionId: args.lastEvent.originalTransactionId,947 currency: args.lastEvent.currency,948 priceAmountMicros: args.lastEvent.priceAmountMicros,949 },950 }951 : {}),952 };953 954 const subscriptionId = args.existing955 ? args.existing._id956 : await ctx.db.insert("subscriptions", {957 projectId: args.projectId,958 startedAt: args.now,959 ...row,960 });961 962 if (args.existing) {963 await ctx.db.patch(args.existing._id, row);964 }965 966 const updatedRow = (await ctx.db.get(subscriptionId))!;967 const afterContribution = statsContributionFor(968 updatedRow,969 afterBillingPeriod,970 args.now,971 );972 await applyStatsTransition(973 ctx,974 args.projectId,975 beforeContribution,976 afterContribution,977 );978 979 return subscriptionId;980}981 982function findSubscriptionByToken(983 ctx: Pick<QueryCtx, "db">,984 projectId: Id<"projects">,985 purchaseToken: string,986): Promise<Doc<"subscriptions"> | null> {987 return ctx.db988 .query("subscriptions")989 .withIndex("by_project_and_token", (q) =>990 q.eq("projectId", projectId).eq("purchaseToken", purchaseToken),991 )992 .unique();993}994 995const MAX_SUBSCRIPTION_TOKEN_ALIAS_HOPS = 64;996 997type SupersedingSubscriptionResolution =998 | { aliased: false }999 | {1000 aliased: true;1001 subscription: Doc<"subscriptions"> | null;1002 productId?: string;1003 };1004 1005async function findSupersedingSubscription(1006 ctx: Pick<QueryCtx, "db">,1007 projectId: Id<"projects">,1008 purchaseToken: string,1009): Promise<SupersedingSubscriptionResolution> {1010 let token = purchaseToken;1011 let productId: string | undefined;1012 const seen = new Set<string>([token]);1013 for (let hop = 0; hop <= MAX_SUBSCRIPTION_TOKEN_ALIAS_HOPS; hop += 1) {1014 const alias = await ctx.db1015 .query("subscriptionTokenAliases")1016 .withIndex("by_project_and_token", (q) =>1017 q.eq("projectId", projectId).eq("purchaseToken", token),1018 )1019 .unique();1020 if (!alias) {1021 return hop === 01022 ? { aliased: false }1023 : {1024 aliased: true,1025 subscription: await findSubscriptionByToken(ctx, projectId, token),1026 productId,1027 };1028 }1029 if (hop === 0) productId = alias.predecessorProductId;1030 if (hop === MAX_SUBSCRIPTION_TOKEN_ALIAS_HOPS) {1031 return { aliased: true, subscription: null, productId };1032 }1033 if (seen.has(alias.successorPurchaseToken)) {1034 return { aliased: true, subscription: null, productId };1035 }1036 token = alias.successorPurchaseToken;1037 seen.add(token);1038 }1039 return { aliased: true, subscription: null, productId };1040}1041 1042async function recordSubscriptionTokenAlias(1043 ctx: MutationCtx,1044 args: {1045 projectId: Id<"projects">;1046 purchaseToken?: string;1047 successorPurchaseToken: string;1048 predecessorProductId?: string;1049 now: number;1050 },1051): Promise<void> {1052 if (1053 !args.purchaseToken ||1054 args.purchaseToken === args.successorPurchaseToken1055 ) {1056 return;1057 }1058 const existing = await ctx.db1059 .query("subscriptionTokenAliases")1060 .withIndex("by_project_and_token", (q) =>1061 q1062 .eq("projectId", args.projectId)1063 .eq("purchaseToken", args.purchaseToken as string),1064 )1065 .unique();1066 if (existing) {1067 if (existing.successorPurchaseToken !== args.successorPurchaseToken) {1068 await ctx.db.patch(existing._id, {1069 successorPurchaseToken: args.successorPurchaseToken,1070 predecessorProductId:1071 existing.predecessorProductId ?? args.predecessorProductId,1072 updatedAt: args.now,1073 });1074 }1075 return;1076 }1077 await ctx.db.insert("subscriptionTokenAliases", {1078 projectId: args.projectId,1079 purchaseToken: args.purchaseToken,1080 successorPurchaseToken: args.successorPurchaseToken,1081 predecessorProductId: args.predecessorProductId,1082 createdAt: args.now,1083 updatedAt: args.now,1084 });1085}1086 1087function preferredReplacementSnapshot(1088 current: Doc<"subscriptions"> | null,1089 linked: Doc<"subscriptions"> | null,1090): Doc<"subscriptions"> | null {1091 if (!current) return linked;1092 // Store timestamps from predecessor and replacement tokens are not a total1093 // order: the predecessor can expire after the replacement becomes active.1094 // Current-token state therefore wins even when it came from verification;1095 // user identity and start history are merged separately during reconciliation.1096 return current;1097}1098 1099export const getSourceProductIdByToken = internalQuery({1100 args: {1101 projectId: v.id("projects"),1102 purchaseToken: v.string(),1103 },1104 returns: v.union(v.string(), v.null()),1105 handler: async (ctx, args) => getSourceProductIdByTokenHandler(ctx, args),1106});1107 1108export async function getSourceProductIdByTokenHandler(1109 ctx: Pick<QueryCtx, "db">,1110 args: { projectId: Id<"projects">; purchaseToken: string },1111): Promise<string | null> {1112 const resolution = await findSupersedingSubscription(1113 ctx,1114 args.projectId,1115 args.purchaseToken,1116 );1117 if (resolution.aliased) return resolution.productId ?? null;1118 const subscription = await findSubscriptionByToken(1119 ctx,1120 args.projectId,1121 args.purchaseToken,1122 );1123 return subscription?.productId ?? null;1124}1125 1126export const getCurrentProductIdByToken = internalQuery({1127 args: {1128 projectId: v.id("projects"),1129 purchaseToken: v.string(),1130 },1131 returns: v.union(v.string(), v.null()),1132 handler: async (ctx, args) => getCurrentProductIdByTokenHandler(ctx, args),1133});1134 1135export async function getCurrentProductIdByTokenHandler(1136 ctx: Pick<QueryCtx, "db">,1137 args: { projectId: Id<"projects">; purchaseToken: string },1138): Promise<string | null> {1139 const resolution = await findSupersedingSubscription(1140 ctx,1141 args.projectId,1142 args.purchaseToken,1143 );1144 if (resolution.aliased) return resolution.subscription?.productId ?? null;1145 const subscription = await findSubscriptionByToken(1146 ctx,1147 args.projectId,1148 args.purchaseToken,1149 );1150 return subscription?.productId ?? null;1151}1152 1153// Look up a product's billing period from the kit-side catalog. We1154// Look up the row for the EXACT (platform, productId) — `products` is1155// keyed by (projectId, platform, productId) precisely because the1156// same SKU can exist on both stores with different billing periods.1157// Earlier behaviour preferred iOS over Android by walking both1158// platforms, which made an Android subscription inherit the iOS1159// period when those rows diverged and skewed `mrrMicros` on both the1160// incremental delta and the next recompute (PR #1241161// (https://github.com/hyodotdev/openiap/pull/124) review). Returns1162// undefined when the product isn't tracked or has no billingPeriod —1163// monthlyMicrosForSub treats that as a P1M fallback.1164async function fetchBillingPeriod(1165 ctx: MutationCtx,1166 projectId: Id<"projects">,1167 platform: SubscriptionPlatform,1168 productId: string,1169): Promise<string | undefined> {1170 const product = await ctx.db1171 .query("products")1172 .withIndex("by_project_and_platform_and_product", (q) =>1173 q1174 .eq("projectId", projectId)1175 .eq("platform", platform)1176 .eq("productId", productId),1177 )1178 .unique();1179 return product?.billingPeriod ?? undefined;1180}1181 1182function coerceEventInput(raw: RawEventInput): SubscriptionEventInput {1183 const appleRenewalPreference =1184 raw.platform === "IOS" &&1185 raw.type === "SubscriptionProductChanged" &&1186 raw.effectiveImmediately !== true;1187 return {1188 type: raw.type,1189 // Apple's renewal preference is the product for the next billing period;1190 // the current transaction remains active until then. Keep the target on1191 // the source event, but do not replace the canonical subscription early.1192 productId: appleRenewalPreference ? undefined : raw.productId,1193 subscriptionState: raw.subscriptionState,1194 expiresAt: raw.expiresAt,1195 renewsAt: raw.renewsAt,1196 willRenew: raw.willRenew,1197 cancellationReason: raw.cancellationReason,1198 currency: appleRenewalPreference ? undefined : raw.currency,1199 priceAmountMicros: appleRenewalPreference1200 ? undefined1201 : raw.priceAmountMicros,1202 };1203}1204 1205function isActive(1206 sub: Doc<"subscriptions">,1207 now: number = Date.now(),1208): boolean {1209 return entitlementActive(sub, now);1210}1211 1212function normalizeHarmonizedPurchaseState(1213 state: string,1214): HarmonizedPurchaseState {1215 const normalized = state.trim().toUpperCase().replace(/-/g, "_");1216 if (normalized in HarmonizedPurchaseState) {1217 return HarmonizedPurchaseState[1218 normalized as keyof typeof HarmonizedPurchaseState1219 ];1220 }1221 return HarmonizedPurchaseState.UNKNOWN;1222}1223 1224// Bind a subscription to a userId. Called by the SDK after a successful1225// receipt validation when the host app knows which user owns the receipt.1226export const bindSubscriptionToUser = internalMutation({1227 args: {1228 projectId: v.id("projects"),1229 purchaseToken: v.string(),1230 userId: v.string(),1231 },1232 returns: v.union(v.id("subscriptions"), v.null()),1233 handler: async (ctx, args) => bindSubscriptionToUserHandler(ctx, args),1234});1235 1236/**1237 * The retained source a re-emitted entitlement event is attributed to. The1238 * stored row is preferred; a pruned one is reconstructed from the snapshot the1239 * subscription keeps. Price is stripped: a webhook already reported it, and1240 * repeating it would put the same money on another event.1241 */1242async function retainedSourceEventFor(1243 ctx: MutationCtx,1244 projectId: Id<"projects">,1245 sub: Doc<"subscriptions">,1246) {1247 if (!sub.lastEventId) return null;1248 const retained = await ctx.db.get(sub.lastEventId);1249 const source =1250 retained?.projectId === projectId1251 ? retained1252 : sub.lastEventSource &&1253 sub.lastEventOccurredAt !== undefined &&1254 sub.lastEventSourceNotificationId1255 ? {1256 _id: sub.lastEventId,1257 type: sub.lastEventSource.type,1258 platform: sub.platform,1259 environment: sub.lastEventSource.environment,1260 productId: sub.lastEventSource.productId,1261 applicationId: sub.lastEventSource.applicationId,1262 transactionId: sub.lastEventSource.transactionId,1263 originalTransactionId: sub.lastEventSource.originalTransactionId,1264 sourceNotificationId: sub.lastEventSourceNotificationId,1265 occurredAt: sub.lastEventOccurredAt,1266 }1267 : null;1268 if (!source) return null;1269 return {1270 ...source,1271 currency: undefined,1272 priceAmountMicros: undefined,1273 amountProvenance: undefined,1274 };1275}1276 1277/**1278 * Moves an existing binding. `bindSubscriptionToUserHandler` refuses this on1279 * purpose — a caller holding a purchase token has not proved it owns the1280 * purchase — so correcting a wrong binding needs an operator-authorized path.1281 * Callers must gate this on a secret key.1282 */1283export async function rebindSubscriptionToUserHandler(1284 ctx: MutationCtx,1285 args: BindSubscriptionToUserArgs,1286): Promise<{ subscriptionId: Id<"subscriptions">; notified: boolean } | null> {1287 if (!isValidSubscriptionUserId(args.userId)) {1288 throw new ConvexError("userId must be nonblank and at most 256 characters");1289 }1290 const project = await assertProjectWritable(ctx, args.projectId);1291 if (await isUserErasureRequested(ctx, project, args.userId)) return null;1292 // Resolve exactly as `bind` does. An operator correcting a wrong binding1293 // usually has the token the customer reported, which on Play may be the one1294 // a replacement superseded.1295 const supersedingResolution = await findSupersedingSubscription(1296 ctx,1297 args.projectId,1298 args.purchaseToken,1299 );1300 const sub = supersedingResolution.aliased1301 ? supersedingResolution.subscription1302 : await findSubscriptionByToken(ctx, args.projectId, args.purchaseToken);1303 if (!sub) return null;1304 if (1305 sub.userId &&1306 sub.userId !== args.userId &&1307 (await isUserErasureRequested(ctx, project, sub.userId))1308 )1309 return null;1310 if (sub.userId === args.userId) {1311 return { subscriptionId: sub._id, notified: true };1312 }1313 const previousUserId = sub.userId;1314 const now = Date.now();1315 1316 // A consumer that gates access on commerce events has no other way to learn1317 // the purchase moved: without these it keeps the wrong user entitled and1318 // never entitles the real one. Decide before writing, so the caller is never1319 // told a rebind succeeded when the notification half of it could not.1320 const entitled = isActive({ ...sub, userId: args.userId }, now);1321 const sourceEvent = entitled1322 ? await retainedSourceEventFor(ctx, args.projectId, sub)1323 : null;1324 await ctx.db.patch(sub._id, {1325 userId: args.userId,1326 accountErased: undefined,1327 updatedAt: now,1328 });1329 if (!entitled) return { subscriptionId: sub._id, notified: true };1330 if (!sourceEvent) {1331 // Every retained trace of the originating notification is gone, so no1332 // event can be attributed. The binding moved; the operator must reconcile1333 // the developer backend by hand.1334 console.warn(1335 "[subscriptions/rebind] moved a binding without emitting entitlement events",1336 { projectId: args.projectId, subscriptionId: sub._id },1337 );1338 return { subscriptionId: sub._id, notified: false };1339 }1340 const snapshot = {1341 state: sub.state,1342 productId: sub.productId,1343 ...(sub.expiresAt !== undefined ? { expiresAt: sub.expiresAt } : {}),1344 ...(sub.renewsAt !== undefined ? { renewsAt: sub.renewsAt } : {}),1345 ...(sub.willRenew !== undefined ? { willRenew: sub.willRenew } : {}),1346 ...(sub.cancellationReason1347 ? { cancellationReason: sub.cancellationReason }1348 : {}),1349 };1350 for (const [userId, active, previouslyActive] of [1351 ...(previousUserId1352 ? ([[previousUserId, false, true]] as const)1353 : ([] as const)),1354 [args.userId, true, false] as const,1355 ]) {1356 await emitCommerceEvent(ctx, {1357 projectId: args.projectId,1358 transition: null,1359 active,1360 previouslyActive,1361 sourceEvent,1362 subscriptionId: sub._id,1363 subscription: { ...snapshot, userId },1364 });1365 }1366 return { subscriptionId: sub._id, notified: true };1367}1368 1369export async function bindSubscriptionToUserHandler(1370 ctx: MutationCtx,1371 args: BindSubscriptionToUserArgs,1372): Promise<Id<"subscriptions"> | null> {1373 if (!isValidSubscriptionUserId(args.userId)) {1374 throw new ConvexError("userId must be nonblank and at most 256 characters");1375 }1376 const project = await assertProjectWritable(ctx, args.projectId);1377 if (await isUserErasureRequested(ctx, project, args.userId)) return null;1378 const supersedingResolution = await findSupersedingSubscription(1379 ctx,1380 args.projectId,1381 args.purchaseToken,1382 );1383 const sub = supersedingResolution.aliased1384 ? supersedingResolution.subscription1385 : await findSubscriptionByToken(ctx, args.projectId, args.purchaseToken);1386 if (!sub) return null;1387 if (sub.userId === args.userId) return sub._id;1388 if (sub.userId) {1389 // Reported the same as an unknown token. A distinct error would tell any1390 // holder of the app-embedded publishable key whether a token exists and1391 // whether someone owns it. The operator still sees the collision.1392 console.warn(1393 "[subscriptions/bind] rejected a bind for an already-bound subscription",1394 { projectId: args.projectId, subscriptionId: sub._id },1395 );1396 return null;1397 }1398 const now = Date.now();1399 // Erasure only unlinked the previous owner; this is a new association, and1400 // the pending-erasure gate above still refuses a user who is being erased.1401 await ctx.db.patch(sub._id, {1402 userId: args.userId,1403 accountErased: undefined,1404 updatedAt: now,1405 });1406 1407 // If the store webhook won the race against verify/bind, its initial event1408 // had no account identity. Emit one correlated entitlement grant after the1409 // binding so a developer backend never has to recover a purchase token from1410 // the public payload. The normal verify -> bind -> webhook path has no1411 // lastEventId yet and emits its grant when that first webhook arrives.1412 if (sub.lastEventId && isActive(sub, now)) {1413 const retainedSource = await ctx.db.get(sub.lastEventId);1414 const sourceEvent =1415 retainedSource?.projectId === args.projectId1416 ? retainedSource1417 : sub.lastEventSource &&1418 sub.lastEventOccurredAt !== undefined &&1419 sub.lastEventSourceNotificationId1420 ? {1421 _id: sub.lastEventId,1422 type: sub.lastEventSource.type,1423 platform: sub.platform,1424 environment: sub.lastEventSource.environment,1425 productId: sub.lastEventSource.productId,1426 applicationId: sub.lastEventSource.applicationId,1427 transactionId: sub.lastEventSource.transactionId,1428 originalTransactionId: sub.lastEventSource.originalTransactionId,1429 currency: sub.lastEventSource.currency,1430 priceAmountMicros: sub.lastEventSource.priceAmountMicros,1431 sourceNotificationId: sub.lastEventSourceNotificationId,1432 occurredAt: sub.lastEventOccurredAt,1433 }1434 : null;1435 if (sourceEvent) {1436 // This only fires after a webhook already emitted the notification's1437 // amount, so repeating it would put the same money on a second event.1438 const entitlementSource = {1439 ...sourceEvent,1440 currency: undefined,1441 priceAmountMicros: undefined,1442 amountProvenance: undefined,1443 };1444 await emitCommerceEvent(ctx, {1445 projectId: args.projectId,1446 transition: null,1447 active: true,1448 previouslyActive: false,1449 sourceEvent: entitlementSource,1450 subscriptionId: sub._id,1451 subscription: {1452 state: sub.state,1453 productId: sub.productId,1454 ...(sub.expiresAt !== undefined ? { expiresAt: sub.expiresAt } : {}),1455 ...(sub.renewsAt !== undefined ? { renewsAt: sub.renewsAt } : {}),1456 ...(sub.willRenew !== undefined ? { willRenew: sub.willRenew } : {}),1457 ...(sub.cancellationReason1458 ? { cancellationReason: sub.cancellationReason }1459 : {}),1460 userId: args.userId,1461 },1462 });1463 }1464 }1465 return sub._id;1466}1467 1468export async function drainSubscriptionUserErasurePage(1469 ctx: MutationCtx,1470 jobId: Id<"subscriptionUserErasureJobs">,1471): Promise<{1472 done: boolean;1473 subscriptionsErased: number;1474 commerceEventsErased: number;1475}> {1476 const job = await ctx.db.get(jobId);1477 if (!job || job.status === "completed" || !job.userId) {1478 return {1479 done: true,1480 subscriptionsErased: job?.subscriptionsErased ?? 0,1481 commerceEventsErased: job?.commerceEventsErased ?? 0,1482 };1483 }1484 1485 const now = Date.now();1486 await ctx.db.patch(jobId, { status: "running", updatedAt: now });1487 const [subscriptions, commerceEvents, purchases] = await Promise.all([1488 ctx.db1489 .query("subscriptions")1490 .withIndex("by_project_and_user", (q) =>1491 q.eq("projectId", job.projectId).eq("userId", job.userId),1492 )1493 .take(USER_ERASURE_BATCH_SIZE),1494 ctx.db1495 .query("commerceEvents")1496 .withIndex("by_project_and_user", (q) =>1497 q.eq("projectId", job.projectId).eq("userId", job.userId),1498 )1499 .take(USER_ERASURE_BATCH_SIZE),1500 ctx.db1501 .query("purchases")1502 .withIndex("by_project_and_app_user", (q) =>1503 q.eq("projectId", job.projectId).eq("appUserId", job.userId),1504 )1505 .take(USER_ERASURE_BATCH_SIZE),1506 ]);1507 1508 for (const subscription of subscriptions) {1509 await ctx.db.patch(subscription._id, {1510 userId: undefined,1511 accountErased: true,1512 });1513 }1514 for (const purchase of purchases) {1515 await ctx.db.patch(purchase._id, {1516 appUserId: undefined,1517 accountErased: true,1518 });1519 }1520 let commerceEventsErasedThisPage = 0;1521 let waitingForClaimedDelivery = false;1522 for (const event of commerceEvents) {1523 const deliveries = await ctx.db1524 .query("outboundDeliveries")1525 .withIndex("by_event", (q) => q.eq("eventId", event._id))1526 .collect();1527 if (deliveries.some((delivery) => delivery.status === "delivering")) {1528 waitingForClaimedDelivery = true;1529 continue;1530 }1531 if (event.eventType.startsWith("entitlement.")) {1532 for (const delivery of deliveries) await ctx.db.delete(delivery._id);1533 await ctx.db.delete(event._id);1534 } else {1535 await ctx.db.patch(event._id, { userId: undefined });1536 }1537 commerceEventsErasedThisPage += 1;1538 }1539 1540 const subscriptionsErased = job.subscriptionsErased + subscriptions.length;1541 const commerceEventsErased =1542 job.commerceEventsErased + commerceEventsErasedThisPage;1543 const done =1544 !waitingForClaimedDelivery &&1545 subscriptions.length < USER_ERASURE_BATCH_SIZE &&1546 purchases.length < USER_ERASURE_BATCH_SIZE &&1547 commerceEvents.length < USER_ERASURE_BATCH_SIZE;1548 1549 if (done) {1550 await ctx.db.patch(jobId, {1551 userId: undefined,1552 status: "completed",1553 subscriptionsErased,1554 commerceEventsErased,1555 updatedAt: now,1556 completedAt: now,1557 });1558 } else {1559 await ctx.db.patch(jobId, {1560 subscriptionsErased,1561 commerceEventsErased,1562 updatedAt: now,1563 });1564 await ctx.scheduler.runAfter(1565 waitingForClaimedDelivery ? 1_000 : 0,1566 internal.subscriptions.internal.drainSubscriptionUserErasureJob,1567 { jobId },1568 );1569 }1570 1571 return { done, subscriptionsErased, commerceEventsErased };1572}1573 1574export const drainSubscriptionUserErasureJob = internalMutation({1575 args: { jobId: v.id("subscriptionUserErasureJobs") },1576 handler: async (ctx, args) =>1577 drainSubscriptionUserErasurePage(ctx, args.jobId),1578});1579 1580export const resumeSubscriptionUserErasureJobs = internalMutation({1581 args: {},1582 returns: v.object({ scheduled: v.number() }),1583 handler: async (ctx) => {1584 const staleBefore = Date.now() - USER_ERASURE_STALE_MS;1585 const [queued, stale] = await Promise.all([1586 ctx.db1587 .query("subscriptionUserErasureJobs")1588 .withIndex("by_status_and_updated", (q) => q.eq("status", "queued"))1589 .take(10),1590 ctx.db1591 .query("subscriptionUserErasureJobs")1592 .withIndex("by_status_and_updated", (q) =>1593 q.eq("status", "running").lt("updatedAt", staleBefore),1594 )1595 .take(10),1596 ]);1597 const jobs = [...queued, ...stale].slice(0, 10);1598 for (const job of jobs) {1599 await ctx.scheduler.runAfter(1600 0,1601 internal.subscriptions.internal.drainSubscriptionUserErasureJob,1602 { jobId: job._id },1603 );1604 }1605 return { scheduled: jobs.length };1606 },1607});1608 1609export async function pruneCompletedSubscriptionUserErasureJobsHandler(1610 ctx: MutationCtx,1611): Promise<{ pruned: number }> {1612 const completedBefore = Date.now() - USER_ERASURE_JOB_RETENTION_MS;1613 const completed = await ctx.db1614 .query("subscriptionUserErasureJobs")1615 .withIndex("by_status_and_updated", (q) =>1616 q.eq("status", "completed").lt("updatedAt", completedBefore),1617 )1618 .take(USER_ERASURE_PRUNER_BATCH_SIZE);1619 1620 for (const job of completed) {1621 await ctx.db.delete(job._id);1622 }1623 if (completed.length === USER_ERASURE_PRUNER_BATCH_SIZE) {1624 await ctx.scheduler.runAfter(1625 0,1626 internal.subscriptions.internal.pruneCompletedSubscriptionUserErasureJobs,1627 {},1628 );1629 }1630 return { pruned: completed.length };1631}1632 1633export const pruneCompletedSubscriptionUserErasureJobs = internalMutation({1634 args: {},1635 returns: v.object({ pruned: v.number() }),1636 handler: async (ctx) => pruneCompletedSubscriptionUserErasureJobsHandler(ctx),1637});1638