Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Appview v1 maintaining device tokens and pushing notifications w/ courier #2073

Merged
merged 6 commits into from
Jan 24, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
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