feat: source-metronome net_balance refactor + cache sidecar

Source-metronome changes:
- Switch to /v1/contracts/customerBalances/getNetBalance endpoint
- Remove deprecated credit_grants (legacy /v1/credits/listGrants) and invoices streams
- Add singleObject resource pattern for single-response endpoints
- Add skipLimit flag for rate_cards (API rejects limit param)
- Split webhook events into balance vs entitlement categories
- Fix webhook backpressure (await push before HTTP 200)

Cache sidecar (apps/cache-sidecar):
- Standalone HTTP process for optimistic balance enforcement
- Two Redis namespaces: sync:* (checkpoint, read-only) and optimistic:* (sidecar-owned)
- GET /v1/balance/:cid — merged checkpoint + optimistic view
- POST /v1/events — record pending event, decrement balance instantly
- Reconciler polls every 5s, discards acknowledged events by watermark
- Deduplication via deterministic sorted set members keyed by event_id
- Fire-and-forget Metronome ingest forwarding
- Integration test suite (25 tests)

Pixel-app rewired to call sidecar instead of Redis/Metronome directly.
E2E script updated for net_balance primary key (customer_id + credit_type_id).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Committed-By-Agent: claude

Author: kdhillon-stripe

apps/cache-sidecar/package.json

@@ -0,0 +1,32 @@
+{
+  "name": "@stripe/sync-cache-sidecar",
+  "version": "0.1.0",
+  "private": true,
+  "description": "HTTP sidecar for optimistic balance enforcement over synced Metronome data",
+  "type": "module",
+  "exports": {
+    ".": {
+      "bun": "./src/index.ts",
+      "types": "./dist/index.d.ts",
+      "import": "./dist/index.js"
+    }
+  },
+  "scripts": {
+    "build": "tsc",
+    "dev": "tsx --watch --conditions bun src/index.ts",
+    "start": "node dist/index.js",
+    "test": "vitest run",
+    "test:watch": "vitest"
+  },
+  "dependencies": {
+    "@hono/node-server": "^1",
+    "hono": "^4",
+    "ioredis": "^5",
+    "zod": "^4.3.6"
+  },
+  "devDependencies": {
+    "@types/node": "^24.10.1",
+    "tsx": "^4",
+    "vitest": "^3.2.4"
+  }
+}

apps/cache-sidecar/src/config.ts

@@ -0,0 +1,25 @@
+import { z } from 'zod'
+
+const ConfigSchema = z.object({
+  REDIS_URL: z.string().default('redis://localhost:56379'),
+  CHECKPOINT_PREFIX: z.string().default('sync:'),
+  OPTIMISTIC_PREFIX: z.string().default('optimistic:'),
+  CREDIT_TYPE_ID: z.string().default('2714e483-4ff1-48e4-9e25-ac732e8f24f2'),
+  METRONOME_API_TOKEN: z.string().optional(),
+  METRONOME_BASE_URL: z.string().default('https://api.metronome.com'),
+  PORT: z.coerce.number().default(4100),
+  RECONCILE_INTERVAL_MS: z.coerce.number().default(5000),
+  WATERMARK_BUFFER_MS: z.coerce.number().default(10000),
+  FIXED_EVENT_COST: z.coerce.number().positive().default(1),
+})
+
+export type Config = z.infer<typeof ConfigSchema>
+
+export function loadConfig(): Config {
+  const result = ConfigSchema.safeParse(process.env)
+  if (!result.success) {
+    console.error('Invalid configuration:', result.error.format())
+    process.exit(1)
+  }
+  return result.data
+}

apps/cache-sidecar/src/index.ts

@@ -0,0 +1,37 @@
+import { serve } from '@hono/node-server'
+import { loadConfig } from './config.js'
+import { FixedCostPricing } from './pricing.js'
+import { Reconciler } from './reconciler.js'
+import { createRedisClient } from './redis.js'
+import { createApp } from './server.js'
+
+const config = loadConfig()
+const redis = createRedisClient(config)
+const pricing = new FixedCostPricing(config.FIXED_EVENT_COST)
+const reconciler = new Reconciler(redis, config)
+
+const app = createApp({ redis, config, pricing })
+
+const server = serve({ fetch: app.fetch, port: config.PORT }, (info) => {
+  console.log(
+    JSON.stringify({
+      msg: 'cache-sidecar started',
+      port: info.port,
+      redis_url: config.REDIS_URL,
+      reconcile_interval_ms: config.RECONCILE_INTERVAL_MS,
+    })
+  )
+})
+
+reconciler.start()
+
+function shutdown() {
+  console.log(JSON.stringify({ msg: 'shutting down' }))
+  reconciler.stop()
+  redis.disconnect()
+  server.close()
+  process.exit(0)
+}
+
+process.on('SIGTERM', shutdown)
+process.on('SIGINT', shutdown)

apps/cache-sidecar/src/pricing.ts

@@ -0,0 +1,15 @@
+export interface PricingStrategy {
+  estimateCost(eventType: string, properties?: Record<string, unknown>): number
+}
+
+/**
+ * Fixed-cost pricing: every event costs the same amount of credits.
+ * Good enough for MVP — swap in a lookup-based strategy later.
+ */
+export class FixedCostPricing implements PricingStrategy {
+  constructor(private readonly cost: number) {}
+
+  estimateCost(_eventType: string, _properties?: Record<string, unknown>): number {
+    return this.cost
+  }
+}

apps/cache-sidecar/src/reconciler.ts

@@ -0,0 +1,103 @@
+import type { Config } from './config.js'
+import type { Redis } from './redis.js'
+import { pendingSetKey, writeOptimisticBalance } from './redis.js'
+import type { CheckpointData, PendingEvent } from './redis.js'
+
+export class Reconciler {
+  private timer: ReturnType<typeof setInterval> | null = null
+
+  constructor(
+    private readonly redis: Redis,
+    private readonly config: Config
+  ) {}
+
+  start(): void {
+    if (this.timer) return
+    console.log(
+      JSON.stringify({
+        msg: 'reconciler started',
+        interval_ms: this.config.RECONCILE_INTERVAL_MS,
+      })
+    )
+    this.timer = setInterval(() => this.reconcile(), this.config.RECONCILE_INTERVAL_MS)
+    // Run immediately on start
+    void this.reconcile()
+  }
+
+  stop(): void {
+    if (this.timer) {
+      clearInterval(this.timer)
+      this.timer = null
+      console.log(JSON.stringify({ msg: 'reconciler stopped' }))
+    }
+  }
+
+  async reconcile(): Promise<void> {
+    try {
+      const pattern = `${this.config.CHECKPOINT_PREFIX}net_balance:*`
+      let cursor = '0'
+      let processed = 0
+
+      do {
+        const [nextCursor, keys] = await this.redis.scan(
+          cursor,
+          'MATCH',
+          pattern,
+          'COUNT',
+          100
+        )
+        cursor = nextCursor
+
+        for (const key of keys) {
+          await this.reconcileKey(key)
+          processed++
+        }
+      } while (cursor !== '0')
+
+      if (processed > 0) {
+        console.log(JSON.stringify({ msg: 'reconcile cycle', keys_processed: processed }))
+      }
+    } catch (err) {
+      console.error(
+        JSON.stringify({
+          msg: 'reconcile error',
+          error: err instanceof Error ? err.message : String(err),
+        })
+      )
+    }
+  }
+
+  private async reconcileKey(key: string): Promise<void> {
+    const raw = await this.redis.get(key)
+    if (!raw) return
+
+    const checkpoint = JSON.parse(raw) as CheckpointData
+    const customerId = checkpoint.customer_id
+
+    const syncedAtMs = Number(checkpoint._synced_at) * 1000
+    const cutoff = syncedAtMs - this.config.WATERMARK_BUFFER_MS
+
+    const pendingKey = pendingSetKey(this.config, customerId)
+
+    // Remove events older than cutoff
+    await this.redis.zremrangebyscore(pendingKey, '-inf', cutoff)
+
+    // Get remaining events and sum costs
+    const members = await this.redis.zrange(pendingKey, 0, -1)
+    let totalCost = 0
+    for (const member of members) {
+      const event = JSON.parse(member) as PendingEvent
+      totalCost += event.estimated_cost
+    }
+
+    // Write optimistic balance
+    const optimisticBalance = checkpoint.balance - totalCost
+    await writeOptimisticBalance(
+      this.redis,
+      this.config,
+      customerId,
+      optimisticBalance,
+      members.length
+    )
+  }
+}

apps/cache-sidecar/src/redis.ts

@@ -0,0 +1,99 @@
+import { Redis } from 'ioredis'
+import type { Config } from './config.js'
+
+export interface CheckpointData {
+  balance: number
+  customer_id: string
+  credit_type_id: string
+  _synced_at: number
+}
+
+export interface PendingEvent {
+  event_id: string
+  event_type: string
+  estimated_cost: number
+  timestamp: number
+  properties?: Record<string, unknown>
+}
+
+export interface OptimisticBalanceHash {
+  value: string
+  computed_at: string
+  event_count: string
+}
+
+export type { Redis }
+
+export function createRedisClient(config: Config): Redis {
+  const client = new Redis(config.REDIS_URL, { maxRetriesPerRequest: 3 })
+  client.on('error', (err) => {
+    console.error(JSON.stringify({ msg: 'redis error', error: err.message }))
+  })
+  return client
+}
+
+export function checkpointKey(config: Config, customerId: string): string {
+  return `${config.CHECKPOINT_PREFIX}net_balance:${customerId}:${config.CREDIT_TYPE_ID}`
+}
+
+export function pendingSetKey(config: Config, customerId: string): string {
+  return `${config.OPTIMISTIC_PREFIX}pending:${customerId}`
+}
+
+export function optimisticBalanceKey(config: Config, customerId: string): string {
+  return `${config.OPTIMISTIC_PREFIX}balance:${customerId}`
+}
+
+export async function getCheckpoint(
+  redis: Redis,
+  config: Config,
+  customerId: string
+): Promise<CheckpointData | null> {
+  const raw = await redis.get(checkpointKey(config, customerId))
+  if (!raw) return null
+  return JSON.parse(raw) as CheckpointData
+}
+
+export async function getPendingEvents(
+  redis: Redis,
+  config: Config,
+  customerId: string
+): Promise<PendingEvent[]> {
+  const members = await redis.zrange(pendingSetKey(config, customerId), 0, -1)
+  return members.map((m) => JSON.parse(m) as PendingEvent)
+}
+
+export async function sumPendingCosts(
+  redis: Redis,
+  config: Config,
+  customerId: string
+): Promise<{ total: number; count: number }> {
+  const events = await getPendingEvents(redis, config, customerId)
+  const total = events.reduce((sum, e) => sum + e.estimated_cost, 0)
+  return { total, count: events.length }
+}
+
+export async function writeOptimisticBalance(
+  redis: Redis,
+  config: Config,
+  customerId: string,
+  value: number,
+  eventCount: number
+): Promise<void> {
+  const key = optimisticBalanceKey(config, customerId)
+  await redis.hset(key, {
+    value: String(value),
+    computed_at: String(Date.now()),
+    event_count: String(eventCount),
+  })
+}
+
+export async function getOptimisticBalance(
+  redis: Redis,
+  config: Config,
+  customerId: string
+): Promise<OptimisticBalanceHash | null> {
+  const data = await redis.hgetall(optimisticBalanceKey(config, customerId))
+  if (!data || !data.value) return null
+  return data as unknown as OptimisticBalanceHash
+}