From bba9d1e3d2f7c91f730ee004cb8c6c0d71bc62bb Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 17 Aug 2026 15:12:13 +0000 Subject: [PATCH] feat(billing): cancel org subscription on identity.org.deleted (FFRNT-174) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit billing-service now reacts to identity.org.deleted by canceling the deleted org's Stripe subscription — the cross-service org-lifecycle slice of FFRNT-174, mirroring the provisioning-service consumers. - new org-deleted.handler.ts: resolve organization -> billing.customers (entity_type='organization') -> subscription, then cancel. cascade 'soft' cancels at period end (reversible), 'hard' cancels immediately. Best-effort + idempotent: no customer / no subscription / already canceled -> no-op. - new org-deleted.consumer.ts: a dedicated TypedConsumer group (-org-deleted) on identity.org.deleted, DLQ via the shared producer; wired + disconnected on shutdown in index.ts. - SubscriptionService.cancelImmediately() for the hard cascade (stripe.subscriptions.cancel), alongside the existing soft cancel(). - tests: soft/hard cascade, no-customer / no-subscription / already- canceled no-ops, and failure propagation to the DLQ. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01HxwMXu8tusGrFcZe6iC9j8 --- services/billing-service/src/index.ts | 14 ++ .../src/kafka/org-deleted.consumer.ts | 33 +++++ .../src/kafka/org-deleted.handler.ts | 75 ++++++++++ .../src/services/subscription.service.ts | 24 ++++ .../tests/kafka/org-deleted.handler.test.ts | 134 ++++++++++++++++++ 5 files changed, 280 insertions(+) create mode 100644 services/billing-service/src/kafka/org-deleted.consumer.ts create mode 100644 services/billing-service/src/kafka/org-deleted.handler.ts create mode 100644 services/billing-service/tests/kafka/org-deleted.handler.test.ts diff --git a/services/billing-service/src/index.ts b/services/billing-service/src/index.ts index 3e4c9e0f8..bb03b3c26 100644 --- a/services/billing-service/src/index.ts +++ b/services/billing-service/src/index.ts @@ -24,6 +24,7 @@ import { MeteringService } from './services/metering.service'; import { KafkaBillingEmitter } from './kafka/producer'; import { startUsageConsumer } from './kafka/consumer'; import { startRefIndexConsumer } from './kafka/ref-index.consumer'; +import { startOrgDeletedConsumer } from './kafka/org-deleted.consumer'; import { HandlerContext } from './handlers/types'; const FLUSH_INTERVAL_SEC = parseInt(process.env.BILLING_METER_FLUSH_INTERVAL_SEC || '60', 10); @@ -100,6 +101,18 @@ async function main() { console.error('[billing-service] ref-index consumer failed to start:', err), ); + // --- identity.org.deleted -> cancel the org's Stripe subscription --- + // Its OWN consumer group: TypedConsumer.run binds a single schema per loop, + // and this reaction must not share offsets with the usage/ref-index consumers. + const orgDeletedConsumer = new TypedConsumer(kafka, `${config.kafka.groupId}-org-deleted`); + startOrgDeletedConsumer( + orgDeletedConsumer, + { customers: customerRepo, subscriptions: subscriptionRepo, subscriptionService }, + producer, + ).catch((err) => + console.error('[billing-service] org-deleted consumer failed to start:', err), + ); + // --- Metering flush loop --- const flushTimer = setInterval(() => { metering.flush().catch((err) => console.error('[billing-service] meter flush error:', err)); @@ -156,6 +169,7 @@ async function main() { try { await consumer.disconnect(); await refIndexConsumer.disconnect(); + await orgDeletedConsumer.disconnect(); await producer.disconnect(); await pool!.end(); } catch (err) { diff --git a/services/billing-service/src/kafka/org-deleted.consumer.ts b/services/billing-service/src/kafka/org-deleted.consumer.ts new file mode 100644 index 000000000..3fb27c841 --- /dev/null +++ b/services/billing-service/src/kafka/org-deleted.consumer.ts @@ -0,0 +1,33 @@ +import { + TypedConsumer, + TypedProducer, + TOPICS, + identityOrgDeletedSchemaV1, + IdentityOrgDeletedPayloadV1, + FuzeEvent, +} from '@fuzefront/shared/dist/kafka'; +import { + handleOrgDeleted, + OrgDeletedHandlerDeps, +} from './org-deleted.handler'; + +/** + * Wires a TypedConsumer to react to `identity.org.deleted` by canceling the + * org's Stripe subscription (see handleOrgDeleted). A handler failure + * dead-letters via the shared DLQ producer (TypedConsumer.run routes + * schema-invalid/handler-failed messages to `.dlq`), so the offset still + * commits and the loop stays healthy. + */ +export async function startOrgDeletedConsumer( + consumer: TypedConsumer, + deps: OrgDeletedHandlerDeps, + dlqProducer?: TypedProducer, +): Promise { + await consumer.connect(); + await consumer.subscribe(TOPICS.IDENTITY_ORG_DELETED); + await consumer.run( + (event: FuzeEvent) => handleOrgDeleted(event, deps), + identityOrgDeletedSchemaV1, + dlqProducer, + ); +} diff --git a/services/billing-service/src/kafka/org-deleted.handler.ts b/services/billing-service/src/kafka/org-deleted.handler.ts new file mode 100644 index 000000000..1c49167a3 --- /dev/null +++ b/services/billing-service/src/kafka/org-deleted.handler.ts @@ -0,0 +1,75 @@ +import { + FuzeEvent, + IdentityOrgDeletedPayloadV1, +} from '@fuzefront/shared/dist/kafka'; +import { CustomerRepository } from '../repositories/customer.repository'; +import { SubscriptionRepository } from '../repositories/subscription.repository'; +import { SubscriptionService } from '../services/subscription.service'; + +/** + * Collaborators the org-deleted reaction needs. Kept as an interface (not the + * concrete wiring) so the handler is unit-testable against in-memory fakes. + */ +export interface OrgDeletedHandlerDeps { + customers: CustomerRepository; + subscriptions: SubscriptionRepository; + subscriptionService: Pick; +} + +/** + * Reacts to `identity.org.deleted` by canceling the deleted org's Stripe + * subscription. The org→subscription link is indirect: billing stores no + * `org_id` on subscriptions, so we resolve + * organization → billing.customers (entity_type='organization') → subscription. + * + * `cascade` picks the cancel semantics: + * - 'soft' (the default org DELETE) → cancel at period end (reversible; the + * org may be reactivated before the period closes). + * - 'hard' → cancel immediately (the org is being purged; nothing left to bill). + * + * Best-effort + idempotent: an org with no billing customer or no subscription + * is a no-op, and an already-canceled subscription is skipped (so a redelivered + * event does not hit Stripe again). The TypedConsumer already validated the + * payload against identityOrgDeletedSchemaV1 before this handler is called. + */ +export async function handleOrgDeleted( + event: FuzeEvent, + deps: OrgDeletedHandlerDeps, +): Promise { + const { organizationId, cascade } = event.payload; + + const customer = await deps.customers.findByEntity('organization', organizationId); + if (!customer) { + console.log( + `[billing-service] org ${organizationId} has no billing customer — nothing to cancel (correlationId=${event.correlationId})`, + ); + return; + } + + const subscription = await deps.subscriptions.findByCustomer(customer.id); + if (!subscription) { + console.log( + `[billing-service] org ${organizationId} has no subscription — nothing to cancel (correlationId=${event.correlationId})`, + ); + return; + } + + if (subscription.status === 'canceled') { + console.log( + `[billing-service] subscription ${subscription.subscriptionId} already canceled — skipping (correlationId=${event.correlationId})`, + ); + return; + } + + if (cascade === 'hard') { + await deps.subscriptionService.cancelImmediately(subscription.subscriptionId); + console.log( + `[billing-service] hard-canceled subscription ${subscription.subscriptionId} for org ${organizationId} (correlationId=${event.correlationId})`, + ); + } else { + await deps.subscriptionService.cancel(subscription.subscriptionId); + console.log( + `[billing-service] soft-canceled (period-end) subscription ${subscription.subscriptionId} for org ${organizationId} (correlationId=${event.correlationId})`, + ); + } +} diff --git a/services/billing-service/src/services/subscription.service.ts b/services/billing-service/src/services/subscription.service.ts index fcdceeda7..3c61c2222 100644 --- a/services/billing-service/src/services/subscription.service.ts +++ b/services/billing-service/src/services/subscription.service.ts @@ -126,6 +126,30 @@ export class SubscriptionService { ); } + /** + * Cancel immediately (hard cancel); Stripe ends the subscription now rather + * than at period end. Used when the owning entity is being purged (e.g. an + * `identity.org.deleted` with `cascade: 'hard'`) and there is nothing left to + * bill for. + */ + async cancelImmediately(stripeSubscriptionId: string): Promise { + const existing = await this.repo.findByStripeId(stripeSubscriptionId); + if (!existing) { + throw new Error(`Subscription not found: ${stripeSubscriptionId}`); + } + const canceled = await this.stripe.subscriptions.cancel( + stripeSubscriptionId, + undefined, + { idempotencyKey: `sub-cancel-now-${stripeSubscriptionId}` }, + ); + return this.repo.upsert( + mapStripeSubscription(canceled, { + customerId: existing.customerId, + planTier: existing.planTier, + }), + ); + } + private async resolvePlanTier(priceId: string): Promise { const plans = await this.plans.getActivePlans(); return plans.find((p) => p.priceId === priceId)?.tierName ?? 'unknown'; diff --git a/services/billing-service/tests/kafka/org-deleted.handler.test.ts b/services/billing-service/tests/kafka/org-deleted.handler.test.ts new file mode 100644 index 000000000..a82561f3a --- /dev/null +++ b/services/billing-service/tests/kafka/org-deleted.handler.test.ts @@ -0,0 +1,134 @@ +import { handleOrgDeleted } from '../../src/kafka/org-deleted.handler'; +import { + FuzeEvent, + TOPICS, + IdentityOrgDeletedPayloadV1, +} from '@fuzefront/shared/dist/kafka'; +import { BillingCustomer, BillingSubscription } from '../../src/types'; + +const ORG_ID = '33333333-3333-3333-3333-333333333333'; +const OWNER_ID = '22222222-2222-2222-2222-222222222222'; + +function deletedEvent( + overrides: Partial = {}, +): FuzeEvent { + return { + version: '1.0', + topic: TOPICS.IDENTITY_ORG_DELETED, + correlationId: 'corr-org-del', + occurredAt: new Date().toISOString(), + payload: { + organizationId: ORG_ID, + slug: 'acme', + ownerId: OWNER_ID, + cascade: 'soft', + ...overrides, + }, + }; +} + +const customer: BillingCustomer = { + id: 'localcust_1', + entityType: 'organization', + entityId: ORG_ID, + stripeCustomerId: 'cus_1', +}; + +function subscription(overrides: Partial = {}): BillingSubscription { + return { + id: 'localsub_1', + customerId: 'localcust_1', + subscriptionId: 'sub_1', + priceId: 'price_pro', + planTier: 'pro', + status: 'active', + seatQuantity: 1, + trialStart: null, + trialEnd: null, + currentPeriodStart: null, + currentPeriodEnd: null, + cancelAtPeriodEnd: false, + canceledAt: null, + ...overrides, + }; +} + +function makeDeps(opts: { + customer?: BillingCustomer | null; + subscription?: BillingSubscription | null; +}) { + const cancel = jest.fn().mockResolvedValue(subscription({ cancelAtPeriodEnd: true })); + const cancelImmediately = jest.fn().mockResolvedValue(subscription({ status: 'canceled' })); + const findByEntity = jest.fn().mockResolvedValue(opts.customer ?? null); + const findByCustomer = jest.fn().mockResolvedValue(opts.subscription ?? null); + return { + deps: { + customers: { findByEntity } as any, + subscriptions: { findByCustomer } as any, + subscriptionService: { cancel, cancelImmediately }, + }, + cancel, + cancelImmediately, + findByEntity, + findByCustomer, + }; +} + +describe('handleOrgDeleted (billing)', () => { + it('soft cascade cancels the subscription at period end', async () => { + const { deps, cancel, cancelImmediately, findByEntity } = makeDeps({ + customer, + subscription: subscription(), + }); + await handleOrgDeleted(deletedEvent({ cascade: 'soft' }), deps); + expect(findByEntity).toHaveBeenCalledWith('organization', ORG_ID); + expect(cancel).toHaveBeenCalledWith('sub_1'); + expect(cancelImmediately).not.toHaveBeenCalled(); + }); + + it('hard cascade cancels the subscription immediately', async () => { + const { deps, cancel, cancelImmediately } = makeDeps({ + customer, + subscription: subscription(), + }); + await handleOrgDeleted(deletedEvent({ cascade: 'hard' }), deps); + expect(cancelImmediately).toHaveBeenCalledWith('sub_1'); + expect(cancel).not.toHaveBeenCalled(); + }); + + it('is a no-op when the org has no billing customer', async () => { + const { deps, cancel, cancelImmediately, findByCustomer } = makeDeps({ + customer: null, + }); + await handleOrgDeleted(deletedEvent(), deps); + expect(findByCustomer).not.toHaveBeenCalled(); + expect(cancel).not.toHaveBeenCalled(); + expect(cancelImmediately).not.toHaveBeenCalled(); + }); + + it('is a no-op when the customer has no subscription', async () => { + const { deps, cancel, cancelImmediately } = makeDeps({ + customer, + subscription: null, + }); + await handleOrgDeleted(deletedEvent(), deps); + expect(cancel).not.toHaveBeenCalled(); + expect(cancelImmediately).not.toHaveBeenCalled(); + }); + + it('skips an already-canceled subscription (idempotent on redelivery)', async () => { + const { deps, cancel, cancelImmediately } = makeDeps({ + customer, + subscription: subscription({ status: 'canceled' }), + }); + await handleOrgDeleted(deletedEvent({ cascade: 'hard' }), deps); + expect(cancel).not.toHaveBeenCalled(); + expect(cancelImmediately).not.toHaveBeenCalled(); + }); + + it('propagates a cancel failure so the consumer can dead-letter', async () => { + const { deps, cancel } = makeDeps({ customer, subscription: subscription() }); + cancel.mockRejectedValueOnce(new Error('stripe down')); + await expect(handleOrgDeleted(deletedEvent(), deps)).rejects.toThrow(/stripe down/); + }); +});