Phase2: Invoicecomputation/AdminpricingUI/Ainvoicemgnt

src/lib/db.ts

@@ -470,6 +470,12 @@ const MIGRATION_SQL = `
     paid_method_detail       TEXT,
     created_at               TIMESTAMPTZ NOT NULL DEFAULT now()
   );
+  -- Phase 2 addition: PDF locale, frozen at issue time so re-rendering
+  -- an old invoice produces an identical document. Defaults to 'de'
+  -- since most pilot customers are Swiss B2B; the generator UI lets
+  -- admin override at issue time.
+  ALTER TABLE invoices
+    ADD COLUMN IF NOT EXISTS locale TEXT NOT NULL DEFAULT 'de';
   CREATE INDEX IF NOT EXISTS idx_invoices_org
     ON invoices(zitadel_org_id, issued_at DESC);
   CREATE INDEX IF NOT EXISTS idx_invoices_status
@@ -2124,3 +2130,424 @@ export async function backfillTenantBillingLifecycle(tenants: {
   }
   return { lifecycleInserted, eventsInserted, suspensionEventsInserted };
 }
+
+// ---------------------------------------------------------------------------
+// Billing — Phase 2: invoice persistence
+// ---------------------------------------------------------------------------
+//
+// Invoice creation is intentionally a single transaction: allocate
+// number, INSERT invoice, INSERT lines, store PDF — all-or-nothing.
+// The Postgres invoice_number_counters row lock serializes
+// concurrent allocators for the same year, producing gapless
+// numbering even under bursts.
+
+import type {
+  Invoice,
+  InvoiceBillingSnapshot,
+  InvoiceDetail,
+  InvoiceDraft,
+  InvoiceLine,
+  InvoiceStatus,
+} from "@/types";
+
+function rowToInvoice(row: any): Invoice {
+  return {
+    id: row.id,
+    invoiceNumber: row.invoice_number,
+    zitadelOrgId: row.zitadel_org_id,
+    periodStart: typeof row.period_start === "string"
+      ? row.period_start
+      : row.period_start.toISOString().split("T")[0],
+    periodEnd: typeof row.period_end === "string"
+      ? row.period_end
+      : row.period_end.toISOString().split("T")[0],
+    issuedAt: row.issued_at?.toISOString?.() ?? row.issued_at,
+    dueAt: typeof row.due_at === "string"
+      ? row.due_at
+      : row.due_at.toISOString().split("T")[0],
+    subtotalChf: Number(row.subtotal_chf),
+    vatRate: Number(row.vat_rate),
+    vatAmountChf: Number(row.vat_amount_chf),
+    totalChf: Number(row.total_chf),
+    status: row.status as InvoiceStatus,
+    locale: row.locale ?? "de",
+    paymentMethod: row.payment_method,
+    billingSnapshot: row.billing_snapshot as InvoiceBillingSnapshot,
+    stripePaymentIntentId: row.stripe_payment_intent_id ?? null,
+    pdfFilename: row.pdf_filename ?? null,
+    hasPdf: row.has_pdf ?? row.pdf_data !== null,
+    adminNotes: row.admin_notes ?? null,
+    paidAt: row.paid_at?.toISOString?.() ?? row.paid_at ?? null,
+    paidBy: row.paid_by ?? null,
+    paidMethodDetail: row.paid_method_detail ?? null,
+    createdAt: row.created_at?.toISOString?.() ?? row.created_at,
+  };
+}
+
+function rowToInvoiceLine(row: any): InvoiceLine {
+  return {
+    id: row.id,
+    invoiceId: row.invoice_id,
+    tenantName: row.tenant_name ?? null,
+    kind: row.kind,
+    description: row.description,
+    quantity: Number(row.quantity),
+    unitLabel: row.unit_label ?? null,
+    unitPriceChf: Number(row.unit_price_chf),
+    amountChf: Number(row.amount_chf),
+    metadata: row.metadata ?? null,
+    displayOrder: row.display_order,
+  };
+}
+
+// Standard SELECT projection that includes a cheap NOT-NULL probe of
+// pdf_data instead of pulling the bytes themselves. Crucial for list
+// endpoints — a few KB per row across hundreds of invoices is wasted
+// network and memory.
+const INVOICE_LIST_COLUMNS = `
+  id, invoice_number, zitadel_org_id, period_start, period_end,
+  issued_at, due_at, subtotal_chf, vat_rate, vat_amount_chf,
+  total_chf, status, locale, payment_method, billing_snapshot,
+  stripe_payment_intent_id, pdf_filename, admin_notes, paid_at,
+  paid_by, paid_method_detail, created_at,
+  (pdf_data IS NOT NULL) AS has_pdf
+`;
+
+/**
+ * Persist a fully-computed invoice draft with its lines and PDF in
+ * a single transaction. Allocates the year-scoped invoice number
+ * inside the same transaction so a rollback restores the counter
+ * (gapless guarantee).
+ *
+ * The caller is responsible for upstream validation:
+ *   - the (org, period) uniqueness (the unique index will reject
+ *     duplicates, but we return a clear error message rather than
+ *     leaking the constraint name)
+ *   - the draft's lines/totals are consistent (compute pipeline
+ *     ensures this)
+ *
+ * `pdfBuffer` is the rendered PDF bytes; pass null if PDF is
+ * generated separately or stored in a side channel. For Phase 2 we
+ * always render synchronously and pass the buffer here.
+ */
+export async function createInvoice(
+  draft: InvoiceDraft,
+  pdfBuffer: Buffer | null,
+  pdfFilename: string | null
+): Promise<Invoice> {
+  await ensureSchema();
+  const pool = getPool();
+  const client = await pool.connect();
+  try {
+    await client.query("BEGIN");
+
+    // Allocate number for the year of period_start. Locking the
+    // counter row prevents concurrent allocators from racing.
+    const year = parseInt(draft.periodStart.slice(0, 4), 10);
+    const counterResult = await client.query(
+      `INSERT INTO invoice_number_counters (year, last_number)
+       VALUES ($1, 1)
+       ON CONFLICT (year) DO UPDATE SET
+         last_number = invoice_number_counters.last_number + 1
+       RETURNING last_number`,
+      [year]
+    );
+    const seq = counterResult.rows[0].last_number;
+    const invoiceNumber = `${year}-${String(seq).padStart(5, "0")}`;
+
+    // Insert invoice row. PDF goes inline as bytea for v1; we can
+    // migrate to MinIO/S3 later if storage gets noisy.
+    const inv = await client.query(
+      `INSERT INTO invoices (
+         invoice_number, zitadel_org_id, period_start, period_end,
+         issued_at, due_at, subtotal_chf, vat_rate, vat_amount_chf,
+         total_chf, status, locale, payment_method, billing_snapshot,
+         pdf_data, pdf_filename
+       ) VALUES (
+         $1, $2, $3::date, $4::date, now(), $5::date, $6, $7, $8, $9,
+         'open', $10, $11, $12::jsonb, $13, $14
+       )
+       RETURNING ${INVOICE_LIST_COLUMNS}`,
+      [
+        invoiceNumber,
+        draft.zitadelOrgId,
+        draft.periodStart,
+        draft.periodEnd,
+        draft.dueAt,
+        draft.subtotalChf,
+        draft.vatRate,
+        draft.vatAmountChf,
+        draft.totalChf,
+        draft.locale,
+        draft.paymentMethod,
+        JSON.stringify(draft.billingSnapshot),
+        pdfBuffer,
+        pdfFilename,
+      ]
+    );
+    const invoiceId = inv.rows[0].id;
+
+    // Insert lines in batch — one INSERT statement is significantly
+    // faster than per-line round-trips, which matters when an invoice
+    // accumulates many ai_usage / skill_usage lines.
+    if (draft.lines.length > 0) {
+      const placeholders: string[] = [];
+      const values: any[] = [];
+      let idx = 1;
+      for (const line of draft.lines) {
+        placeholders.push(
+          `($${idx++}, $${idx++}, $${idx++}, $${idx++}, $${idx++}, $${idx++}, $${idx++}, $${idx++}, $${idx++}::jsonb, $${idx++})`
+        );
+        values.push(
+          invoiceId,
+          line.tenantName,
+          line.kind,
+          line.description,
+          line.quantity,
+          line.unitLabel,
+          line.unitPriceChf,
+          line.amountChf,
+          line.metadata ? JSON.stringify(line.metadata) : null,
+          line.displayOrder
+        );
+      }
+      await client.query(
+        `INSERT INTO invoice_lines (
+           invoice_id, tenant_name, kind, description, quantity,
+           unit_label, unit_price_chf, amount_chf, metadata, display_order
+         ) VALUES ${placeholders.join(", ")}`,
+        values
+      );
+    }
+
+    await client.query("COMMIT");
+    return rowToInvoice(inv.rows[0]);
+  } catch (e: any) {
+    await client.query("ROLLBACK").catch(() => undefined);
+    // Translate the uniqueness violation into a user-friendly error.
+    // 23505 = unique_violation in Postgres.
+    if (e?.code === "23505" && /uniq_invoices_org_period/.test(e?.constraint ?? "")) {
+      const month = draft.periodStart.slice(0, 7);
+      throw new Error(
+        `An invoice already exists for this org and billing period (${month}). ` +
+          `Delete the existing invoice first if you want to regenerate.`
+      );
+    }
+    throw e;
+  } finally {
+    client.release();
+  }
+}
+
+export async function getInvoiceById(id: string): Promise<Invoice | null> {
+  await ensureSchema();
+  const result = await getPool().query(
+    `SELECT ${INVOICE_LIST_COLUMNS} FROM invoices WHERE id = $1`,
+    [id]
+  );
+  return result.rows.length > 0 ? rowToInvoice(result.rows[0]) : null;
+}
+
+export async function getInvoiceDetail(
+  id: string
+): Promise<InvoiceDetail | null> {
+  const invoice = await getInvoiceById(id);
+  if (!invoice) return null;
+  const lines = await getPool().query(
+    `SELECT * FROM invoice_lines WHERE invoice_id = $1
+     ORDER BY display_order, id`,
+    [id]
+  );
+  return { invoice, lines: lines.rows.map(rowToInvoiceLine) };
+}
+
+/**
+ * Fetch the PDF bytes for an invoice. Returns null if no PDF was
+ * stored (shouldn't happen in v1; defensive against partial state).
+ */
+export async function getInvoicePdf(
+  id: string
+): Promise<{ data: Buffer; filename: string } | null> {
+  await ensureSchema();
+  const result = await getPool().query(
+    "SELECT pdf_data, pdf_filename, invoice_number FROM invoices WHERE id = $1",
+    [id]
+  );
+  if (result.rows.length === 0) return null;
+  const row = result.rows[0];
+  if (!row.pdf_data) return null;
+  return {
+    data: row.pdf_data,
+    filename: row.pdf_filename ?? `${row.invoice_number}.pdf`,
+  };
+}
+
+/**
+ * List invoices, optionally filtered. Used by the admin invoice
+ * list page and (Phase 3) the customer-facing /billing page.
+ *
+ * The customer-facing call site MUST pass `zitadelOrgId` to scope
+ * results — this helper does not enforce that itself.
+ */
+export async function listInvoices(filters: {
+  zitadelOrgId?: string;
+  status?: InvoiceStatus;
+  /** Inclusive YYYY-MM filter on period_start. */
+  periodMonth?: string;
+  limit?: number;
+} = {}): Promise<Invoice[]> {
+  await ensureSchema();
+  const where: string[] = [];
+  const values: any[] = [];
+  let idx = 1;
+  if (filters.zitadelOrgId) {
+    where.push(`zitadel_org_id = $${idx++}`);
+    values.push(filters.zitadelOrgId);
+  }
+  if (filters.status) {
+    where.push(`status = $${idx++}`);
+    values.push(filters.status);
+  }
+  if (filters.periodMonth) {
+    where.push(`to_char(period_start, 'YYYY-MM') = $${idx++}`);
+    values.push(filters.periodMonth);
+  }
+  const limit = filters.limit ?? 200;
+  const sql =
+    `SELECT ${INVOICE_LIST_COLUMNS} FROM invoices ` +
+    (where.length > 0 ? `WHERE ${where.join(" AND ")} ` : "") +
+    `ORDER BY issued_at DESC LIMIT $${idx}`;
+  values.push(limit);
+  const result = await getPool().query(sql, values);
+  return result.rows.map(rowToInvoice);
+}
+
+/**
+ * Sweep open invoices past their due date to `overdue` status.
+ * Cheap idempotent UPDATE; safe to call on every admin list view
+ * to keep status fresh without a dedicated cron.
+ */
+export async function syncOverdueInvoices(): Promise<number> {
+  await ensureSchema();
+  const result = await getPool().query(
+    `UPDATE invoices
+        SET status = 'overdue'
+      WHERE status = 'open'
+        AND due_at < CURRENT_DATE`
+  );
+  return result.rowCount ?? 0;
+}
+
+export async function markInvoicePaid(
+  id: string,
+  opts: { paidBy: string; paidMethodDetail?: string | null; paidAt?: Date }
+): Promise<Invoice | null> {
+  await ensureSchema();
+  const result = await getPool().query(
+    `UPDATE invoices
+        SET status = 'paid',
+            paid_at = COALESCE($2::timestamptz, now()),
+            paid_by = $3,
+            paid_method_detail = $4
+      WHERE id = $1
+        AND status IN ('open', 'overdue')
+      RETURNING ${INVOICE_LIST_COLUMNS}`,
+    [
+      id,
+      opts.paidAt ?? null,
+      opts.paidBy,
+      opts.paidMethodDetail ?? null,
+    ]
+  );
+  return result.rows.length > 0 ? rowToInvoice(result.rows[0]) : null;
+}
+
+/**
+ * Hard delete an invoice and its lines (CASCADE).
+ *
+ * This is the testing tool — Swiss bookkeeping requires immutable
+ * invoices in production, but during pilot/testing we need to
+ * iterate. The gap left in the invoice number sequence is
+ * intentional and documented; no attempt to "recycle" numbers.
+ *
+ * Reminders (and their PDFs) cascade-delete via the FK.
+ */
+export async function deleteInvoice(id: string): Promise<boolean> {
+  await ensureSchema();
+  const result = await getPool().query(
+    "DELETE FROM invoices WHERE id = $1 RETURNING id",
+    [id]
+  );
+  return (result.rowCount ?? 0) > 0;
+}
+
+/**
+ * Has this tenant ever been billed a setup fee? Drives the
+ * compute pipeline's "include setup line on first invoice"
+ * decision. Looks at invoice_lines directly so it survives org
+ * billing config edits.
+ */
+export async function tenantHasSetupFeeBilled(
+  tenantName: string
+): Promise<boolean> {
+  await ensureSchema();
+  const result = await getPool().query(
+    `SELECT 1 FROM invoice_lines
+      WHERE tenant_name = $1 AND kind = 'tenant_setup'
+      LIMIT 1`,
+    [tenantName]
+  );
+  return result.rows.length > 0;
+}
+
+/**
+ * Aggregate open balance per org for the admin overview. Returns
+ * orgs with at least one open or overdue invoice; orgs in good
+ * standing don't appear.
+ */
+export async function getOrgOpenBalances(): Promise<{
+  zitadelOrgId: string;
+  openCount: number;
+  overdueCount: number;
+  totalOpenChf: number;
+}[]> {
+  await ensureSchema();
+  const result = await getPool().query(
+    `SELECT
+        zitadel_org_id,
+        COUNT(*) FILTER (WHERE status = 'open') AS open_count,
+        COUNT(*) FILTER (WHERE status = 'overdue') AS overdue_count,
+        SUM(total_chf) FILTER (WHERE status IN ('open', 'overdue')) AS total_open
+      FROM invoices
+      WHERE status IN ('open', 'overdue')
+      GROUP BY zitadel_org_id
+      ORDER BY total_open DESC`
+  );
+  return result.rows.map((r) => ({
+    zitadelOrgId: r.zitadel_org_id,
+    openCount: Number(r.open_count),
+    overdueCount: Number(r.overdue_count),
+    totalOpenChf: Number(r.total_open),
+  }));
+}
+
+/**
+ * Update the stored PDF for an invoice. Used by the two-pass
+ * compute pipeline: insert invoice with empty PDF → render PDF with
+ * the allocated invoice number → write bytes back.
+ *
+ * Could be merged into createInvoice via a render callback in a
+ * future cleanup, but two passes are simpler and the extra UPDATE
+ * is cheap.
+ */
+export async function updateInvoicePdf(
+  invoiceId: string,
+  pdfBuffer: Buffer,
+  filename: string
+): Promise<void> {
+  await ensureSchema();
+  await getPool().query(
+    "UPDATE invoices SET pdf_data = $2, pdf_filename = $3 WHERE id = $1",
+    [invoiceId, pdfBuffer, filename]
+  );
+}