Appview v1 maintaining device tokens and pushing notifications w/ courier

.github/workflows/build-and-push-bsky-aws.yaml

@@ -3,7 +3,7 @@ on:
   push:
     branches:
       - main
-      - appview-v1-sync-mutes
+      - appview-v1-courier
 env:
   REGISTRY: ${{ secrets.AWS_ECR_REGISTRY_USEAST2_PACKAGES_REGISTRY }}
   USERNAME: ${{ secrets.AWS_ECR_REGISTRY_USEAST2_PACKAGES_USERNAME }}

packages/bsky/package.json

@@ -29,16 +29,16 @@
     "test:log": "tail -50 test.log | pino-pretty",
     "test:updateSnapshot": "jest --updateSnapshot",
     "migration:create": "ts-node ./bin/migration-create.ts",
-    "buf:gen": "buf generate ../bsync/proto"
+    "buf:gen": "buf generate ../bsync/proto && buf generate ./proto"
   },
   "dependencies": {
     "@atproto/api": "workspace:^",
     "@atproto/common": "workspace:^",
     "@atproto/crypto": "workspace:^",
-    "@atproto/syntax": "workspace:^",
     "@atproto/identity": "workspace:^",
     "@atproto/lexicon": "workspace:^",
     "@atproto/repo": "workspace:^",
+    "@atproto/syntax": "workspace:^",
     "@atproto/xrpc-server": "workspace:^",
     "@bufbuild/protobuf": "^1.5.0",
     "@connectrpc/connect": "^1.1.4",
@@ -55,6 +55,7 @@
     "ioredis": "^5.3.2",
     "kysely": "^0.22.0",
     "multiformats": "^9.9.0",
+    "murmurhash": "^2.0.1",
     "p-queue": "^6.6.2",
     "pg": "^8.10.0",
     "pino": "^8.15.0",

packages/bsky/proto/courier.proto

@@ -0,0 +1,56 @@
+syntax = "proto3";
+
+package courier;
+option go_package = "./;courier";
+
+import "google/protobuf/struct.proto";
+import "google/protobuf/timestamp.proto";
+
+//
+// Messages
+//
+
+// Ping
+message PingRequest {}
+message PingResponse {}
+
+// Notifications
+
+enum AppPlatform {
+  APP_PLATFORM_UNSPECIFIED = 0;
+  APP_PLATFORM_IOS = 1;
+  APP_PLATFORM_ANDROID = 2;
+  APP_PLATFORM_WEB = 3;
+}
+
+message Notification {
+  string id = 1;
+  string recipient_did = 2;
+  string title = 3;
+  string message = 4;
+  string collapse_key = 5;
+  bool always_deliver = 6;
+  google.protobuf.Timestamp timestamp = 7;
+  google.protobuf.Struct additional = 8;
+}
+
+message PushNotificationsRequest {
+  repeated Notification notifications = 1;
+}
+
+message PushNotificationsResponse {}
+
+message RegisterDeviceTokenRequest {
+  string did = 1;
+  string token = 2;
+  string app_id = 3;
+  AppPlatform platform = 4;
+}
+
+message RegisterDeviceTokenResponse {}
+
+service Service {
+  rpc Ping(PingRequest) returns (PingResponse);
+  rpc PushNotifications(PushNotificationsRequest) returns (PushNotificationsResponse);
+  rpc RegisterDeviceToken(RegisterDeviceTokenRequest) returns (RegisterDeviceTokenResponse);
+}

packages/bsky/src/api/app/bsky/notification/registerPush.ts

@@ -1,29 +1,63 @@
+import assert from 'node:assert'
 import { InvalidRequestError } from '@atproto/xrpc-server'
 import { Server } from '../../../../lexicon'
 import AppContext from '../../../../context'
 import { Platform } from '../../../../notifications'
+import { CourierClient } from '../../../../courier'
+import { AppPlatform } from '../../../../proto/courier_pb'
 
 export default function (server: Server, ctx: AppContext) {
   server.app.bsky.notification.registerPush({
     auth: ctx.authVerifier.standard,
-    handler: async ({ auth, input }) => {
+    handler: async ({ req, auth, input }) => {
       const { token, platform, serviceDid, appId } = input.body
       const did = auth.credentials.iss
       if (serviceDid !== auth.credentials.aud) {
         throw new InvalidRequestError('Invalid serviceDid.')
       }
-      const { notifServer } = ctx
       if (platform !== 'ios' && platform !== 'android' && platform !== 'web') {
         throw new InvalidRequestError(
           'Unsupported platform: must be "ios", "android", or "web".',
         )
       }
-      await notifServer.registerDeviceForPushNotifications(
-        did,
-        token,
-        platform as Platform,
-        appId,
-      )
+
+      const db = ctx.db.getPrimary()
+
+      const registerDeviceWithAppview = async () => {
+        await ctx.services
+          .actor(db)
+          .registerPushDeviceToken(did, token, platform as Platform, appId)
+      }
+
+      const registerDeviceWithCourier = async (
+        courierClient: CourierClient,
+      ) => {
+        await courierClient.registerDeviceToken({
+          did,
+          token,
+          platform:
+            platform === 'ios'
+              ? AppPlatform.IOS
+              : platform === 'android'
+              ? AppPlatform.ANDROID
+              : AppPlatform.WEB,
+          appId,
+        })
+      }
+
+      if (ctx.cfg.courierOnlyRegistration) {
+        assert(ctx.courierClient)
+        await registerDeviceWithCourier(ctx.courierClient)
+      } else {
+        await registerDeviceWithAppview()
+        if (ctx.courierClient) {
+          try {
+            await registerDeviceWithCourier(ctx.courierClient)
+          } catch (err) {
+            req.log.warn(err, 'failed to register device token with courier')
+          }
+        }
+      }
     },
   })
 }

packages/bsky/src/config.ts

@@ -36,6 +36,11 @@ export interface ServerConfigValues {
   bsyncHttpVersion?: '1.1' | '2'
   bsyncIgnoreBadTls?: boolean
   bsyncOnlyMutes?: boolean
+  courierUrl?: string
+  courierApiKey?: string
+  courierHttpVersion?: '1.1' | '2'
+  courierIgnoreBadTls?: boolean
+  courierOnlyRegistration?: boolean
   adminPassword: string
   moderatorPassword: string
   triagePassword: string
@@ -100,6 +105,18 @@ export class ServerConfig {
     const bsyncOnlyMutes = process.env.BSKY_BSYNC_ONLY_MUTES === 'true'
     assert(!bsyncOnlyMutes || bsyncUrl, 'bsync-only mutes requires a bsync url')
     assert(bsyncHttpVersion === '1.1' || bsyncHttpVersion === '2')
+    const courierUrl = process.env.BSKY_COURIER_URL || undefined
+    const courierApiKey = process.env.BSKY_COURIER_API_KEY || undefined
+    const courierHttpVersion = process.env.BSKY_COURIER_HTTP_VERSION || '2'
+    const courierIgnoreBadTls =
+      process.env.BSKY_COURIER_IGNORE_BAD_TLS === 'true'
+    const courierOnlyRegistration =
+      process.env.BSKY_COURIER_ONLY_REGISTRATION === 'true'
+    assert(
+      !courierOnlyRegistration || courierUrl,
+      'courier-only registration requires a courier url',
+    )
+    assert(courierHttpVersion === '1.1' || courierHttpVersion === '2')
     const dbPrimaryPostgresUrl =
       overrides?.dbPrimaryPostgresUrl || process.env.DB_PRIMARY_POSTGRES_URL
     let dbReplicaPostgresUrls = overrides?.dbReplicaPostgresUrls
@@ -169,6 +186,11 @@ export class ServerConfig {
       bsyncHttpVersion,
       bsyncIgnoreBadTls,
       bsyncOnlyMutes,
+      courierUrl,
+      courierApiKey,
+      courierHttpVersion,
+      courierIgnoreBadTls,
+      courierOnlyRegistration,
       adminPassword,
       moderatorPassword,
       triagePassword,
@@ -305,6 +327,26 @@ export class ServerConfig {
     return this.cfg.bsyncIgnoreBadTls
   }
 
+  get courierUrl() {
+    return this.cfg.courierUrl
+  }
+
+  get courierApiKey() {
+    return this.cfg.courierApiKey
+  }
+
+  get courierHttpVersion() {
+    return this.cfg.courierHttpVersion
+  }
+
+  get courierIgnoreBadTls() {
+    return this.cfg.courierIgnoreBadTls
+  }
+
+  get courierOnlyRegistration() {
+    return this.cfg.courierOnlyRegistration
+  }
+
   get adminPassword() {
     return this.cfg.adminPassword
   }

packages/bsky/src/context.ts

@@ -10,10 +10,10 @@ import { Services } from './services'
 import DidRedisCache from './did-cache'
 import { BackgroundQueue } from './background'
 import { MountedAlgos } from './feed-gen/types'
-import { NotificationServer } from './notifications'
 import { Redis } from './redis'
 import { AuthVerifier } from './auth-verifier'
 import { BsyncClient } from './bsync'
+import { CourierClient } from './courier'
 
 export class AppContext {
   constructor(
@@ -29,8 +29,8 @@ export class AppContext {
       backgroundQueue: BackgroundQueue
       searchAgent?: AtpAgent
       bsyncClient?: BsyncClient
+      courierClient?: CourierClient
       algos: MountedAlgos
-      notifServer: NotificationServer
       authVerifier: AuthVerifier
     },
   ) {}
@@ -71,10 +71,6 @@ export class AppContext {
     return this.opts.redis
   }
 
-  get notifServer(): NotificationServer {
-    return this.opts.notifServer
-  }
-
   get searchAgent(): AtpAgent | undefined {
     return this.opts.searchAgent
   }
@@ -83,6 +79,10 @@ export class AppContext {
     return this.opts.bsyncClient
   }
 
+  get courierClient(): CourierClient | undefined {
+    return this.opts.courierClient
+  }
+
   get authVerifier(): AuthVerifier {
     return this.opts.authVerifier
   }

packages/bsky/src/courier.ts

@@ -0,0 +1,41 @@
+import { Service } from './proto/courier_connect'
+import {
+  Code,
+  ConnectError,
+  PromiseClient,
+  createPromiseClient,
+  Interceptor,
+} from '@connectrpc/connect'
+import {
+  createConnectTransport,
+  ConnectTransportOptions,
+} from '@connectrpc/connect-node'
+
+export type CourierClient = PromiseClient<typeof Service>
+
+export const createCourierClient = (
+  opts: ConnectTransportOptions,
+): CourierClient => {
+  const transport = createConnectTransport(opts)
+  return createPromiseClient(Service, transport)
+}
+
+export { Code }
+
+export const isCourierError = (
+  err: unknown,
+  code?: Code,
+): err is ConnectError => {
+  if (err instanceof ConnectError) {
+    return !code || err.code === code
+  }
+  return false
+}
+
+export const authWithApiKey =
+  (apiKey: string): Interceptor =>
+  (next) =>
+  (req) => {
+    req.header.set('authorization', `Bearer ${apiKey}`)
+    return next(req)
+  }

packages/bsky/src/index.ts

@@ -29,12 +29,12 @@ import {
 } from './image/invalidator'
 import { BackgroundQueue } from './background'
 import { MountedAlgos } from './feed-gen/types'
-import { NotificationServer } from './notifications'
 import { AtpAgent } from '@atproto/api'
 import { Keypair } from '@atproto/crypto'
 import { Redis } from './redis'
 import { AuthVerifier } from './auth-verifier'
-import { authWithApiKey, createBsyncClient } from './bsync'
+import { authWithApiKey as bsyncAuth, createBsyncClient } from './bsync'
+import { authWithApiKey as courierAuth, createCourierClient } from './courier'
 
 export type { ServerConfigValues } from './config'
 export type { MountedAlgos } from './feed-gen/types'
@@ -113,7 +113,6 @@ export class BskyAppView {
 
     const backgroundQueue = new BackgroundQueue(db.getPrimary())
 
-    const notifServer = new NotificationServer(db.getPrimary())
     const searchAgent = config.searchEndpoint
       ? new AtpAgent({ service: config.searchEndpoint })
       : undefined
@@ -142,7 +141,18 @@ export class BskyAppView {
           httpVersion: config.bsyncHttpVersion ?? '2',
           nodeOptions: { rejectUnauthorized: !config.bsyncIgnoreBadTls },
           interceptors: config.bsyncApiKey
-            ? [authWithApiKey(config.bsyncApiKey)]
+            ? [bsyncAuth(config.bsyncApiKey)]
+            : [],
+        })
+      : undefined
+
+    const courierClient = config.courierUrl
+      ? createCourierClient({
+          baseUrl: config.courierUrl,
+          httpVersion: config.courierHttpVersion ?? '2',
+          nodeOptions: { rejectUnauthorized: !config.courierIgnoreBadTls },
+          interceptors: config.courierApiKey
+            ? [courierAuth(config.courierApiKey)]
             : [],
         })
       : undefined
@@ -159,8 +169,8 @@ export class BskyAppView {
       backgroundQueue,
       searchAgent,
       bsyncClient,
+      courierClient,
       algos,
-      notifServer,
       authVerifier,
     })