RevenueCat sync in bsync

packages/bsync/package.json

@@ -29,17 +29,23 @@
     "@atproto/syntax": "workspace:^",
     "@bufbuild/protobuf": "^1.5.0",
     "@connectrpc/connect": "^1.1.4",
+    "@connectrpc/connect-express": "^1.1.4",
     "@connectrpc/connect-node": "^1.1.4",
+    "compression": "^1.7.4",
+    "express": "^4.21.1",
     "http-terminator": "^3.2.0",
     "kysely": "^0.22.0",
     "pg": "^8.10.0",
     "pino-http": "^8.2.1",
-    "typed-emitter": "^2.1.0"
+    "typed-emitter": "^2.1.0",
+    "zod": "^3.23.8"
   },
   "devDependencies": {
     "@bufbuild/buf": "^1.28.1",
     "@bufbuild/protoc-gen-es": "^1.5.0",
     "@connectrpc/protoc-gen-connect-es": "^1.1.4",
+    "@types/compression": "^1.7.5",
+    "@types/express": "^4.17.13",
     "@types/pg": "^8.6.6",
     "get-port": "^5.1.1",
     "jest": "^28.1.2",

packages/bsync/src/api/health.ts

@@ -0,0 +1,12 @@
+import express from 'express'
+import AppContext from '../context'
+
+export const createRouter = (ctx: AppContext): express.Router => {
+  const router = express.Router()
+
+  router.get('/_health', async function (_req, res) {
+    res.send({ version: ctx.cfg.service.version })
+  })
+
+  return router
+}

packages/bsync/src/api/revenueCat.ts

@@ -0,0 +1,88 @@
+import express, { RequestHandler } from 'express'
+import { AppContext } from '..'
+import { rcEventBodySchema, RevenueCatClient } from '../purchases'
+import { addPurchaseOperation, RcEventBody } from '../purchases'
+import { isValidDid } from '../routes/util'
+import { httpLogger as log } from '..'
+
+type AppContextWithRevenueCatClient = AppContext & {
+  revenueCatClient: RevenueCatClient
+}
+
+const auth =
+  (ctx: AppContextWithRevenueCatClient): RequestHandler =>
+  (req: express.Request, res: express.Response, next: express.NextFunction) =>
+    ctx.revenueCatClient.isWebhookAuthorizationValid(
+      req.header('Authorization'),
+    )
+      ? next()
+      : res.status(403).send({
+          success: false,
+          error: 'Forbidden: invalid authentication for RevenueCat webhook',
+        })
+
+const webhookHandler =
+  (ctx: AppContextWithRevenueCatClient): RequestHandler =>
+  async (req, res) => {
+    const { revenueCatClient } = ctx
+
+    let body: RcEventBody
+    try {
+      body = rcEventBodySchema.parse(req.body)
+    } catch (error) {
+      log.error({ error }, 'RevenueCat webhook body schema validation failed')
+
+      return res.status(400).send({
+        success: false,
+        error: 'Bad request: body schema validation failed',
+      })
+    }
+
+    const { app_user_id: actorDid } = body.event
+
+    if (!isValidDid(actorDid)) {
+      log.error({ actorDid }, 'RevenueCat webhook got invalid DID')
+
+      return res.status(400).send({
+        success: false,
+        error: 'Bad request: invalid DID in app_user_id',
+      })
+    }
+
+    try {
+      const entitlements =
+        await revenueCatClient.getEntitlementIdentifiers(actorDid)
+
+      const id = await addPurchaseOperation(ctx.db, actorDid, entitlements)
+
+      res.send({ success: true, operationId: id })
+    } catch (error) {
+      log.error({ error }, 'Error while processing RevenueCat webhook')
+
+      res.status(500).send({
+        success: false,
+        error:
+          'Internal server error: an error happened while processing the request',
+      })
+    }
+  }
+
+const assertAppContextWithRevenueCatClient: (
+  ctx: AppContext,
+) => asserts ctx is AppContextWithRevenueCatClient = (ctx: AppContext) => {
+  if (!ctx.revenueCatClient) {
+    throw new Error(
+      'RevenueCat webhook was tried to be set up without configuring a RevenueCat client.',
+    )
+  }
+}
+
+export const createRouter = (ctx: AppContext): express.Router => {
+  assertAppContextWithRevenueCatClient(ctx)
+
+  const router = express.Router()
+  router.use(auth(ctx))
+  router.use(express.json())
+  router.post('/', webhookHandler(ctx))
+  return router
+}

packages/bsync/src/config.ts

@@ -23,17 +23,33 @@ export const envToCfg = (env: ServerEnvironment): ServerConfig => {
     apiKeys: new Set(env.apiKeys),
   }
 
+  let revenueCatCfg: RevenueCatConfig | undefined
+  if (env.revenueCatV1ApiKey) {
+    assert(env.revenueCatV1ApiUrl, 'missing revenue cat v1 api url')
+    assert(
+      env.revenueCatWebhookAuthorization,
+      'missing revenue cat webhook authorization',
+    )
+    revenueCatCfg = {
+      v1ApiKey: env.revenueCatV1ApiKey,
+      v1ApiUrl: env.revenueCatV1ApiUrl,
+      webhookAuthorization: env.revenueCatWebhookAuthorization,
+    }
+  }
+
   return {
     service: serviceCfg,
     db: dbCfg,
     auth: authCfg,
+    revenueCat: revenueCatCfg,
   }
 }
 
 export type ServerConfig = {
   service: ServiceConfig
   db: DatabaseConfig
   auth: AuthConfig
+  revenueCat?: RevenueCatConfig
 }
 
 type ServiceConfig = {
@@ -55,6 +71,12 @@ type AuthConfig = {
   apiKeys: Set<string>
 }
 
+type RevenueCatConfig = {
+  v1ApiUrl: string
+  v1ApiKey: string
+  webhookAuthorization: string
+}
+
 export const readEnv = (): ServerEnvironment => {
   return {
     // service
@@ -70,6 +92,12 @@ export const readEnv = (): ServerEnvironment => {
     dbMigrate: envBool('BSYNC_DB_MIGRATE'),
     // secrets
     apiKeys: envList('BSYNC_API_KEYS'),
+    // revenue cat
+    revenueCatV1ApiKey: envStr('BSYNC_REVENUE_CAT_V1_API_KEY'),
+    revenueCatV1ApiUrl: envStr('BSYNC_REVENUE_CAT_V1_API_URL'),
+    revenueCatWebhookAuthorization: envStr(
+      'BSYNC_REVENUE_CAT_WEBHOOK_AUTHORIZATION',
+    ),
   }
 }
 
@@ -87,4 +115,8 @@ export type ServerEnvironment = {
   dbMigrate?: boolean
   // secrets
   apiKeys: string[]
+  // revenue cat
+  revenueCatV1ApiUrl?: string
+  revenueCatV1ApiKey?: string
+  revenueCatWebhookAuthorization?: string
 }

packages/bsync/src/context.ts

@@ -4,21 +4,26 @@ import Database from './db'
 import { createMuteOpChannel } from './db/schema/mute_op'
 import { createNotifOpChannel } from './db/schema/notif_op'
 import { EventEmitter } from 'stream'
+import { RevenueCatClient } from './purchases'
+import { createPurchaseOpChannel } from './db/schema/purchase_op'
 
 export type AppContextOptions = {
   db: Database
+  revenueCatClient: RevenueCatClient | undefined
   cfg: ServerConfig
   shutdown: AbortSignal
 }
 
 export class AppContext {
   db: Database
+  revenueCatClient: RevenueCatClient | undefined
   cfg: ServerConfig
   shutdown: AbortSignal
   events: TypedEventEmitter<AppEvents>
 
   constructor(opts: AppContextOptions) {
     this.db = opts.db
+    this.revenueCatClient = opts.revenueCatClient
     this.cfg = opts.cfg
     this.shutdown = opts.shutdown
     this.events = new EventEmitter() as TypedEventEmitter<AppEvents>
@@ -36,7 +41,17 @@ export class AppContext {
       poolMaxUses: cfg.db.poolMaxUses,
       poolIdleTimeoutMs: cfg.db.poolIdleTimeoutMs,
     })
-    return new AppContext({ db, cfg, shutdown, ...overrides })
+
+    let revenueCatClient: RevenueCatClient | undefined
+    if (cfg.revenueCat) {
+      revenueCatClient = new RevenueCatClient({
+        v1ApiKey: cfg.revenueCat.v1ApiKey,
+        v1ApiUrl: cfg.revenueCat.v1ApiUrl,
+        webhookAuthorization: cfg.revenueCat.webhookAuthorization,
+      })
+    }
+
+    return new AppContext({ db, revenueCatClient, cfg, shutdown, ...overrides })
   }
 }
 
@@ -45,4 +60,5 @@ export default AppContext
 export type AppEvents = {
   [createMuteOpChannel]: () => void
   [createNotifOpChannel]: () => void
+  [createPurchaseOpChannel]: () => void
 }

packages/bsync/src/db/migrations/20241205T030533572Z-purchases.ts

@@ -0,0 +1,24 @@
+import { Kysely, sql } from 'kysely'
+
+export async function up(db: Kysely<unknown>): Promise<void> {
+  await db.schema
+    .createTable('purchase_op')
+    .addColumn('id', 'bigserial', (col) => col.primaryKey())
+    .addColumn('actorDid', 'varchar', (col) => col.notNull())
+    .addColumn('entitlements', 'jsonb', (col) => col.notNull())
+    .addColumn('createdAt', 'timestamptz', (col) =>
+      col.notNull().defaultTo(sql`CURRENT_TIMESTAMP`),
+    )
+    .execute()
+  await db.schema
+    .createTable('purchase_item')
+    .addColumn('actorDid', 'varchar', (col) => col.primaryKey())
+    .addColumn('entitlements', 'jsonb', (col) => col.notNull())
+    .addColumn('fromId', 'bigint', (col) => col.notNull())
+    .execute()
+}
+
+export async function down(db: Kysely<unknown>): Promise<void> {
+  await db.schema.dropTable('purchase_item').execute()
+  await db.schema.dropTable('purchase_op').execute()
+}

packages/bsync/src/db/migrations/index.ts

@@ -4,3 +4,4 @@
 
 export * as _20240108T220751294Z from './20240108T220751294Z-init'
 export * as _20240717T224303472Z from './20240717T224303472Z-notif-ops'
+export * as _20241205T030533572Z from './20241205T030533572Z-purchases'

packages/bsync/src/db/schema/index.ts

@@ -3,11 +3,15 @@ import * as muteOp from './mute_op'
 import * as muteItem from './mute_item'
 import * as notifOp from './notif_op'
 import * as notifItem from './notif_item'
+import * as purchaseOp from './purchase_op'
+import * as purchaseItem from './purchase_item'
 
 export type DatabaseSchemaType = muteItem.PartialDB &
   muteOp.PartialDB &
   notifItem.PartialDB &
-  notifOp.PartialDB
+  notifOp.PartialDB &
+  purchaseItem.PartialDB &
+  purchaseOp.PartialDB
 
 export type DatabaseSchema = Kysely<DatabaseSchemaType>
 

packages/bsync/src/db/schema/purchase_item.ts

@@ -0,0 +1,14 @@
+import { ColumnType, Selectable } from 'kysely'
+
+export interface PurchaseItem {
+  actorDid: string
+  // https://github.com/kysely-org/kysely/issues/137
+  entitlements: ColumnType<string[], string, string>
+  fromId: number
+}
+
+export type PurchaseItemEntry = Selectable<PurchaseItem>
+
+export const tableName = 'purchase_item'
+
+export type PartialDB = { [tableName]: PurchaseItem }

packages/bsync/src/db/schema/purchase_op.ts

@@ -0,0 +1,17 @@
+import { ColumnType, GeneratedAlways, Selectable } from 'kysely'
+
+export interface PurchaseOp {
+  id: GeneratedAlways<number>
+  actorDid: string
+  // https://github.com/kysely-org/kysely/issues/137
+  entitlements: ColumnType<string[], string, string>
+  createdAt: GeneratedAlways<Date>
+}
+
+export type PurchaseOpEntry = Selectable<PurchaseOp>
+
+export const tableName = 'purchase_op'
+
+export type PartialDB = { [tableName]: PurchaseOp }
+
+export const createPurchaseOpChannel = 'purchase_op_create' // used with listen/notify