enqueue community posts for moderation

packages/bsky/src/api/community/blacksky/feed/submitPost.ts

@@ -1,24 +1,31 @@
-import { InvalidRequestError, AuthRequiredError, Server } from '@atproto/xrpc-server'
+import { subsystemLogger } from '@atproto/common'
+import {
+  InvalidRequestError,
+  AuthRequiredError,
+  Server,
+} from '@atproto/xrpc-server'
 import { AppContext } from '../../../../context.js'
-import { AtUriString, DidString, CidString, AtIdentifierString } from '@atproto/lex'
+import { AtUriString, CidString } from '@atproto/lex'
+import {
+  getServiceEndpoint,
+  unpackIdentityServices,
+} from '../../../../data-plane/client/util.js'
 import { community } from '../../../../lexicons/index.js'
+import { findBlobMetadata } from '../../../../util/find-blob-refs.js'
+
+const logger = subsystemLogger('bsky:moderation')
 
 const COMMUNITY_POST_COLLECTION = 'community.blacksky.feed.post'
 
 export default function (server: Server, ctx: AppContext) {
   server.add(community.blacksky.feed.submitPost, {
     auth: ctx.authVerifier.standard,
     handler: async ({ input, auth }) => {
-      console.log('[submitPost] START', { hasAuth: !!auth })
-
       const requesterDid = auth.credentials.iss
-      console.log('[submitPost] requesterDid:', requesterDid)
 
-      console.log('[submitPost] checking membership...')
       const { isMember } = await ctx.dataplane.checkCommunityMembership({
         did: requesterDid,
       })
-      console.log('[submitPost] membership check result:', { isMember })
 
       if (!isMember) {
         throw new AuthRequiredError(
@@ -40,16 +47,6 @@ export default function (server: Server, ctx: AppContext) {
         expectedCid,
       } = input.body
 
-      console.log('[submitPost] input:', {
-        rkey,
-        text: text?.substring(0, 50),
-        hasReply: !!reply,
-        hasEmbed: !!embed,
-        langs,
-        createdAt,
-        expectedCid,
-      })
-
       // Validate reply cascade
       if (reply) {
         const rootUri = reply.root.uri
@@ -71,9 +68,7 @@ export default function (server: Server, ctx: AppContext) {
       }
 
       const uri = `at://${requesterDid}/${COMMUNITY_POST_COLLECTION}/${rkey}` as AtUriString
-      console.log('[submitPost] generated uri:', uri)
 
-      console.log('[submitPost] calling dataplane.submitCommunityPost...')
       const { cid, cidVerified } = await ctx.dataplane.submitCommunityPost({
         uri,
         rkey,
@@ -91,8 +86,6 @@ export default function (server: Server, ctx: AppContext) {
         createdAt,
         expectedCid: expectedCid ?? '',
       })
-      console.log('[submitPost] dataplane result:', { cid, cidVerified })
-
       // If client provided expectedCid but it didn't match, reject
       if (expectedCid && !cidVerified) {
         throw new InvalidRequestError(
@@ -101,7 +94,41 @@ export default function (server: Server, ctx: AppContext) {
         )
       }
 
-      console.log('[submitPost] SUCCESS')
+      // Enqueue for moderation (fire-and-forget with internal retry)
+      if (ctx.moderationClient) {
+        let pdsEndpoint: string | undefined
+        try {
+          const identity = await ctx.dataplane.getIdentityByDid({
+            did: requesterDid,
+          })
+          const services = unpackIdentityServices(identity.services)
+          pdsEndpoint = getServiceEndpoint(services, {
+            id: 'atproto_pds',
+            type: 'AtprotoPersonalDataServer',
+          })
+        } catch (err) {
+          logger.warn(
+            { err, did: requesterDid },
+            'failed to resolve PDS endpoint for moderation enqueue',
+          )
+        }
+
+        if (pdsEndpoint) {
+          const blobs = embed ? findBlobMetadata(embed) : []
+          const blobCids = blobs.map((blob) => blob.cid)
+          ctx.moderationClient
+            .enqueue({
+              did: requesterDid,
+              collection: COMMUNITY_POST_COLLECTION,
+              rkey,
+              pdsEndpoint,
+              blobCids: blobCids.length > 0 ? blobCids : undefined,
+              blobs: blobs.length > 0 ? blobs : undefined,
+            })
+            .catch(() => {}) // error already logged inside client
+        }
+      }
+
       return {
         encoding: 'application/json' as const,
         body: { uri: uri as AtUriString, cid: cid as CidString },

packages/bsky/src/config.ts

@@ -67,6 +67,8 @@ export interface ServerConfigValues {
   suggestionsApiKey?: string
   topicsUrl?: string
   topicsApiKey?: string
+  moderationUrl?: string
+  moderationApiKey?: string
   cdnUrl?: string
   videoPlaylistUrlPattern?: string
   videoThumbnailUrlPattern?: string
@@ -148,6 +150,8 @@ export class ServerConfig {
     const handleResolveNameservers = envList(
       process.env.BSKY_HANDLE_RESOLVE_NAMESERVERS,
     )
+    const moderationUrl = process.env.MODERATION_URL || undefined
+    const moderationApiKey = process.env.MODERATION_API_KEY || undefined
     const cdnUrl = process.env.BSKY_CDN_URL || process.env.BSKY_IMG_URI_ENDPOINT
     // Values 0 through 16
     const etcdHosts =
@@ -377,6 +381,8 @@ export class ServerConfig {
       rolodexApiKey,
       rolodexHttpVersion,
       rolodexIgnoreBadTls,
+      moderationUrl,
+      moderationApiKey,
       blobRateLimitBypassKey,
       blobRateLimitBypassHostname,
       adminPasswords,
@@ -540,6 +546,14 @@ export class ServerConfig {
     return this.cfg.topicsApiKey
   }
 
+  get moderationUrl() {
+    return this.cfg.moderationUrl
+  }
+
+  get moderationApiKey() {
+    return this.cfg.moderationApiKey
+  }
+
   get cdnUrl() {
     return this.cfg.cdnUrl
   }

packages/bsky/src/context.ts

@@ -13,6 +13,7 @@ import { DataPlaneClient, HostList } from './data-plane/client/index.js'
 import { FeatureGatesClient } from './feature-gates/index.js'
 import { Hydrator } from './hydration/hydrator.js'
 import { KwsClient } from './kws.js'
+import { ModerationClient } from './moderation-client.js'
 import { httpLogger as log } from './logger.js'
 import { PeerModConfig } from './peer-mod.js'
 import { RolodexClient } from './rolodex.js'
@@ -46,6 +47,7 @@ export class AppContext {
       featureGatesClient: FeatureGatesClient
       blobDispatcher: Dispatcher
       kwsClient: KwsClient | undefined
+      moderationClient: ModerationClient | undefined
       peerModConfig: PeerModConfig
     },
   ) {}
@@ -134,6 +136,10 @@ export class AppContext {
     return this.opts.kwsClient
   }
 
+  get moderationClient(): ModerationClient | undefined {
+    return this.opts.moderationClient
+  }
+
   reqLabelers(req: express.Request): ParsedLabelers {
     const val = req.header('atproto-accept-labelers')
     let parsed: ParsedLabelers | null

packages/bsky/src/index.ts

@@ -7,7 +7,7 @@ import { Etcd3 } from 'etcd3'
 import express from 'express'
 // eslint-disable-next-line import/default
 import httpTerminator from 'http-terminator'
-import { DAY, SECOND } from '@atproto/common'
+import { DAY, SECOND, subsystemLogger } from '@atproto/common'
 import { Keypair } from '@atproto/crypto'
 import { IdResolver } from '@atproto/identity'
 import { Client } from '@atproto/lex'
@@ -39,6 +39,7 @@ import { Hydrator } from './hydration/hydrator.js'
 import * as imageServer from './image/server.js'
 import { ImageUriBuilder } from './image/uri.js'
 import { createKwsClient } from './kws.js'
+import { ModerationClient } from './moderation-client.js'
 import { loggerMiddleware } from './logger.js'
 import { readPeerModConfig } from './peer-mod.js'
 import {
@@ -217,6 +218,20 @@ export class BskyAppView {
 
     const kwsClient = config.kws ? createKwsClient(config.kws) : undefined
 
+    let moderationClient: ModerationClient | undefined
+    if (config.moderationUrl) {
+      if (!config.moderationApiKey) {
+        subsystemLogger('bsky').warn(
+          'MODERATION_URL is set but MODERATION_API_KEY is missing — moderation client disabled',
+        )
+      } else {
+        moderationClient = new ModerationClient({
+          baseUrl: config.moderationUrl,
+          apiKey: config.moderationApiKey,
+        })
+      }
+    }
+
     const entrywayJwtPublicKey = config.entrywayJwtPublicKeyHex
       ? createPublicKeyObject(config.entrywayJwtPublicKeyHex)
       : undefined
@@ -252,6 +267,8 @@ export class BskyAppView {
       featureGatesClient,
       blobDispatcher,
       kwsClient,
+      community-posts-moderation
+      moderationClient,
       peerModConfig,
     })
 

packages/bsky/src/moderation-client.ts

@@ -0,0 +1,97 @@
+import { subsystemLogger } from '@atproto/common'
+
+const logger = subsystemLogger('bsky:moderation')
+
+export interface ModerationEnqueueParams {
+  did: string
+  collection: string
+  rkey: string
+  pdsEndpoint: string
+  blobCids?: string[]
+  blobs?: ModerationBlobMetadata[]
+}
+
+export interface ModerationBlobMetadata {
+  cid: string
+  mimeType: string
+  size?: number
+}
+
+export class ModerationClient {
+  private baseUrl: string
+  private apiKey: string
+  private maxRetries: number
+  private baseDelayMs: number
+
+  constructor(opts: {
+    baseUrl: string
+    apiKey: string
+    maxRetries?: number
+    baseDelayMs?: number
+  }) {
+    this.baseUrl = opts.baseUrl
+    this.apiKey = opts.apiKey
+    this.maxRetries = opts.maxRetries ?? 5
+    this.baseDelayMs = opts.baseDelayMs ?? 1000
+  }
+
+  async enqueue(params: ModerationEnqueueParams): Promise<void> {
+    const body = {
+      did: params.did,
+      collection: params.collection,
+      rkey: params.rkey,
+      pds_endpoint: params.pdsEndpoint,
+      blob_cids: params.blobCids,
+      blobs: params.blobs?.map((blob) => ({
+        cid: blob.cid,
+        mime_type: blob.mimeType,
+        ...(blob.size === undefined ? {} : { size: blob.size }),
+      })),
+    }
+
+    for (let attempt = 0; attempt <= this.maxRetries; attempt++) {
+      try {
+        const resp = await fetch(`${this.baseUrl}/enqueue`, {
+          method: 'POST',
+          headers: {
+            'Content-Type': 'application/json',
+            'X-Moderation-Key': this.apiKey,
+          },
+          body: JSON.stringify(body),
+        })
+        if (resp.status === 202) {
+          logger.info(
+            { did: params.did, rkey: params.rkey },
+            'moderation enqueue succeeded',
+          )
+          return
+        }
+        throw new Error(`Unexpected status: ${resp.status}`)
+      } catch (err) {
+        if (attempt < this.maxRetries) {
+          const delay = this.baseDelayMs * Math.pow(2, attempt)
+          logger.warn(
+            {
+              err,
+              attempt,
+              maxRetries: this.maxRetries,
+              delayMs: delay,
+              did: params.did,
+              rkey: params.rkey,
+            },
+            'moderation enqueue failed, retrying',
+          )
+          await new Promise((resolve) => setTimeout(resolve, delay))
+        }
+      }
+    }
+    logger.error(
+      {
+        did: params.did,
+        rkey: params.rkey,
+        collection: params.collection,
+      },
+      'CRITICAL: moderation enqueue failed after all retries - post not scanned',
+    )
+  }
+}

packages/bsky/src/util/find-blob-refs.ts

@@ -0,0 +1,27 @@
+import {
+  LexValue,
+  enumBlobRefs,
+  getBlobCidString,
+  getBlobMime,
+  getBlobSize,
+} from '@atproto/lex'
+import type { BlobRef } from '@atproto/lex'
+
+export interface BlobMetadata {
+  cid: string
+  mimeType: string
+  size?: number
+}
+
+export const findBlobRefs = (val: unknown): BlobRef[] =>
+  Array.from(enumBlobRefs(val as LexValue, { strict: false, allowLegacy: true }))
+
+export const findBlobMetadata = (val: unknown): BlobMetadata[] =>
+  findBlobRefs(val).map((ref) => {
+    const size = getBlobSize(ref)
+    return {
+      cid: getBlobCidString(ref),
+      mimeType: getBlobMime(ref),
+      ...(size === undefined ? {} : { size }),
+    }
+  })