Phase1: Schema + skill event tracking

src/lib/db.ts

@@ -272,6 +272,252 @@ const MIGRATION_SQL = `
   END $$;
   CREATE INDEX IF NOT EXISTS idx_support_ticket_comments_ticket
     ON support_ticket_comments(ticket_id, created_at);
+
+  -- =========================================================================
+  -- Billing — Phase 1: pricing, lifecycle, and events
+  -- =========================================================================
+  --
+  -- This block introduces the schema for the consolidated billing
+  -- subsystem. Phase 1 lands the schema + writes the lifecycle and
+  -- skill-event rows that downstream phases consume; invoice
+  -- generation, PDF rendering, Stripe wiring and reminders are added
+  -- in later phases. The invoice/line/reminder tables are created
+  -- here so all billing schema lives in a single migration block,
+  -- but no code writes to them until Phase 2.
+  --
+  -- Money columns: NUMERIC(10,2) for CHF amounts (max ~99 million.99
+  -- which is fine for the foreseeable future) and NUMERIC(10,5) for
+  -- per-unit prices (Threema messages and skill-days can have
+  -- sub-rappen unit prices — 0.00012 CHF/message is the kind of
+  -- granularity we want to keep precise across multiplication).
+  --
+  -- All timestamps are TIMESTAMPTZ. Billing-day computations use UTC
+  -- by convention (matches all other stored timestamps; avoids DST
+  -- ambiguity around month boundaries).
+
+  -- Single-row platform pricing config. The id=1 CHECK and explicit
+  -- DEFAULT 1 make accidental multi-row inserts impossible, and
+  -- callers can always SELECT * FROM platform_pricing WHERE id = 1.
+  -- (Could use a settings KV table; this shape is clearer for
+  -- typed columns that the admin UI binds to directly.)
+  CREATE TABLE IF NOT EXISTS platform_pricing (
+    id                       INT PRIMARY KEY DEFAULT 1 CHECK (id = 1),
+    tenant_monthly_fee_chf   NUMERIC(10,2) NOT NULL DEFAULT 0,
+    tenant_setup_fee_chf     NUMERIC(10,2) NOT NULL DEFAULT 0,
+    -- Single price per Threema message; the relay reports in+out
+    -- separately but the spec ("billed per Message") treats both
+    -- directions identically. If asymmetric pricing is needed
+    -- later, split into in/out columns — additive change.
+    threema_message_chf      NUMERIC(10,5) NOT NULL DEFAULT 0,
+    -- Default VAT rate for CH/LI customers, as percent (e.g. 8.10).
+    -- Foreign customers' effective rate is computed at invoice time
+    -- from billing address + VAT number (reverse-charge logic).
+    vat_rate_chli            NUMERIC(5,2) NOT NULL DEFAULT 8.10,
+    updated_at               TIMESTAMPTZ NOT NULL DEFAULT now()
+  );
+  -- Ensure the single row exists. ON CONFLICT DO NOTHING is idempotent
+  -- on every migration run.
+  INSERT INTO platform_pricing (id) VALUES (1) ON CONFLICT DO NOTHING;
+
+  -- Per-package optional daily price. Any package id can have a row;
+  -- the admin UI in Phase 2 will only expose skill-category packages
+  -- because that's what the spec called for, but nothing here
+  -- prevents pricing other categories later.
+  --
+  -- "skill" naming follows the spec ("custom pricing also for skills").
+  CREATE TABLE IF NOT EXISTS skill_pricing (
+    skill_id                 TEXT PRIMARY KEY,
+    daily_price_chf          NUMERIC(10,5) NOT NULL,
+    created_at               TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at               TIMESTAMPTZ NOT NULL DEFAULT now()
+  );
+
+  -- One row per tenant. `created_at` anchors first-month proration;
+  -- `deleted_at` (nullable, stamped on delete) anchors last-month
+  -- proration. The PiecedTenant CR is the source of truth for
+  -- existence, but once the CR is deleted we lose its
+  -- creationTimestamp — so we mirror those two bookends here.
+  CREATE TABLE IF NOT EXISTS tenant_billing_lifecycle (
+    tenant_name              TEXT PRIMARY KEY,
+    zitadel_org_id           TEXT NOT NULL,
+    created_at               TIMESTAMPTZ NOT NULL,
+    deleted_at               TIMESTAMPTZ
+  );
+  CREATE INDEX IF NOT EXISTS idx_tenant_billing_lifecycle_org
+    ON tenant_billing_lifecycle(zitadel_org_id);
+
+  -- Skill enable/disable events. One row per state change; same-day
+  -- toggles still produce multiple rows, and the billing computation
+  -- collapses to distinct UTC days at compute time. This append-only
+  -- log preserves history for audit and lets us re-bill historical
+  -- months reproducibly.
+  --
+  -- `skill_id` is the package id from PACKAGE_CATALOG. We store
+  -- events for ALL package toggles, not just skill-category — the
+  -- channel/core toggles are cheap to record and may become billable
+  -- in the future without a schema change.
+  CREATE TABLE IF NOT EXISTS tenant_skill_events (
+    id                       BIGSERIAL PRIMARY KEY,
+    tenant_name              TEXT NOT NULL,
+    zitadel_org_id           TEXT NOT NULL,
+    skill_id                 TEXT NOT NULL,
+    event_kind               TEXT NOT NULL CHECK (event_kind IN ('enabled', 'disabled')),
+    occurred_at              TIMESTAMPTZ NOT NULL DEFAULT now()
+  );
+  CREATE INDEX IF NOT EXISTS idx_tenant_skill_events_tenant_skill
+    ON tenant_skill_events(tenant_name, skill_id, occurred_at);
+  CREATE INDEX IF NOT EXISTS idx_tenant_skill_events_org_time
+    ON tenant_skill_events(zitadel_org_id, occurred_at);
+
+  -- Suspend/resume transitions. Same shape as skill events. Reading
+  -- these in order reconstructs the suspended-state segments for
+  -- monthly fee proration: a tenant in 'suspended' state pays no
+  -- monthly fee for the days it was suspended.
+  --
+  -- The portal commands the transition (PATCH spec.suspend); the
+  -- operator observes and stamps PiecedTenantStatus.suspendedAt
+  -- after reconcile. We record the event at command time — billing
+  -- is monthly so the few-second reconcile lag is irrelevant.
+  CREATE TABLE IF NOT EXISTS tenant_suspension_events (
+    id                       BIGSERIAL PRIMARY KEY,
+    tenant_name              TEXT NOT NULL,
+    zitadel_org_id           TEXT NOT NULL,
+    event_kind               TEXT NOT NULL CHECK (event_kind IN ('suspended','resumed')),
+    occurred_at              TIMESTAMPTZ NOT NULL DEFAULT now()
+  );
+  CREATE INDEX IF NOT EXISTS idx_tenant_suspension_events_tenant
+    ON tenant_suspension_events(tenant_name, occurred_at);
+
+  -- Per-org billing configuration. Distinct from org_billing
+  -- (address/VAT/email): that table is customer-editable, this one
+  -- is admin-controlled and holds the payment posture.
+  --
+  -- Defaults: a new org has pay_by_invoice = false (must use a
+  -- credit card per the onboarding gate in Phase 4) and auto
+  -- billing/reminders enabled. Admin can flip pay_by_invoice on
+  -- per customer, after which approval no longer requires a card.
+  CREATE TABLE IF NOT EXISTS org_billing_config (
+    zitadel_org_id           TEXT PRIMARY KEY,
+    pay_by_invoice           BOOLEAN NOT NULL DEFAULT FALSE,
+    stripe_customer_id       TEXT,
+    auto_invoice_enabled     BOOLEAN NOT NULL DEFAULT TRUE,
+    auto_reminders_enabled   BOOLEAN NOT NULL DEFAULT TRUE,
+    created_at               TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at               TIMESTAMPTZ NOT NULL DEFAULT now()
+  );
+
+  -- Stripe payment methods. Populated by the Phase 4 webhook handler.
+  -- Created in Phase 1 so all billing schema is together; rows are
+  -- empty until Phase 4 ships.
+  CREATE TABLE IF NOT EXISTS org_payment_methods (
+    id                       UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    zitadel_org_id           TEXT NOT NULL,
+    stripe_payment_method_id TEXT NOT NULL UNIQUE,
+    brand                    TEXT,
+    last4                    TEXT,
+    exp_month                INT,
+    exp_year                 INT,
+    is_default               BOOLEAN NOT NULL DEFAULT FALSE,
+    created_at               TIMESTAMPTZ NOT NULL DEFAULT now()
+  );
+  CREATE INDEX IF NOT EXISTS idx_org_payment_methods_org
+    ON org_payment_methods(zitadel_org_id);
+  -- At most one default payment method per org. Partial unique index
+  -- so non-default rows don't conflict with each other.
+  CREATE UNIQUE INDEX IF NOT EXISTS uniq_org_payment_methods_default
+    ON org_payment_methods(zitadel_org_id) WHERE is_default = TRUE;
+
+  -- Gapless per-year invoice number counter (Art. 957a OR
+  -- compliance). A Postgres SEQUENCE would be faster but allows
+  -- gaps on rollback; this counter table is SELECT FOR UPDATE-able
+  -- and produces gapless numbers when the invoice insert is in the
+  -- same transaction. Populated lazily — the first invoice of each
+  -- year inserts its row.
+  CREATE TABLE IF NOT EXISTS invoice_number_counters (
+    year                     INT PRIMARY KEY,
+    last_number              INT NOT NULL DEFAULT 0
+  );
+
+  -- Issued invoices. Immutable once status leaves 'draft'.
+  -- billing_snapshot captures the address/VAT/email at issue time
+  -- so subsequent edits to org_billing don't mutate historical
+  -- invoices.
+  CREATE TABLE IF NOT EXISTS invoices (
+    id                       UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    invoice_number           TEXT NOT NULL UNIQUE,
+    zitadel_org_id           TEXT NOT NULL,
+    -- Billing period as DATEs (not timestamps): a calendar month.
+    -- period_end is the last day of the month, inclusive.
+    period_start             DATE NOT NULL,
+    period_end               DATE NOT NULL,
+    issued_at                TIMESTAMPTZ NOT NULL DEFAULT now(),
+    due_at                   DATE NOT NULL,
+    subtotal_chf             NUMERIC(10,2) NOT NULL,
+    vat_rate                 NUMERIC(5,2) NOT NULL,
+    vat_amount_chf           NUMERIC(10,2) NOT NULL,
+    total_chf                NUMERIC(10,2) NOT NULL,
+    status                   TEXT NOT NULL DEFAULT 'open' CHECK (
+      status IN ('draft','open','paid','overdue','void','uncollectible')
+    ),
+    billing_snapshot         JSONB NOT NULL,
+    payment_method           TEXT NOT NULL CHECK (payment_method IN ('invoice','card')),
+    stripe_payment_intent_id TEXT,
+    pdf_data                 BYTEA,
+    pdf_filename             TEXT,
+    admin_notes              TEXT,
+    paid_at                  TIMESTAMPTZ,
+    paid_by                  TEXT,
+    paid_method_detail       TEXT,
+    created_at               TIMESTAMPTZ NOT NULL DEFAULT now()
+  );
+  CREATE INDEX IF NOT EXISTS idx_invoices_org
+    ON invoices(zitadel_org_id, issued_at DESC);
+  CREATE INDEX IF NOT EXISTS idx_invoices_status
+    ON invoices(status, due_at);
+  -- One invoice per org per billing month — protects the monthly
+  -- cron from double-issuing if it gets retried mid-run.
+  CREATE UNIQUE INDEX IF NOT EXISTS uniq_invoices_org_period
+    ON invoices(zitadel_org_id, period_start);
+
+  -- Invoice line items. `kind` lets the PDF renderer group lines
+  -- (all monthly fees together, all AI usage together, etc.) and
+  -- the admin UI filter by category.
+  CREATE TABLE IF NOT EXISTS invoice_lines (
+    id                       UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    invoice_id               UUID NOT NULL REFERENCES invoices(id) ON DELETE CASCADE,
+    -- NULL for org-wide items; tenant name for per-tenant breakdowns.
+    tenant_name              TEXT,
+    kind                     TEXT NOT NULL CHECK (kind IN (
+      'tenant_monthly','tenant_setup','ai_usage','threema_messages','skill_usage','adjustment'
+    )),
+    description              TEXT NOT NULL,
+    quantity                 NUMERIC(12,4) NOT NULL DEFAULT 1,
+    unit_label               TEXT,
+    unit_price_chf           NUMERIC(10,5) NOT NULL,
+    amount_chf               NUMERIC(10,2) NOT NULL,
+    -- Per-kind audit metadata (e.g. {proration_days, days_in_month}
+    -- for tenant_monthly; {in_count, out_count} for threema_messages).
+    metadata                 JSONB,
+    display_order            INT NOT NULL DEFAULT 0
+  );
+  CREATE INDEX IF NOT EXISTS idx_invoice_lines_invoice
+    ON invoice_lines(invoice_id, display_order);
+
+  -- Reminders fired against open/overdue invoices. Level 3 = final.
+  -- One PDF per reminder, stored alongside.
+  CREATE TABLE IF NOT EXISTS invoice_reminders (
+    id                       UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    invoice_id               UUID NOT NULL REFERENCES invoices(id) ON DELETE CASCADE,
+    level                    INT NOT NULL CHECK (level IN (1, 2, 3)),
+    sent_at                  TIMESTAMPTZ NOT NULL DEFAULT now(),
+    sent_by                  TEXT NOT NULL,
+    pdf_data                 BYTEA,
+    pdf_filename             TEXT,
+    email_sent_to            TEXT,
+    UNIQUE (invoice_id, level)
+  );
+  CREATE INDEX IF NOT EXISTS idx_invoice_reminders_invoice
+    ON invoice_reminders(invoice_id, level);
 `;
 
 let migrated = false;
@@ -1336,3 +1582,545 @@ export async function updateSupportTicket(
   );
   return result.rows.length > 0 ? rowToSupportTicket(result.rows[0]) : null;
 }
+
+// ---------------------------------------------------------------------------
+// Billing — Phase 1: pricing, lifecycle, and skill events
+// ---------------------------------------------------------------------------
+//
+// All helpers are intentionally narrow CRUD — no business logic.
+// Higher-level operations (compute monthly proration, build an
+// invoice from raw signals) belong in a future lib/billing.ts
+// introduced by Phase 2.
+//
+// Hook callers should treat every write here as best-effort: if
+// recording a lifecycle/event row fails, log and continue — never
+// fail the underlying K8s mutation. Drift is corrected by the
+// idempotent backfill helper at the bottom of this section.
+
+import type {
+  PlatformPricing,
+  SkillPricing,
+  OrgBillingConfig,
+  TenantBillingLifecycle,
+  TenantSkillEvent,
+  TenantSuspensionEvent,
+} from "@/types";
+
+// --- platform_pricing ------------------------------------------------------
+
+function rowToPlatformPricing(row: any): PlatformPricing {
+  return {
+    tenantMonthlyFeeChf: Number(row.tenant_monthly_fee_chf),
+    tenantSetupFeeChf: Number(row.tenant_setup_fee_chf),
+    threemaMessageChf: Number(row.threema_message_chf),
+    vatRateChli: Number(row.vat_rate_chli),
+    updatedAt: row.updated_at?.toISOString?.() ?? row.updated_at,
+  };
+}
+
+/**
+ * Read the single-row platform pricing config. The migration seeds
+ * the row with zeros on first run, so this never returns null on a
+ * properly migrated database.
+ */
+export async function getPlatformPricing(): Promise<PlatformPricing> {
+  await ensureSchema();
+  const result = await getPool().query(
+    "SELECT * FROM platform_pricing WHERE id = 1"
+  );
+  if (result.rows.length === 0) {
+    // Defensive: re-seed if the row went missing (manual DELETE,
+    // partial restore, etc.). The migration's INSERT ON CONFLICT
+    // should have ensured this, but a stale row state shouldn't
+    // crash the helper.
+    await getPool().query(
+      "INSERT INTO platform_pricing (id) VALUES (1) ON CONFLICT DO NOTHING"
+    );
+    const retry = await getPool().query(
+      "SELECT * FROM platform_pricing WHERE id = 1"
+    );
+    return rowToPlatformPricing(retry.rows[0]);
+  }
+  return rowToPlatformPricing(result.rows[0]);
+}
+
+/**
+ * Update one or more pricing fields. Pass only the fields to change;
+ * unspecified fields are left as-is. `updated_at` is always refreshed.
+ */
+export async function updatePlatformPricing(changes: {
+  tenantMonthlyFeeChf?: number;
+  tenantSetupFeeChf?: number;
+  threemaMessageChf?: number;
+  vatRateChli?: number;
+}): Promise<PlatformPricing> {
+  await ensureSchema();
+  const sets: string[] = ["updated_at = now()"];
+  const values: any[] = [];
+  let idx = 1;
+  if (changes.tenantMonthlyFeeChf !== undefined) {
+    sets.push(`tenant_monthly_fee_chf = $${idx++}`);
+    values.push(changes.tenantMonthlyFeeChf);
+  }
+  if (changes.tenantSetupFeeChf !== undefined) {
+    sets.push(`tenant_setup_fee_chf = $${idx++}`);
+    values.push(changes.tenantSetupFeeChf);
+  }
+  if (changes.threemaMessageChf !== undefined) {
+    sets.push(`threema_message_chf = $${idx++}`);
+    values.push(changes.threemaMessageChf);
+  }
+  if (changes.vatRateChli !== undefined) {
+    sets.push(`vat_rate_chli = $${idx++}`);
+    values.push(changes.vatRateChli);
+  }
+  // Only updated_at would change → still execute so the caller's
+  // intent ("touch the row") isn't silently dropped, but make it
+  // an explicit no-op if no fields were provided.
+  if (sets.length === 1 && values.length === 0) {
+    return getPlatformPricing();
+  }
+  const result = await getPool().query(
+    `UPDATE platform_pricing SET ${sets.join(", ")} WHERE id = 1 RETURNING *`,
+    values
+  );
+  return rowToPlatformPricing(result.rows[0]);
+}
+
+// --- skill_pricing ---------------------------------------------------------
+
+function rowToSkillPricing(row: any): SkillPricing {
+  return {
+    skillId: row.skill_id,
+    dailyPriceChf: Number(row.daily_price_chf),
+    createdAt: row.created_at?.toISOString?.() ?? row.created_at,
+    updatedAt: row.updated_at?.toISOString?.() ?? row.updated_at,
+  };
+}
+
+export async function listSkillPricing(): Promise<SkillPricing[]> {
+  await ensureSchema();
+  const result = await getPool().query(
+    "SELECT * FROM skill_pricing ORDER BY skill_id"
+  );
+  return result.rows.map(rowToSkillPricing);
+}
+
+export async function getSkillPricing(
+  skillId: string
+): Promise<SkillPricing | null> {
+  await ensureSchema();
+  const result = await getPool().query(
+    "SELECT * FROM skill_pricing WHERE skill_id = $1",
+    [skillId]
+  );
+  return result.rows.length > 0 ? rowToSkillPricing(result.rows[0]) : null;
+}
+
+/**
+ * Upsert a daily price for a package. Setting a price activates
+ * usage-based billing for the (tenant, skill) pair: every UTC day
+ * the package was enabled in the billing month is one unit on the
+ * invoice.
+ */
+export async function setSkillPricing(
+  skillId: string,
+  dailyPriceChf: number
+): Promise<SkillPricing> {
+  await ensureSchema();
+  const result = await getPool().query(
+    `INSERT INTO skill_pricing (skill_id, daily_price_chf)
+     VALUES ($1, $2)
+     ON CONFLICT (skill_id) DO UPDATE SET
+       daily_price_chf = EXCLUDED.daily_price_chf,
+       updated_at      = now()
+     RETURNING *`,
+    [skillId, dailyPriceChf]
+  );
+  return rowToSkillPricing(result.rows[0]);
+}
+
+/**
+ * Remove the price for a package. Day-tracking events continue to
+ * be recorded (cheap append-only) but the package becomes free
+ * effective immediately. Historical invoices already issued are
+ * unaffected.
+ */
+export async function removeSkillPricing(skillId: string): Promise<void> {
+  await ensureSchema();
+  await getPool().query("DELETE FROM skill_pricing WHERE skill_id = $1", [
+    skillId,
+  ]);
+}
+
+// --- tenant_billing_lifecycle ---------------------------------------------
+
+function rowToTenantBillingLifecycle(row: any): TenantBillingLifecycle {
+  return {
+    tenantName: row.tenant_name,
+    zitadelOrgId: row.zitadel_org_id,
+    createdAt: row.created_at?.toISOString?.() ?? row.created_at,
+    deletedAt: row.deleted_at?.toISOString?.() ?? row.deleted_at ?? null,
+  };
+}
+
+export async function getTenantBillingLifecycle(
+  tenantName: string
+): Promise<TenantBillingLifecycle | null> {
+  await ensureSchema();
+  const result = await getPool().query(
+    "SELECT * FROM tenant_billing_lifecycle WHERE tenant_name = $1",
+    [tenantName]
+  );
+  return result.rows.length > 0
+    ? rowToTenantBillingLifecycle(result.rows[0])
+    : null;
+}
+
+/**
+ * Record a tenant's creation for billing purposes. Idempotent on
+ * `tenant_name` — re-running with a different created_at is a no-op
+ * so re-approvals don't move the proration anchor. Pair with
+ * recordInitialSkillEvents() at the same call site.
+ */
+export async function recordTenantCreated(
+  tenantName: string,
+  zitadelOrgId: string,
+  createdAt?: Date
+): Promise<void> {
+  await ensureSchema();
+  await getPool().query(
+    `INSERT INTO tenant_billing_lifecycle (tenant_name, zitadel_org_id, created_at)
+     VALUES ($1, $2, COALESCE($3::timestamptz, now()))
+     ON CONFLICT (tenant_name) DO NOTHING`,
+    [tenantName, zitadelOrgId, createdAt ?? null]
+  );
+}
+
+/**
+ * Stamp deletion timestamp. Idempotent — calling twice keeps the
+ * first deletion's timestamp. Pair with closing-skill-disabled
+ * events at the same call site if you want the events log to
+ * reflect that everything is now off.
+ *
+ * We deliberately don't delete the row — the lifecycle record is
+ * needed for any final invoice covering the deletion month.
+ */
+export async function recordTenantDeleted(
+  tenantName: string,
+  deletedAt?: Date
+): Promise<void> {
+  await ensureSchema();
+  await getPool().query(
+    `UPDATE tenant_billing_lifecycle
+       SET deleted_at = COALESCE(deleted_at, COALESCE($2::timestamptz, now()))
+     WHERE tenant_name = $1`,
+    [tenantName, deletedAt ?? null]
+  );
+}
+
+// --- tenant_skill_events --------------------------------------------------
+
+function rowToTenantSkillEvent(row: any): TenantSkillEvent {
+  return {
+    id: String(row.id),
+    tenantName: row.tenant_name,
+    zitadelOrgId: row.zitadel_org_id,
+    skillId: row.skill_id,
+    eventKind: row.event_kind as "enabled" | "disabled",
+    occurredAt: row.occurred_at?.toISOString?.() ?? row.occurred_at,
+  };
+}
+
+/**
+ * Append a batch of enabled/disabled events. Single multi-row INSERT
+ * so all rows share an effectively-identical occurred_at when
+ * `occurredAt` is omitted (otherwise they'd be a few microseconds
+ * apart, which would skew the "is this skill on for day D" computation
+ * around midnight UTC).
+ *
+ * Empty arrays are a no-op (no SQL fired).
+ */
+export async function recordSkillEvents(
+  tenantName: string,
+  zitadelOrgId: string,
+  added: string[],
+  removed: string[],
+  occurredAt?: Date
+): Promise<void> {
+  if (added.length === 0 && removed.length === 0) return;
+  await ensureSchema();
+  const at = occurredAt ?? new Date();
+  // Build placeholders. Each row uses 5 placeholders.
+  const values: any[] = [];
+  const rows: string[] = [];
+  let idx = 1;
+  for (const skillId of added) {
+    rows.push(`($${idx++}, $${idx++}, $${idx++}, 'enabled', $${idx++})`);
+    values.push(tenantName, zitadelOrgId, skillId, at);
+  }
+  for (const skillId of removed) {
+    rows.push(`($${idx++}, $${idx++}, $${idx++}, 'disabled', $${idx++})`);
+    values.push(tenantName, zitadelOrgId, skillId, at);
+  }
+  await getPool().query(
+    `INSERT INTO tenant_skill_events
+       (tenant_name, zitadel_org_id, skill_id, event_kind, occurred_at)
+     VALUES ${rows.join(", ")}`,
+    values
+  );
+}
+
+/**
+ * Read events for a tenant within a half-open interval — `from`
+ * inclusive, `to` exclusive. Used by Phase 2's billing computation
+ * to collapse to billable days. Returned in chronological order.
+ */
+export async function listSkillEventsForTenant(
+  tenantName: string,
+  from: Date,
+  to: Date
+): Promise<TenantSkillEvent[]> {
+  await ensureSchema();
+  const result = await getPool().query(
+    `SELECT * FROM tenant_skill_events
+      WHERE tenant_name = $1
+        AND occurred_at >= $2
+        AND occurred_at <  $3
+      ORDER BY occurred_at, id`,
+    [tenantName, from, to]
+  );
+  return result.rows.map(rowToTenantSkillEvent);
+}
+
+/**
+ * Read the most recent event for each (tenant, skill) pair as of
+ * a moment in time. Phase 2 uses this to know which skills were
+ * enabled at the start of a billing window.
+ *
+ * Implemented with DISTINCT ON for a single round-trip; the
+ * (tenant_name, skill_id, occurred_at) index supports the sort.
+ */
+export async function getSkillStateAt(
+  tenantName: string,
+  asOf: Date
+): Promise<Record<string, "enabled" | "disabled">> {
+  await ensureSchema();
+  const result = await getPool().query(
+    `SELECT DISTINCT ON (skill_id) skill_id, event_kind
+       FROM tenant_skill_events
+      WHERE tenant_name = $1 AND occurred_at <= $2
+      ORDER BY skill_id, occurred_at DESC, id DESC`,
+    [tenantName, asOf]
+  );
+  const out: Record<string, "enabled" | "disabled"> = {};
+  for (const row of result.rows) {
+    out[row.skill_id] = row.event_kind as "enabled" | "disabled";
+  }
+  return out;
+}
+
+// --- tenant_suspension_events ---------------------------------------------
+
+function rowToTenantSuspensionEvent(row: any): TenantSuspensionEvent {
+  return {
+    id: String(row.id),
+    tenantName: row.tenant_name,
+    zitadelOrgId: row.zitadel_org_id,
+    eventKind: row.event_kind as "suspended" | "resumed",
+    occurredAt: row.occurred_at?.toISOString?.() ?? row.occurred_at,
+  };
+}
+
+export async function recordSuspensionEvent(
+  tenantName: string,
+  zitadelOrgId: string,
+  eventKind: "suspended" | "resumed",
+  occurredAt?: Date
+): Promise<void> {
+  await ensureSchema();
+  await getPool().query(
+    `INSERT INTO tenant_suspension_events
+       (tenant_name, zitadel_org_id, event_kind, occurred_at)
+     VALUES ($1, $2, $3, COALESCE($4::timestamptz, now()))`,
+    [tenantName, zitadelOrgId, eventKind, occurredAt ?? null]
+  );
+}
+
+export async function listSuspensionEventsForTenant(
+  tenantName: string,
+  from: Date,
+  to: Date
+): Promise<TenantSuspensionEvent[]> {
+  await ensureSchema();
+  const result = await getPool().query(
+    `SELECT * FROM tenant_suspension_events
+      WHERE tenant_name = $1
+        AND occurred_at >= $2
+        AND occurred_at <  $3
+      ORDER BY occurred_at, id`,
+    [tenantName, from, to]
+  );
+  return result.rows.map(rowToTenantSuspensionEvent);
+}
+
+// --- org_billing_config ---------------------------------------------------
+
+function rowToOrgBillingConfig(row: any): OrgBillingConfig {
+  return {
+    zitadelOrgId: row.zitadel_org_id,
+    payByInvoice: row.pay_by_invoice,
+    stripeCustomerId: row.stripe_customer_id ?? null,
+    autoInvoiceEnabled: row.auto_invoice_enabled,
+    autoRemindersEnabled: row.auto_reminders_enabled,
+    createdAt: row.created_at?.toISOString?.() ?? row.created_at,
+    updatedAt: row.updated_at?.toISOString?.() ?? row.updated_at,
+  };
+}
+
+/**
+ * Get config for an org, auto-creating with defaults if missing.
+ * Returning the row (vs null) simplifies callers: the gate logic in
+ * Phase 4 ("approve only if pay_by_invoice OR has card") doesn't
+ * need a "what if no row" branch.
+ *
+ * Defaults are baked into the table's column defaults, so the
+ * INSERT here only needs the primary key.
+ */
+export async function getOrgBillingConfig(
+  zitadelOrgId: string
+): Promise<OrgBillingConfig> {
+  await ensureSchema();
+  await getPool().query(
+    `INSERT INTO org_billing_config (zitadel_org_id)
+     VALUES ($1)
+     ON CONFLICT (zitadel_org_id) DO NOTHING`,
+    [zitadelOrgId]
+  );
+  const result = await getPool().query(
+    "SELECT * FROM org_billing_config WHERE zitadel_org_id = $1",
+    [zitadelOrgId]
+  );
+  return rowToOrgBillingConfig(result.rows[0]);
+}
+
+export async function updateOrgBillingConfig(
+  zitadelOrgId: string,
+  changes: {
+    payByInvoice?: boolean;
+    stripeCustomerId?: string | null;
+    autoInvoiceEnabled?: boolean;
+    autoRemindersEnabled?: boolean;
+  }
+): Promise<OrgBillingConfig> {
+  await ensureSchema();
+  // Ensure row exists first — mirrors getOrgBillingConfig's
+  // auto-create.
+  await getPool().query(
+    `INSERT INTO org_billing_config (zitadel_org_id)
+     VALUES ($1)
+     ON CONFLICT (zitadel_org_id) DO NOTHING`,
+    [zitadelOrgId]
+  );
+  const sets: string[] = ["updated_at = now()"];
+  const values: any[] = [zitadelOrgId];
+  let idx = 2;
+  if (changes.payByInvoice !== undefined) {
+    sets.push(`pay_by_invoice = $${idx++}`);
+    values.push(changes.payByInvoice);
+  }
+  if (changes.stripeCustomerId !== undefined) {
+    sets.push(`stripe_customer_id = $${idx++}`);
+    values.push(changes.stripeCustomerId);
+  }
+  if (changes.autoInvoiceEnabled !== undefined) {
+    sets.push(`auto_invoice_enabled = $${idx++}`);
+    values.push(changes.autoInvoiceEnabled);
+  }
+  if (changes.autoRemindersEnabled !== undefined) {
+    sets.push(`auto_reminders_enabled = $${idx++}`);
+    values.push(changes.autoRemindersEnabled);
+  }
+  const result = await getPool().query(
+    `UPDATE org_billing_config
+        SET ${sets.join(", ")}
+      WHERE zitadel_org_id = $1
+      RETURNING *`,
+    values
+  );
+  return rowToOrgBillingConfig(result.rows[0]);
+}
+
+// --- Backfill -------------------------------------------------------------
+//
+// Idempotent one-time bootstrap for tenants that existed before
+// Phase 1 shipped. Phase 2 wraps this in an admin endpoint; for now
+// it can be invoked from a one-off node script.
+//
+// For each PiecedTenant CR:
+//   - If no tenant_billing_lifecycle row exists, insert one with
+//     created_at = metadata.creationTimestamp.
+//   - If no tenant_skill_events row exists for the tenant, insert
+//     'enabled' events at the same timestamp for every package
+//     currently in spec.packages.
+// Tenants suspended at backfill time get a 'suspended' event at
+// status.suspendedAt (operator-stamped); resumed tenants get nothing
+// extra (default state is "running").
+
+/**
+ * Returns a count of lifecycle rows inserted and skill events
+ * recorded — both expected to be zero on a second run.
+ */
+export async function backfillTenantBillingLifecycle(tenants: {
+  name: string;
+  zitadelOrgId: string;
+  createdAt: Date;
+  packages: string[];
+  suspendedAt: Date | null;
+}[]): Promise<{ lifecycleInserted: number; eventsInserted: number; suspensionEventsInserted: number }> {
+  await ensureSchema();
+  let lifecycleInserted = 0;
+  let eventsInserted = 0;
+  let suspensionEventsInserted = 0;
+  for (const t of tenants) {
+    // Lifecycle row — idempotent.
+    const existing = await getTenantBillingLifecycle(t.name);
+    if (!existing) {
+      await recordTenantCreated(t.name, t.zitadelOrgId, t.createdAt);
+      lifecycleInserted++;
+    }
+    // Initial skill events — only if the tenant has zero events at
+    // all. We don't want to add to an active event stream.
+    const eventsRow = await getPool().query(
+      "SELECT 1 FROM tenant_skill_events WHERE tenant_name = $1 LIMIT 1",
+      [t.name]
+    );
+    if (eventsRow.rows.length === 0 && t.packages.length > 0) {
+      await recordSkillEvents(
+        t.name,
+        t.zitadelOrgId,
+        t.packages,
+        [],
+        t.createdAt
+      );
+      eventsInserted += t.packages.length;
+    }
+    // Suspension state — only if the tenant has zero suspension
+    // events. If it's currently suspended, record one 'suspended'
+    // event at the operator-stamped time so proration sees it.
+    const susRow = await getPool().query(
+      "SELECT 1 FROM tenant_suspension_events WHERE tenant_name = $1 LIMIT 1",
+      [t.name]
+    );
+    if (susRow.rows.length === 0 && t.suspendedAt) {
+      await recordSuspensionEvent(
+        t.name,
+        t.zitadelOrgId,
+        "suspended",
+        t.suspendedAt
+      );
+      suspensionEventsInserted++;
+    }
+  }
+  return { lifecycleInserted, eventsInserted, suspensionEventsInserted };
+}