✨ Apply queue filter on database query when fetching subjects

bluesky-social · Dec 20, 2024 · 8596172 · 8596172
1 parent c559e7e
commit 8596172
Show file tree

Hide file tree

Showing 9 changed files with 73 additions and 0 deletions.
diff --git a/lexicons/tools/ozone/moderation/queryStatuses.json b/lexicons/tools/ozone/moderation/queryStatuses.json
@@ -8,6 +8,14 @@
       "parameters": {
         "type": "params",
         "properties": {
+          "queueCount": {
+            "type": "integer",
+            "description": "Number of queues being used by moderators. Subjects will be split among all queues."
+          },
+          "queueIndex": {
+            "type": "integer",
+            "description": "Index of the queue to fetch subjects from. Works only when queueCount value is specified."
+          },
           "includeAllUserRecords": {
             "type": "boolean",
             "description": "All subjects, or subjects from given 'collections' param, belonging to the account specified in the 'subject' param will be returned."

diff --git a/packages/api/src/client/lexicons.ts b/packages/api/src/client/lexicons.ts
@@ -12377,6 +12377,16 @@ export const schemaDict = {
         parameters: {
           type: 'params',
           properties: {
+            queueCount: {
+              type: 'integer',
+              description:
+                'The number of subjects to return from the queue. If not specified, all subjects are returned.',
+            },
+            queueIndex: {
+              type: 'integer',
+              description:
+                'The index of the first subject to return from the queue. If not specified, the first subject is returned.',
+            },
             includeAllUserRecords: {
               type: 'boolean',
               description:

diff --git a/packages/api/src/client/types/tools/ozone/moderation/queryStatuses.ts b/packages/api/src/client/types/tools/ozone/moderation/queryStatuses.ts
@@ -9,6 +9,10 @@ import { CID } from 'multiformats/cid'
 import * as ToolsOzoneModerationDefs from './defs'
 
 export interface QueryParams {
+  /** The number of subjects to return from the queue. If not specified, all subjects are returned. */
+  queueCount?: number
+  /** The index of the first subject to return from the queue. If not specified, the first subject is returned. */
+  queueIndex?: number
   /** All subjects, or subjects from given 'collections' param, belonging to the account specified in the 'subject' param will be returned. */
   includeAllUserRecords?: boolean
   /** The subject to get the status for. */

diff --git a/packages/ozone/src/api/moderation/queryStatuses.ts b/packages/ozone/src/api/moderation/queryStatuses.ts
@@ -33,6 +33,8 @@ export default function (server: Server, ctx: AppContext) {
         excludeTags = [],
         collections = [],
         subjectType,
+        queueCount,
+        queueIndex,
       } = params
       const db = ctx.db
       const modService = ctx.modService(db)
@@ -63,6 +65,8 @@ export default function (server: Server, ctx: AppContext) {
         excludeTags,
         collections,
         subjectType,
+        queueCount,
+        queueIndex,
       })
       const subjectStatuses = results.statuses.map((status) =>
         modService.views.formatSubjectStatus(status),

diff --git a/packages/ozone/src/lexicon/lexicons.ts b/packages/ozone/src/lexicon/lexicons.ts
@@ -12377,6 +12377,16 @@ export const schemaDict = {
         parameters: {
           type: 'params',
           properties: {
+            queueCount: {
+              type: 'integer',
+              description:
+                'The number of subjects to return from the queue. If not specified, all subjects are returned.',
+            },
+            queueIndex: {
+              type: 'integer',
+              description:
+                'The index of the first subject to return from the queue. If not specified, the first subject is returned.',
+            },
             includeAllUserRecords: {
               type: 'boolean',
               description:

diff --git a/packages/ozone/src/lexicon/types/tools/ozone/moderation/queryStatuses.ts b/packages/ozone/src/lexicon/types/tools/ozone/moderation/queryStatuses.ts
@@ -10,6 +10,10 @@ import { HandlerAuth, HandlerPipeThrough } from '@atproto/xrpc-server'
 import * as ToolsOzoneModerationDefs from './defs'
 
 export interface QueryParams {
+  /** The number of subjects to return from the queue. If not specified, all subjects are returned. */
+  queueCount?: number
+  /** The index of the first subject to return from the queue. If not specified, the first subject is returned. */
+  queueIndex?: number
   /** All subjects, or subjects from given 'collections' param, belonging to the account specified in the 'subject' param will be returned. */
   includeAllUserRecords?: boolean
   /** The subject to get the status for. */

diff --git a/packages/ozone/src/mod-service/index.ts b/packages/ozone/src/mod-service/index.ts
@@ -831,6 +831,8 @@ export class ModerationService {
   }
 
   async getSubjectStatuses({
+    queueCount,
+    queueIndex,
     includeAllUserRecords,
     cursor,
     limit = 50,
@@ -858,6 +860,8 @@ export class ModerationService {
     collections,
     subjectType,
   }: {
+    queueCount?: number
+    queueIndex?: number
     includeAllUserRecords?: boolean
     cursor?: string
     limit?: number
@@ -909,6 +913,21 @@ export class ModerationService {
       builder = builder.where('recordPath', '!=', '')
     }
 
+    // Only fetch items that belongs to the specified queue when specified
+    if (
+      !subject &&
+      queueCount &&
+      queueIndex &&
+      queueCount >= 0 &&
+      queueIndex < queueCount
+    ) {
+      builder = builder.where(
+        sql`ABS(HASHTEXT(did)) % ${queueCount}`,
+        '=',
+        queueIndex,
+      )
+    }
+
     // If subjectType is set to 'account' let that take priority and ignore collections filter
     if (collections.length && subjectType !== 'account') {
       builder = builder.where('recordPath', '!=', '').where((qb) => {

diff --git a/packages/pds/src/lexicon/lexicons.ts b/packages/pds/src/lexicon/lexicons.ts
@@ -12377,6 +12377,16 @@ export const schemaDict = {
         parameters: {
           type: 'params',
           properties: {
+            queueCount: {
+              type: 'integer',
+              description:
+                'The number of subjects to return from the queue. If not specified, all subjects are returned.',
+            },
+            queueIndex: {
+              type: 'integer',
+              description:
+                'The index of the first subject to return from the queue. If not specified, the first subject is returned.',
+            },
             includeAllUserRecords: {
               type: 'boolean',
               description:

diff --git a/packages/pds/src/lexicon/types/tools/ozone/moderation/queryStatuses.ts b/packages/pds/src/lexicon/types/tools/ozone/moderation/queryStatuses.ts
@@ -10,6 +10,10 @@ import { HandlerAuth, HandlerPipeThrough } from '@atproto/xrpc-server'
 import * as ToolsOzoneModerationDefs from './defs'
 
 export interface QueryParams {
+  /** The number of subjects to return from the queue. If not specified, all subjects are returned. */
+  queueCount?: number
+  /** The index of the first subject to return from the queue. If not specified, the first subject is returned. */
+  queueIndex?: number
   /** All subjects, or subjects from given 'collections' param, belonging to the account specified in the 'subject' param will be returned. */
   includeAllUserRecords?: boolean
   /** The subject to get the status for. */