feat: Added support for invoice payment object (#245)

Author: KaranPradhan266

README.md

@@ -106,6 +106,7 @@ To deploy the sync-engine to a Supabase Edge Function, follow this [guide](./doc
 - [x] `invoice.overpaid` 🟢
 - [x] `invoice.will_be_due` 🟢
 - [x] `invoice.voided` 🟢
+- [x] `invoice_payment.paid` 🟢
 - [ ] `issuing_authorization.request`
 - [ ] `issuing_card.created`
 - [ ] `issuing_cardholder.created`

docs/index.md

@@ -71,6 +71,7 @@ This project synchronizes your Stripe account to a PostgreSQL database. It can b
 - [x] `invoice.overpaid` 🟢
 - [x] `invoice.will_be_due` 🟢
 - [x] `invoice.voided` 🟢
+- [x] `invoice_payment.paid` 🟢
 - [ ] `issuing_authorization.request`
 - [ ] `issuing_card.created`
 - [ ] `issuing_cardholder.created`

packages/fastify-app/src/test/helpers/mockStripe.ts

@@ -277,6 +277,30 @@ export const mockStripe = {
       },
     })),
   },
+  invoicePayments: {
+    retrieve: vitest.fn((id) =>
+      Promise.resolve({
+        id: id,
+        object: 'invoice_payment',
+        amount_paid: 2000,
+        amount_requested: 2000,
+        created: 1391288554,
+        currency: 'usd',
+        invoice: 'in_103Q0w2eZvKYlo2C5PYwf6Wf',
+        is_default: true,
+        livemode: false,
+        payment: {
+          type: 'payment_intent',
+          payment_intent: 'pi_103Q0w2eZvKYlo2C364X582Z',
+        },
+        status: 'paid',
+        status_transitions: {
+          canceled_at: null,
+          paid_at: 1391288554,
+        },
+      })
+    ),
+  },
   subscriptions: {
     retrieve: vitest.fn((id) =>
       Promise.resolve({

packages/fastify-app/src/test/stripe/invoice_payment_paid.json

@@ -0,0 +1,35 @@
+{
+  "id": "evt_test_invoice_payment_paid",
+  "object": "event",
+  "api_version": "2023-10-16",
+  "created": 1642649111,
+  "data": {
+    "object": {
+      "id": "inpay_1M3USa2eZvKYlo2CBjuwbq0N",
+      "object": "invoice_payment",
+      "amount_paid": 2000,
+      "amount_requested": 2000,
+      "created": 1391288554,
+      "currency": "usd",
+      "invoice": "in_103Q0w2eZvKYlo2C5PYwf6Wf",
+      "is_default": true,
+      "livemode": false,
+      "payment": {
+        "type": "payment_intent",
+        "payment_intent": "pi_103Q0w2eZvKYlo2C364X582Z"
+      },
+      "status": "paid",
+      "status_transitions": {
+        "canceled_at": null,
+        "paid_at": 1391288554
+      }
+    }
+  },
+  "livemode": false,
+  "pending_webhooks": 1,
+  "request": {
+    "id": null,
+    "idempotency_key": null
+  },
+  "type": "invoice_payment.paid"
+}

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

@@ -125,6 +125,7 @@ describe('POST /webhooks', () => {
     'refund_failed.json',
     'refund_updated.json',
     'checkout_session_completed.json',
+    'invoice_payment_paid.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
@@ -147,7 +148,7 @@ describe('POST /webhooks', () => {
     })
 
     if (response.statusCode != 200) {
-      logger.error('error: ', response.body)
+      logger.error({ responseBody: response.body }, 'error')
     }
     expect(response.statusCode).toBe(200)
 
@@ -192,7 +193,7 @@ describe('POST /webhooks', () => {
     })
 
     if (response.statusCode != 200) {
-      logger.error('error: ', response.body)
+      logger.error({ responseBody: response.body }, 'error')
     }
     expect(response.statusCode).toBe(200)
     expect(JSON.parse(response.body)).toMatchObject({ received: true })

packages/sync-engine/src/database/migrations/0040_invoice_payments.sql

@@ -0,0 +1,25 @@
+create table
+    if not exists "stripe"."invoice_payments" (
+        "id" text primary key,
+        object text,
+        amount_paid bigint,
+        amount_requested bigint,
+        created integer,
+        currency text,
+        invoice text,
+        is_default boolean,
+        livemode boolean,
+        payment jsonb,
+        status text,
+        status_transitions jsonb,
+        last_synced_at timestamptz,
+        updated_at timestamptz default timezone('utc'::text, now()) not null
+    );
+
+create index stripe_invoice_payments_invoice_idx on "stripe"."invoice_payments" using btree (invoice);
+
+create trigger handle_updated_at
+    before update
+    on stripe.invoice_payments
+    for each row
+    execute procedure set_updated_at();

packages/sync-engine/src/schemas/invoice_payment.ts

@@ -0,0 +1,18 @@
+import { EntitySchema } from './types'
+
+export const invoicePaymentSchema: EntitySchema = {
+  properties: [
+    'id',
+    'object',
+    'amount_paid',
+    'amount_requested',
+    'created',
+    'currency',
+    'invoice',
+    'is_default',
+    'livemode',
+    'payment',
+    'status',
+    'status_transitions',
+  ],
+} as const

packages/sync-engine/src/stripeSync.ts

@@ -32,6 +32,7 @@ import { reviewSchema } from './schemas/review'
 import { refundSchema } from './schemas/refund'
 import { activeEntitlementSchema } from './schemas/active_entitlement'
 import { featureSchema } from './schemas/feature'
+import { invoicePaymentSchema } from './schemas/invoice_payment'
 import type { PoolConfig } from 'pg'
 
 function getUniqueIds<T>(entries: T[], key: string): string[] {
@@ -553,6 +554,24 @@ export class StripeSync {
         )
         break
       }
+      case 'invoice_payment.paid': {
+        const { entity: invoicePayment, refetched } = await this.fetchOrUseWebhookData(
+          event.data.object as Stripe.InvoicePayment,
+          (id) => this.stripe.invoicePayments.retrieve(id)
+        )
+
+        this.config.logger?.info(
+          `Received webhook ${event.id}: ${event.type} for invoicePayment ${invoicePayment.id}`
+        )
+
+        await this.upsertInvoicePayments(
+          [invoicePayment],
+          false,
+          this.getSyncTimestamp(event, refetched)
+        )
+
+        break
+      }
       default:
         throw new Error('Unhandled webhook event')
     }
@@ -627,6 +646,10 @@ export class StripeSync {
       return this.stripe.reviews.retrieve(stripeId).then((it) => this.upsertReviews([it]))
     } else if (stripeId.startsWith('re_')) {
       return this.stripe.refunds.retrieve(stripeId).then((it) => this.upsertRefunds([it]))
+    } else if (stripeId.startsWith('inpay_')) {
+      return this.stripe.invoicePayments
+        .retrieve(stripeId)
+        .then((it) => this.upsertInvoicePayments([it]))
     } else if (stripeId.startsWith('feat_')) {
       return this.stripe.entitlements.features
         .retrieve(stripeId)
@@ -1280,6 +1303,43 @@ export class StripeSync {
     )
   }
 
+  async upsertInvoicePayments(
+    invoicePayments: Stripe.InvoicePayment[],
+    backfillRelatedEntities?: boolean,
+    syncTimestamp?: string
+  ): Promise<Stripe.InvoicePayment[]> {
+    if (backfillRelatedEntities ?? this.config.backfillRelatedEntities) {
+      const invoiceIds = getUniqueIds(invoicePayments, 'invoice')
+      const paymentIntentIds: string[] = []
+      const chargeIds: string[] = []
+
+      for (const invoicePayment of invoicePayments) {
+        const payment = invoicePayment.payment as
+          | { type?: string; payment_intent?: string; charge?: string }
+          | undefined
+
+        if (payment?.type === 'payment_intent' && payment.payment_intent) {
+          paymentIntentIds.push(payment.payment_intent.toString())
+        } else if (payment?.type === 'charge' && payment.charge) {
+          chargeIds.push(payment.charge.toString())
+        }
+      }
+
+      await Promise.all([
+        this.backfillInvoices(invoiceIds),
+        this.backfillPaymentIntents([...new Set(paymentIntentIds)]),
+        this.backfillCharges([...new Set(chargeIds)]),
+      ])
+    }
+
+    return this.postgresClient.upsertManyWithTimestampProtection(
+      invoicePayments,
+      'invoice_payments',
+      invoicePaymentSchema,
+      syncTimestamp
+    )
+  }
+
   async upsertPlans(
     plans: Stripe.Plan[],
     backfillRelatedEntities?: boolean,