From 7a657e61153f9f52a98cc64bfdbd9076f8fe5850 Mon Sep 17 00:00:00 2001 From: Rishi Balakrishnan Date: Mon, 22 Jun 2026 13:35:38 -0400 Subject: [PATCH] enqueue community posts for moderation --- .../api/community/blacksky/feed/submitPost.ts | 71 +++++++++----- packages/bsky/src/config.ts | 14 +++ packages/bsky/src/context.ts | 6 ++ packages/bsky/src/index.ts | 18 +++- packages/bsky/src/moderation-client.ts | 97 +++++++++++++++++++ packages/bsky/src/util/find-blob-refs.ts | 27 ++++++ 6 files changed, 210 insertions(+), 23 deletions(-) create mode 100644 packages/bsky/src/moderation-client.ts create mode 100644 packages/bsky/src/util/find-blob-refs.ts diff --git a/packages/bsky/src/api/community/blacksky/feed/submitPost.ts b/packages/bsky/src/api/community/blacksky/feed/submitPost.ts index 2eab0dc4e..9db4f9025 100644 --- a/packages/bsky/src/api/community/blacksky/feed/submitPost.ts +++ b/packages/bsky/src/api/community/blacksky/feed/submitPost.ts @@ -1,7 +1,19 @@ -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' @@ -9,16 +21,11 @@ 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 }, diff --git a/packages/bsky/src/config.ts b/packages/bsky/src/config.ts index 36324e075..677724b1d 100644 --- a/packages/bsky/src/config.ts +++ b/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 } diff --git a/packages/bsky/src/context.ts b/packages/bsky/src/context.ts index 2643b0159..03dc25ce6 100644 --- a/packages/bsky/src/context.ts +++ b/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 { RolodexClient } from './rolodex.js' import { StashClient } from './stash.js' @@ -45,6 +46,7 @@ export class AppContext { featureGatesClient: FeatureGatesClient blobDispatcher: Dispatcher kwsClient: KwsClient | undefined + moderationClient: ModerationClient | undefined }, ) {} @@ -128,6 +130,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 diff --git a/packages/bsky/src/index.ts b/packages/bsky/src/index.ts index 205b7f18a..aa5e33857 100644 --- a/packages/bsky/src/index.ts +++ b/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 { authWithApiKey as rolodexAuth, @@ -216,6 +217,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 @@ -249,6 +264,7 @@ export class BskyAppView { featureGatesClient, blobDispatcher, kwsClient, + moderationClient, }) const server = createServer([], { diff --git a/packages/bsky/src/moderation-client.ts b/packages/bsky/src/moderation-client.ts new file mode 100644 index 000000000..21271288e --- /dev/null +++ b/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 { + 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', + ) + } +} diff --git a/packages/bsky/src/util/find-blob-refs.ts b/packages/bsky/src/util/find-blob-refs.ts new file mode 100644 index 000000000..a985ba7a9 --- /dev/null +++ b/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 }), + } + })