RevenueCat sync in bsync

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('BSKY_REVENUE_CAT_V1_API_KEY'),
+    revenueCatV1ApiUrl: envStr('BSKY_REVENUE_CAT_V1_API_URL'),
+    revenueCatWebhookAuthorization: envStr(
+      'BSKY_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,25 @@ 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 './subscriptions'
 
 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 +40,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 })
   }
 }
 

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

@@ -0,0 +1,24 @@
+import { Kysely, sql } from 'kysely'
+
+export async function up(db: Kysely<unknown>): Promise<void> {
+  await db.schema
+    .createTable('subs_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('subs_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('subs_item').execute()
+  await db.schema.dropTable('subs_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-subs'

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 subsOp from './subs_op'
+import * as subsItem from './subs_item'
 
 export type DatabaseSchemaType = muteItem.PartialDB &
   muteOp.PartialDB &
   notifItem.PartialDB &
-  notifOp.PartialDB
+  notifOp.PartialDB &
+  subsItem.PartialDB &
+  subsOp.PartialDB
 
 export type DatabaseSchema = Kysely<DatabaseSchemaType>
 

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

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

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

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

packages/bsync/src/index.ts

@@ -8,6 +8,10 @@ import { ServerConfig } from './config'
 import routes from './routes'
 import { createMuteOpChannel } from './db/schema/mute_op'
 import { createNotifOpChannel } from './db/schema/notif_op'
+import {
+  isRevenueCatWebhookUrl,
+  revenueCatWebhookHandler,
+} from './subscriptions'
 
 export * from './config'
 export * from './client'
@@ -50,6 +54,9 @@ export class BsyncService {
         res.setHeader('content-type', 'application/json')
         return res.end(JSON.stringify({ version: cfg.service.version }))
       }
+      if (isRevenueCatWebhookUrl(req.url)) {
+        return revenueCatWebhookHandler(ctx, req, res)
+      }
       handler(req, res)
     })
     return new BsyncService({ ctx, server, ac })

packages/bsync/src/subscriptions/index.ts

@@ -0,0 +1,2 @@
+export * from './revenueCatClient'
+export * from './revenueCatWebhookHandler'

packages/bsync/src/subscriptions/revenueCatClient.ts

@@ -0,0 +1,73 @@
+type Config = {
+  v1ApiKey: string
+  v1ApiUrl: string
+  webhookAuthorization: string
+}
+
+// Reference: https://www.revenuecat.com/docs/api-v1#tag/customers
+export type GetSubscriberResponse = {
+  subscriber: Subscriber
+}
+
+export type Subscriber = {
+  entitlements: {
+    [entitlementIdentifier: string]: Entitlement
+  }
+}
+
+export type Entitlement = {
+  expires_date: string
+}
+
+export class RevenueCatClient {
+  private v1ApiKey: string
+  private v1ApiUrl: string
+  private webhookAuthorization: string
+
+  constructor({ v1ApiKey, v1ApiUrl, webhookAuthorization }: Config) {
+    this.v1ApiKey = v1ApiKey
+    this.v1ApiUrl = v1ApiUrl
+    this.webhookAuthorization = webhookAuthorization
+  }
+
+  private async fetch<T extends object>(
+    path: string,
+    method: string = 'GET',
+  ): Promise<T> {
+    const url = new URL(path, this.v1ApiUrl)
+    const res = await fetch(url, {
+      method,
+      headers: {
+        Authorization: `Bearer ${this.v1ApiKey}`,
+      },
+    })
+
+    if (!res.ok) {
+      throw new Error(`Failed to fetch ${path}: ${res.statusText}`)
+    }
+
+    return res.json() as T
+  }
+
+  private getSubscriber(did: string): Promise<GetSubscriberResponse> {
+    return this.fetch<GetSubscriberResponse>(
+      `/subscribers/${encodeURIComponent(did)}`,
+    )
+  }
+
+  async getEntitlementIdentifiers(did: string): Promise<string[]> {
+    const { subscriber } = await this.getSubscriber(did)
+
+    const now = Date.now()
+    return Object.entries(subscriber.entitlements)
+      .filter(
+        ([_, entitlement]) =>
+          now < new Date(entitlement.expires_date).valueOf(),
+      )
+      .map(([entitlementIdentifier]) => entitlementIdentifier)
+  }
+
+  isWebhookAuthorizationValid(authorization: string | undefined): boolean {
+    return authorization === this.webhookAuthorization
+  }
+}