Send mark-read-generic notification on updateSeen (#2567)

* send `mark-read-generic` notification on `updateSeen`

add `reason`

add `recipientDid`

push `mark-read-generic` notification on `updateSeen`

add `client_controlled`

* unique id, change `alwaysDeliver` to false

* use murmur id

* organize import

* bsky: fix tests, making courier config optional. fix unread count query.

---------

Co-authored-by: Devin Ivy <[email protected]>

Author: haileyok

packages/bsky/package.json

@@ -51,6 +51,7 @@
     "key-encoder": "^2.0.3",
     "kysely": "^0.22.0",
     "multiformats": "^9.9.0",
+    "murmurhash": "^2.0.1",
     "p-queue": "^6.6.2",
     "pg": "^8.10.0",
     "pino": "^8.21.0",

packages/bsky/proto/courier.proto

@@ -32,6 +32,7 @@ message Notification {
   bool always_deliver = 6;
   google.protobuf.Timestamp timestamp = 7;
   google.protobuf.Struct additional = 8;
+  bool client_controlled = 9;
 }
 
 message PushNotificationsRequest {

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

@@ -1,4 +1,7 @@
-import { InvalidRequestError } from '@atproto/xrpc-server'
+import {
+  InvalidRequestError,
+  MethodNotImplementedError,
+} from '@atproto/xrpc-server'
 import { Server } from '../../../../lexicon'
 import AppContext from '../../../../context'
 import { AppPlatform } from '../../../../proto/courier_pb'
@@ -7,6 +10,11 @@ export default function (server: Server, ctx: AppContext) {
   server.app.bsky.notification.registerPush({
     auth: ctx.authVerifier.standard,
     handler: async ({ auth, input }) => {
+      if (!ctx.courierClient) {
+        throw new MethodNotImplementedError(
+          'This service is not configured to support push token registration.',
+        )
+      }
       const { token, platform, serviceDid, appId } = input.body
       const did = auth.credentials.iss
       if (serviceDid !== auth.credentials.aud) {

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

@@ -1,4 +1,5 @@
-import { Timestamp } from '@bufbuild/protobuf'
+import murmur from 'murmurhash'
+import { Struct, Timestamp } from '@bufbuild/protobuf'
 import { Server } from '../../../../lexicon'
 import AppContext from '../../../../context'
 
@@ -20,7 +21,29 @@ export default function (server: Server, ctx: AppContext) {
           timestamp: Timestamp.fromDate(seenAt),
           priority: true,
         }),
+        ctx.courierClient?.pushNotifications({
+          notifications: [
+            {
+              id: getNotifId(viewer, seenAt),
+              clientControlled: true,
+              recipientDid: viewer,
+              alwaysDeliver: false,
+              collapseKey: 'mark-read-generic',
+              timestamp: Timestamp.fromDate(new Date()),
+              additional: Struct.fromJson({
+                reason: 'mark-read-generic',
+              }),
+            },
+          ],
+        }),
       ])
     },
   })
 }
+
+function getNotifId(viewer: string, seenAt: Date) {
+  const key = ['mark-read-generic', viewer, seenAt.getTime().toString()].join(
+    '::',
+  )
+  return murmur.v3(key).toString(16)
+}

packages/bsky/src/config.ts

@@ -17,7 +17,7 @@ export interface ServerConfigValues {
   bsyncApiKey?: string
   bsyncHttpVersion?: '1.1' | '2'
   bsyncIgnoreBadTls?: boolean
-  courierUrl: string
+  courierUrl?: string
   courierApiKey?: string
   courierHttpVersion?: '1.1' | '2'
   courierIgnoreBadTls?: boolean
@@ -92,7 +92,6 @@ export class ServerConfig {
     const bsyncIgnoreBadTls = process.env.BSKY_BSYNC_IGNORE_BAD_TLS === 'true'
     assert(bsyncHttpVersion === '1.1' || bsyncHttpVersion === '2')
     const courierUrl = process.env.BSKY_COURIER_URL || undefined
-    assert(courierUrl)
     const courierApiKey = process.env.BSKY_COURIER_API_KEY || undefined
     const courierHttpVersion = process.env.BSKY_COURIER_HTTP_VERSION || '2'
     const courierIgnoreBadTls =

packages/bsky/src/context.ts

@@ -30,7 +30,7 @@ export class AppContext {
       signingKey: Keypair
       idResolver: IdResolver
       bsyncClient: BsyncClient
-      courierClient: CourierClient
+      courierClient: CourierClient | undefined
       authVerifier: AuthVerifier
       featureGates: FeatureGates
     },
@@ -76,7 +76,7 @@ export class AppContext {
     return this.opts.bsyncClient
   }
 
-  get courierClient(): CourierClient {
+  get courierClient(): CourierClient | undefined {
     return this.opts.courierClient
   }
 

packages/bsky/src/data-plane/server/routes/notifs.ts

@@ -116,7 +116,7 @@ export default (db: Database): Partial<ServiceImpl<typeof Service>> => ({
             .selectFrom('follow')
             .select(sql<boolean>`${true}`.as('val'))
             .where('creator', '=', actorDid)
-            .whereRef('subjectDid', '=', ref('notif.author')),
+            .whereRef('subjectDid', '=', ref('notification.author')),
         ),
       )
       .executeTakeFirst()

packages/bsky/src/index.ts

@@ -110,14 +110,16 @@ export class BskyAppView {
       interceptors: config.bsyncApiKey ? [bsyncAuth(config.bsyncApiKey)] : [],
     })
 
-    const courierClient = createCourierClient({
-      baseUrl: config.courierUrl,
-      httpVersion: config.courierHttpVersion ?? '2',
-      nodeOptions: { rejectUnauthorized: !config.courierIgnoreBadTls },
-      interceptors: config.courierApiKey
-        ? [courierAuth(config.courierApiKey)]
-        : [],
-    })
+    const courierClient = config.courierUrl
+      ? createCourierClient({
+          baseUrl: config.courierUrl,
+          httpVersion: config.courierHttpVersion ?? '2',
+          nodeOptions: { rejectUnauthorized: !config.courierIgnoreBadTls },
+          interceptors: config.courierApiKey
+            ? [courierAuth(config.courierApiKey)]
+            : [],
+        })
+      : undefined
 
     const entrywayJwtPublicKey = config.entrywayJwtPublicKeyHex
       ? createPublicKeyObject(config.entrywayJwtPublicKeyHex)

packages/bsky/src/proto/courier_pb.ts

@@ -175,6 +175,11 @@ export class Notification extends Message<Notification> {
    */
   additional?: Struct
 
+  /**
+   * @generated from field: bool client_controlled = 9;
+   */
+  clientControlled = false
+
   constructor(data?: PartialMessage<Notification>) {
     super()
     proto3.util.initPartial(data, this)
@@ -206,6 +211,12 @@ export class Notification extends Message<Notification> {
     },
     { no: 7, name: 'timestamp', kind: 'message', T: Timestamp },
     { no: 8, name: 'additional', kind: 'message', T: Struct },
+    {
+      no: 9,
+      name: 'client_controlled',
+      kind: 'scalar',
+      T: 8 /* ScalarType.BOOL */,
+    },
   ])
 
   static fromBinary(

packages/dev-env/src/bsky.ts

@@ -62,7 +62,6 @@ export class TestBsky {
       dataplaneHttpVersion: '1.1',
       bsyncUrl: `http://localhost:${bsyncPort}`,
       bsyncHttpVersion: '1.1',
-      courierUrl: 'https://fake.example',
       modServiceDid: cfg.modServiceDid ?? 'did:example:invalidMod',
       labelsFromIssuerDids: [EXAMPLE_LABELER],
       ...cfg,