Initial import of zappier-edge from zapier monorepo

src/db/billing-repo.ts

@@ -0,0 +1,68 @@
+import Database from 'better-sqlite3';
+
+export interface BillingReportRepo {
+  getReportedCents(customerId: string, period: string): number;
+  upsertReportedCents(customerId: string, period: string, cumulativeCents: number): void;
+}
+
+export interface JobLockRepo {
+  tryAcquireLock(name: string, ttlMs: number): boolean;
+  releaseLock(name: string): void;
+}
+
+export class SqliteBillingReportRepo implements BillingReportRepo, JobLockRepo {
+  constructor(
+    private db: Database.Database,
+    private now: () => number = Date.now,
+  ) {
+    this.db.exec(`
+      CREATE TABLE IF NOT EXISTS billing_reports (
+        customer_id TEXT NOT NULL,
+        period TEXT NOT NULL,
+        reported_cents INTEGER NOT NULL,
+        reported_at_ms INTEGER NOT NULL,
+        PRIMARY KEY (customer_id, period)
+      )
+    `);
+    this.db.exec(`
+      CREATE TABLE IF NOT EXISTS job_locks (
+        name TEXT PRIMARY KEY,
+        acquired_at_ms INTEGER NOT NULL
+      )
+    `);
+  }
+
+  getReportedCents(customerId: string, period: string): number {
+    const row = this.db
+      .prepare('SELECT reported_cents FROM billing_reports WHERE customer_id = ? AND period = ?')
+      .get(customerId, period) as { reported_cents: number } | undefined;
+    return row?.reported_cents ?? 0;
+  }
+
+  upsertReportedCents(customerId: string, period: string, cumulativeCents: number): void {
+    this.db
+      .prepare(
+        `INSERT INTO billing_reports (customer_id, period, reported_cents, reported_at_ms)
+         VALUES (?, ?, ?, ?)
+         ON CONFLICT(customer_id, period) DO UPDATE SET
+           reported_cents = excluded.reported_cents,
+           reported_at_ms = excluded.reported_at_ms`,
+      )
+      .run(customerId, period, cumulativeCents, this.now());
+  }
+
+  tryAcquireLock(name: string, ttlMs: number): boolean {
+    const result = this.db
+      .prepare(
+        `INSERT INTO job_locks (name, acquired_at_ms) VALUES (?, ?)
+         ON CONFLICT(name) DO UPDATE SET acquired_at_ms = excluded.acquired_at_ms
+         WHERE job_locks.acquired_at_ms <= excluded.acquired_at_ms - ?`,
+      )
+      .run(name, this.now(), ttlMs);
+    return result.changes === 1;
+  }
+
+  releaseLock(name: string): void {
+    this.db.prepare('DELETE FROM job_locks WHERE name = ?').run(name);
+  }
+}