Clean up block & mute handling for push notifs (bluesky-social#1520)

clean up block & mute handling for push notifs

packages/bsky/src/notifications.ts

@@ -1,13 +1,13 @@
 import axios from 'axios'
-import { Insertable } from 'kysely'
+import { Insertable, sql } from 'kysely'
 import TTLCache from '@isaacs/ttlcache'
 import { AtUri } from '@atproto/api'
 import { MINUTE, chunkArray } from '@atproto/common'
 import Database from './db/primary'
 import { Notification } from './db/tables/notification'
 import { NotificationPushToken as PushToken } from './db/tables/notification-push-token'
 import logger from './indexer/logger'
-import { notSoftDeletedClause } from './db/util'
+import { notSoftDeletedClause, valuesList } from './db/util'
 import { ids } from './lexicon/lexicons'
 import { retryHttp } from './util/retry'
 
@@ -188,7 +188,7 @@ export class NotificationServer {
     const allUris = [...subjectUris, ...recordUris]
 
     // gather potential display data for notifications in batch
-    const [authors, posts] = await Promise.all([
+    const [authors, posts, blocksAndMutes] = await Promise.all([
       this.db.db
         .selectFrom('actor')
         .leftJoin('profile', 'profile.creator', 'actor.did')
@@ -207,6 +207,7 @@ export class NotificationServer {
         .where('post.uri', 'in', allUris.length ? allUris : [''])
         .select(['post.uri as uri', 'text'])
         .execute(),
+      this.findBlocksAndMutes(notifs),
     ])
 
     const authorsByDid = authors.reduce((acc, author) => {
@@ -233,8 +234,12 @@ export class NotificationServer {
       const postRecord = postsByUri[recordUri]
       const postSubject = subjectUri ? postsByUri[subjectUri] : null
 
-      // if no display name, dont send notification
-      if (!author) {
+      // if blocked or muted, don't send notification
+      const shouldFilter = blocksAndMutes.some(
+        (pair) => pair.author === notif.author && pair.receiver === notif.did,
+      )
+      if (shouldFilter || !author) {
+        // if no display name, dont send notification
         continue
       }
       // const author = displayName.displayName
@@ -304,6 +309,48 @@ export class NotificationServer {
 
     return results
   }
+
+  async findBlocksAndMutes(notifs: InsertableNotif[]) {
+    const pairs = notifs.map((n) => ({ author: n.author, receiver: n.did }))
+    const { ref } = this.db.db.dynamic
+    const blockQb = this.db.db
+      .selectFrom('actor_block')
+      .where((outer) =>
+        outer
+          .where((qb) =>
+            qb
+              .whereRef('actor_block.creator', '=', ref('author'))
+              .whereRef('actor_block.subjectDid', '=', ref('receiver')),
+          )
+          .orWhere((qb) =>
+            qb
+              .whereRef('actor_block.creator', '=', ref('receiver'))
+              .whereRef('actor_block.subjectDid', '=', ref('author')),
+          ),
+      )
+      .select(['creator', 'subjectDid'])
+    const muteQb = this.db.db
+      .selectFrom('mute')
+      .whereRef('mute.subjectDid', '=', ref('author'))
+      .whereRef('mute.mutedByDid', '=', ref('receiver'))
+      .selectAll()
+    const muteListQb = this.db.db
+      .selectFrom('list_item')
+      .innerJoin('list_mute', 'list_mute.listUri', 'list_item.listUri')
+      .whereRef('list_mute.mutedByDid', '=', ref('receiver'))
+      .whereRef('list_item.subjectDid', '=', ref('author'))
+      .select('list_item.subjectDid')
+
+    const values = valuesList(pairs.map((p) => sql`${p.author}, ${p.receiver}`))
+    const filterPairs = await this.db.db
+      .selectFrom(values.as(sql`pair (author, receiver)`))
+      .whereExists(muteQb)
+      .orWhereExists(muteListQb)
+      .orWhereExists(blockQb)
+      .selectAll()
+      .execute()
+    return filterPairs as { author: string; receiver: string }[]
+  }
 }
 
 const isRecent = (isoTime: string, timeDiff: number): boolean => {

packages/bsky/tests/notification-server.test.ts

@@ -1,4 +1,4 @@
-import AtpAgent from '@atproto/api'
+import AtpAgent, { AtUri } from '@atproto/api'
 import { TestNetwork } from '@atproto/dev-env'
 import { SeedClient } from './seeds/client'
 import basicSeed from './seeds/basic'
@@ -8,6 +8,7 @@ import { Database } from '../src'
 describe('notification server', () => {
   let network: TestNetwork
   let agent: AtpAgent
+  let pdsAgent: AtpAgent
   let sc: SeedClient
   let notifServer: NotificationServer
 
@@ -19,7 +20,7 @@ describe('notification server', () => {
       dbPostgresSchema: 'bsky_notification_server',
     })
     agent = network.bsky.getClient()
-    const pdsAgent = network.pds.getClient()
+    pdsAgent = network.pds.getClient()
     sc = new SeedClient(pdsAgent)
     await basicSeed(sc)
     await network.processAll()
@@ -111,6 +112,86 @@ describe('notification server', () => {
       expect(attrs[0].title).toEqual('bobby liked your post')
     })
 
+    it('filters notifications that violate blocks', async () => {
+      const db = network.bsky.ctx.db.getPrimary()
+      const notif = await getLikeNotification(db, alice)
+      if (!notif) throw new Error('no notification found')
+      const blockRef = await pdsAgent.api.app.bsky.graph.block.create(
+        { repo: alice },
+        { subject: notif.author, createdAt: new Date().toISOString() },
+        sc.getHeaders(alice),
+      )
+      await network.processAll()
+      // verify inverse of block
+      const flippedNotif = {
+        ...notif,
+        did: notif.author,
+        author: notif.did,
+      }
+      const attrs = await notifServer.getNotificationDisplayAttributes([
+        notif,
+        flippedNotif,
+      ])
+      expect(attrs.length).toBe(0)
+      const uri = new AtUri(blockRef.uri)
+      await pdsAgent.api.app.bsky.graph.block.delete(
+        { repo: alice, rkey: uri.rkey },
+        sc.getHeaders(alice),
+      )
+      await network.processAll()
+    })
+
+    it('filters notifications that violate mutes', async () => {
+      const db = network.bsky.ctx.db.getPrimary()
+      const notif = await getLikeNotification(db, alice)
+      if (!notif) throw new Error('no notification found')
+      await pdsAgent.api.app.bsky.graph.muteActor(
+        { actor: notif.author },
+        { headers: sc.getHeaders(alice), encoding: 'application/json' },
+      )
+      const attrs = await notifServer.getNotificationDisplayAttributes([notif])
+      expect(attrs.length).toBe(0)
+      await pdsAgent.api.app.bsky.graph.unmuteActor(
+        { actor: notif.author },
+        { headers: sc.getHeaders(alice), encoding: 'application/json' },
+      )
+    })
+
+    it('filters notifications that violate mutelists', async () => {
+      const db = network.bsky.ctx.db.getPrimary()
+      const notif = await getLikeNotification(db, alice)
+      if (!notif) throw new Error('no notification found')
+      const listRef = await pdsAgent.api.app.bsky.graph.list.create(
+        { repo: alice },
+        {
+          name: 'mute',
+          purpose: 'app.bsky.graph.defs#modlist',
+          createdAt: new Date().toISOString(),
+        },
+        sc.getHeaders(alice),
+      )
+      await pdsAgent.api.app.bsky.graph.listitem.create(
+        { repo: alice },
+        {
+          subject: notif.author,
+          list: listRef.uri,
+          createdAt: new Date().toISOString(),
+        },
+        sc.getHeaders(alice),
+      )
+      await network.processAll()
+      await pdsAgent.api.app.bsky.graph.muteActorList(
+        { list: listRef.uri },
+        { headers: sc.getHeaders(alice), encoding: 'application/json' },
+      )
+      const attrs = await notifServer.getNotificationDisplayAttributes([notif])
+      expect(attrs.length).toBe(0)
+      await pdsAgent.api.app.bsky.graph.unmuteActorList(
+        { list: listRef.uri },
+        { headers: sc.getHeaders(alice), encoding: 'application/json' },
+      )
+    })
+
     it('prepares notification to be sent', async () => {
       const db = network.bsky.ctx.db.getPrimary()
       const notif = await getLikeNotification(db, alice)