fix(data-export): handle bigint hourly cursors

services/user-data-export/src/source-adapters.test.ts

@@ -2178,8 +2178,8 @@ describe('source adapters', () => {
 
 describe('microdollar usage hourly', () => {
   // The coalesced key columns are SELECTed as well as ordered on, which SELECT DISTINCT
-  // requires, so a row carries both the raw values and the cursor values. `key_2` comes
-  // back as a number here and as text elsewhere, since a bigint arrives as either.
+  // requires, so a row carries both the raw values and the cursor values. Depending on the
+  // active pg type parser, bigint columns arrive as bigint, number, or text.
   const HOURLY_ROW = {
     cursor_owner: 'org-9',
     project_id: 'project-a',
@@ -2294,6 +2294,42 @@ describe('microdollar usage hourly', () => {
     expect(page?.nextCursor).toEqual({ key: ['org-9', 'project-a', '-1', 'NL'] });
   });
 
+  it('normalizes bigint values returned by the shared pg type parser', async () => {
+    const { adapters } = harness([
+      {
+        ...HOURLY_ROW,
+        vercel_ip_country_id: 16n,
+        key_2: 16n,
+      },
+    ]);
+
+    const page = await requireAdapter(adapters, 'microdollar_usage_hourly').readPage?.({
+      ...READ_PAGE_INPUT,
+      limit: 1,
+    });
+    const byField = new Map(page?.records.map(record => [record.field, record.value]));
+
+    expect(byField.get('vercel_ip_country_id')).toBe('16');
+    expect(page?.nextCursor).toEqual({ key: ['org-9', 'project-a', '16', 'NL'] });
+  });
+
+  it('normalizes the bigint null sentinel returned by the shared pg type parser', async () => {
+    const { adapters } = harness([
+      {
+        ...HOURLY_ROW,
+        vercel_ip_country_id: null,
+        key_2: -1n,
+      },
+    ]);
+
+    const page = await requireAdapter(adapters, 'microdollar_usage_hourly').readPage?.({
+      ...READ_PAGE_INPUT,
+      limit: 1,
+    });
+
+    expect(page?.nextCursor).toEqual({ key: ['org-9', 'project-a', '-1', 'NL'] });
+  });
+
   it('scopes each read to one owner column and never both', async () => {
     const { adapters, warehouseCalls } = harness([HOURLY_ROW]);
     const adapter = requireAdapter(adapters, 'microdollar_usage_hourly');

services/user-data-export/src/source-adapters.ts

@@ -1328,6 +1328,19 @@ function warehouseText(value: unknown): string | null {
   return typeof value === 'string' ? value : null;
 }
 
+/** A PostgreSQL integer transported as text, number, or bigint by the active type parser. */
+function warehouseIntegerText(value: unknown): string | null {
+  if (value === null) return null;
+  if (typeof value === 'bigint' || typeof value === 'string') return String(value);
+  return typeof value === 'number' && Number.isFinite(value) ? String(value) : null;
+}
+
+function requiredCursorValue(value: unknown): string {
+  if (typeof value === 'bigint' || typeof value === 'string') return String(value);
+  if (typeof value === 'number' && Number.isFinite(value)) return String(value);
+  throw new Error('Replica row has invalid cursor');
+}
+
 /**
  * A `numeric` column, read as a number.
  *
@@ -1939,15 +1952,9 @@ export function createSourceAdapters(queries: SourceAdapterQueries): SourceAdapt
         ).then(result =>
           result.map(row => ({
             project_id: warehouseText(row.project_id),
-            vercel_ip_country_id:
-              typeof row.vercel_ip_country_id === 'number'
-                ? String(row.vercel_ip_country_id)
-                : warehouseText(row.vercel_ip_country_id),
+            vercel_ip_country_id: warehouseIntegerText(row.vercel_ip_country_id),
             vercel_ip_country: warehouseText(row.vercel_ip_country),
-            key: [0, 1, 2, 3].map(index => {
-              const value = row[`key_${index}`];
-              return typeof value === 'number' ? String(value) : requiredString(value, 'cursor');
-            }),
+            key: [0, 1, 2, 3].map(index => requiredCursorValue(row[`key_${index}`])),
           }))
         );
         return {