Config to start notifications daemon from a specific did (#1922)

config to start notifications daemon from a specific did
bluesky-social · Dec 5, 2023 · 49e7f98 · 49e7f98
1 parent f9fd3e6
commit 49e7f98
Show file tree

Hide file tree

Showing 5 changed files with 37 additions and 8 deletions.
diff --git a/packages/bsky/src/daemon/config.ts b/packages/bsky/src/daemon/config.ts
@@ -4,6 +4,7 @@ export interface DaemonConfigValues {
   version: string
   dbPostgresUrl: string
   dbPostgresSchema?: string
+  notificationsDaemonFromDid?: string
 }
 
 export class DaemonConfig {
@@ -15,11 +16,16 @@ export class DaemonConfig {
       overrides?.dbPostgresUrl || process.env.DB_PRIMARY_POSTGRES_URL
     const dbPostgresSchema =
       overrides?.dbPostgresSchema || process.env.DB_POSTGRES_SCHEMA
+    const notificationsDaemonFromDid =
+      overrides?.notificationsDaemonFromDid ||
+      process.env.BSKY_NOTIFS_DAEMON_FROM_DID ||
+      undefined
     assert(dbPostgresUrl)
     return new DaemonConfig({
       version,
       dbPostgresUrl,
       dbPostgresSchema,
+      notificationsDaemonFromDid,
       ...stripUndefineds(overrides ?? {}),
     })
   }
@@ -35,6 +41,10 @@ export class DaemonConfig {
   get dbPostgresSchema() {
     return this.cfg.dbPostgresSchema
   }
+
+  get notificationsDaemonFromDid() {
+    return this.cfg.notificationsDaemonFromDid
+  }
 }
 
 function stripUndefineds(

diff --git a/packages/bsky/src/daemon/index.ts b/packages/bsky/src/daemon/index.ts
@@ -43,9 +43,11 @@ export class BskyDaemon {
   }
 
   async start() {
-    const { db } = this.ctx
+    const { db, cfg } = this.ctx
     const pool = db.pool
-    this.notifications.run()
+    this.notifications.run({
+      startFromDid: cfg.notificationsDaemonFromDid,
+    })
     this.dbStatsInterval = setInterval(() => {
       dbLogger.info(
         {

diff --git a/packages/bsky/src/daemon/notifications.ts b/packages/bsky/src/daemon/notifications.ts
@@ -47,4 +47,8 @@ export class NotificationsDaemon {
   }
 }
 
-type RunOptions = { forever?: boolean; batchSize?: number }
+type RunOptions = {
+  forever?: boolean
+  batchSize?: number
+  startFromDid?: string
+}
diff --git a/packages/bsky/src/services/actor/index.ts b/packages/bsky/src/services/actor/index.ts
@@ -147,24 +147,34 @@ export class ActorService {
   }
 
   async *all(
-    opts: { batchSize?: number; forever?: boolean; cooldownMs?: number } = {},
+    opts: {
+      batchSize?: number
+      forever?: boolean
+      cooldownMs?: number
+      startFromDid?: string
+    } = {},
   ) {
-    const { cooldownMs = 1000, batchSize = 1000, forever = false } = opts
+    const {
+      cooldownMs = 1000,
+      batchSize = 1000,
+      forever = false,
+      startFromDid,
+    } = opts
     const baseQuery = this.db.db
       .selectFrom('actor')
       .selectAll()
       .orderBy('did')
       .limit(batchSize)
     while (true) {
-      let cursor: ActorResult | undefined
+      let cursor = startFromDid
       do {
         const actors = cursor
-          ? await baseQuery.where('did', '>', cursor.did).execute()
+          ? await baseQuery.where('did', '>', cursor).execute()
           : await baseQuery.execute()
         for (const actor of actors) {
           yield actor
         }
-        cursor = actors.at(-1)
+        cursor = actors.at(-1)?.did
       } while (cursor)
       if (forever) {
         await wait(cooldownMs)

diff --git a/services/bsky/daemon.js b/services/bsky/daemon.js
@@ -18,6 +18,7 @@ const main = async () => {
     version: env.version,
     dbPostgresUrl: env.dbPostgresUrl,
     dbPostgresSchema: env.dbPostgresSchema,
+    notificationsDaemonFromDid: env.notificationsDaemonFromDid,
   })
   const daemon = BskyDaemon.create({ db, cfg })
   await daemon.start()
@@ -34,6 +35,8 @@ const getEnv = () => ({
   dbPoolSize: maybeParseInt(process.env.DB_POOL_SIZE),
   dbPoolMaxUses: maybeParseInt(process.env.DB_POOL_MAX_USES),
   dbPoolIdleTimeoutMs: maybeParseInt(process.env.DB_POOL_IDLE_TIMEOUT_MS),
+  notificationsDaemonFromDid:
+    process.env.BSKY_NOTIFS_DAEMON_FROM_DID || undefined,
 })
 
 const maybeParseInt = (str) => {