diff --git a/client/src/webpages/dashboard/mrt/ManualReviewQueuesDashboard.tsx b/client/src/webpages/dashboard/mrt/ManualReviewQueuesDashboard.tsx index 08e4103..919accb 100644 --- a/client/src/webpages/dashboard/mrt/ManualReviewQueuesDashboard.tsx +++ b/client/src/webpages/dashboard/mrt/ManualReviewQueuesDashboard.tsx @@ -1,10 +1,11 @@ import { StarFilled, TapFilled } from '@/icons'; import AngleDoubleRight from '@/icons/lni/Direction/angle-double-right.svg?react'; -import GridAlt from '@/icons/lnif/Design/grid-alt.svg?react'; import Star from '@/icons/lni/Web and Technology/star.svg?react'; +import GridAlt from '@/icons/lnif/Design/grid-alt.svg?react'; import { gql } from '@apollo/client'; import Button from 'antd/lib/button'; import Checkbox from 'antd/lib/checkbox'; +import Input from 'antd/lib/input'; import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; import { Helmet } from 'react-helmet-async'; import { Link, useNavigate } from 'react-router-dom'; @@ -18,6 +19,7 @@ import TabBar from '../components/TabBar'; import { ColumnProps, DefaultColumnFilter } from '../components/table/filters'; import { dateSort, integerSort, stringSort } from '../components/table/sort'; import Table from '../components/table/Table'; +import CopyTextComponent from '@/components/common/CopyTextComponent'; import { GQLUserPermission, @@ -116,41 +118,61 @@ type DeleteAllJobsModalInfo = { visible: boolean; }; +function getQueueName( + queues: ReadonlyArray<{ id: string; name: string }> | null | undefined, + id: string | undefined, +): string | null { + if (queues == null || id == null) return null; + return queues.find((it) => it.id === id)?.name ?? null; +} + const MRTQueuesDashboardTabs = ['DEFAULT', 'APPEALS'] as const; type MRTQueuesDashboardTab = (typeof MRTQueuesDashboardTabs)[number]; // Utility function to format time ago const formatTimeAgo = (date: string | Date | null | undefined): string => { if (!date) return 'N/A'; - + const now = new Date().getTime(); const then = new Date(date).getTime(); const diffMs = now - then; const diffMins = Math.floor(diffMs / 60000); const diffHours = Math.floor(diffMs / 3600000); const diffDays = Math.floor(diffMs / 86400000); - + if (diffMins < 1) return 'Just now'; - if (diffMins < 60) return `${diffMins} minute${diffMins !== 1 ? 's' : ''} ago`; - if (diffHours < 24) return `${diffHours} hour${diffHours !== 1 ? 's' : ''} ago`; + if (diffMins < 60) + return `${diffMins} minute${diffMins !== 1 ? 's' : ''} ago`; + if (diffHours < 24) + return `${diffHours} hour${diffHours !== 1 ? 's' : ''} ago`; return `${diffDays} day${diffDays !== 1 ? 's' : ''} ago`; }; // Get color class based on age const getAgeColorClass = (date: string | Date | null | undefined): string => { if (!date) return 'text-gray-500'; - + const now = new Date().getTime(); const then = new Date(date).getTime(); const diffHours = (now - then) / 3600000; - + if (diffHours > 24) return 'text-red-600 font-semibold'; // Over 1 day - red if (diffHours > 4) return 'text-orange-600'; // Over 4 hours - orange return 'text-green-600'; // Under 4 hours - green }; // Column visibility configuration -type ColumnId = 'favoriteQueues' | 'id' | 'name' | 'description' | 'oldestTaskAge' | 'pendingJobCount' | 'startReviewing' | 'mutations' | 'deleteJobs' | 'previewJobs'; +type ColumnId = + | 'favoriteQueues' + | 'id' + | 'name' + | 'description' + | 'oldestTaskAge' + | 'pendingJobCount' + | 'startReviewing' + | 'mutations' + | 'deleteJobs' + | 'previewJobs'; const COLUMN_VISIBILITY_STORAGE_KEY = 'mrt-queues-column-visibility'; @@ -197,7 +219,9 @@ export default function ManualReviewQueuesDashboard() { }); // Column visibility state - const [columnVisibility, setColumnVisibility] = useState>(() => { + const [columnVisibility, setColumnVisibility] = useState< + Record + >(() => { try { const stored = localStorage.getItem(COLUMN_VISIBILITY_STORAGE_KEY); if (stored) { @@ -212,14 +236,17 @@ export default function ManualReviewQueuesDashboard() { // Save column visibility to localStorage whenever it changes useEffect(() => { try { - localStorage.setItem(COLUMN_VISIBILITY_STORAGE_KEY, JSON.stringify(columnVisibility)); + localStorage.setItem( + COLUMN_VISIBILITY_STORAGE_KEY, + JSON.stringify(columnVisibility), + ); } catch (e) { // Failed to save to localStorage } }, [columnVisibility]); const toggleColumnVisibility = useCallback((columnId: ColumnId) => { - setColumnVisibility(prev => ({ ...prev, [columnId]: !prev[columnId] })); + setColumnVisibility((prev) => ({ ...prev, [columnId]: !prev[columnId] })); }, []); const [columnsMenuVisible, setColumnsMenuVisible] = useState(false); @@ -271,6 +298,10 @@ export default function ManualReviewQueuesDashboard() { const [modalInfo, setModalInfo] = useState(null); const [deleteAllJobsModalInfo, setDeleteAllJobsModalInfo] = useState(null); + const [deleteAllJobsConfirmText, setDeleteAllJobsConfirmText] = useState(''); + useEffect(() => { + setDeleteAllJobsConfirmText(''); + }, [deleteAllJobsModalInfo?.id, deleteAllJobsModalInfo?.visible]); const [deleteAllJobsFromQueue] = useGQLDeleteAllJobsFromQueueMutation({ onError: () => {}, onCompleted: async () => refetch(), @@ -321,14 +352,21 @@ export default function ManualReviewQueuesDashboard() { /> ); + const deleteAllJobsTargetName = getQueueName( + queues, + deleteAllJobsModalInfo?.id, + ); + // Trim both sides; the comparison is otherwise exact (case- and + // whitespace-sensitive in the middle of the string). + const deleteAllJobsConfirmEnabled = + deleteAllJobsTargetName != null && + deleteAllJobsConfirmText.trim() === deleteAllJobsTargetName.trim(); const deleteAllJobsModal = ( it.id === modalInfo.id)!.name - }'` + : `Delete All Jobs From '${deleteAllJobsTargetName}'` } visible={deleteAllJobsModalInfo?.visible ?? false} footer={[ @@ -339,7 +377,9 @@ export default function ManualReviewQueuesDashboard() { }, { title: 'Confirm', + disabled: !deleteAllJobsConfirmEnabled, onClick: () => { + if (!deleteAllJobsConfirmEnabled) return; deleteAllJobsFromQueue({ variables: { queueId: deleteAllJobsModalInfo!.id }, }); @@ -348,8 +388,27 @@ export default function ManualReviewQueuesDashboard() { }, ]} > - Are you sure you want to delete all jobs from this queue? You can't undo - this action. For larger queues, this can take a few minutes. +
+

+ You are about to delete all jobs from{' '} + {deleteAllJobsTargetName ?? 'this queue'}. This wipes + the queue's pending items in Redis and{' '} + cannot be undone from the UI. Recovery is only + possible by re-running the items through{' '} + npm run recover-mrt-queue on the server. +

+

+ To confirm, type the queue name{' '} + {deleteAllJobsTargetName ?? ''} below: +

+ setDeleteAllJobsConfirmText(event.target.value)} + placeholder={deleteAllJobsTargetName ?? 'queue name'} + aria-label="Type the queue name to confirm deletion" + autoFocus + /> +
); @@ -412,83 +471,106 @@ export default function ManualReviewQueuesDashboard() { ); const columns = useMemo( - () => filterNullOrUndefined([ - columnVisibility.favoriteQueues ? { - Header: '', - accessor: 'favoriteQueues', - canSort: false, - } : undefined, - columnVisibility.id ? { - Header: 'ID', - accessor: 'id', - Filter: (props: ColumnProps) => - DefaultColumnFilter({ - columnProps: props, - accessor: 'id', - placeholder: 'Queue ID', - }), - filter: 'text', - sortType: stringSort, - } : undefined, - columnVisibility.name ? { - Header: 'Name', - accessor: 'name', - Filter: (props: ColumnProps) => - DefaultColumnFilter({ - columnProps: props, - accessor: 'name', - placeholder: 'My Queue', - }), - filter: 'text', - sortType: stringSort, - } : undefined, - columnVisibility.description ? { - Header: 'Description', - accessor: 'description', - Filter: (props: ColumnProps) => - DefaultColumnFilter({ - columnProps: props, - accessor: 'description', - }), - filter: 'text', - sortType: stringSort, - } : undefined, - columnVisibility.oldestTaskAge ? { - Header: 'Oldest Task Age', - accessor: 'oldestTaskAge', - sortType: dateSort('oldestJobCreatedAt'), - } : undefined, - columnVisibility.pendingJobCount ? { - Header: 'Pending Jobs', - accessor: 'pendingJobCount', - sortType: integerSort, - } : undefined, - columnVisibility.startReviewing ? { - Header: '', - accessor: 'startReviewing', - canSort: false, - } : undefined, - columnVisibility.mutations ? { - Header: '', - accessor: 'mutations', - canSort: false, - } : undefined, - userHasPermissions(data?.me?.permissions, [ - GQLUserPermission.EditMrtQueues, - ]) && columnVisibility.deleteJobs - ? { - Header: '', - accessor: 'deleteJobs', - canSort: false, - } - : undefined, - columnVisibility.previewJobs ? { - Header: '', - accessor: 'previewJobs', - canSort: false, - } : undefined, - ]), - [data?.me?.permissions, columnVisibility], + () => + filterNullOrUndefined([ + columnVisibility.favoriteQueues + ? { + Header: '', + accessor: 'favoriteQueues', + canSort: false, + } + : undefined, + columnVisibility.id + ? { + Header: 'ID', + accessor: 'id', + Filter: (props: ColumnProps) => + DefaultColumnFilter({ + columnProps: props, + accessor: 'id', + placeholder: 'Queue ID', + }), + filter: 'text', + sortType: stringSort, + } + : undefined, + columnVisibility.name + ? { + Header: 'Name', + accessor: 'name', + Filter: (props: ColumnProps) => + DefaultColumnFilter({ + columnProps: props, + accessor: 'name', + placeholder: 'My Queue', + }), + filter: 'text', + sortType: stringSort, + } + : undefined, + columnVisibility.description + ? { + Header: 'Description', + accessor: 'description', + Filter: (props: ColumnProps) => + DefaultColumnFilter({ + columnProps: props, + accessor: 'description', + }), + filter: 'text', + sortType: stringSort, + } + : undefined, + columnVisibility.oldestTaskAge + ? { + Header: 'Oldest Task Age', + accessor: 'oldestTaskAge', + sortType: dateSort('oldestJobCreatedAt'), + } + : undefined, + columnVisibility.pendingJobCount + ? { + Header: 'Pending Jobs', + accessor: 'pendingJobCount', + sortType: integerSort, + } + : undefined, + columnVisibility.startReviewing + ? { + Header: '', + accessor: 'startReviewing', + canSort: false, + } + : undefined, + columnVisibility.mutations + ? { + Header: '', + accessor: 'mutations', + canSort: false, + } + : undefined, + userHasPermissions(data?.me?.permissions, [ + GQLUserPermission.ManageOrg, + ]) && columnVisibility.deleteJobs + ? { + Header: '', + accessor: 'deleteJobs', + canSort: false, + } + : undefined, + previewJobsViewEnabled && + userHasPermissions(data?.me?.permissions, [ + GQLUserPermission.EditMrtQueues, + ]) && + columnVisibility.previewJobs + ? { + Header: '', + accessor: 'previewJobs', + canSort: false, + } + : undefined, + ]), + [data?.me?.permissions, columnVisibility, previewJobsViewEnabled], ); const dataValues = useMemo( () => @@ -496,7 +578,14 @@ export default function ManualReviewQueuesDashboard() { ? filterNullOrUndefined(queues) .filter((it) => it.isAppealsQueue === (selectedTab === 'APPEALS')) .map( - ({ id, name, description, pendingJobCount, isDefaultQueue, oldestJobCreatedAt }) => { + ({ + id, + name, + description, + pendingJobCount, + isDefaultQueue, + oldestJobCreatedAt, + }) => { const rulesForQueue = routingRules?.filter((it) => it.destinationQueue.id === id) ?? []; @@ -566,7 +655,7 @@ export default function ManualReviewQueuesDashboard() { /> ), ...(userHasPermissions(data?.me?.permissions, [ - GQLUserPermission.EditMrtQueues, + GQLUserPermission.ManageOrg, ]) ? { deleteJobs: ( @@ -656,7 +745,7 @@ export default function ManualReviewQueuesDashboard() { /> ), - id:
{values.id}
, + id: , name: (
{values.name} @@ -689,7 +778,8 @@ export default function ManualReviewQueuesDashboard() { return ; } - const visibleColumnsCount = Object.values(columnVisibility).filter(Boolean).length; + const visibleColumnsCount = + Object.values(columnVisibility).filter(Boolean).length; // Columns button component const columnsButton = ( @@ -700,7 +790,9 @@ export default function ManualReviewQueuesDashboard() { ? 'bg-white text-gray-600 hover:bg-white hover:text-gray-600' : 'bg-gray-600 text-white border-none hover:bg-gray-500' }`} - icon={} + icon={ + + } onClick={() => setColumnsMenuVisible(!columnsMenuVisible)} > Columns @@ -712,12 +804,22 @@ export default function ManualReviewQueuesDashboard() {
{(Object.keys(columnLabels) as ColumnId[]) .filter((columnId) => { - // Only show deleteJobs and previewJobs if user has permissions - if (columnId === 'deleteJobs' || (columnId === 'previewJobs' && previewJobsViewEnabled)) { + // deleteJobs is admin-only (irreversible); previewJobs uses + // the regular queue-edit permission and requires the feature + // flag — both must be true to show the toggle. + if (columnId === 'deleteJobs') { return userHasPermissions(data?.me?.permissions, [ - GQLUserPermission.EditMrtQueues, + GQLUserPermission.ManageOrg, ]); } + if (columnId === 'previewJobs') { + return ( + previewJobsViewEnabled && + userHasPermissions(data?.me?.permissions, [ + GQLUserPermission.EditMrtQueues, + ]) + ); + } return true; }) .map((columnId) => ( @@ -801,7 +903,12 @@ export default function ManualReviewQueuesDashboard() { {tabs.length > 1 ? tabBar : null} { /* @ts-ignore */ - +
} {deleteModal} {deleteAllJobsModal} diff --git a/docker-compose.yaml b/docker-compose.yaml index 6ca192a..3ec1f7b 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -1,17 +1,30 @@ services: redis: image: redis:8.6.1-trixie + restart: unless-stopped + command: + - redis-server + - --appendonly + - 'yes' + - --maxmemory + - '256mb' + - --maxmemory-policy + - noeviction + ports: + - '127.0.0.1:6379:6379' volumes: - redis_data:/data - ports: - - '6379:6379' + deploy: + resources: + limits: + memory: 384M # Runs all migrations from scratch, after clearing the database(s). migrations: image: node:24.14.1-bullseye-slim command: bash -c 'set -e - npm i && ( [ "$CI" = "true" ] && npm run db:clean -- --env staging || true ) - && for db in api-server-pg scylla clickhouse; do npm run db:create -- --db "$$db" --env staging; npm run db:update -- --db "$$db" --env staging; done' + npm i && ( [ "$CI" = "true" ] && npm run db:clean -- --env staging || true ) + && for db in api-server-pg scylla clickhouse; do npm run db:create -- --db "$$db" --env staging; npm run db:update -- --db "$$db" --env staging; done' working_dir: /src env_file: ./.env.githubci environment: @@ -104,8 +117,8 @@ services: redis: condition: service_started - # Regenerate GraphQL types and fail if the working tree drifts. - # Mirrors the `check_generated_graphql` CI job. + # Regenerate GraphQL types and fail if the working tree drifts. + # Mirrors the `check_generated_graphql` CI job. # node_modules is in a dedicated volume so the install doesn't clobber the host's node_modules. codegen-check: build: @@ -114,10 +127,10 @@ services: target: server_base working_dir: /src command: bash -c 'set -e - && git config --global --add safe.directory /src - && npm ci - && npm run generate - && [[ -z "$(git status --porcelain)" ]]' + && git config --global --add safe.directory /src + && npm ci + && npm run generate + && [[ -z "$(git status --porcelain)" ]]' volumes: - .:/src - codegen_node_modules:/src/node_modules @@ -175,7 +188,8 @@ services: CLICKHOUSE_USER: default CLICKHOUSE_PASSWORD: 'clickhouse' healthcheck: - test: ['CMD-SHELL', 'clickhouse-client --host localhost --query "SELECT 1"'] + test: + ['CMD-SHELL', 'clickhouse-client --host localhost --query "SELECT 1"'] interval: 5s timeout: 3s retries: 5 diff --git a/server/bin/README.md b/server/bin/README.md index 22db77e..345d46a 100644 --- a/server/bin/README.md +++ b/server/bin/README.md @@ -21,6 +21,7 @@ npm run get-invite -- --email "user@example.com" ### Output The script will display: + - Invite details (email, role, org ID, created date) - **Signup URL** - The full URL to complete signup @@ -87,6 +88,7 @@ All parameters are required: ### Output The script will output: + - Organization ID - Organization details (name, email, website) - Admin user ID and details @@ -152,8 +154,87 @@ The script performs the following actions: ### Troubleshooting If the script fails: + - Check that your database connection is configured correctly in `.env` - Verify that the organization name and email don't already exist - Ensure the website URL is valid (must start with `http://` or `https://`) - Check the console output for specific error messages +--- + +## recover-mrt-queue.ts + +Re-enqueues items into a Manual Review Tool queue after Redis loss. + +Pending MRT job payloads only live in Redis (BullMQ). When a queue is +obliterated — intentionally via the "Delete All Jobs" button, or +unintentionally via a Redis cluster reset / data loss — the items disappear +from the moderator's queue. The list of WHICH items were enqueued is +preserved in Postgres (`manual_review_tool.job_creations`); rebuilt +`reportHistory` is sourced from the data warehouse table +`REPORTING_SERVICE.REPORTS`; item bodies are re-fetched via +`ItemInvestigationService.getItemByIdentifier`, which cascades through +Scylla (`item_submission_by_thread`), the org's Partial Items endpoint, +and the data warehouse (6-month lookback) — using whichever source still +has the item. + +### Usage + +Get the `--queueId` from the MRT queues dashboard (the "ID" column has a +copy-to-clipboard button next to each id). The mode is auto-detected from +the queue's role in `ncmec_org_settings`, so the same command works for +both default and NCMEC queues: + +```bash +# Dry run -- prints what would be re-enqueued, makes no changes +npm run recover-mrt-queue -- \ + --orgId "" \ + --queueId "" + +# Actually re-enqueue +npm run recover-mrt-queue -- \ + --orgId "" \ + --queueId "" \ + --apply +``` + +### Parameters + +- `--orgId` (required): Organization id whose queue is being recovered. +- `--queueId` (required): MRT queue id to recover into. Must already exist. +- `--mode default|ncmec` (optional): job kind to re-enqueue. Defaults to + auto-detect: `ncmec` if `--queueId` matches the org's + `ncmec_org_settings.default_ncmec_queue_id`, else `default`. Pass + explicitly only to override (rare; only useful for orgs whose routing + rules send DEFAULT jobs into the NCMEC queue, or vice versa). The + configuration banner prints both the chosen mode and its source. +- `--since ""` (default: 30 days ago): only consider + `job_creations` rows after this timestamp. +- `--limit ` (default `10000`, max `100000`): cap on number of items. +- `--apply` (default off): actually call enqueue. Without this flag, the + script is a dry-run and prints the items it would re-enqueue. +- `--no-report-history` (default off): skip rebuilding `reportHistory` from + the data warehouse (useful if the warehouse is unavailable). + +### Safety + +- Dry-run by default. `--apply` is required to make any changes. +- Items that already have a decision in + `manual_review_tool.manual_review_decisions` are filtered out. +- BullMQ dedupes by `(itemTypeId, itemId)` per queue, so re-running the + script on a partially-recovered queue is safe. +- Per-item enqueue errors are logged and counted but do not abort the run. +- The script validates that `--orgId` and `--queueId` look like opaque ids + before issuing any DB queries. + +### What it does NOT recover + +- The original BullMQ `JobId` values are not preserved — recovered jobs + receive new ids derived from the item identifier. +- Item field data is whatever the cascade above can find. If the item has + been hard-deleted from Scylla, the org has no Partial Items endpoint, + and the warehouse 6-month lookback misses, the item is logged as + skipped and counted in the final summary. +- Report history is best-effort: only inbound `submitReport` rows that made + it into `REPORTING_SERVICE.REPORTS` are restored. Rule-driven enqueues + (`ENQUEUE_TO_MRT`) never had a report history to begin with. diff --git a/server/bin/recover-mrt-queue.ts b/server/bin/recover-mrt-queue.ts new file mode 100644 index 0000000..a90a3f5 --- /dev/null +++ b/server/bin/recover-mrt-queue.ts @@ -0,0 +1,444 @@ +#!/usr/bin/env node +/* eslint-disable no-console */ +/** + * Re-enqueues items into a Manual Review Tool queue after Redis loss. + * + * Pending MRT job payloads only live in Redis (BullMQ); when a queue is + * obliterated the items disappear from the moderator's queue. We rebuild + * each job from: + * - `manual_review_tool.job_creations` (Postgres) -- the list of items + * ever enqueued and their policy ids + * - `REPORTING_SERVICE.REPORTS` (warehouse) -- the rebuilt reportHistory + * - `ItemInvestigationService.getItemByIdentifier` -- the item body, + * which cascades Scylla -> Partial Items endpoint -> warehouse + * + * Dry-run by default. Pass `--apply` to actually enqueue. Re-enqueueing is + * idempotent: BullMQ dedups by (itemTypeId, itemId) per queue, and items + * with an existing row in `manual_review_decisions` are filtered out. + * + * Mode (`default` vs `ncmec`) is auto-detected from whether --queueId + * matches `ncmec_org_settings.default_ncmec_queue_id`. Pass --mode only to + * override (rare). + * + * Usage: + * npm run recover-mrt-queue -- \ + * --orgId "" \ + * --queueId "" \ + * [--mode default|ncmec] \ + * [--since "2026-04-15T00:00:00Z"] \ + * [--limit 5000] \ + * [--apply] \ + * [--no-report-history] + */ +import yargs from 'yargs'; +import { hideBin } from 'yargs/helpers'; + +import getBottle from '../iocContainer/index.js'; +import { + jobIdToGuid, + type JobId, + type ReportHistory, +} from '../services/manualReviewToolService/index.js'; +import { jsonStringify } from '../utils/encoding.js'; +import { + isRecoveryMode, + itemKey, + loadReportHistories, + RECOVERY_MODES, + runEnqueueWorkers, + type Candidate, + type RecoveryMode, +} from './recoverMrtQueueLib.js'; + +const DEFAULT_LOOKBACK_DAYS = 30; +const DEFAULT_LIMIT = 10000; +const ENQUEUE_CONCURRENCY = 10; + +const argv = await yargs(hideBin(process.argv)) + .strict() + .options({ + orgId: { + type: 'string', + demandOption: true, + description: 'Organization id whose queue is being recovered', + }, + queueId: { + type: 'string', + demandOption: true, + description: 'MRT queue id to recover into (must exist)', + }, + mode: { + type: 'string', + choices: RECOVERY_MODES, + description: + 'Job kind to re-enqueue. Defaults to auto-detect: "ncmec" if --queueId matches the org\'s default_ncmec_queue_id, otherwise "default". Pass explicitly to override (rare).', + }, + since: { + type: 'string', + description: `ISO timestamp; only consider job_creations rows after this. Default: ${DEFAULT_LOOKBACK_DAYS} days ago.`, + }, + limit: { + type: 'number', + default: DEFAULT_LIMIT, + description: 'Maximum number of items to attempt to recover', + }, + apply: { + type: 'boolean', + default: false, + description: + 'Actually call enqueue. Without this flag the script is a dry-run.', + }, + 'no-report-history': { + type: 'boolean', + default: false, + description: + 'Skip rebuilding reportHistory from the data warehouse (useful if the warehouse is unavailable or the job kind is not report-driven).', + }, + }) + .help() + .parse(); + +// Fail fast on obviously bad ids before issuing DB queries. SQL safety +// itself comes from Kysely parameterization, not this check. +const ID_PATTERN = /^[A-Za-z0-9_-]{1,64}$/; +function assertIdShape(name: string, value: string) { + if (!ID_PATTERN.test(value)) { + throw new Error( + `Invalid ${name}: ${jsonStringify(value)}. Expected 1-64 chars of [A-Za-z0-9_-].`, + ); + } +} + +assertIdShape('orgId', argv.orgId); +assertIdShape('queueId', argv.queueId); + +if (!Number.isInteger(argv.limit) || argv.limit <= 0 || argv.limit > 100_000) { + throw new Error( + `--limit must be a positive integer <= 100000, got ${argv.limit}`, + ); +} + +const since = (() => { + if (argv.since == null) { + return new Date(Date.now() - DEFAULT_LOOKBACK_DAYS * 24 * 60 * 60 * 1000); + } + const parsed = new Date(argv.since); + if (Number.isNaN(parsed.getTime())) { + throw new Error(`--since must be a valid ISO timestamp, got ${argv.since}`); + } + return parsed; +})(); + +function banner(label: string) { + console.log('\n' + '═'.repeat(60)); + console.log(label); + console.log('═'.repeat(60)); +} + +type ResolvedMode = { + mode: RecoveryMode; + detectedMode: RecoveryMode; + defaultNcmecQueueId: string | null; + modeSource: string; +}; + +async function resolveMode( + container: Awaited>['container'], +): Promise { + const ncmecSettings = await container.NcmecService.getNcmecOrgSettings( + argv.orgId, + ); + const defaultNcmecQueueId = ncmecSettings?.defaultNcmecQueueId ?? null; + const detectedMode: RecoveryMode = + defaultNcmecQueueId != null && defaultNcmecQueueId === argv.queueId + ? 'ncmec' + : 'default'; + const explicitMode = isRecoveryMode(argv.mode) ? argv.mode : undefined; + const mode: RecoveryMode = explicitMode ?? detectedMode; + const modeSource = + explicitMode != null + ? `operator override (auto-detect would have been "${detectedMode}")` + : 'auto-detected'; + return { mode, detectedMode, defaultNcmecQueueId, modeSource }; +} + +async function loadCandidates( + container: Awaited>['container'], +): Promise { + banner('Loading candidates from manual_review_tool.job_creations'); + + // Page newest-first, dedupe by (item_type_id, item_id), and check decisions + // per page so we stop on undecided count not distinct count, which could + // be all-decided. + const PAGE_SIZE = 1000; + // Chunked to stay under Postgres's 65,535 bind-parameter cap. + const DECISION_CHUNK = 500; + const byItem = new Map(); + const checkedGuids = new Set(); + const decidedGuids = new Set(); + let totalRowsLoaded = 0; + let cursorCreatedAt: Date | null = null; + let cursorId: string | null = null; + + const countUndecided = () => { + let n = 0; + for (const c of byItem.values()) { + if (!decidedGuids.has(jobIdToGuid(c.latestJobId))) n++; + } + return n; + }; + + while (true) { + let q = container.KyselyPg.selectFrom('manual_review_tool.job_creations') + .select(['id', 'item_id', 'item_type_id', 'created_at', 'policy_ids']) + .where('org_id', '=', argv.orgId) + .where('queue_id', '=', argv.queueId) + .where('created_at', '>=', since); + + if (cursorCreatedAt != null && cursorId != null) { + const cAt = cursorCreatedAt; + const cId = cursorId; + q = q.where((eb) => + eb.or([ + eb('created_at', '<', cAt), + eb.and([eb('created_at', '=', cAt), eb('id', '<', cId)]), + ]), + ); + } + + const page = await q + .orderBy('created_at', 'desc') + .orderBy('id', 'desc') + .limit(PAGE_SIZE) + .execute(); + + if (page.length === 0) break; + totalRowsLoaded += page.length; + + for (const row of page) { + const key = `${row.item_type_id}\x00${row.item_id}`; + const existing = byItem.get(key); + if ( + existing == null || + new Date(row.created_at).getTime() > existing.latestCreatedAt.getTime() + ) { + byItem.set(key, { + itemId: row.item_id, + itemTypeId: row.item_type_id, + latestJobId: row.id as JobId, + latestCreatedAt: new Date(row.created_at), + policyIds: row.policy_ids ?? [], + }); + } + } + + const last = page[page.length - 1]; + cursorCreatedAt = new Date(last.created_at); + cursorId = last.id; + + // Resolve decisions for any guids we haven't checked yet, chunked to + // stay under Postgres's 65,535 bind-parameter ceiling. + const newGuids: string[] = []; + for (const c of byItem.values()) { + const guid = jobIdToGuid(c.latestJobId); + if (!checkedGuids.has(guid)) { + checkedGuids.add(guid); + newGuids.push(guid); + } + } + for (let i = 0; i < newGuids.length; i += DECISION_CHUNK) { + const chunk = newGuids.slice(i, i + DECISION_CHUNK); + const decisionRows = await container.KyselyPg.selectFrom( + 'manual_review_tool.manual_review_decisions', + ) + .select(['id']) + .where('org_id', '=', argv.orgId) + .where('id', 'in', chunk) + .execute(); + for (const r of decisionRows) decidedGuids.add(r.id); + } + + if (countUndecided() >= argv.limit) break; + if (page.length < PAGE_SIZE) break; + } + + console.log( + `Loaded ${totalRowsLoaded} job_creations rows across paginated reads`, + ); + console.log(`Deduplicated to ${byItem.size} distinct items`); + + if (byItem.size === 0) return []; + + const candidates = Array.from(byItem.values()) + .filter((c) => !decidedGuids.has(jobIdToGuid(c.latestJobId))) + .slice(0, argv.limit); + console.log( + `Filtered out ${decidedGuids.size} items that already have a decision (latest enqueue).`, + ); + console.log(`Candidates to recover: ${candidates.length}`); + return candidates; +} + +async function maybeRebuildHistories( + container: Awaited>['container'], + candidates: readonly Candidate[], + mode: RecoveryMode, +): Promise> { + const out = new Map(); + if (argv['no-report-history'] || mode !== 'default') return out; + + banner('Rebuilding reportHistory from REPORTING_SERVICE.REPORTS'); + try { + await loadReportHistories(container, argv.orgId, candidates, out); + const withHistory = Array.from(out.values()).filter( + (h) => h.length > 0, + ).length; + console.log( + `Rebuilt history for ${withHistory}/${candidates.length} items`, + ); + } catch (e: unknown) { + console.warn( + '\nWARNING: failed to rebuild reportHistory from data warehouse, continuing without it:', + e instanceof Error ? e.message : e, + ); + } + return out; +} + +function printSample( + candidates: readonly Candidate[], + reportHistoryByItem: Map, +) { + banner('Sample of candidates (first 20)'); + for (const c of candidates.slice(0, 20)) { + const histCount = reportHistoryByItem.get(itemKey(c))?.length ?? 0; + console.log( + ` type=${c.itemTypeId} item=${c.itemId} enqueuedAt=${c.latestCreatedAt.toISOString()} policies=${c.policyIds.length} history=${histCount}`, + ); + } + if (candidates.length > 20) { + console.log(` ... and ${candidates.length - 20} more`); + } +} + +async function main() { + const bottle = await getBottle(); + const container = bottle.container; + + try { + // Bypassing permissioning is safe here: back-office script run by an + // operator on their own infra. Org ownership is still validated. + const queue = + await container.ManualReviewToolService.getQueueForOrgAndDangerouslyBypassPermissioning( + { orgId: argv.orgId, queueId: argv.queueId }, + ); + if (queue == null) { + console.error( + `\nQueue ${argv.queueId} not found for org ${argv.orgId}. Aborting.`, + ); + process.exit(2); + } + + const { mode, defaultNcmecQueueId, modeSource } = + await resolveMode(container); + + banner('Recovery configuration'); + console.log(`Org id: ${argv.orgId}`); + console.log(`Queue id: ${argv.queueId}`); + console.log(`Queue name: ${queue.name}`); + console.log(`Mode: ${mode} (${modeSource})`); + console.log(`Since: ${since.toISOString()}`); + console.log(`Limit: ${argv.limit}`); + console.log( + `Mutating? ${argv.apply ? 'YES (--apply)' : 'no (dry run)'}`, + ); + console.log(`Rebuild history? ${argv['no-report-history'] ? 'no' : 'yes'}`); + + const pendingNow = + await container.ManualReviewToolService.getPendingJobCount({ + orgId: argv.orgId, + queueId: argv.queueId, + }); + console.log(`Pending now: ${pendingNow}`); + if (pendingNow > 0) { + console.warn( + '\nWARNING: queue is not empty. Recovery is still safe (BullMQ dedupes per item) but you may end up re-fetching item data unnecessarily. Consider stopping here unless you intentionally want to top-up.', + ); + } + + // NcmecService always routes to the org's `default_ncmec_queue_id`. If + // --mode ncmec was forced onto a different queue, items would silently + // land in the wrong place, so refuse up front. + if (mode === 'ncmec') { + if (defaultNcmecQueueId == null) { + console.error( + `\nOrg ${argv.orgId} has no default_ncmec_queue_id configured. NCMEC recovery cannot route items. Aborting.`, + ); + process.exit(2); + } + if (defaultNcmecQueueId !== argv.queueId) { + console.error( + `\n--queueId (${argv.queueId}) does not match this org's default NCMEC queue (${defaultNcmecQueueId}). NcmecService always routes to the configured default queue, so recovery would land items in the wrong place. Re-run with --queueId ${defaultNcmecQueueId}, or update ncmec_org_settings first.`, + ); + process.exit(2); + } + } + + const candidates = await loadCandidates(container); + if (candidates.length === 0) { + console.log('Nothing to recover. Exiting.'); + await container.closeSharedResourcesForShutdown(); + process.exit(0); + } + + const reportHistoryByItem = await maybeRebuildHistories( + container, + candidates, + mode, + ); + printSample(candidates, reportHistoryByItem); + + if (!argv.apply) { + banner('Dry run -- no changes made'); + console.log( + `Re-run with --apply to actually re-enqueue ${candidates.length} items.`, + ); + await container.closeSharedResourcesForShutdown(); + process.exit(0); + } + + // Concurrency-bounded so we don't hammer item-data sources or BullMQ. + banner('Re-enqueueing items'); + const stats = await runEnqueueWorkers( + container, + argv.orgId, + argv.queueId, + candidates, + mode, + reportHistoryByItem, + ENQUEUE_CONCURRENCY, + ); + + banner('Done'); + console.log(`Enqueued: ${stats.enqueued}`); + console.log(`Skipped: ${stats.skipped}`); + console.log(`Failed: ${stats.failed}`); + + await container.closeSharedResourcesForShutdown(); + process.exit(stats.failed > 0 ? 1 : 0); + } catch (error: unknown) { + console.error('\nError running recovery script:\n'); + console.error(error); + try { + await container.closeSharedResourcesForShutdown(); + } catch (shutdownError) { + console.error('Error during shutdown:', shutdownError); + } + process.exit(1); + } +} + +main().catch((error: unknown) => { + console.error('Unhandled error:', error); + process.exit(1); +}); diff --git a/server/bin/recoverMrtQueueLib.ts b/server/bin/recoverMrtQueueLib.ts new file mode 100644 index 0000000..ec0ca05 --- /dev/null +++ b/server/bin/recoverMrtQueueLib.ts @@ -0,0 +1,290 @@ +/* eslint-disable no-console */ +/** + * Helpers for `bin/recover-mrt-queue.ts`. Split out so the entry-point file + * stays under the 500-line per-file lint cap; not intended for general reuse. + */ +import { type ItemIdentifier } from '@roostorg/types'; +import { v1 as uuidv1 } from 'uuid'; + +import type getBottle from '../iocContainer/index.js'; +import { itemSubmissionToItemSubmissionWithTypeIdentifier } from '../services/itemProcessingService/index.js'; +import { + type JobId, + type ManualReviewJobInput, + type ReportHistory, +} from '../services/manualReviewToolService/index.js'; +import { toCorrelationId } from '../utils/correlationIds.js'; + +export type Container = Awaited>['container']; + +export const RECOVERY_MODES = ['default', 'ncmec'] as const; +export type RecoveryMode = (typeof RECOVERY_MODES)[number]; + +export function isRecoveryMode(value: unknown): value is RecoveryMode { + return ( + typeof value === 'string' && + (RECOVERY_MODES as readonly string[]).includes(value) + ); +} + +export type Candidate = { + itemId: string; + itemTypeId: string; + /** Latest job_creations.id (an external JobId) for this item in this queue. */ + latestJobId: JobId; + /** Latest job_creations.created_at for this item in this queue. */ + latestCreatedAt: Date; + policyIds: readonly string[]; +}; + +export function itemKey(c: Pick): string { + return `${c.itemTypeId}\x00${c.itemId}`; +} + +export type EnqueueResult = 'enqueued' | 'skipped'; +export type EnqueueStats = { + enqueued: number; + skipped: number; + failed: number; +}; + +/** + * Shape of a row from `REPORTING_SERVICE.REPORTS`. All fields are nullable + * because warehouse rows aren't guaranteed to be fully populated. + */ +type ReportsWarehouseRow = { + org_id: string | null; + request_id: string | null; + reporter_user_id: string | null; + reporter_user_item_type_id: string | null; + reporter_kind: string | null; + reported_at: string | null; + policy_id: string | null; + reported_for_reason: string | null; + reported_item_id: string | null; + reported_item_type_id: string | null; +}; + +/** + * Bulk-fetch report history rows from the data warehouse for every candidate. + * Mutates `out` in place. Chunks queries to stay under CH's parameter limit. + */ +export async function loadReportHistories( + container: Container, + orgId: string, + candidates: readonly Candidate[], + out: Map, +): Promise { + const CHUNK = 500; + for (let i = 0; i < candidates.length; i += CHUNK) { + const chunk = candidates.slice(i, i + CHUNK); + const itemIds = chunk.map((c) => c.itemId); + const itemTypeIds = chunk.map((c) => c.itemTypeId); + // Positional `?` placeholders. The IN-list match here is the cross-product + // of the two columns, so we filter the (item_id, item_type_id) tuple in + // JS (CH lacks a portable IN-tuple across providers). + const rows = (await container.DataWarehouse.query( + ` + SELECT + org_id, + request_id, + reporter_user_id, + reporter_user_item_type_id, + reporter_kind, + reported_at, + policy_id, + reported_for_reason, + reported_item_id, + reported_item_type_id + FROM REPORTING_SERVICE.REPORTS + WHERE org_id = ? + AND reported_item_id IN (${itemIds.map(() => '?').join(', ')}) + AND reported_item_type_id IN (${itemTypeIds.map(() => '?').join(', ')}) + ORDER BY reported_at ASC + `, + container.Tracer, + [orgId, ...itemIds, ...itemTypeIds], + )) as readonly ReportsWarehouseRow[]; + + const validKeys = new Set(chunk.map((c) => itemKey(c))); + for (const row of rows) { + const itemId = row.reported_item_id ?? ''; + const itemTypeId = row.reported_item_type_id ?? ''; + const key = `${itemTypeId}\x00${itemId}`; + if (!validKeys.has(key)) continue; + + // The reportId surfaced via the API is the segment after `submit-report:` + // in the warehouse's correlation-style request_id; fall back to the raw + // value if the format ever changes so we don't lose the row entirely. + const requestId = row.request_id ?? ''; + const reportId = requestId.includes(':') + ? requestId.slice(requestId.indexOf(':') + 1) + : requestId; + + const reportedAt = + row.reported_at != null ? new Date(row.reported_at) : new Date(0); + if (Number.isNaN(reportedAt.getTime())) continue; + + const reporterId: ItemIdentifier | undefined = + row.reporter_kind === 'user' && + row.reporter_user_id != null && + row.reporter_user_item_type_id != null + ? { + id: row.reporter_user_id, + typeId: row.reporter_user_item_type_id, + } + : undefined; + + const entry = { + reporterId, + reason: row.reported_for_reason ?? undefined, + reportId, + reportedAt, + policyId: row.policy_id ?? undefined, + }; + + const existing = out.get(key); + if (existing) { + existing.push(entry); + } else { + out.set(key, [entry]); + } + } + } +} + +/** + * Re-enqueue one candidate. DEFAULT goes through `ManualReviewToolService.enqueue`; + * NCMEC goes through `NcmecService.enqueueForHumanReviewIfApplicable`, which + * always routes to the org's configured default NCMEC queue. + */ +export async function enqueueOne( + container: Container, + orgId: string, + queueId: string, + candidate: Candidate, + mode: RecoveryMode, + reportHistoryByItem: Map, +): Promise { + const itemIdentifier: ItemIdentifier = { + id: candidate.itemId, + typeId: candidate.itemTypeId, + }; + + // `job_creations` only stores ids, so we re-fetch the body via + // ItemInvestigationService, which cascades Scylla -> Partial Items endpoint + // -> warehouse content-API requests. If all three miss the item is gone + // and we skip. + const itemResult = + await container.ItemInvestigationService.getItemByIdentifier({ + orgId, + itemIdentifier, + latestSubmissionOnly: true, + }).catch((e: unknown) => { + console.warn( + `[${candidate.itemTypeId}/${candidate.itemId}] item lookup failed:`, + e instanceof Error ? e.message : e, + ); + return null; + }); + if (itemResult?.latestSubmission == null) { + console.warn( + `[${candidate.itemTypeId}/${candidate.itemId}] no item data found in Scylla, partial-items endpoint, or warehouse -- skipping`, + ); + return 'skipped'; + } + + // The adapter returns a full `ItemSubmission`; enqueue expects the + // slimmer `ItemSubmissionWithTypeIdentifier`. + const itemSubmission = itemSubmissionToItemSubmissionWithTypeIdentifier( + itemResult.latestSubmission, + ); + + // Synthetic correlation id makes this run greppable in logs/traces. + const correlationId = toCorrelationId<'manual-action-run'>({ + type: 'manual-action-run', + id: `recover-mrt-queue:${uuidv1()}`, + }); + + if (mode === 'ncmec') { + const result = + await container.NcmecService.enqueueForHumanReviewIfApplicable({ + orgId, + createdAt: candidate.latestCreatedAt, + item: itemSubmission, + correlationId, + enqueueSource: 'MRT_JOB', + enqueueSourceInfo: { kind: 'MRT_JOB' }, + reenqueuedFrom: { jobId: candidate.latestJobId }, + }); + return result.status === 'ENQUEUED' ? 'enqueued' : 'skipped'; + } + + // `reportedForReasons` is mirrored from the rebuilt history so the reviewer + // UI shows reporter+reason without restoring the legacy field. + const history = reportHistoryByItem.get(itemKey(candidate)) ?? []; + const input: ManualReviewJobInput = { + orgId, + correlationId, + createdAt: candidate.latestCreatedAt, + enqueueSource: 'MRT_JOB', + enqueueSourceInfo: { kind: 'MRT_JOB' }, + reenqueuedFrom: { jobId: candidate.latestJobId }, + payload: { + kind: 'DEFAULT', + item: itemSubmission, + reportHistory: history, + reportedForReasons: history.map((h) => ({ + reporterId: h.reporterId, + reason: h.reason, + })), + }, + policyIds: [...candidate.policyIds], + }; + + await container.ManualReviewToolService.enqueue(input, queueId); + return 'enqueued'; +} + +/** + * Drain `candidates` through a fixed-size pool of workers. Per-item errors + * are caught so a single bad item doesn't abort the whole run. + */ +export async function runEnqueueWorkers( + container: Container, + orgId: string, + queueId: string, + candidates: readonly Candidate[], + mode: RecoveryMode, + reportHistoryByItem: Map, + concurrency: number, +): Promise { + const stats: EnqueueStats = { enqueued: 0, skipped: 0, failed: 0 }; + let cursor = 0; + + const worker = async (): Promise => { + while (cursor < candidates.length) { + const candidate = candidates[cursor++]; + try { + const result = await enqueueOne( + container, + orgId, + queueId, + candidate, + mode, + reportHistoryByItem, + ); + stats[result]++; + } catch (e: unknown) { + stats.failed++; + console.warn( + `[${candidate.itemTypeId}/${candidate.itemId}] enqueue failed:`, + e instanceof Error ? e.message : e, + ); + } + } + }; + + await Promise.all(Array.from({ length: concurrency }, worker)); + return stats; +} diff --git a/server/graphql/modules/manualReviewTool.ts b/server/graphql/modules/manualReviewTool.ts index 41fada9..9660da2 100644 --- a/server/graphql/modules/manualReviewTool.ts +++ b/server/graphql/modules/manualReviewTool.ts @@ -2402,6 +2402,11 @@ const Mutation: GQLMutationResolvers = { if (user == null) { throw unauthenticatedError('Authenticated user required'); } + // Admin-only: irreversible (pending payloads only live in Redis). + // Recovery is via `server/bin/recover-mrt-queue.ts`. + if (!user.getPermissions().includes(UserPermission.MANAGE_ORG)) { + throw forbiddenError('Only org admins can delete all jobs from a queue'); + } try { await context.services.ManualReviewToolService.deleteAllJobsFromQueue({ orgId: user.orgId, diff --git a/server/package.json b/server/package.json index 3a20651..c593b9d 100644 --- a/server/package.json +++ b/server/package.json @@ -18,7 +18,8 @@ "lint": "eslint \"./**/*.{ts,tsx,js}\"", "runWorkerOrJob": "node --loader ts-node/esm --require dotenv/config bin/run-worker-or-job.ts", "create-org": "node --loader ts-node/esm --require dotenv/config bin/create-org-and-user.ts", - "get-invite": "node --loader ts-node/esm --require dotenv/config bin/get-invite-token.ts" + "get-invite": "node --loader ts-node/esm --require dotenv/config bin/get-invite-token.ts", + "recover-mrt-queue": "node --loader ts-node/esm --require dotenv/config bin/recover-mrt-queue.ts" }, "author": "Roostorg", "license": "ISC", diff --git a/server/services/manualReviewToolService/index.ts b/server/services/manualReviewToolService/index.ts index 741e6c7..7afc685 100644 --- a/server/services/manualReviewToolService/index.ts +++ b/server/services/manualReviewToolService/index.ts @@ -7,6 +7,7 @@ import { export { ManualReviewToolService, + type JobId, type ManualReviewJob, type ManualReviewJobOrAppeal, type ManualReviewAppealJob, @@ -20,9 +21,13 @@ export { type ContentManualReviewJobPayload, type UserManualReviewJobPayload, type NcmecManualReviewJobPayload, + type ReportHistory, } from './manualReviewToolService.js'; -export { type ManualReviewQueue } from './modules/QueueOperations.js'; +export { + type ManualReviewQueue, + jobIdToGuid, +} from './modules/QueueOperations.js'; export { type RoutingRule } from './modules/JobRouting.js'; diff --git a/server/services/manualReviewToolService/modules/QueueOperations.test.ts b/server/services/manualReviewToolService/modules/QueueOperations.test.ts index c7590db..f3d2713 100644 --- a/server/services/manualReviewToolService/modules/QueueOperations.test.ts +++ b/server/services/manualReviewToolService/modules/QueueOperations.test.ts @@ -8,6 +8,7 @@ import createMrtQueue from '../../../test/fixtureHelpers/createMrtQueue.js'; import createOrg from '../../../test/fixtureHelpers/createOrg.js'; import createUser from '../../../test/fixtureHelpers/createUser.js'; import { makeTestWithFixture } from '../../../test/utils.js'; +import { UserPermission } from '../../userManagementService/index.js'; import { bullJobIdtoExternalJobId, itemIdToBullJobId, @@ -201,4 +202,33 @@ describe('QueueOperations', () => { ).toEqual(false); }, ); + + // Regression: `deleteAllJobsFromQueue` is irreversible and used to accept + // EDIT_MRT_QUEUES (held by moderator managers) -- that gap accidentally + // cleared a production queue. It now requires MANAGE_ORG. + testWithQueueAndActions()( + 'deleteAllJobsFromQueue rejects EDIT_MRT_QUEUES without MANAGE_ORG', + async ({ org, queue, mrtService }) => { + await expect( + mrtService.deleteAllJobsFromQueue({ + orgId: org.id, + queueId: queue.id, + userPermissions: [UserPermission.EDIT_MRT_QUEUES], + }), + ).rejects.toMatchObject({ name: 'DeleteAllJobsUnauthorizedError' }); + }, + ); + + testWithQueueAndActions()( + 'deleteAllJobsFromQueue accepts MANAGE_ORG', + async ({ org, queue, mrtService }) => { + await expect( + mrtService.deleteAllJobsFromQueue({ + orgId: org.id, + queueId: queue.id, + userPermissions: [UserPermission.MANAGE_ORG], + }), + ).resolves.toBeUndefined(); + }, + ); }); diff --git a/server/services/manualReviewToolService/modules/QueueOperations.ts b/server/services/manualReviewToolService/modules/QueueOperations.ts index c42f98a..8be9249 100644 --- a/server/services/manualReviewToolService/modules/QueueOperations.ts +++ b/server/services/manualReviewToolService/modules/QueueOperations.ts @@ -837,7 +837,9 @@ export default class QueueOperations { userPermissions: readonly UserPermission[]; }) { const { orgId, queueId, userPermissions } = opts; - if (!userPermissions.includes(UserPermission.EDIT_MRT_QUEUES)) { + // Admin-only (MANAGE_ORG). `obliterate` wipes pending payloads from + // Redis irreversibly; recovery is via `server/bin/recover-mrt-queue.ts`. + if (!userPermissions.includes(UserPermission.MANAGE_ORG)) { throw makeDeleteAllJobsInsufficientPermissionsError({ shouldErrorSpan: true, });