fix: prevent synced invoices from getting updated by outdated webhooks

Author: ignaciodob

apps/node-fastify/src/test/test-utils.ts

@@ -11,7 +11,7 @@ export async function fetchInvoicesFromDatabase(postgresClient: PostgresClient,
 
   const placeholders = invoiceIds.map((_, index) => `$${index + 1}`).join(',');
   const result = await postgresClient.query(
-    `SELECT id, invoice_number, customer_id, total, currency, status, updated_at FROM orb.invoices WHERE id IN (${placeholders})`,
+    `SELECT id, invoice_number, customer_id, total, currency, status, updated_at, last_synced_at FROM orb.invoices WHERE id IN (${placeholders})`,
     invoiceIds
   );
   return result.rows;

apps/node-fastify/src/test/webhooks.test.ts

@@ -182,6 +182,10 @@ describe('POST /webhooks', () => {
 
     const webhookData = JSON.parse(payload);
     webhookData.type = webhookType;
+    
+    const webhookTimestamp = new Date('2025-01-15T10:30:00.000Z').toISOString();
+    webhookData.created_at = webhookTimestamp;
+    
     payload = JSON.stringify(webhookData);
 
     const invoiceId = webhookData.invoice.id;
@@ -200,6 +204,10 @@ describe('POST /webhooks', () => {
     expect(invoice.total).toBe(webhookData.invoice.amount_due);
     expect(invoice.currency).toBe(webhookData.invoice.currency);
     expect(invoice.status).toBe(webhookData.invoice.status);
+    
+    // Verify that last_synced_at gets set to the webhook timestamp for new invoices
+    expect(invoice.last_synced_at).toBeDefined();
+    expect(new Date(invoice.last_synced_at).toISOString()).toBe(webhookTimestamp);
   });
 
   it('should update an existing invoice when webhook arrives', async () => {
@@ -229,8 +237,14 @@ describe('POST /webhooks', () => {
       status: initialStatus,
     };
 
-    // Store the initial invoice in the database
-    await syncInvoices(postgresClient, [initialInvoiceData]);
+    // Store the initial invoice in the database with an old timestamp
+    const oldTimestamp = new Date('2025-01-10T10:00:00.000Z').toISOString();
+    await syncInvoices(postgresClient, [initialInvoiceData], oldTimestamp);
+
+    // Verify the initial invoice was created with the old timestamp
+    const [initialInvoice] = await fetchInvoicesFromDatabase(orbSync.postgresClient, [invoiceId]);
+    expect(initialInvoice).toBeDefined();
+    expect(new Date(initialInvoice.last_synced_at).toISOString()).toBe(oldTimestamp);
 
     // Now update the webhook data with new values
     const updatedAmount = 1500;
@@ -240,6 +254,10 @@ describe('POST /webhooks', () => {
     webhookData.invoice.status = updatedStatus;
     webhookData.invoice.paid_at = new Date().toISOString();
 
+    // Set a newer webhook timestamp that should trigger an update
+    const newWebhookTimestamp = new Date('2025-01-15T10:30:00.000Z').toISOString();
+    webhookData.created_at = newWebhookTimestamp;
+
     payload = JSON.stringify(webhookData);
 
     // Send the webhook with updated data
@@ -252,13 +270,80 @@ describe('POST /webhooks', () => {
     });
 
     // Verify that the invoice was updated in the database
-    const [invoice] = await fetchInvoicesFromDatabase(orbSync.postgresClient, [invoiceId]);
-    expect(invoice).toBeDefined();
-    expect(Number(invoice.total)).toBe(updatedAmount);
-    expect(invoice.status).toBe(updatedStatus);
+    const [updatedInvoice] = await fetchInvoicesFromDatabase(orbSync.postgresClient, [invoiceId]);
+    expect(updatedInvoice).toBeDefined();
+    expect(Number(updatedInvoice.total)).toBe(updatedAmount);
+    expect(updatedInvoice.status).toBe(updatedStatus);
 
     // Verify that the updated_at timestamp was changed
-    expect(invoice.updated_at).toBeDefined();
-    expect(new Date(invoice.updated_at).getTime()).toBeGreaterThan(new Date(webhookData.invoice.created_at).getTime());
+    expect(updatedInvoice.updated_at).toBeDefined();
+    expect(new Date(updatedInvoice.updated_at).getTime()).toBeGreaterThan(new Date(webhookData.invoice.created_at).getTime());
+    
+    // Verify that last_synced_at was updated to the new webhook timestamp
+    expect(updatedInvoice.last_synced_at).toBeDefined();
+    expect(new Date(updatedInvoice.last_synced_at).toISOString()).toBe(newWebhookTimestamp);
+    
+    // Verify that the new timestamp is newer than the old timestamp
+    expect(new Date(updatedInvoice.last_synced_at).getTime()).toBeGreaterThan(new Date(oldTimestamp).getTime());
+  });
+
+  it('should NOT update invoice when webhook timestamp is older than last_synced_at', async () => {
+    let payload = loadWebhookPayload('invoice');
+    const postgresClient = orbSync.postgresClient;
+
+    const webhookData = JSON.parse(payload);
+    const invoiceId = webhookData.invoice.id;
+    await deleteTestData(orbSync.postgresClient, 'invoices', [invoiceId]);
+
+    webhookData.type = 'invoice.payment_succeeded';
+
+    // Insert an invoice with a "new" timestamp and known values
+    const originalAmount = 2000;
+    const originalStatus = 'paid';
+    webhookData.invoice.amount_due = originalAmount.toString();
+    webhookData.invoice.total = originalAmount.toString();
+    webhookData.invoice.status = originalStatus;
+
+    const newTimestamp = new Date('2025-01-15T10:30:00.000Z').toISOString();
+    const initialInvoiceData = {
+      ...webhookData.invoice,
+      amount_due: originalAmount.toString(),
+      total: originalAmount.toString(),
+      status: originalStatus,
+    };
+    await syncInvoices(postgresClient, [initialInvoiceData], newTimestamp);
+
+    // Verify the invoice was created with the new timestamp
+    const [initialInvoice] = await fetchInvoicesFromDatabase(orbSync.postgresClient, [invoiceId]);
+    expect(initialInvoice).toBeDefined();
+    expect(Number(initialInvoice.total)).toBe(originalAmount);
+    expect(initialInvoice.status).toBe(originalStatus);
+    expect(new Date(initialInvoice.last_synced_at).toISOString()).toBe(newTimestamp);
+
+    // Now attempt to update with an older webhook timestamp and different values
+    const outdatedAmount = 1000;
+    const outdatedStatus = 'pending';
+    webhookData.invoice.amount_due = outdatedAmount.toString();
+    webhookData.invoice.total = outdatedAmount.toString();
+    webhookData.invoice.status = outdatedStatus;
+    webhookData.invoice.paid_at = undefined;
+    const oldWebhookTimestamp = new Date('2025-01-10T10:00:00.000Z').toISOString();
+    webhookData.created_at = oldWebhookTimestamp;
+    payload = JSON.stringify(webhookData);
+
+    // Send the webhook with the outdated data
+    const response = await sendWebhookRequest(payload);
+    expect(response.statusCode).toBe(200);
+    const data = response.json();
+    expect(data).toMatchObject({ received: true });
+
+    // Fetch the invoice again and verify it was NOT updated
+    const [afterWebhookInvoice] = await fetchInvoicesFromDatabase(orbSync.postgresClient, [invoiceId]);
+    expect(afterWebhookInvoice).toBeDefined();
+    // Data should remain unchanged
+    expect(Number(afterWebhookInvoice.total)).toBe(originalAmount);
+    expect(afterWebhookInvoice.status).toBe(originalStatus);
+    // last_synced_at should remain the new timestamp
+    expect(new Date(afterWebhookInvoice.last_synced_at).toISOString()).toBe(newTimestamp);
   });
 });

db/migrations/0009_add_last_synced_at_columns.sql

@@ -0,0 +1,21 @@
+-- Add last_synced_at column to all relevant entity tables
+-- This column tracks when each entity was last synchronized with Orb.
+-- It enables timestamp-based protection to prevent old webhooks from overwriting newer data.
+
+alter table orb.invoices
+add column last_synced_at timestamptz;
+
+alter table orb.customers
+add column last_synced_at timestamptz;
+
+alter table orb.subscriptions
+add column last_synced_at timestamptz;
+
+alter table orb.credit_notes
+add column last_synced_at timestamptz;
+
+alter table orb.plans
+add column last_synced_at timestamptz;
+
+alter table orb.billable_metrics
+add column last_synced_at timestamptz;

packages/orb-sync-lib/src/database/postgres.ts

@@ -47,6 +47,51 @@ export class PostgresClient {
     return results.flatMap((it) => it.rows);
   }
 
+  async upsertManyWithTimestampProtection<
+    T extends {
+      [Key: string]: any; // eslint-disable-line @typescript-eslint/no-explicit-any
+    },
+  >(
+    entries: T[],
+    table: string,
+    tableSchema: JsonSchema,
+    syncTimestamp: string
+  ): Promise<T[]> {
+    if (!entries.length) return [];
+
+    // Max 5 in parallel to avoid exhausting connection pool
+    const chunkSize = 5;
+    const results: pg.QueryResult<T>[] = [];
+
+    for (let i = 0; i < entries.length; i += chunkSize) {
+      const chunk = entries.slice(i, i + chunkSize);
+
+      const queries: Promise<pg.QueryResult<T>>[] = [];
+      chunk.forEach((entry) => {
+        // Inject the values
+        const cleansed = this.cleanseArrayField(entry, tableSchema);
+        // Add last_synced_at to the cleansed data for SQL parameter binding
+        cleansed.last_synced_at = syncTimestamp;
+        
+        const upsertSql = this.constructUpsertWithTimestampProtectionSql(
+          this.config.schema,
+          table,
+          tableSchema
+        );
+
+        const prepared = sql(upsertSql, {
+          useNullForMissing: true,
+        })(cleansed);
+
+        queries.push(this.pool.query(prepared.text, prepared.values));
+      });
+
+      results.push(...(await Promise.all(queries)));
+    }
+
+    return results.flatMap((it) => it.rows);
+  }
+
   private constructUpsertSql = (schema: string, table: string, tableSchema: JsonSchema): string => {
     const conflict = 'id';
     const properties = tableSchema.properties;
@@ -72,6 +117,35 @@ export class PostgresClient {
       ;`;
   };
 
+  private constructUpsertWithTimestampProtectionSql = (
+    schema: string,
+    table: string,
+    tableSchema: JsonSchema
+  ): string => {
+    const conflict = 'id';
+    const properties = tableSchema.properties;
+
+    return `
+      INSERT INTO "${schema}"."${table}" (
+        ${Object.keys(properties)
+          .map((x) => `"${x}"`)
+          .join(',')}
+      )
+      VALUES (
+        ${Object.keys(properties)
+          .map((x) => `:${x}`)
+          .join(',')}
+      )
+      ON CONFLICT (${conflict}) DO UPDATE SET
+        ${Object.keys(properties)
+          .filter((x) => x !== 'last_synced_at')
+          .map((x) => `"${x}" = EXCLUDED."${x}"`)
+          .join(',')},
+        last_synced_at = :last_synced_at
+      WHERE "${table}"."last_synced_at" IS NULL 
+         OR "${table}"."last_synced_at" < :last_synced_at;`;
+  };
+
   /**
    * Updates a subscription's billing cycle dates, provided that the current end date is in the past (i.e. the subscription
    * data in the database being stale).

packages/orb-sync-lib/src/orb-sync.ts

@@ -173,7 +173,7 @@ export class OrbSync {
         const invoice = webhook.invoice;
         this.config.logger?.info(`Received webhook ${webhook.id}: ${parsedData.type} for invoice ${invoice.id}`);
 
-        await syncInvoices(this.postgresClient, [invoice]);
+        await syncInvoices(this.postgresClient, [invoice], webhook.created_at);
 
         const billingCycle = getBillingCycleFromInvoice(invoice);
         if (billingCycle && invoice.subscription) {
@@ -201,7 +201,7 @@ export class OrbSync {
 
         this.config.logger?.info(`Received webhook ${webhook.id}: ${webhook.type} for invoice ${webhook.invoice.id}`);
 
-        await syncInvoices(this.postgresClient, [webhook.invoice]);
+        await syncInvoices(this.postgresClient, [webhook.invoice], webhook.created_at);
         break;
       }
 

packages/orb-sync-lib/src/schemas/invoice.ts

@@ -43,6 +43,7 @@ export const invoiceSchema: JsonSchema = {
     sync_failed_at: { type: 'string' },
     voided_at: { type: 'string' },
     will_auto_issue: { type: 'boolean' },
+    last_synced_at: { type: 'string' },
   },
   required: ['id'],
 } as const;

packages/orb-sync-lib/src/sync/invoices.ts

@@ -6,15 +6,19 @@ import { Invoice } from 'orb-billing/resources';
 
 const TABLE = 'invoices';
 
-export async function syncInvoices(postgresClient: PostgresClient, invoices: Invoice[]) {
-  return postgresClient.upsertMany(
+export async function syncInvoices(postgresClient: PostgresClient, invoices: Invoice[], syncTimestamp?: string) {
+  const timestamp = syncTimestamp || new Date().toISOString();
+
+  return postgresClient.upsertManyWithTimestampProtection(
     invoices.map((invoice) => ({
       ...invoice,
       customer_id: invoice.customer.id,
       subscription_id: invoice.subscription?.id,
+      last_synced_at: timestamp,
     })),
     TABLE,
-    invoiceSchema
+    invoiceSchema,
+    timestamp
   );
 }