NNO Billing & Metering

-- Platform subscription record
CREATE TABLE subscriptions (
  id                  TEXT PRIMARY KEY,  -- NanoID
  platform_id         TEXT NOT NULL UNIQUE,
  tier                TEXT NOT NULL,     -- 'starter' | 'growth' | 'scale'
  status              TEXT NOT NULL,     -- 'active' | 'trialing' | 'past_due' | 'canceled'
  stripe_customer_id  TEXT NOT NULL,
  stripe_sub_id       TEXT NOT NULL,
  billing_email       TEXT NOT NULL,
  current_period_start INTEGER NOT NULL,
  current_period_end   INTEGER NOT NULL,
  cancel_at_period_end INTEGER NOT NULL DEFAULT 0,
  -- Phase 2 additions:
  trial_start         INTEGER,           -- Unix ms — trial period start (NULL = no trial)
  trial_end           INTEGER,           -- Unix ms — trial period end (NULL = no trial)
  auto_finalize       INTEGER NOT NULL DEFAULT 1,  -- 0 = require manual review before finalizing invoice
  created_at          INTEGER NOT NULL,
  updated_at          INTEGER NOT NULL
);

-- Daily usage snapshots per platform (or per entity when entity_id is set)
-- migrations/0002_add_entity_metering.sql adds entity_id for per-tenant granularity
CREATE TABLE usage_snapshots (
  id           TEXT PRIMARY KEY,
  platform_id  TEXT NOT NULL,
  entity_id    TEXT,              -- NULL = platform-level aggregate; non-NULL = per-tenant (added by migration 0002)
  snapshot_date TEXT NOT NULL,   -- 'YYYY-MM-DD'
  -- Cloudflare Worker invocations
  worker_invocations  INTEGER NOT NULL DEFAULT 0,
  worker_errors       INTEGER NOT NULL DEFAULT 0,
  -- D1 operations
  d1_read_rows        INTEGER NOT NULL DEFAULT 0,
  d1_write_rows       INTEGER NOT NULL DEFAULT 0,
  -- R2 storage and operations
  r2_storage_bytes    INTEGER NOT NULL DEFAULT 0,
  r2_class_a_ops      INTEGER NOT NULL DEFAULT 0,
  r2_class_b_ops      INTEGER NOT NULL DEFAULT 0,
  -- KV operations
  kv_reads            INTEGER NOT NULL DEFAULT 0,
  kv_writes           INTEGER NOT NULL DEFAULT 0,
  -- Metadata
  collected_at        INTEGER NOT NULL
  -- Migration 0001 created this table with an inline UNIQUE(platform_id, snapshot_date) constraint.
  -- Migration 0002 added entity_id and idx_snapshots_platform_entity_date, but SQLite cannot drop
  -- inline constraints without recreating the table. Migration 0004 (0004_fix_snapshot_unique.sql)
  -- performed the table recreation, removing the old 2-col constraint and leaving uniqueness
  -- enforced solely by idx_snapshots_platform_entity_date:
  -- UNIQUE INDEX idx_snapshots_platform_entity_date
  --   ON usage_snapshots(platform_id, COALESCE(entity_id, ''), snapshot_date)
);

-- Monthly invoice records
-- migrations/0002_add_entity_metering.sql adds entity_id for per-tenant invoicing
CREATE TABLE invoices (
  id              TEXT PRIMARY KEY,
  platform_id     TEXT NOT NULL,
  entity_id       TEXT,           -- NULL = platform-level invoice; non-NULL = per-tenant (added by migration 0002)
  stripe_invoice_id TEXT,
  period_start    INTEGER NOT NULL,
  period_end      INTEGER NOT NULL,
  status          TEXT NOT NULL,  -- 'draft' | 'open' | 'paid' | 'void' | 'uncollectible'
  tier            TEXT NOT NULL,
  base_amount     INTEGER NOT NULL,   -- in cents
  overage_amount  INTEGER NOT NULL,   -- in cents
  total_amount    INTEGER NOT NULL,   -- in cents
  line_items      TEXT NOT NULL,      -- JSON array of LineItem
  pdf_url         TEXT,
  due_date        INTEGER,
  paid_at         INTEGER,
  created_at      INTEGER NOT NULL,
  updated_at      INTEGER NOT NULL
);

-- Usage threshold alerts
CREATE TABLE usage_alerts (
  id           TEXT PRIMARY KEY,
  platform_id  TEXT NOT NULL,
  resource     TEXT NOT NULL,  -- 'worker_invocations' | 'd1_read_rows' | etc.
  threshold_pct INTEGER NOT NULL,  -- 50 | 75 | 90 | 100
  triggered_at INTEGER NOT NULL,
  notified_at  INTEGER,  -- NULL = not yet sent
  period       TEXT NOT NULL  -- 'YYYY-MM' billing period
);

-- Stripe webhook event log (idempotency)
CREATE TABLE stripe_events (
  stripe_event_id TEXT PRIMARY KEY,
  event_type      TEXT NOT NULL,
  processed_at    INTEGER NOT NULL,
  payload         TEXT NOT NULL    -- JSON
);

CREATE INDEX idx_snapshots_platform_date ON usage_snapshots(platform_id, snapshot_date);
CREATE INDEX idx_invoices_platform ON invoices(platform_id, status);
CREATE INDEX idx_alerts_platform ON usage_alerts(platform_id, period);

# Injected into each platform Worker's wrangler.toml by NNO Provisioning
[[analytics_engine_datasets]]
binding = "ANALYTICS"
dataset = "{platformId}-usage"

// Injected into every platform Worker by the NNO feature SDK
export function recordUsageEvent(
  analytics: AnalyticsEngineDataset,
  platformId: string,
  entityId: string,
  featureId: string,
  responseStatus: number
): void {
  analytics.writeDataPoint({
    blobs: [platformId, entityId, featureId, String(responseStatus)],
    doubles: [1],   // invocation count
    indexes: [platformId],
  });
}

async function fetchWorkerInvocations(
  cfApiToken: string,
  cfAccountId: string,
  platformId: string,
  date: string,    // 'YYYY-MM-DD'
  dayAfter: string // 'YYYY-MM-DD' — next calendar day
): Promise<number> {
  // doubles[0] = 1 (invocation count written by each platform Worker's CFAE emitter)
  const query =
    `SELECT SUM(_sample_interval * doubles[0]) AS invocations ` +
    `FROM ${platformId}_usage ` +
    `WHERE timestamp >= toDateTime('${date} 00:00:00') ` +
    `AND timestamp < toDateTime('${dayAfter} 00:00:00')`;

  const response = await fetch(
    `https://api.cloudflare.com/client/v4/accounts/${cfAccountId}/analytics_engine/sql`,
    {
      method: 'POST',
      headers: {
        Authorization: `Bearer ${cfApiToken}`,
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({ query }),
    }
  );

  const result = await response.json<{ data: { invocations: number }[] }>();
  return result.data[0]?.invocations ?? 0;
}

# D1 usage — Cloudflare GraphQL Analytics API
{
  viewer {
    accounts(filter: { accountTag: "<accountId>" }) {
      d1AnalyticsAdaptiveGroups(
        filter: { date_geq: "<date>", date_lt: "<dayAfter>" }
        limit: 10000
      ) {
        sum { readQueries writeQueries }
        dimensions { databaseName }
      }
    }
  }
}

# R2 storage — Cloudflare GraphQL Analytics API
{
  viewer {
    accounts(filter: { accountTag: "<accountId>" }) {
      r2StorageAdaptiveGroups(
        filter: { date_geq: "<date>", date_lt: "<dayAfter>" }
        limit: 10000
      ) {
        max { payloadSize }
        dimensions { bucketName }
      }
    }
  }
}

# KV operations — Cloudflare GraphQL Analytics API
{
  viewer {
    accounts(filter: { accountTag: "<accountId>" }) {
      kvOperationsAdaptiveGroups(
        filter: { date_geq: "<date>", date_lt: "<dayAfter>" }
        limit: 10000
      ) {
        sum { requests }
        dimensions { namespaceId requestType }
      }
    }
  }
}

# services/billing/wrangler.toml
[triggers]
crons = ["0 2 * * *",    # 02:00 UTC daily — usage snapshot
         "0 3 * * *",    # 03:00 UTC daily — trial expiry check
         "0 6 1 * *"]    # 06:00 UTC on 1st — monthly invoice generation

Cron fires at 02:00 UTC
         │
         ▼
For each active platform in subscriptions table:
  1. Fetch all platform resources from NNO Registry
     (CF resource IDs for all Workers, D1s, R2 buckets, KV namespaces)
         │
         ▼
  2. Collect usage from Cloudflare APIs in parallel:
     ├─ CFAE: Worker invocations (yesterday's date)
     ├─ CF API: D1 read/write rows per database
     ├─ CF API: R2 storage + ops per bucket
     └─ CF API: KV reads/writes per namespace
         │
         ▼
  3. Aggregate totals across all entities for the platform
         │
         ▼
  4. Upsert row in usage_snapshots (idempotent)
         │
         ▼
  5. Check threshold alerts:
     ├─ Calculate MTD (month-to-date) usage vs. tier limits
     ├─ If 50%/75%/90%/100% thresholds crossed → insert usage_alerts
     └─ If new alert → queue notification email

-- MTD worker invocations for a platform in the current billing period
SELECT SUM(worker_invocations) AS mtd_invocations
FROM   usage_snapshots
WHERE  platform_id   = ?
  AND  snapshot_date >= ?   -- current_period_start date
  AND  snapshot_date <  ?   -- today

crons = ["0 2 * * *",     # daily snapshot job
         "0 3 * * *",     # trial expiry check
         "0 6 1 * *"]     # monthly invoice job (1st of month, 06:00 UTC)

async function generateInvoice(
  platformId: string,
  period: { start: number; end: number }
): Promise<Invoice> {
  const subscription = await getSubscription(platformId);
  const tierLimits  = TIER_LIMITS[subscription.tier];

  // 1. Sum all daily snapshots for the billing period
  const usage = await sumPeriodUsage(platformId, period);

  // 2. Calculate overages per resource
  const overageItems: LineItem[] = [];

  for (const [resource, limit] of Object.entries(tierLimits)) {
    const used   = usage[resource] ?? 0;
    const excess = Math.max(0, used - limit);
    if (excess > 0) {
      const rate   = OVERAGE_RATES[resource];
      const amount = Math.ceil((excess / rate.unit) * rate.cents);
      overageItems.push({
        resource,
        used,
        limit,
        excess,
        rate_description: rate.description,
        amount_cents: amount,
      });
    }
  }

  const baseAmount    = TIER_BASE_PRICES[subscription.tier];
  const overageAmount = overageItems.reduce((s, i) => s + i.amount_cents, 0);
  const totalAmount   = baseAmount + overageAmount;

  // 3. Create draft invoice in Stripe
  const stripeInvoice = await createStripeInvoice({
    customerId: subscription.stripe_customer_id,
    baseAmount,
    overageItems,
    period,
  });

  // 4. Record in billing D1
  return await insertInvoice({
    platformId,
    stripeInvoiceId: stripeInvoice.id,
    period,
    tier: subscription.tier,
    baseAmount,
    overageAmount,
    totalAmount,
    lineItems: overageItems,
  });
}

[
  {
    "resource": "worker_invocations",
    "used": 8500000,
    "limit": 5000000,
    "excess": 3500000,
    "rate_description": "$0.30 per 1M invocations",
    "amount_cents": 105
  },
  {
    "resource": "d1_read_rows",
    "used": 30000000,
    "limit": 25000000,
    "excess": 5000000,
    "rate_description": "$0.001 per 1M rows",
    "amount_cents": 1
  }
]

Client signs up on NNO Portal
         │
         ▼
Billing Service: create Stripe Customer
         │
         ▼
Billing Service: create Stripe Subscription
  → { subscription_id, current_period_start, current_period_end }
         │
         ▼
Upsert in subscriptions table
         │
         ▼ (end of billing period)
Billing Service:
  1. Generate invoice (flat tier + overages)
  2. Add Stripe Invoice Items for overages
  3. Finalise Stripe invoice → charge customer
         │
         ▼
Stripe webhook → /webhooks:
  invoice.paid    → mark invoice as paid in billing D1
  invoice.payment_failed → mark as past_due, send alert
  customer.subscription.deleted → mark subscription as canceled

app.post('/webhooks', async (c) => {
  const payload   = await c.req.text();
  const signature = c.req.header('stripe-signature') ?? '';

  let event: Stripe.Event;
  try {
    event = stripe.webhooks.constructEvent(
      payload, signature, c.env.STRIPE_WEBHOOK_SECRET
    );
  } catch {
    return c.json({ error: 'Invalid signature' }, 400);
  }

  // Idempotency: check if event already processed
  const existing = await c.env.DB
    .prepare('SELECT 1 FROM stripe_events WHERE stripe_event_id = ?')
    .bind(event.id)
    .first();
  if (existing) return c.json({ ok: true });

  // Process event
  await handleStripeEvent(c.env, event);

  // Record as processed
  await c.env.DB
    .prepare('INSERT INTO stripe_events (stripe_event_id, event_type, processed_at, payload) VALUES (?, ?, ?, ?)')
    .bind(event.id, event.type, Date.now(), JSON.stringify(event))
    .run();

  return c.json({ received: true, type: event.type });
});

Subject: ⚠️ Neutrino Usage Alert — Worker Invocations at 75%

Your platform AcmeCorp (k3m9p2xw7q) has used 75% of its
included Worker invocations for this billing period.

  Used:     37.5M / 50M invocations
  Period:   February 2026

If you exceed your limit, overage charges of $0.30/M will apply.

→ View usage details: https://portal.nno.app/billing
→ Upgrade plan: https://portal.nno.app/billing/upgrade

# services/billing/wrangler.toml

name = "nno-k3m9p2xw7q-billing"
main = "src/index.ts"
compatibility_date = "2024-09-13"
compatibility_flags = ["nodejs_compat"]

# Cron triggers:
#   02:00 UTC daily  — usage snapshot
#   03:00 UTC daily  — trial expiry check (cancel overdue trials)
#   06:00 UTC on 1st — monthly invoice generation
[triggers]
crons = ["0 2 * * *", "0 3 * * *", "0 6 1 * *"]

# ============================================================================
# PRODUCTION (default — no --env flag)
# ============================================================================
[[analytics_engine_datasets]]
binding = "NNO_METRICS"
dataset = "nno_metrics"

# PLATFORM_ID (no NNO_ prefix) is correct for the billing service.
# Billing is a standalone NNO operator service — it does not embed a platform
# ID in session payloads, so the shorter name is used by design.
# (Contrast with services/auth and services/iam which use NNO_PLATFORM_ID
# because those services populate session.platformId.)
[vars]
PLATFORM_ID  = "k3m9p2xw7q"
SERVICE_NAME = "billing"
ENVIRONMENT  = "prod"

[[d1_databases]]
binding       = "DB"
database_name = "nno-k3m9p2xw7q-billing-db"
database_id   = "a9240e7a-41f6-4694-ac89-642295707277"

[[routes]]
pattern = "billing.svc.nno.app"
custom_domain = true

# ============================================================================
# STG ENVIRONMENT
# ============================================================================
[env.stg]
name = "nno-k3m9p2xw7q-billing-stg"

[env.stg.vars]
PLATFORM_ID  = "k3m9p2xw7q"
SERVICE_NAME = "billing"
ENVIRONMENT  = "stg"

[[env.stg.routes]]
pattern = "billing.svc.stg.nno.app"
custom_domain = true

[[env.stg.analytics_engine_datasets]]
binding = "NNO_METRICS"
dataset = "nno_metrics"

[[env.stg.d1_databases]]
binding       = "DB"
database_name = "nno-k3m9p2xw7q-billing-db-stg"
database_id   = "cf8ae831-0091-4fc6-943f-b7b713c4d8af"