fix(billing): hard-fail missing planVersion + zero-cost short-circuit + meter API options shape (#3581)

🟠 D: in meterMode, missing (planId, ratioVersion) snapshot now throws instead
of silently using ratio=1. For consumers like trawl with wizard×5/generate×3,
silent fallback to ×1 would substantially under-charge. Backfill migration
ensures (planId, version) snapshots exist for all configured plans.

🟠 E: unitsFromCosts({}) returns 0 instead of METER_RUN_BASE. attribute() with
zero-cost costsOverride skips the idempotency key write entirely, preserving
the key for a future non-zero attribution at the same step.

🟡 E: attribute() now accepts {costs, planId, ratioVersion, stepKey,
costsOverride} options for explicit per-call overrides. Backward compatible —
history-object path still works.

ensureSeeded() now detects version drift on existing active plans and
re-seeds rather than silently skipping. WARN log on drift for audit trail.

Author: PierreBrisorgueil

modules/billing/migrations/20260502110000-backfill-plan-version-snapshots.js

@@ -0,0 +1,139 @@
+/**
+ * Module dependencies
+ */
+import mongoose from 'mongoose';
+import config from '../../../config/index.js';
+
+const isPlainObject = (value) =>
+  value != null && typeof value === 'object' && !Array.isArray(value);
+
+const isNonEmptyRatios = (value) => isPlainObject(value) && Object.keys(value).length > 0;
+
+const configuredPlans = () => {
+  const byPlanId = new Map();
+
+  const definitions = config?.billing?.planDefinitions;
+  if (Array.isArray(definitions)) {
+    for (const def of definitions) {
+      if (def?.planId) byPlanId.set(def.planId, def);
+    }
+  }
+
+  const legacyPlans = config?.billing?.plans;
+  if (isPlainObject(legacyPlans)) {
+    for (const [planId, def] of Object.entries(legacyPlans)) {
+      byPlanId.set(planId, { ...(def ?? {}), planId });
+    }
+  }
+
+  return byPlanId;
+};
+
+const canonicalSnapshot = (planId, plansById) => {
+  const plan = plansById.get(planId);
+  if (!plan) return null;
+
+  const version = plan.version ?? config?.billing?.meter?.ratioVersion ?? null;
+  if (!version || !isNonEmptyRatios(plan.ratios) || typeof plan.meterQuota !== 'number') {
+    return null;
+  }
+
+  return {
+    version,
+    ratios: plan.ratios,
+    meterQuota: plan.meterQuota,
+  };
+};
+
+/**
+ * Migration: Backfill BillingPlan version snapshots from billing config.
+ *
+ * Repairs legacy BillingPlan docs missing a version or frozen ratios by applying
+ * the canonical configured snapshot for the same planId. Idempotent: documents
+ * already at the canonical version with non-empty ratios are skipped.
+ *
+ * @returns {Promise<void>}
+ */
+export async function up() {
+  const collection = mongoose.connection.db.collection('billingplans');
+  const plansById = configuredPlans();
+  const BATCH_SIZE = 500;
+  let processed = 0;
+  let skipped = 0;
+
+  const cursor = collection.find(
+    {
+      $or: [
+        { version: { $exists: false } },
+        { version: null },
+        { ratios: { $exists: false } },
+        { ratios: null },
+        { ratios: {} },
+      ],
+    },
+    { projection: { _id: 1, planId: 1, version: 1, ratios: 1 } },
+  );
+
+  const ops = [];
+
+  for await (const doc of cursor) {
+    const canonical = canonicalSnapshot(doc.planId, plansById);
+    if (!canonical) {
+      console.warn(
+        `[migration] backfill-plan-version-snapshots: missing canonical config for ${doc.planId}; skipping`,
+      );
+      skipped += 1;
+      continue;
+    }
+
+    if (doc.version === canonical.version && isNonEmptyRatios(doc.ratios)) {
+      skipped += 1;
+      continue;
+    }
+
+    ops.push({
+      updateOne: {
+        filter: { _id: doc._id },
+        update: {
+          $set: {
+            version: canonical.version,
+            ratios: canonical.ratios,
+            meterQuota: canonical.meterQuota,
+          },
+        },
+      },
+    });
+
+    if (ops.length >= BATCH_SIZE) {
+      await collection.bulkWrite(ops, { ordered: false });
+      processed += ops.length;
+      ops.length = 0;
+    }
+  }
+
+  if (ops.length > 0) {
+    await collection.bulkWrite(ops, { ordered: false });
+    processed += ops.length;
+  }
+
+  if (processed > 0 || skipped > 0) {
+    console.info(
+      `[migration] backfill-plan-version-snapshots: backfilled ${processed} documents, skipped ${skipped}`,
+    );
+  }
+}
+
+/**
+ * Down: intentionally no-op.
+ *
+ * Reverting version snapshots can corrupt historical attribution because meter
+ * calculations depend on immutable (planId, version) ratios. Keep consistency
+ * and require manual intervention for any rollback.
+ *
+ * @returns {void}
+ */
+export function down() {
+  console.warn(
+    '[migration] backfill-plan-version-snapshots DOWN: no-op; preserving BillingPlan snapshots',
+  );
+}

modules/billing/services/billing.meter.service.js

@@ -6,7 +6,7 @@ import BillingPlanService from './billing.plan.service.js';
 
 /**
  * Floor charge per run — configurable via config.billing.meter.runBaseUnits.
- * Guarantees every attributed event costs at least 1 unit regardless of costs.
+ * Applies only when no cost data is available at all.
  */
 export const METER_RUN_BASE = config?.billing?.meter?.runBaseUnits ?? 1;
 
@@ -17,11 +17,12 @@ export const METER_RUN_BASE = config?.billing?.meter?.runBaseUnits ?? 1;
  *              dollar for each feature key.
  *
  *              Formula per key: floor(cost[key] * ratio[key] * dollarsToUnitRatio)
- *              Total: max(sum(per-key units), METER_RUN_BASE)
+ *              Total: sum(per-key units). Empty or zero-only cost maps return 0.
+ *              METER_RUN_BASE applies only when costs is null/undefined.
  *
- *              When getPlanByVersion returns null (version mismatch), logs a WARN and
- *              falls back to ratio=1 for all features. Check logs if charges look flat —
- *              this indicates meter.ratioVersion vs planDefinitions version drift.
+ *              When meterMode is enabled and getPlanByVersion returns null
+ *              (version mismatch), throws so attribution cannot silently bill with
+ *              ratio=1. When meterMode is disabled, keeps the legacy ratio=1 fallback.
  *
  * @param {Object} costs - Feature-keyed cost map: { featureKey: usdCost }.
  * @param {string} planId - Logical plan identifier (e.g. "pro").
@@ -30,15 +31,26 @@ export const METER_RUN_BASE = config?.billing?.meter?.runBaseUnits ?? 1;
  */
 // biome-ignore lint/correctness/useQwikValidLexicalScope: false positive — Node.js service, not Qwik
 const unitsFromCosts = async (costs, planId, ratioVersion) => {
-  if (!costs || typeof costs !== 'object') {
+  if (costs == null || typeof costs !== 'object') {
     return { totalUnits: METER_RUN_BASE, breakdown: {} };
   }
 
   const dollarsToUnitRatio = config?.billing?.meter?.dollarsToUnitRatio ?? 1000;
 
+  const hasBillableCost = Object.values(costs).some(
+    (cost) => typeof cost === 'number' && Number.isFinite(cost) && cost > 0,
+  );
+  if (!hasBillableCost) {
+    return { totalUnits: 0, breakdown: {} };
+  }
+
   // Fetch the frozen plan snapshot for the given version
   const plan = await BillingPlanService.getPlanByVersion(planId, ratioVersion);
   if (!plan) {
+    if (config?.billing?.meterMode) {
+      throw new Error(`[billing.meter] missing plan version snapshot for (${planId}, ${ratioVersion})`);
+    }
+
     // WARN: version mismatch likely — costs.config emits a version that has no matching BillingPlan.
     // Charges will use ratio=1 (flat) for all features. Align meter.ratioVersion + planDefinitions[].version
     // + downstream cost-config emitted version to resolve. See billing README — Version Namespace Contract.
@@ -63,9 +75,7 @@ const unitsFromCosts = async (costs, planId, ratioVersion) => {
     }
   }
 
-  const totalUnits = Math.max(rawTotal, METER_RUN_BASE);
-
-  return { totalUnits, breakdown };
+  return { totalUnits: rawTotal, breakdown };
 };
 
 /**
@@ -136,23 +146,45 @@ const capBreakdown = (breakdown, cappedUnits, originalTotal) => {
  *   Use 'initial' for the first (and often only) charge per history.
  *   Use a distinct value (e.g. 'digest', 'fix:1', 'fix:2') for subsequent attributions on
  *   the same history after cost-impacting mutations (setDigest, setFixCost).
- *   Downstream callers must pass ONLY the incremental cost delta in history.costs for each step.
+ *   Downstream callers must pass ONLY the incremental cost delta in history.costs for each step,
+ *   or use options.costsOverride to pass the delta directly.
  *   Format: alphanumeric, colon, hyphen, underscore only, 1-64 chars (e.g. 'initial', 'digest', 'fix:1').
  *   null/undefined → defaults to 'initial'. Any other invalid value → throws Error.
  *   No-op when config.billing.meterMode is false (validation is skipped in that case).
+ * @param {Object|null|undefined} [options.costsOverride=null] - Optional cost map to charge
+ *   instead of history.costs. Use for delta charging:
+ *   `attribute(history, orgId, { stepKey: 'digest', costsOverride: { digest: 0.5 } })`.
+ *   Empty or zero-only overrides return `{ applied: false, reason: 'zero_cost_skipped' }`
+ *   before writing the idempotency key.
+ * @param {Object|null|undefined} [options.costs=null] - Alias cost map used when
+ *   options.costsOverride is not provided. Prefer costsOverride for new delta-charge callers.
+ * @param {string|null|undefined} [options.planId=null] - Optional planId override for paths
+ *   where the history plan should not be used:
+ *   `attribute(history, orgId, { planId: 'override', ratioVersion: '2026.05' })`.
+ * @param {string|null|undefined} [options.ratioVersion=null] - Optional ratio snapshot override
+ *   paired with options.planId. Falls back to history.planVersion when omitted.
  * @returns {Promise<{applied: boolean, meterUsed: number, extrasConsumed: number, reason?: string}>}
- *   `reason` is present only when `applied` is true but extras were exhausted:
- *   `{ applied: true, meterUsed, extrasConsumed: 0, reason: 'extras_exhausted' }`.
+ *   `reason` is present for zero-cost skips and exhausted extras:
+ *   `{ applied: false, meterUsed: 0, extrasConsumed: 0, reason: 'zero_cost_skipped' }`
+ *   or `{ applied: true, meterUsed, extrasConsumed: 0, reason: 'extras_exhausted' }`.
  * @throws {Error} If stepKey is non-null/undefined and does not match the expected format.
  */
 // biome-ignore lint/correctness/useQwikValidLexicalScope: false positive — Node.js service, not Qwik
-const attribute = async (history, organizationId, { stepKey = 'initial' } = {}) => {
+const attribute = async (history, organizationId, options = {}) => {
   // Fast-path: when metering is disabled, skip all validation and return immediately.
   // This preserves the documented no-op behavior for callers passing any stepKey when meterMode=false.
   if (!config?.billing?.meterMode) {
     return { applied: false, meterUsed: 0, extrasConsumed: 0 };
   }
 
+  const {
+    stepKey = 'initial',
+    costs: explicitCosts = null,
+    costsOverride = null,
+    planId: explicitPlanId = null,
+    ratioVersion: explicitRatioVersion = null,
+  } = options ?? {};
+
   // Validate stepKey: must be a non-empty alphanumeric+colon+hyphen+underscore string (1-64 chars).
   // Silent fallback to 'initial' was removed — it collides with the actual initial attribution
   // and silently drops subsequent step charges (e.g. 'digest', 'fix:1').
@@ -170,14 +202,15 @@ const attribute = async (history, organizationId, { stepKey = 'initial' } = {})
   const { default: BillingUsageService } = await import('./billing.usage.service.js');
   const { default: BillingExtraService } = await import('./billing.extra.service.js');
 
-  const planId = history.planId ?? config?.billing?.plans?.[0] ?? 'pro';
-  const ratioVersion = history.planVersion ?? null;
+  const planId = explicitPlanId ?? history.planId ?? config?.billing?.plans?.[0] ?? 'pro';
+  const ratioVersion = explicitRatioVersion ?? history.planVersion ?? null;
+  const costs = costsOverride ?? explicitCosts ?? history.costs;
 
   let totalUnits = METER_RUN_BASE;
   let breakdown = {};
 
-  if (history.costs && ratioVersion) {
-    ({ totalUnits, breakdown } = await unitsFromCosts(history.costs, planId, ratioVersion));
+  if (costs && ratioVersion) {
+    ({ totalUnits, breakdown } = await unitsFromCosts(costs, planId, ratioVersion));
   }
 
   const maxUnits = config?.billing?.meter?.maxUnitsPerOperation ?? Infinity;
@@ -190,6 +223,10 @@ const attribute = async (history, organizationId, { stepKey = 'initial' } = {})
     );
   }
 
+  if (cappedUnits === 0) {
+    return { applied: false, meterUsed: 0, extrasConsumed: 0, reason: 'zero_cost_skipped' };
+  }
+
   const idempotencyKey = `${history._id?.toString?.() ?? String(history._id)}:${validatedStepKey}`;
 
   const result = await BillingUsageService.incrementMeter(

modules/billing/services/billing.plan.service.js

@@ -162,6 +162,44 @@ const bumpVersionWithRetry = async (planId, fields, { maxAttempts = 3 } = {}) =>
   throw lastErr;
 };
 
+const isDuplicateKeyError = (err) =>
+  err.code === 11000 || (err.message && err.message.includes('E11000'));
+
+/**
+ * @desc Resolve the configured immutable version for a plan definition.
+ *       Returns null when config does not pin a version and count-derived
+ *       backward-compat behavior should be used for first-time seeding.
+ * @param {Object} planDef - Configured plan definition without planId.
+ * @returns {string|null} Configured version, or null.
+ */
+const configuredVersion = (planDef) => planDef.version ?? config?.billing?.meter?.ratioVersion ?? null;
+
+/**
+ * @desc Create a configured plan snapshot, deriving a legacy vN version only
+ *       when config did not specify one.
+ * @param {string} planId - Logical plan identifier.
+ * @param {Object} planDef - Configured plan definition without planId.
+ * @param {string|null} [version=null] - Pre-resolved version to use.
+ * @returns {Promise<Object>} The created BillingPlan document.
+ */
+const createConfiguredPlan = async (planId, planDef, version = null) => {
+  let resolvedVersion = version;
+  if (!resolvedVersion) {
+    const total = await BillingPlanRepository.count(planId);
+    resolvedVersion = `v${total + 1}`;
+  }
+
+  return BillingPlanRepository.create({
+    planId,
+    version: resolvedVersion,
+    meterQuota: planDef.meterQuota ?? 0,
+    ratios: planDef.ratios ?? { default: 1 },
+    effectiveFrom: new Date(),
+    effectiveUntil: null,
+    active: true,
+  });
+};
+
 /**
  * @function ensureSeeded
  * @description Upsert BillingPlan docs from config.billing.planDefinitions.
@@ -196,38 +234,39 @@ const ensureSeeded = async () => {
 
   for (const def of definitions) {
     const { planId, ...planDef } = def;
+    const targetVersion = configuredVersion(planDef);
     const existing = await BillingPlanRepository.findActive(planId);
     if (existing) {
-      skipped += 1;
+      if (!targetVersion || existing.version === targetVersion) {
+        skipped += 1;
+        continue;
+      }
+
+      console.info(
+        `[billing.plan] version drift detected for ${planId}: active=${existing.version}, config=${targetVersion}; re-seeding`,
+      );
+      try {
+        await BillingPlanRepository.deactivateAll(planId, new Date());
+        await createConfiguredPlan(planId, planDef, targetVersion);
+        cache.delete(planId);
+        seeded += 1;
+      } catch (err) {
+        if (isDuplicateKeyError(err)) {
+          skipped += 1;
+          continue;
+        }
+        throw err;
+      }
       continue;
     }
 
     try {
-      // Version resolution priority:
-      // 1. Explicit version in planDefinitions entry (e.g. '2026.05' — YYYY.MM contract)
-      // 2. config.billing.meter.ratioVersion (canonical version emitted by attribute())
-      // 3. Derived from count (v${total + 1}) — full backward compat for projects without version config
-      let version = planDef.version ?? config?.billing?.meter?.ratioVersion ?? null;
-      if (!version) {
-        const total = await BillingPlanRepository.count(planId);
-        version = `v${total + 1}`;
-      }
-
-      await BillingPlanRepository.create({
-        planId,
-        version,
-        meterQuota: planDef.meterQuota ?? 0,
-        ratios: planDef.ratios ?? { default: 1 },
-        effectiveFrom: new Date(),
-        effectiveUntil: null,
-        active: true,
-      });
+      await createConfiguredPlan(planId, planDef, targetVersion);
       cache.delete(planId);
       seeded += 1;
     } catch (err) {
       // E11000: concurrent pod beat us to the insert — treat as skip, not fatal.
-      const isE11000 = err.code === 11000 || (err.message && err.message.includes('E11000'));
-      if (isE11000) {
+      if (isDuplicateKeyError(err)) {
         skipped += 1;
         continue;
       }