Merge pull request #177 from supabase/chore/add-upsert-timestamp-protection

chore: add upsert timestamp protection

Author: ignaciodob

packages/fastify-app/src/test/webhooks.test.ts

@@ -1,7 +1,7 @@
 'use strict'
 import { FastifyInstance } from 'fastify'
 import { createHmac } from 'node:crypto'
-import { runMigrations } from '@supabase/stripe-sync-engine'
+import { PostgresClient, runMigrations } from '@supabase/stripe-sync-engine'
 import { beforeAll, describe, test, expect, afterAll, vitest } from 'vitest'
 import { getConfig } from '../utils/config'
 import { createServer } from '../app'
@@ -12,6 +12,11 @@ import { StripeSync } from '@supabase/stripe-sync-engine'
 const unixtime = Math.floor(new Date().getTime() / 1000)
 const stripeWebhookSecret = getConfig().stripeWebhookSecret
 
+const postgresClient = new PostgresClient({
+  databaseUrl: getConfig().databaseUrl,
+  schema: getConfig().schema,
+})
+
 describe('POST /webhooks', () => {
   let server: FastifyInstance
 
@@ -35,22 +40,31 @@ describe('POST /webhooks', () => {
     await server.close()
   })
 
+  function getTableName(entityType: string): string {
+    if (entityType.includes('.')) {
+      // Handle cases where entityType has a prefix (e.g., "radar.early_fraud_warning")
+      return entityType.split('.').pop() || entityType
+    }
+    return entityType
+  }
+
+  async function deleteTestData(entityType: string, entityId: string) {
+    const tableName = getTableName(entityType)
+    await postgresClient.query(`DELETE FROM stripe.${tableName}s WHERE id = $1`, [entityId])
+  }
+
   test.each([
     'customer_updated.json',
     'customer_deleted.json',
     'customer_tax_id_created.json',
-    'customer_tax_id_deleted.json',
     'customer_tax_id_updated.json',
     'product_created.json',
-    'product_deleted.json',
     'product_updated.json',
     'price_created.json',
-    'price_deleted.json',
     'price_updated.json',
     'subscription_created.json',
     'subscription_deleted.json',
     'subscription_updated.json',
-    'invoice_deleted.json',
     'invoice_paid.json',
     'invoice_updated.json',
     'invoice_finalized.json',
@@ -77,32 +91,81 @@ describe('POST /webhooks', () => {
     'payment_method_automatically_updated.json',
     'payment_method_detached.json',
     'payment_method_updated.json',
-    'charge_dispute_closed',
-    'charge_dispute_created',
-    'charge_dispute_funds_reinstated',
-    'charge_dispute_funds_withdrawn',
-    'charge_dispute_updated',
-    'plan_created',
-    'plan_deleted',
-    'plan_updated',
-    'payment_intent_amount_capturable_updated',
-    'payment_intent_canceled',
-    'payment_intent_created',
-    'payment_intent_partially_funded',
-    'payment_intent_payment_failed',
-    'payment_intent_processing',
-    'payment_intent_requires_action',
-    'payment_intent_succeeded',
-    'credit_note_created',
-    'credit_note_updated',
-    'credit_note_voided',
-    'early_fraud_warning_created',
-    'early_fraud_warning_updated',
-    'review_closed',
-    'review_opened',
-    'refund_created',
-    'refund_failed',
-    'refund_updated',
+    'charge_dispute_closed.json',
+    'charge_dispute_created.json',
+    'charge_dispute_funds_reinstated.json',
+    'charge_dispute_funds_withdrawn.json',
+    'charge_dispute_updated.json',
+    'plan_created.json',
+    'plan_updated.json',
+    'payment_intent_amount_capturable_updated.json',
+    'payment_intent_canceled.json',
+    'payment_intent_created.json',
+    'payment_intent_partially_funded.json',
+    'payment_intent_payment_failed.json',
+    'payment_intent_processing.json',
+    'payment_intent_requires_action.json',
+    'payment_intent_succeeded.json',
+    'credit_note_created.json',
+    'credit_note_updated.json',
+    'credit_note_voided.json',
+    'early_fraud_warning_created.json',
+    'early_fraud_warning_updated.json',
+    'review_closed.json',
+    'review_opened.json',
+    'refund_created.json',
+    'refund_failed.json',
+    'refund_updated.json',
+  ])('event %s is upserted', async (jsonFile) => {
+    const eventBody = await import(`./stripe/${jsonFile}`).then(({ default: myData }) => myData)
+    // Update the event body created timestamp to be the current time
+    eventBody.created = unixtime
+    const signature = createHmac('sha256', stripeWebhookSecret)
+      .update(`${unixtime}.${JSON.stringify(eventBody)}`, 'utf8')
+      .digest('hex')
+    const entity = eventBody.data.object
+    const entityId = entity.id
+    const entityType = entity.object
+    await deleteTestData(entityType, entityId)
+
+    const response = await server.inject({
+      url: `/webhooks`,
+      method: 'POST',
+      headers: {
+        'stripe-signature': `t=${unixtime},v1=${signature},v0=ff`,
+      },
+      payload: eventBody,
+    })
+
+    if (response.statusCode != 200) {
+      logger.error('error: ', response.body)
+    }
+    expect(response.statusCode).toBe(200)
+
+    const tableName = getTableName(entityType)
+    const result = await postgresClient.query(`SELECT * FROM stripe.${tableName}s WHERE id = $1`, [
+      entityId,
+    ])
+
+    const rows = result.rows
+    expect(rows.length).toBe(1)
+
+    const dbEntity = rows[0]
+    expect(dbEntity.id).toBe(entityId)
+
+    const syncTimestamp = new Date(eventBody.created * 1000).toISOString()
+    expect(dbEntity.last_synced_at.toISOString()).toBe(syncTimestamp)
+  })
+
+  test.each([
+    'customer_tax_id_deleted.json',
+    'product_deleted.json',
+    'price_deleted.json',
+    'invoice_deleted.json',
+    'plan_deleted.json',
+    'refund_created.json',
+    'refund_failed.json',
+    'refund_updated.json',
   ])('process event %s', async (jsonFile) => {
     const eventBody = await import(`./stripe/${jsonFile}`).then(({ default: myData }) => myData)
     const signature = createHmac('sha256', stripeWebhookSecret)
@@ -124,4 +187,88 @@ describe('POST /webhooks', () => {
     expect(response.statusCode).toBe(200)
     expect(JSON.parse(response.body)).toMatchObject({ received: true })
   })
+
+  test('webhook with older timestamp does not override newer data', async () => {
+    const eventBody = await import('./stripe/charge_updated.json').then(
+      ({ default: myData }) => myData
+    )
+    const entity = eventBody.data.object
+    const entityId = entity.id
+    const entityType = entity.object
+    const tableName = getTableName(entityType)
+
+    // Clean up any existing test data
+    await deleteTestData(entityType, entityId)
+
+    // First, send a webhook with current timestamp (newer data)
+    const newerTimestamp = unixtime
+    const newerEventBody = { ...eventBody, created: newerTimestamp }
+    const newerSignature = createHmac('sha256', stripeWebhookSecret)
+      .update(`${newerTimestamp}.${JSON.stringify(newerEventBody)}`, 'utf8')
+      .digest('hex')
+
+    const newerResponse = await server.inject({
+      url: `/webhooks`,
+      method: 'POST',
+      headers: {
+        'stripe-signature': `t=${newerTimestamp},v1=${newerSignature},v0=ff`,
+      },
+      payload: newerEventBody,
+    })
+
+    expect(newerResponse.statusCode).toBe(200)
+
+    // Verify the newer data was stored
+    const newerResult = await postgresClient.query(
+      `SELECT * FROM stripe.${tableName}s WHERE id = $1`,
+      [entityId]
+    )
+    expect(newerResult.rows.length).toBe(1)
+    const newerDbEntity = newerResult.rows[0]
+    const newerSyncTimestamp = new Date(newerTimestamp * 1000).toISOString()
+    expect(newerDbEntity.last_synced_at.toISOString()).toBe(newerSyncTimestamp)
+
+    // Now send a webhook with an older timestamp and different paid value (should not override)
+    const olderTimestamp = newerTimestamp - 60 // 1 minute older
+    const olderEventBody = {
+      ...eventBody,
+      created: olderTimestamp,
+      data: {
+        ...eventBody.data,
+        object: {
+          ...eventBody.data.object,
+          paid: !eventBody.data.object.paid, // Flip the paid value
+        },
+      },
+    }
+    const olderSignature = createHmac('sha256', stripeWebhookSecret)
+      .update(`${olderTimestamp}.${JSON.stringify(olderEventBody)}`, 'utf8')
+      .digest('hex')
+
+    const olderResponse = await server.inject({
+      url: `/webhooks`,
+      method: 'POST',
+      headers: {
+        'stripe-signature': `t=${olderTimestamp},v1=${olderSignature},v0=ff`,
+      },
+      payload: olderEventBody,
+    })
+
+    expect(olderResponse.statusCode).toBe(200)
+
+    // Verify the data still has the newer timestamp and newer paid value (not overridden)
+    const olderResult = await postgresClient.query(
+      `SELECT * FROM stripe.${tableName}s WHERE id = $1`,
+      [entityId]
+    )
+    expect(olderResult.rows.length).toBe(1)
+    const olderDbEntity = olderResult.rows[0]
+    expect(olderDbEntity.last_synced_at.toISOString()).toBe(newerSyncTimestamp)
+    expect(olderDbEntity.last_synced_at.toISOString()).not.toBe(
+      new Date(olderTimestamp * 1000).toISOString()
+    )
+    // Verify the paid field still reflects the newer webhook's value
+    expect(olderDbEntity.paid).toBe(newerEventBody.data.object.paid)
+    expect(olderDbEntity.paid).not.toBe(olderEventBody.data.object.paid)
+  })
 })

packages/sync-engine/src/database/migrations/0033_add_last_synced_at.sql

@@ -0,0 +1,85 @@
+-- Add last_synced_at column to all Stripe tables for tracking sync status
+
+-- Charges
+alter table "stripe"."charges"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Coupons
+alter table "stripe"."coupons"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Credit Notes
+alter table "stripe"."credit_notes"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Customers
+alter table "stripe"."customers"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Disputes
+alter table "stripe"."disputes"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Early Fraud Warnings
+alter table "stripe"."early_fraud_warnings"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Events
+alter table "stripe"."events"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Invoices
+alter table "stripe"."invoices"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Payment Intents
+alter table "stripe"."payment_intents"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Payment Methods
+alter table "stripe"."payment_methods"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Payouts
+alter table "stripe"."payouts"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Plans
+alter table "stripe"."plans"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Prices
+alter table "stripe"."prices"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Products
+alter table "stripe"."products"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Refunds
+alter table "stripe"."refunds"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Reviews
+alter table "stripe"."reviews"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Setup Intents
+alter table "stripe"."setup_intents"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Subscription Items
+alter table "stripe"."subscription_items"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Subscription Schedules
+alter table "stripe"."subscription_schedules"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Subscriptions
+alter table "stripe"."subscriptions"
+add column IF NOT EXISTS "last_synced_at" timestamptz;
+
+-- Tax IDs
+alter table "stripe"."tax_ids"
+add column IF NOT EXISTS "last_synced_at" timestamptz;

packages/sync-engine/src/database/migrations/0034_remove_foreign_keys.sql

@@ -0,0 +1,13 @@
+-- Remove all foreign key constraints
+
+ALTER TABLE "stripe"."subscriptions" DROP CONSTRAINT IF EXISTS "subscriptions_customer_fkey";
+
+ALTER TABLE "stripe"."prices" DROP CONSTRAINT IF EXISTS "prices_product_fkey";
+
+ALTER TABLE "stripe"."invoices" DROP CONSTRAINT IF EXISTS "invoices_customer_fkey";
+
+ALTER TABLE "stripe"."invoices" DROP CONSTRAINT IF EXISTS "invoices_subscription_fkey";
+
+ALTER TABLE "stripe"."subscription_items" DROP CONSTRAINT IF EXISTS "subscription_items_price_fkey";
+
+ALTER TABLE "stripe"."subscription_items" DROP CONSTRAINT IF EXISTS "subscription_items_subscription_fkey";