fix: prevent synced invoices from getting updated by outdated webhooks (#68)

ignaciodob · web-flow · commit 479da7583ed7 · 2025-07-01T09:54:57.000+08:00
This PR introduces a general safeguard for upserting - the `upsertManyWithTimestampProtection` function. This function uses a SQL statement with a `WHERE` clause that only updates records when the incoming `last_synced_at` timestamp is newer than the existing one. This prevents older webhook events from overwriting newer data, ensuring data consistency. This pattern can easily be applied to other entities by using the same function in their respective sync functions.

It also:
- Adds a new migration to add the last_synced_at column to all entities: `invoices`, `customers`, `subscriptions`, `credit_notes`, `plans`, and `billable_metrics`
- Adds the `syncTimestamp` parameter to the `syncInvoices` method allows webhook handlers to pass the webhook's `created_at` timestamp, which is then used as the `last_synced_at` value for timestamp-based protection
- Adds assertions and tests to cover this functionality, including a new test case where we check that an invoice is not updated with an out-of-date webhook.
diff --git a/apps/node-fastify/src/test/test-utils.ts b/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;
diff --git a/apps/node-fastify/src/test/webhooks.test.ts b/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,135 @@ 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);
+  });
+
+  it('should ignore outdated webhook for invoice when last_synced_at is newer, even with multiple invoices', async () => {
+    const postgresClient = orbSync.postgresClient;
+    let payload = loadWebhookPayload('invoice');
+    const webhookData = JSON.parse(payload);
+
+    webhookData.type = 'invoice.payment_succeeded';
+
+    // Prepare two invoices with different ids
+    const invoiceId1 = webhookData.invoice.id;
+    const invoiceId2 = invoiceId1 + '_other';
+
+    // Insert both invoices with different last_synced_at values
+    const webhookInvoice1Timestamp = new Date('2025-01-10T09:30:00.000Z').toISOString(); // This is older than invoice1Timestamp, so it should not update invoice1
+    const invoice1Timestamp = new Date('2025-01-10T10:00:00.000Z').toISOString();
+    const invoice2Timestamp = new Date('2025-01-10T10:30:00.000Z').toISOString();
+
+    // Invoice 1: will be updated
+    const invoiceData1 = {
+      ...webhookData.invoice,
+      id: invoiceId1,
+      total: '1500',
+      status: 'paid',
+    };
+
+    // Invoice 2 data is irrelevant for this test
+    const invoiceData2 = {
+      ...webhookData.invoice,
+      id: invoiceId2
+    };
+
+    await deleteTestData(postgresClient, 'invoices', [invoiceId1, invoiceId2]);
+    await syncInvoices(postgresClient, [invoiceData1], invoice1Timestamp);
+    await syncInvoices(postgresClient, [invoiceData2], invoice2Timestamp);
+
+    // Now, send a webhook for invoiceId1 with an older timestamp and outdated values
+    webhookData.invoice.id = invoiceId1;
+    webhookData.invoice.total = '1000';
+    webhookData.invoice.status = 'pending';
+    webhookData.created_at = webhookInvoice1Timestamp;
+    payload = JSON.stringify(webhookData);
+
+    const response = await sendWebhookRequest(payload);
+    expect(response.statusCode).toBe(200);
+
+    // Fetch both invoices
+    const [updatedInvoice1] = await fetchInvoicesFromDatabase(postgresClient, [invoiceId1]);
+
+    // Invoice 1 should NOT be updated because the webhook timestamp is older than the invoice's last_synced_at
+    expect(updatedInvoice1).toBeDefined();
+    expect(updatedInvoice1.total).toBe('1500');
+    expect(updatedInvoice1.status).toBe('paid');
+    expect(new Date(updatedInvoice1.last_synced_at).toISOString()).toBe(invoice1Timestamp);
+
   });
 });
diff --git a/packages/orb-sync-lib/src/database/postgres.ts b/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,38 @@ export class PostgresClient {
       ;`;
   };
 
+  private constructUpsertWithTimestampProtectionSql = (
+    schema: string,
+    table: string,
+    tableSchema: JsonSchema
+  ): string => {
+    const conflict = 'id';
+    const properties = tableSchema.properties;
+
+    // The WHERE clause in ON CONFLICT DO UPDATE only applies to the conflicting row
+    // (the row being updated), not to all rows in the table. PostgreSQL ensures that
+    // the condition is evaluated only for the specific row that conflicts with the INSERT.
+    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).
diff --git a/packages/orb-sync-lib/src/orb-sync.ts b/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;
       }
 
diff --git a/packages/orb-sync-lib/src/schemas/invoice.ts b/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;
diff --git a/packages/orb-sync-lib/src/sync/invoices.ts b/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
   );
 }
 
