Skip to content

Commit

Permalink
Appview v1 maintaining device tokens and pushing notifications w/ cou…
Browse files Browse the repository at this point in the history
…rier (#2073)

* add courier proto to bsky, build

* update registerPush on appview to support registering device tokens with courier

* setup bsky notifications to send to either gorush or courier

* wire courier push into indexer, test

* courier push retries

* tidy and build
  • Loading branch information
devinivy authored Jan 24, 2024
1 parent d108310 commit 6a318b9
Show file tree
Hide file tree
Showing 19 changed files with 1,059 additions and 196 deletions.
2 changes: 1 addition & 1 deletion .github/workflows/build-and-push-bsky-aws.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -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 }}
Expand Down
5 changes: 3 additions & 2 deletions packages/bsky/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -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",
Expand Down
56 changes: 56 additions & 0 deletions packages/bsky/proto/courier.proto
Original file line number Diff line number Diff line change
@@ -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);
}
50 changes: 42 additions & 8 deletions packages/bsky/src/api/app/bsky/notification/registerPush.ts
Original file line number Diff line number Diff line change
@@ -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')
}
}
}
},
})
}
42 changes: 42 additions & 0 deletions packages/bsky/src/config.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -169,6 +186,11 @@ export class ServerConfig {
bsyncHttpVersion,
bsyncIgnoreBadTls,
bsyncOnlyMutes,
courierUrl,
courierApiKey,
courierHttpVersion,
courierIgnoreBadTls,
courierOnlyRegistration,
adminPassword,
moderatorPassword,
triagePassword,
Expand Down Expand Up @@ -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
}
Expand Down
12 changes: 6 additions & 6 deletions packages/bsky/src/context.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -29,8 +29,8 @@ export class AppContext {
backgroundQueue: BackgroundQueue
searchAgent?: AtpAgent
bsyncClient?: BsyncClient
courierClient?: CourierClient
algos: MountedAlgos
notifServer: NotificationServer
authVerifier: AuthVerifier
},
) {}
Expand Down Expand Up @@ -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
}
Expand All @@ -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
}
Expand Down
41 changes: 41 additions & 0 deletions packages/bsky/src/courier.ts
Original file line number Diff line number Diff line change
@@ -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)
}
20 changes: 15 additions & 5 deletions packages/bsky/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -159,8 +169,8 @@ export class BskyAppView {
backgroundQueue,
searchAgent,
bsyncClient,
courierClient,
algos,
notifServer,
authVerifier,
})

Expand Down
Loading

0 comments on commit 6a318b9

Please sign in to comment.