fix(billing): plan-change forced reset + atomic usage+extras outbox

modules/billing/README.md

@@ -89,6 +89,20 @@ step. Passing the cumulative total will double-charge costs already attributed i
 **Backward compat**: callers that only ever attribute once (no multi-step) continue to work
 unchanged — the default `stepKey='initial'` makes the idempotency key `${history._id}:initial`.
 
+## Plan-change semantics
+
+When Stripe `plan.changed` webhook fires, devkit calls `forceRotateForPlanChange(orgId, { preserveUsage: true })` by default:
+- Updates `meterQuota` and `planVersion` snapshot to the new plan
+- Preserves `meterUsed` (no refund, no double-charge)
+
+Consumers wanting clean-break behavior on downgrade should pass `{ preserveUsage: false }`.
+
+## Extras debit reliability
+
+`attribute()` returns optimistically after usage increment + outbox row insert. Extras debit happens out of band; if it fails, cron `retry-pending-extras-debit` reconciles within 5min. After 5 failed attempts, the outbox row is marked `failed` and event `billing.extras_debit.exhausted` is emitted for alerting.
+
+Consumers should NOT retry on `applied: true` — the outbox handles eventual consistency.
+
 ## Stripe — `automatic_tax` flag
 
 ```js

modules/billing/crons/README.md

@@ -15,13 +15,15 @@ No `node-cron` dependency — orchestration is handled by Kubernetes CronJob man
 | `billing.weeklyReset.js` | Reset meter counters for orgs whose billing period rolled over | Daily `0 1 * * *` |
 | `billing.extrasExpiration.js` | Expire topup ledger entries past their `expiresAt` date | Daily `0 2 * * *` |
 | `billing.dunningSweep.js` | Downgrade stale `past_due` subs (>14d) to `unpaid` + `free` | Daily `0 3 * * *` |
+| `retry-pending-extras-debit.cron.js` | Retry pending extras debits from the meter outbox | Every 5 minutes `*/5 * * * *` |
 
 ## Usage
 
 ```sh
 NODE_ENV=production node modules/billing/crons/billing.weeklyReset.js
 NODE_ENV=production node modules/billing/crons/billing.extrasExpiration.js
 NODE_ENV=production node modules/billing/crons/billing.dunningSweep.js
+NODE_ENV=production node modules/billing/crons/retry-pending-extras-debit.cron.js
 ```
 
 Exit code 0 = success (or meterMode disabled). Exit code 1 = at least one error or fatal failure.

modules/billing/crons/retry-pending-extras-debit.cron.js

@@ -0,0 +1,49 @@
+/**
+ * Cron script — retry pending extras debits from the meter outbox.
+ *
+ * No-op when config.billing.meterMode === false (default).
+ * Intended to run as a Kubernetes CronJob every 5 minutes.
+ *
+ * Usage:
+ *   NODE_ENV=production node modules/billing/crons/retry-pending-extras-debit.cron.js
+ */
+
+process.env.NODE_ENV = process.env.NODE_ENV || 'development';
+
+const [{ default: config }, { default: mongooseService }] = await Promise.all([
+  import('../../../config/index.js'),
+  import('../../../lib/services/mongoose.js'),
+]);
+
+if (!config?.billing?.meterMode) {
+  console.log('[billing.retryPendingExtrasDebit] meterMode disabled — skipping.');
+  process.exit(0);
+}
+
+const { randomInt } = await import('node:crypto');
+const jitterMs = randomInt(0, 60_000);
+await new Promise((resolve) => setTimeout(resolve, jitterMs));
+
+try {
+  await mongooseService.loadModels();
+  await mongooseService.connect();
+
+  const { default: BillingMeterOutboxService } = await import('../services/billing.meter.outbox.service.js');
+  const result = await BillingMeterOutboxService.retryPendingExtrasDebits(5 * 60 * 1000, 100);
+
+  console.log(
+    `[billing.retryPendingExtrasDebit] done — scanned: ${result.scanned}, committed: ${result.committed}, failedAttempts: ${result.failedAttempts}, exhausted: ${result.exhausted}`,
+  );
+  if (result.exhausted > 0) {
+    // Exhausted rows are a handled business outcome (alert event already emitted),
+    // not an operational cron failure — log for visibility without failing the job.
+    console.warn(`[billing.retryPendingExtrasDebit] exhausted rows (alert emitted): ${result.exhausted}`);
+  }
+  process.exitCode = 0;
+} catch (err) {
+  console.error('[billing.retryPendingExtrasDebit] fatal:', err);
+  process.exitCode = 1;
+} finally {
+  await mongooseService.disconnect?.();
+}
+process.exit(process.exitCode ?? 0);

modules/billing/lib/events.js

@@ -9,6 +9,10 @@ import { EventEmitter } from 'events';
  * Events:
  *   - `plan.changed`    — emitted when a subscription's plan changes
  *     Payload: { organizationId, previousPlan, newPlan, subscription, isDowngrade }
+ *   - `billing.plan_change.rotated` — emitted after current-week meter snapshot refresh
+ *     Payload: { organizationId, oldQuota, newQuota, oldVersion, newVersion, preserveUsage }
+ *   - `billing.extras_debit.exhausted` — emitted when outbox extras debit retries fail 5 times
+ *     Payload: { organizationId, idempotencyKey, extrasUnits, attempts, lastError }
  *   - `payment.failed`  — emitted when an invoice payment fails (pastDueSince set on first failure)
  *     Payload: { organizationId }
  */

modules/billing/models/billing.meter.outbox.model.mongoose.js

@@ -0,0 +1,40 @@
+/**
+ * Module dependencies
+ */
+import mongoose from 'mongoose';
+
+const Schema = mongoose.Schema;
+
+/**
+ * Meter outbox model.
+ *
+ * Stores deferred extras debits created after meter usage crosses plan quota.
+ * Pending rows are retried by the billing retry-pending-extras-debit cron.
+ */
+const BillingMeterOutboxMongoose = new Schema({
+  organizationId: { type: Schema.ObjectId, required: true, index: true },
+  idempotencyKey: { type: String, required: true, unique: true },
+  extrasUnits: { type: Number, required: true },
+  status: { type: String, enum: ['pending', 'committed', 'failed'], default: 'pending', index: true },
+  attempts: { type: Number, default: 0 },
+  lastError: { type: String, default: null },
+  lastAttemptedAt: { type: Date, default: null },
+  createdAt: { type: Date, default: () => new Date() },
+});
+
+BillingMeterOutboxMongoose.index({ status: 1, lastAttemptedAt: 1 });
+
+/**
+ * Returns the hex string representation of the document ObjectId.
+ * @returns {string} Hex string of the ObjectId.
+ */
+function addID() {
+  return this._id.toHexString();
+}
+
+BillingMeterOutboxMongoose.virtual('id').get(addID);
+BillingMeterOutboxMongoose.set('toJSON', {
+  virtuals: true,
+});
+
+mongoose.model('BillingMeterOutbox', BillingMeterOutboxMongoose);

modules/billing/models/billing.meter.outbox.schema.js

@@ -0,0 +1,35 @@
+/**
+ * Module dependencies
+ */
+import { z } from 'zod';
+
+/**
+ * BillingMeterOutbox Zod schema — mirrors billing.meter.outbox.model.mongoose.js
+ */
+
+const objectIdRegex = /^[a-f\d]{24}$/i;
+
+const BillingMeterOutboxStatus = z.enum(['pending', 'committed', 'failed']);
+
+const BillingMeterOutbox = z.object({
+  organizationId: z.string().trim().regex(objectIdRegex, 'organizationId must be a valid ObjectId'),
+  idempotencyKey: z.string().trim().min(1, 'idempotencyKey is required'),
+  extrasUnits: z.number().int().min(1, 'extrasUnits must be >= 1'),
+  status: BillingMeterOutboxStatus.default('pending'),
+  attempts: z.number().int().min(0).default(0),
+  lastError: z.string().nullable().default(null),
+  lastAttemptedAt: z.coerce.date().nullable().default(null),
+  createdAt: z.coerce.date().optional(),
+});
+
+const BillingMeterOutboxCreate = z.object({
+  organizationId: z.string().trim().regex(objectIdRegex, 'organizationId must be a valid ObjectId'),
+  idempotencyKey: z.string().trim().min(1, 'idempotencyKey is required'),
+  extrasUnits: z.number().int().min(1, 'extrasUnits must be >= 1'),
+});
+
+export default {
+  BillingMeterOutboxStatus,
+  BillingMeterOutbox,
+  BillingMeterOutboxCreate,
+};

modules/billing/repositories/billing.meter.outbox.repository.js

@@ -0,0 +1,123 @@
+/**
+ * Module dependencies
+ */
+import mongoose from 'mongoose';
+
+/**
+ * @function BillingMeterOutbox
+ * @description Lazily resolves the BillingMeterOutbox Mongoose model.
+ *              Deferred to keep unit tests importable before model registration.
+ * @returns {import('mongoose').Model} The registered BillingMeterOutbox model.
+ */
+// biome-ignore lint/correctness/useQwikValidLexicalScope: false positive — Node.js repository, not Qwik
+const BillingMeterOutbox = () => mongoose.model('BillingMeterOutbox');
+
+/**
+ * @function create
+ * @description Insert a pending outbox row for a deferred extras debit.
+ * @param {Object} payload - Outbox row fields.
+ * @param {string} payload.organizationId - Organization ObjectId.
+ * @param {string} payload.idempotencyKey - Usage attribution idempotency key.
+ * @param {number} payload.extrasUnits - Extras units to debit.
+ * @param {Object} [options={}] - Optional write options.
+ * @param {import('mongoose').ClientSession} [options.session] - Optional Mongo session.
+ * @returns {Promise<Object>} Inserted outbox document.
+ */
+// biome-ignore lint/correctness/useQwikValidLexicalScope: false positive — Node.js repository, not Qwik
+const create = async ({ organizationId, idempotencyKey, extrasUnits }, options = {}) => {
+  const docs = await BillingMeterOutbox().create(
+    [{
+      organizationId,
+      idempotencyKey,
+      extrasUnits,
+      status: 'pending',
+    }],
+    options.session ? { session: options.session } : undefined,
+  );
+  return docs[0];
+};
+
+/**
+ * @function findPendingDue
+ * @description Return pending outbox rows whose last attempt is due for retry.
+ *              Rows with lastAttemptedAt=null are due immediately.
+ * @param {number} [thresholdMs=300000] - Retry backoff threshold in milliseconds.
+ * @param {number} [limit=100] - Maximum rows to return.
+ * @returns {Promise<Object[]>} Pending due outbox rows.
+ */
+// biome-ignore lint/correctness/useQwikValidLexicalScope: false positive — Node.js repository, not Qwik
+const findPendingDue = (thresholdMs = 5 * 60 * 1000, limit = 100) => {
+  const dueBefore = new Date(Date.now() - thresholdMs);
+  return BillingMeterOutbox()
+    .find({
+      status: 'pending',
+      $or: [
+        { lastAttemptedAt: null },
+        { lastAttemptedAt: { $lt: dueBefore } },
+      ],
+    })
+    .sort({ lastAttemptedAt: 1, createdAt: 1 })
+    .limit(limit)
+    .lean();
+};
+
+/**
+ * @function markCommitted
+ * @description Mark an outbox row as committed after a successful extras debit.
+ *              The `status:'pending'` filter makes this idempotent: committed or
+ *              failed rows are immutable and concurrent calls are no-ops.
+ * @param {string} id - Outbox row id.
+ * @returns {Promise<Object>} Mongo update result.
+ */
+// biome-ignore lint/correctness/useQwikValidLexicalScope: false positive — Node.js repository, not Qwik
+const markCommitted = (id) =>
+  BillingMeterOutbox().updateOne(
+    { _id: id, status: 'pending' },
+    { $set: { status: 'committed', lastError: null, lastAttemptedAt: new Date() } },
+  );
+
+/**
+ * @function markFailedAttempt
+ * @description Record a failed debit attempt. The fifth failed attempt exhausts
+ *              the row and moves it to failed status atomically. The status
+ *              transition uses `{ status: 'pending' }` as a filter on the
+ *              exhaustion update so that concurrent cron runs cannot emit
+ *              duplicate exhausted events.
+ * @param {string} id - Outbox row id.
+ * @param {Error|string} error - Failure to record.
+ * @returns {Promise<Object|null>} Updated outbox row after failure accounting.
+ */
+// biome-ignore lint/correctness/useQwikValidLexicalScope: false positive — Node.js repository, not Qwik
+const markFailedAttempt = async (id, error) => {
+  const message = error?.message ?? String(error);
+  const doc = await BillingMeterOutbox().findOneAndUpdate(
+    { _id: id, status: 'pending' },
+    {
+      $inc: { attempts: 1 },
+      $set: {
+        lastError: message,
+        lastAttemptedAt: new Date(),
+      },
+    },
+    { returnDocument: 'after' },
+  ).lean();
+
+  if (!doc) return null;
+  if (doc.attempts >= 5) {
+    // Atomic exhaustion transition: filter on status:'pending' ensures only
+    // the first concurrent caller wins the status flip and owns the event emit.
+    return BillingMeterOutbox().findOneAndUpdate(
+      { _id: id, status: 'pending' },
+      { $set: { status: 'failed' } },
+      { returnDocument: 'after' },
+    ).lean();
+  }
+  return doc;
+};
+
+export default {
+  create,
+  findPendingDue,
+  markCommitted,
+  markFailedAttempt,
+};