feat(billing): cancel org subscription on identity.org.deleted (FFRNT-174)

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) {

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 `<topic>.dlq`), so the offset still
+ * commits and the loop stays healthy.
+ */
+export async function startOrgDeletedConsumer(
+  consumer: TypedConsumer,
+  deps: OrgDeletedHandlerDeps,
+  dlqProducer?: TypedProducer,
+): Promise<void> {
+  await consumer.connect();
+  await consumer.subscribe(TOPICS.IDENTITY_ORG_DELETED);
+  await consumer.run<IdentityOrgDeletedPayloadV1>(
+    (event: FuzeEvent<IdentityOrgDeletedPayloadV1>) => handleOrgDeleted(event, deps),
+    identityOrgDeletedSchemaV1,
+    dlqProducer,
+  );
+}

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<SubscriptionService, 'cancel' | 'cancelImmediately'>;
+}
+
+/**
+ * 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<IdentityOrgDeletedPayloadV1>,
+  deps: OrgDeletedHandlerDeps,
+): Promise<void> {
+  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})`,
+    );
+  }
+}

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<BillingSubscription> {
+    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<string> {
     const plans = await this.plans.getActivePlans();
     return plans.find((p) => p.priceId === priceId)?.tierName ?? 'unknown';

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<IdentityOrgDeletedPayloadV1> = {},
+): FuzeEvent<IdentityOrgDeletedPayloadV1> {
+  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> = {}): 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/);
+  });
+});