Initial import of zappier-edge from zapier monorepo

tests/report-usage-idempotency.test.ts

@@ -0,0 +1,245 @@
+import Database from 'better-sqlite3';
+import { Customer } from '../src/auth';
+import { METER_EVENT_NAME, MeterEventClient } from '../src/billing/stripe';
+import { SqliteBillingReportRepo } from '../src/db/billing-repo';
+import { reportMonthlyUsage } from '../src/jobs/report-usage';
+import { TierConfig } from '../src/pricing';
+import { UsageEntry } from '../src/usage';
+
+const SINCE = new Date('2026-07-01T00:00:00Z');
+
+const customer: Customer = {
+  id: 'cust_1',
+  name: 'Acme',
+  tierId: 'pro',
+  apiKey: 'key_1',
+  stripeCustomerId: 'cus_123',
+};
+
+const customerB: Customer = {
+  id: 'cust_2',
+  name: 'Beta',
+  tierId: 'pro',
+  apiKey: 'key_2',
+  stripeCustomerId: 'cus_456',
+};
+
+const tier: TierConfig = { id: 'pro', name: 'Pro', multiplier: 1, monthlyCreditCents: 0 };
+
+const entry = (cents: number, customerId = 'cust_1', timestamp = '2026-07-10T00:00:00Z'): UsageEntry => ({
+  customerId,
+  endpointId: 'storage',
+  cents,
+  metadataBytes: 0,
+  attachmentBytes: 0,
+  timestamp: new Date(timestamp),
+});
+
+interface MeterCall {
+  eventName: string;
+  customerId: string;
+  value: string;
+  identifier?: string;
+}
+
+function makeHarness(opts: {
+  entries: UsageEntry[];
+  customers?: Customer[];
+  throwFor?: (stripeCustomerId: string) => boolean;
+  since?: Date;
+  onMeterEvent?: (call: MeterCall) => void;
+  now?: () => number;
+  existingRepo?: SqliteBillingReportRepo;
+}) {
+  const calls: MeterCall[] = [];
+  let shouldThrow = false;
+  const client: MeterEventClient = {
+    createMeterEvent: async (params) => {
+      if (shouldThrow || opts.throwFor?.(params.customerId)) throw new Error('stripe down');
+      calls.push(params);
+      opts.onMeterEvent?.(params);
+    },
+  };
+  const billingRepo =
+    opts.existingRepo ?? new SqliteBillingReportRepo(new Database(':memory:'), opts.now);
+  const logs: string[] = [];
+  const run = (overrides: { entries?: UsageEntry[]; since?: Date } = {}) =>
+    reportMonthlyUsage({
+      client,
+      usage: {
+        listFor: (customerId: string) =>
+          (overrides.entries ?? opts.entries).filter((e) => e.customerId === customerId),
+      },
+      customers: { list: () => opts.customers ?? [customer] },
+      tiers: [tier],
+      billingRepo,
+      locks: billingRepo,
+      since: overrides.since ?? opts.since ?? SINCE,
+      log: (m) => logs.push(m),
+    });
+  return {
+    run,
+    calls,
+    logs,
+    billingRepo,
+    setShouldThrow: (v: boolean) => {
+      shouldThrow = v;
+    },
+  };
+}
+
+describe('reportMonthlyUsage idempotency', () => {
+  it('reports the full billable amount on first run and upserts the ledger', async () => {
+    const h = makeHarness({ entries: [entry(500)] });
+    await h.run();
+    expect(h.calls).toEqual([
+      {
+        eventName: METER_EVENT_NAME,
+        customerId: 'cus_123',
+        value: '500',
+        identifier: 'cus_123:2026-07:500',
+      },
+    ]);
+    expect(h.billingRepo.getReportedCents('cust_1', '2026-07')).toBe(500);
+  });
+
+  it('sends no meter event when the same month is re-run with unchanged usage', async () => {
+    const h = makeHarness({ entries: [entry(500)] });
+    await h.run();
+    await h.run();
+    expect(h.calls).toHaveLength(1);
+    expect(h.billingRepo.getReportedCents('cust_1', '2026-07')).toBe(500);
+    expect(h.logs[1]).toBe('skip cust_1 2026-07 (already reported 500c)');
+  });
+
+  it('reports only the delta when usage grows within the same period', async () => {
+    const entries = [entry(500)];
+    const h = makeHarness({ entries });
+    await h.run();
+    entries.push(entry(300));
+    await h.run();
+    expect(h.calls).toEqual([
+      {
+        eventName: METER_EVENT_NAME,
+        customerId: 'cus_123',
+        value: '500',
+        identifier: 'cus_123:2026-07:500',
+      },
+      {
+        eventName: METER_EVENT_NAME,
+        customerId: 'cus_123',
+        value: '300',
+        identifier: 'cus_123:2026-07:800',
+      },
+    ]);
+    expect(h.billingRepo.getReportedCents('cust_1', '2026-07')).toBe(800);
+  });
+
+  it('does not touch the ledger when the Stripe call fails, so a retry reports the full delta', async () => {
+    const h = makeHarness({ entries: [entry(500)] });
+    h.setShouldThrow(true);
+    await expect(h.run()).rejects.toThrow('stripe down');
+    expect(h.billingRepo.getReportedCents('cust_1', '2026-07')).toBe(0);
+    h.setShouldThrow(false);
+    await h.run();
+    expect(h.calls).toHaveLength(1);
+    expect(h.calls[0].value).toBe('500');
+    expect(h.billingRepo.getReportedCents('cust_1', '2026-07')).toBe(500);
+  });
+
+  it('logs "nothing to report" instead of "already reported 0c" for a never-reported customer', async () => {
+    const h = makeHarness({ entries: [] });
+    await h.run();
+    expect(h.calls).toHaveLength(0);
+    expect(h.logs).toEqual(['skip cust_1 2026-07 (nothing to report)']);
+  });
+
+  it('reports the full amount again after month rollover (prior period does not leak)', async () => {
+    const h = makeHarness({ entries: [entry(500)] });
+    await h.run();
+    expect(h.calls[0]).toMatchObject({ value: '500', identifier: 'cus_123:2026-07:500' });
+
+    const augustEntries = [entry(700, 'cust_1', '2026-08-10T00:00:00Z')];
+    await h.run({ entries: augustEntries, since: new Date('2026-08-01T00:00:00Z') });
+    expect(h.calls).toHaveLength(2);
+    expect(h.calls[1]).toEqual({
+      eventName: METER_EVENT_NAME,
+      customerId: 'cus_123',
+      value: '700',
+      identifier: 'cus_123:2026-08:700',
+    });
+    expect(h.billingRepo.getReportedCents('cust_1', '2026-08')).toBe(700);
+    expect(h.billingRepo.getReportedCents('cust_1', '2026-07')).toBe(500);
+  });
+
+  it('keeps customer A reported in the ledger when customer B’s Stripe call fails', async () => {
+    const h = makeHarness({
+      entries: [entry(500), entry(300, 'cust_2')],
+      customers: [customer, customerB],
+      throwFor: (id) => id === 'cus_456',
+    });
+    await expect(h.run()).rejects.toThrow('stripe down');
+    expect(h.calls).toHaveLength(1);
+    expect(h.billingRepo.getReportedCents('cust_1', '2026-07')).toBe(500);
+    expect(h.billingRepo.getReportedCents('cust_2', '2026-07')).toBe(0);
+
+    // Retry: A is skipped (delta 0), B reports its full delta.
+    // Reuse the same repo so the ledger persists across "process restarts".
+    const h2 = makeHarness({
+      entries: [entry(500), entry(300, 'cust_2')],
+      customers: [customer, customerB],
+      existingRepo: h.billingRepo,
+    });
+    await h2.run();
+    expect(h2.calls).toEqual([
+      {
+        eventName: METER_EVENT_NAME,
+        customerId: 'cus_456',
+        value: '300',
+        identifier: 'cus_456:2026-07:300',
+      },
+    ]);
+  });
+
+  it('aborts a second invocation while the lock is held, sending nothing', async () => {
+    let innerLogs: string[] = [];
+    let innerCalls = 0;
+    const h = makeHarness({
+      entries: [entry(500)],
+      onMeterEvent: () => {
+        // Simulate a concurrent invocation (e.g. manual run while cron is in flight).
+        void innerRun();
+      },
+    });
+    const innerRun = async () => {
+      innerCalls++;
+      await h.run();
+    };
+    // Capture logs of the inner run separately by swapping the log sink is
+    // overkill; the shared harness appends to the same array, which is fine:
+    // the inner run must only add the abort line and no meter call.
+    await h.run();
+    innerLogs = h.logs.filter((l) => l.includes('abort'));
+    expect(innerCalls).toBe(1);
+    expect(h.calls).toHaveLength(1); // only the outer run reported
+    expect(innerLogs).toHaveLength(1);
+  });
+
+  it('takes over a stale lock left by a crashed run', async () => {
+    let nowMs = 1_000_000;
+    const h = makeHarness({ entries: [entry(500)], now: () => nowMs });
+    // Simulate a crashed run that acquired the lock and never released it.
+    expect(h.billingRepo.tryAcquireLock('report-usage', 3_600_000)).toBe(true);
+
+    // Within the TTL the job aborts.
+    await h.run();
+    expect(h.calls).toHaveLength(0);
+    expect(h.logs.some((l) => l.includes('abort'))).toBe(true);
+
+    // After the TTL the lock is stale and the run proceeds.
+    nowMs += 3_600_001;
+    await h.run();
+    expect(h.calls).toHaveLength(1);
+    expect(h.billingRepo.getReportedCents('cust_1', '2026-07')).toBe(500);
+  });
+});