Merge pull request #2115 from bluesky-social/signup-queue-notify

Signup queue notifications

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

@@ -3,7 +3,7 @@ on:
   push:
     branches:
       - main
-      - signup-queueing-take2
+      - signup-queue-notify
 env:
   REGISTRY: ${{ secrets.AWS_ECR_REGISTRY_USEAST2_PACKAGES_REGISTRY }}
   USERNAME: ${{ secrets.AWS_ECR_REGISTRY_USEAST2_PACKAGES_USERNAME }}

packages/pds/buf.gen.yaml

@@ -0,0 +1,12 @@
+version: v1
+plugins:
+  - plugin: es
+    opt:
+      - target=ts
+      - import_extension=.ts
+    out: src/proto
+  - plugin: connect-es
+    opt:
+      - target=ts
+      - import_extension=.ts
+    out: src/proto

packages/pds/package.json

@@ -29,7 +29,8 @@
     "test:sqlite": "jest --testPathIgnorePatterns /tests/proxied/*",
     "test:log": "tail -50 test.log | pino-pretty",
     "test:updateSnapshot": "jest --updateSnapshot",
-    "migration:create": "ts-node ./bin/migration-create.ts"
+    "migration:create": "ts-node ./bin/migration-create.ts",
+    "buf:gen": "buf generate ./proto"
   },
   "dependencies": {
     "@atproto/api": "workspace:^",
@@ -42,6 +43,9 @@
     "@atproto/syntax": "workspace:^",
     "@atproto/xrpc": "workspace:^",
     "@atproto/xrpc-server": "workspace:^",
+    "@bufbuild/protobuf": "^1.5.0",
+    "@connectrpc/connect": "^1.1.4",
+    "@connectrpc/connect-node": "^1.1.4",
     "@did-plc/lib": "^0.0.1",
     "better-sqlite3": "^7.6.2",
     "bytes": "^3.1.2",
@@ -66,6 +70,7 @@
     "pg": "^8.10.0",
     "pino": "^8.15.0",
     "pino-http": "^8.2.1",
+    "rate-limiter-flexible": "^2.4.1",
     "sharp": "^0.32.6",
     "twilio": "^4.20.1",
     "typed-emitter": "^2.1.0",
@@ -77,6 +82,9 @@
     "@atproto/bsky": "workspace:^",
     "@atproto/dev-env": "workspace:^",
     "@atproto/lex-cli": "workspace:^",
+    "@bufbuild/buf": "^1.28.1",
+    "@bufbuild/protoc-gen-es": "^1.5.0",
+    "@connectrpc/protoc-gen-connect-es": "^1.1.4",
     "@did-plc/server": "^0.0.1",
     "@types/cors": "^2.8.12",
     "@types/disposable-email": "^0.2.0",

packages/pds/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/pds/src/config/config.ts

@@ -211,6 +211,16 @@ export const envToCfg = (env: ServerEnvironment): ServerConfig => {
       }
     : { enabled: false }
 
+  const courierHttpVersion = env.courierHttpVersion ?? '2'
+  assert(courierHttpVersion === '1.1' || courierHttpVersion === '2')
+  const activatorCfg: ServerConfig['activator'] = {
+    courierUrl: env.courierUrl,
+    courierHttpVersion,
+    courierIgnoreBadTls: env.courierIgnoreBadTls,
+    courierApiKey: env.courierApiKey,
+    emailsPerDay: env.activatorEmailsPerDay,
+  }
+
   const crawlersCfg: ServerConfig['crawlers'] = env.crawlers ?? []
 
   return {
@@ -227,6 +237,7 @@ export const envToCfg = (env: ServerEnvironment): ServerConfig => {
     modService: modServiceCfg,
     redis: redisCfg,
     rateLimits: rateLimitsCfg,
+    activator: activatorCfg,
     crawlers: crawlersCfg,
   }
 }
@@ -245,6 +256,7 @@ export type ServerConfig = {
   modService: ModServiceConfig
   redis: RedisScratchConfig | null
   rateLimits: RateLimitsConfig
+  activator: ActivatorConfig
   crawlers: string[]
 }
 
@@ -355,6 +367,14 @@ export type RateLimitsConfig =
     }
   | { enabled: false }
 
+export type ActivatorConfig = {
+  courierUrl?: string
+  courierHttpVersion?: '1.1' | '2'
+  courierIgnoreBadTls?: boolean
+  courierApiKey?: string
+  emailsPerDay?: number
+}
+
 export type BksyAppViewConfig = {
   url: string
   did: string

packages/pds/src/config/env.ts

@@ -87,6 +87,13 @@ export const readEnv = (): ServerEnvironment => {
     redisScratchAddress: envStr('PDS_REDIS_SCRATCH_ADDRESS'),
     redisScratchPassword: envStr('PDS_REDIS_SCRATCH_PASSWORD'),
 
+    // activator
+    courierUrl: envStr('PDS_COURIER_URL'),
+    courierHttpVersion: envStr('PDS_COURIER_HTTP_VERSION'),
+    courierIgnoreBadTls: envBool('PDS_COURIER_IGNORE_BAD_TLS'),
+    courierApiKey: envStr('PDS_COURIER_API_KEY'),
+    activatorEmailsPerDay: envInt('PDS_ACTIVATOR_EMAILS_PER_DAY'),
+
     // crawlers
     crawlers: envList('PDS_CRAWLERS'),
 
@@ -202,6 +209,13 @@ export type ServerEnvironment = {
   redisScratchAddress?: string
   redisScratchPassword?: string
 
+  // activator
+  courierUrl?: string
+  courierHttpVersion?: string
+  courierIgnoreBadTls?: boolean
+  courierApiKey?: string
+  activatorEmailsPerDay?: number
+
   // crawler
   crawlers?: string[]
 

packages/pds/src/context.ts

@@ -1,5 +1,6 @@
 import * as nodemailer from 'nodemailer'
 import { Redis } from 'ioredis'
+import { RateLimiterRedis } from 'rate-limiter-flexible'
 import * as plc from '@did-plc/lib'
 import * as crypto from '@atproto/crypto'
 import { IdResolver } from '@atproto/identity'
@@ -25,6 +26,8 @@ import { TwilioClient } from './twilio'
 import assert from 'assert'
 import { SignupLimiter } from './signup-queue/limiter'
 import { SignupActivator } from './signup-queue/activator'
+import { createCourierClient, authWithApiKey as courierAuth } from './courier'
+import { DAY } from '@atproto/common'
 
 export type AppContextOptions = {
   db: Database
@@ -231,7 +234,33 @@ export class AppContext {
     }
 
     const signupLimiter = new SignupLimiter(db)
-    const signupActivator = new SignupActivator(db)
+    const courierClient = cfg.activator.courierUrl
+      ? createCourierClient({
+          baseUrl: cfg.activator.courierUrl,
+          httpVersion: cfg.activator.courierHttpVersion ?? '2',
+          nodeOptions: {
+            rejectUnauthorized: !cfg.activator.courierIgnoreBadTls,
+          },
+          interceptors: cfg.activator.courierApiKey
+            ? [courierAuth(cfg.activator.courierApiKey)]
+            : [],
+        })
+      : undefined
+    const limiter =
+      cfg.activator.emailsPerDay && redisScratch
+        ? new RateLimiterRedis({
+            storeClient: redisScratch,
+            duration: DAY / 1000,
+            points: cfg.activator.emailsPerDay,
+          })
+        : undefined
+
+    const signupActivator = new SignupActivator({
+      db,
+      mailer,
+      courierClient,
+      limiter,
+    })
 
     const pdsAgents = new PdsAgents()
 

packages/pds/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/pds/src/mailer/index.ts

@@ -26,6 +26,7 @@ export class ServerMailer {
       deleteAccount: this.compile('delete-account'),
       confirmEmail: this.compile('confirm-email'),
       updateEmail: this.compile('update-email'),
+      accountActivated: this.compile('account-activated'),
     }
   }
 
@@ -65,6 +66,16 @@ export class ServerMailer {
     })
   }
 
+  async sendAccountActivated(
+    params: { handle: string },
+    mailOpts: Mail.Options,
+  ) {
+    return this.sendTemplate('accountActivated', params, {
+      subject: 'Your Bluesky Account is Activated!',
+      ...mailOpts,
+    })
+  }
+
   private async sendTemplate(templateName, params, mailOpts: Mail.Options) {
     const html = this.templates[templateName]({
       ...params,