Merge pull request #4906 from civitai/feat/image-scanning-ingestion

feat(ingestion): scan images with imageScanning behind a Flipt flag
This commit is contained in:
Briant Diehl
2026-09-17 11:06:44 -06:00
committed by GitHub
44 changed files with 2650 additions and 3359 deletions
-2
View File
@@ -212,8 +212,6 @@ SHOPIFY_ADMIN_TOKEN=
FLIPT_URL=""
FLIPT_FETCHER_SECRET=placeholder
IMAGE_SCANNER_NEW=false
# App Blocks — per-app generation spend/velocity ABSOLUTE CEILINGS (incident knobs).
# 🔴 These are UPPER BOUNDS, not the limit an app receives. Each app's actual
# ceilings come from its server-owned `spendTier` (+ any moderator per-app
+6 -15
View File
@@ -438,22 +438,12 @@ Both use `GREATEST` semantics (never lower, only raise) and lock `userNsfwLevel`
#### 4. Webhook Integration
**File**: `src/pages/api/webhooks/image-scan-result.ts` (image scanning)
**File**: `src/pages/api/webhooks/image-scan-result.ts` reads the flag; the scan processors (`image-scan-result.service.ts`, `image-scanning-result.service.ts`) do the fan-out
```typescript
// Feature flag gated
const featureFlags = getFeatureFlagsLazy({ req });
if (featureFlags.articleImageScanning) {
// Find articles using this image
const articleConnections = await dbWrite.imageConnection.findMany({
where: { imageId: image.id, entityType: 'Article' },
});
// Debounced updates
for (const { entityId } of articleConnections) {
await debounceArticleUpdate(entityId);
}
}
articleImageScanning: getFeatureFlagsLazy({ req }).articleImageScanning, // webhook → processor input
// in the processor, once the verdict is written:
if (articleImageScanning) await fanOutArticleImageUpdates(imageId); // content + cover articles → debounceArticleUpdate
```
**File**: `src/pages/api/webhooks/text-moderation-result.ts` (text moderation)
@@ -993,7 +983,8 @@ console.log('Article Image Scanning:', flags.articleImageScanning);
- `src/server/services/article.service.ts` - Main article operations, image linking, scan status
- `src/server/services/article-content-cleanup.service.ts` - Image/media extraction from content (server-side)
- `src/server/services/nsfwLevels.service.ts` - NSFW level calculation (cover + content images)
- `src/pages/api/webhooks/image-scan-result.ts` - Image scan webhook handler
- `src/pages/api/webhooks/image-scan-result.ts` - Image scan webhook handler (reads the flag, routes by workflow step)
- `src/server/services/image-scan-result.service.ts`, `image-scanning-result.service.ts`, `image-scan-pipeline.ts` - scan processing + article fan-out
- `src/server/utils/webhook-debounce.ts` - Redis-based debouncing logic
- `src/utils/article-helpers.ts` - Image extraction (client-side), shared helpers
+1 -1
View File
@@ -391,7 +391,7 @@ articleImageScanning: [], // was ['public']
```
**Effect**:
- New articles still save, still submit to text moderation, but `debounceArticleUpdate` in `image-scan-result.ts` short-circuits
- New articles still save, still submit to text moderation, but the scan processors skip `fanOutArticleImageUpdates`, so no article update is queued
- Articles already in `Processing` will **not** auto-transition to `Published` — they need manual intervention or the flag re-enabled
- Existing NSFW levels stay as-is
+3 -3
View File
@@ -187,11 +187,11 @@ one) and **Net-new** (moderation the moderator app doesn't cover — new domains
### 13. knights "down-leveled" review (`services/image.service.ts`)
- [ ] `getDownleveledImages` (ClickHouse `knights_new_order_downleveled` + PG join) + `addToNewOrderQueue`
- [ ] `getDownleveledImages` (ClickHouse `knights_new_order_downleveled` + PG join) (`image.service.ts`) + `addToNewOrderQueue` (`image-scan-pipeline.ts`)
### 14. scan-result ingestion pipeline internals (`services/image-scan-result.service.ts`)
### 14. scan-result ingestion pipeline internals (`services/image-scan-pipeline.ts`, `services/image-scan-result.service.ts`)
- [ ] `resolveScanOutcome`, `auditScanResults`, `markImageScanError`, `blockImageFromRating`, `getAssociatedEntities`, `evaluateImageModRules`, `isExemptFromAiVerification`, `processTags`
- [ ] `resolveScanOutcome`, `auditScanResults`, `markImageScanError`, `getAssociatedEntities`, `evaluateImageModRules`, `processTags` (pipeline); `blockImageFromRating`, `isExemptFromAiVerification` (image-scan-result.service)
---
+2 -2
View File
@@ -43,8 +43,8 @@ ORDER BY (scanner, label, contentHash, version);
Column semantics:
- **`version`** (per-label) — policyHash from `result.policyHash` for XGuard scans; hardcoded `'1'` for image scans until the orchestrator team surfaces per-result version info on the `mediaRating` step.
- **`modelVersion`** (workflow-level) — scanner/model version stamp, sourced from `workflow.metadata.version`. Hardcoded `'1'` everywhere today.
- **`version`** (per-label) — policyHash from `result.policyHash` for XGuard scans; for image scans a fixed `'1'` (`mediaRating` lane) or `'2'` (`imageScanning` lane) until the orchestrator surfaces per-result version info.
- **`modelVersion`** (workflow-level) — scanner/model version stamp, sourced from `workflow.metadata.version` (default `'1'`); image scans write `'1'` or `'2'` by lane, as above.
Both kept as separate columns so when the orchestrator starts returning per-label version info on `mediaRating`, we can populate `version` independently of `modelVersion`.
+7 -7
View File
@@ -25,7 +25,7 @@ What gets authored in phase 1:
The orchestrator stamps each evaluation with a `policyHash` derived from the exact policy text. That hash becomes the `version` column on `scanner_label_results`, which is what makes A/B comparison in phase 3 work without civitai-side bookkeeping.
For image scanners, "defining a label" means adding a new `include*` flag to the `mediaRating` step input and a corresponding output field (see [image-scan-result.service.ts](src/server/services/image-scan-result.service.ts) and the `TagSource` enum). Versioning is currently a `'1'` placeholder until the orchestrator surfaces per-result version info.
For image scanners, "defining a label" means adding a new `include*` flag to the `mediaRating` step input and a corresponding output field (see [image-scan-result.service.ts](src/server/services/image-scan-result.service.ts) and the `TagSource` enum). Versioning is a fixed stamp until the orchestrator surfaces per-result version info: `'1'` for the `mediaRating` lane, `'2'` for the `imageScanning` lane (different models, so the two must not aggregate).
### Phase 2 — Provide a dataset
@@ -217,12 +217,12 @@ The `topK` distributions, bounding boxes, face geometry, and confidences for non
`wdTagging` and `mediaHash` are unchanged.
**Code changes required in [image-scan-result.service.ts](src/server/services/image-scan-result.service.ts):**
**Code changes required in [image-scan-result.service.ts](src/server/services/image-scan-result.service.ts) (step types) and [image-scan-pipeline.ts](src/server/services/image-scan-pipeline.ts) (tags + audit):**
1. Extend the `MediaRatingStep` TypeScript type (line 74) — add the four new optional output fields.
2. Extend `tagsWithSource` (line 251) to emit `minor`/`ai`/`anime` tags from the new fields when present.
1. Extend the `MediaRatingStep` TypeScript type in `image-scan-result.service.ts` — add the four new optional output fields.
2. Extend `tagsWithSource` (`buildAndInsertScanTags` in `image-scan-pipeline.ts`) to emit `minor`/`ai`/`anime` tags from the new fields when present.
3. Add `AiRecognition` + `AnimeRecognition` to the `TagSource` Prisma enum (or accept lossy provenance and reuse `Computed`).
4. Decide whether `auditScanResults` branches on the new `MinorDetection`-sourced `minor` tag (stronger signal than the hand-curated `tagsNeedingReview` list) or just appends `minor` to that list.
4. Decide whether `auditScanResults` (`image-scan-pipeline.ts`) branches on the new `MinorDetection`-sourced `minor` tag (stronger signal than the hand-curated `tagsNeedingReview` list) or just appends `minor` to that list.
5. Forward the full untruncated `mediaRating.output` to the audit log so the per-detection / topK / non-triggered confidences are preserved for tuning.
---
@@ -254,13 +254,13 @@ The `topK` distributions, bounding boxes, face geometry, and confidences for non
### Image ingestion
- **Webhook**: [src/pages/api/webhooks/image-scan-result.ts](src/pages/api/webhooks/image-scan-result.ts) — handles legacy POST + new orchestrator workflow format
- **Webhook**: [src/pages/api/webhooks/image-scan-result.ts](src/pages/api/webhooks/image-scan-result.ts) — orchestrator workflow/job events only; routes `imageScanning` workflows to [image-scanning-result.service.ts](src/server/services/image-scanning-result.service.ts) and `wdTagging` + `mediaRating` workflows to `processImageScanWorkflow`. Which step is submitted is chosen in `createImageIngestionRequest` by the Flipt flag `image-ingestion-image-scanning`.
- **Processor**: [image-scan-result.service.ts](src/server/services/image-scan-result.service.ts) → `processImageScanWorkflow` — parses `wdTagging`/`mediaRating`/`mediaHash` steps. The `mediaRating` type definition is what needs to be extended for the new fields.
- **Storage**:
- `Image` table: `nsfwLevel: Int`, `minor: Boolean`, `poi: Boolean`, `needsReview: String?`, `blockedFor: String?`, `ingestion: ImageIngestionStatus`, `scanJobs: Json?`
- `TagsOnImageDetails`: `automated`, `disabled`, `needsReview`, `confidence`, `source: TagSource`
- `ImageTagForReview`: review queue (per-image, per-tag)
- **Review trigger conditions** (current logic in [image-scan-result.ts](src/pages/api/webhooks/image-scan-result.ts:697-733)): `child-10/13/15` + realistic, POI word-list match, `nsfwLevel === Blocked`, moderator-specific tags.
- **Review trigger conditions** (current logic in `auditScanResults`, [image-scan-pipeline.ts](src/server/services/image-scan-pipeline.ts)): `child-10/13/15` + realistic, POI word-list match, `nsfwLevel === Blocked`, moderator-specific tags.
- **Tag rules**: [src/server/utils/tag-rules.ts](src/server/utils/tag-rules.ts) — replacements, appends, computed combos.
### Cross-cutting
+6 -4
View File
@@ -4,6 +4,8 @@ Investigation of errors logged by the `/api/webhooks/image-scan-result` handler
`processImageScanResult` (`src/server/services/image-scan-result.service.ts`), which became the
default scan-result path on 2026-07-22.
> **2026-09 note:** `processImageScanResult` is gone. The webhook now dispatches to `processImageScanWorkflow` (wdTagging + mediaRating) or `processImageScanningWorkflow` (imageScanning), and the shared stages moved to `image-scan-pipeline.ts`. Errors log as `name: 'image-scan-result'` or `'image-scanning-result'` by lane; submit failures log as `'image-ingestion'` or `'image-scanning-ingestion'`.
## Where failures are logged
| Signal | Dataset | Shape |
@@ -181,9 +183,9 @@ await signalClient
This is already the house pattern — `src/server/services/referral.service.ts` and
`src/server/auth/session-invalidation.ts` both catch-and-log around `signalClient.send`.
Apply the same change to the legacy path at `src/pages/api/webhooks/image-scan-result.ts` (the
unguarded `await signalClient.send` in the `data.ingestion !== 'Blocked'` branch) — that call is
why timeout/signal-500 errors also appear on days before the 07-22 cutover.
The legacy (non-orchestrator) path in `src/pages/api/webhooks/image-scan-result.ts` that also sent
this signal unguarded has since been removed; it is why timeout/signal-500 errors also appear on days
before the 07-22 cutover.
### Tradeoff
@@ -326,5 +328,5 @@ Images rejected as "too large" end up terminally `Error` and unscanned. That is
downscaling before submit, or raising the orchestrator's limit. Worth a follow-up decision.
`markImageScanSubmitFailure` duplicates the jsonb shape written by `markImageScanError` in
`image-scan-result.service.ts`. They differ (submit failures have no `workflowId`), but the two
`image-scan-pipeline.ts`. They differ (submit failures have no `workflowId`), but the two
should probably share a helper before a third caller appears.
+2 -3
View File
@@ -166,8 +166,7 @@ replica-lag miss, so a plain `dbRead` is safe.
branches). When there's no post it does a `dbRead` lookup by `thumbnailImageId`, then calls
`updateModel3DNsfwLevels` directly (a **synchronous** recompute — not a JobQueue enqueue).
- Call sites (each just passes `{ imageId, postId }`, replacing the removed `queueModel3D…`):
- `src/pages/api/webhooks/image-scan-result.ts` — Scanned + Blocked
- `src/server/services/image-scan-result.service.ts` — Scanned + Blocked branches
- `src/server/services/image-scan-pipeline.ts` `applyIngestionSideEffects` — Scanned + Blocked branches (shared by both scan pipelines; the legacy webhook body is gone)
- `src/server/services/image.service.ts` `updateImageNsfwLevel` — moderator branch (also
added `postId` to the `findUnique` select)
@@ -195,7 +194,7 @@ Model3D trigger) and had the image trigger read it (zero lookup). A review found
soundness gap: `Image.metadata` is rewritten **wholesale** in several places, so the flag
could be silently and permanently dropped:
- `src/server/services/image-scan-result.service.ts` (~line 735): `"metadata" =
- `src/server/services/image-scan-pipeline.ts` (`resolveScanOutcome`): `"metadata" =
COALESCE(<modRule.metadata>::jsonb, "metadata")` **replaces** the column when a mod rule
carries a metadata payload — on the scan path, same UPDATE as the nsfwLevel change.
- `src/server/services/image.service.ts` `updateImageNsfwLevel` (~line 6747) and the
+1 -1
View File
@@ -111,7 +111,7 @@ Re-read off the live ticket 2026-08-21; the section above had captured the middl
every gallery image via `UPDATE "Image" SET minor = …, poi = …`, queueing those ids into the image
index. `poi` is on `modelUpsertSchema` beside `minor` and `upsertModel` calls that fan-out, so
flipping poi today already does all of it. Image-level `poi` also excludes from the image search
index and is re-derived from the parent model at scan time (`image-scan-result.service.ts:1089`).
index and is re-derived from the parent model at scan time (`getAssociatedEntities` in `image-scan-pipeline.ts`).
What `poi` genuinely lacks against `minor` is narrower, and is a build list rather than a question:
**no dedicated setter / `moderatorProcedure`**, **no snapshot** (so no rollback), **absent from
@@ -170,7 +170,8 @@ and builds; that is all that is established. Before any of it is trusted in prod
## 4. Known-open, decided or deferred
- **`aiNsfwLevel` / `aiModel` exist in production but not in `schema.full.prisma`.**
`src/pages/api/webhooks/image-scan-result.ts` writes both, and Front Page Audit reads `aiNsfwLevel`
nothing in this repo writes them any more (the only writer was the legacy scanner branch of
`image-scan-result.ts`, removed), and Front Page Audit reads `aiNsfwLevel`
through raw `sql` because it cannot be selected as a typed column. It is the scanner's own rating, and
disagreement with `nsfwLevel` is the strongest signal that a row needs a human — so it is worth
having. Either add both to the schema and regenerate, or accept the raw read as permanent.
-1
View File
@@ -118,7 +118,6 @@ const IO_CALL_NAMES = new Set([
'fetch',
// image ingestion / scanner
'ingestImage',
'ingestImageBulk',
'createImageIngestionRequest',
// orchestrator
'submitWorkflow',
+1 -1
View File
@@ -140,7 +140,7 @@
"@civitai/generation-metadata": "^0.3.0",
"@civitai/moderation": "workspace:*",
"@civitai/next-axiom": "^0.17.0",
"@civitai/orchestration-client": "0.2.0-beta.104",
"@civitai/orchestration-client": "0.2.0-beta.106",
"@civitai/shared": "workspace:*",
"@clavata/sdk": "^0.2.3",
"@clickhouse/client": "^0.2.2",
-9
View File
@@ -2064,15 +2064,6 @@ export const REDIS_SYS_KEYS = {
CREATION_BLOCKED_TAGS: 'system:creation-blocked-tags',
LIVE_FEATURE_FLAGS: 'system:live-feature-flags',
SUSPICIOUS_AUDIT_MATCHES: 'system:suspicious-audit-matches',
/*
Runtime toggle for the new image ingestion path (createImageIngestionRequest
with the expanded mediaRating step). Read by image.service.ts before
routing to the new vs legacy scanner. Accepts '1'/'true' to enable,
'0'/'false' to disable. If the key is missing, the first call seeds it to
'false' so the toggle is discoverable in Redis. Lets ops flip without a
deploy.
*/
IMAGE_SCANNER_NEW: 'system:image-scanner-new',
/*
Per-run image cap for the remove-deleted-user-images job. Set to '0' to
pause the drain without a deploy. Missing key means the job's compiled
+1 -1
View File
@@ -586,7 +586,7 @@ export const imageScanWebhookCounter = registerCounterWithLabels({
export const imageScanSubmittedCounter = registerCounterWithLabels({
name: 'image_scan_submitted_total',
help: 'ingestImage() scan submissions by lane (new|legacy) and result (success|failed)',
help: 'ingestImage() scan submissions by lane (new = wdTagging+mediaRating | imageScanning) and result (success|failed)',
labelNames: ['lane', 'result'] as const,
});
+5 -5
View File
@@ -71,8 +71,8 @@ importers:
specifier: ^0.17.0
version: 0.17.0(next@16.3.1(@babel/core@7.29.0)(@opentelemetry/api@1.9.0)(@playwright/test@1.57.0)(@types/node@24.13.3)(react-dom@18.3.1(react@18.3.1))(react@18.3.1)(sass@1.90.0))
'@civitai/orchestration-client':
specifier: 0.2.0-beta.104
version: 0.2.0-beta.104
specifier: 0.2.0-beta.106
version: 0.2.0-beta.106
'@civitai/shared':
specifier: workspace:*
version: link:packages/civitai-shared
@@ -2225,8 +2225,8 @@ packages:
peerDependencies:
next: ^12.1.4 || ^13 || ^14
'@civitai/orchestration-client@0.2.0-beta.104':
resolution: {integrity: sha512-V7EKGBMi3YjtOfkxXOmogJhSbM83t9OCJOtxH6QSpwzsE7JZH85gtoMW6sSvF7pDLj9Ywim68Pmlq2M0Qd47sg==}
'@civitai/orchestration-client@0.2.0-beta.106':
resolution: {integrity: sha512-PfDmcOXhwgknRitkyeE21/oJMnu6+eDSYyLWL2UgdZScoi5adRnEhJNfV+nWIBHzKz62whwEW+BhG8uxTTzsZw==}
engines: {git: '>=2.11.0', node: '>=18.0.0', npm: '>=7.19.0', yarn: '>=1.7.0'}
'@clavata/sdk@0.2.3':
@@ -13450,7 +13450,7 @@ snapshots:
next: 16.3.1(@babel/core@7.29.0)(@opentelemetry/api@1.9.0)(@playwright/test@1.57.0)(@types/node@24.13.3)(react-dom@18.3.1(react@18.3.1))(react@18.3.1)(sass@1.90.0)
whatwg-fetch: 3.6.20
'@civitai/orchestration-client@0.2.0-beta.104':
'@civitai/orchestration-client@0.2.0-beta.106':
dependencies:
'@hey-api/client-fetch': 0.1.14
rfc6902: 5.2.0
@@ -0,0 +1,142 @@
import type { NextApiRequest, NextApiResponse } from 'next';
import { beforeEach, describe, expect, it, vi } from 'vitest';
import type * as ImageScanningResultService from '~/server/services/image-scanning-result.service';
const mocks = vi.hoisted(() => ({
getWorkflow: vi.fn(),
captureJobFailureReason: vi.fn(),
processImageScanWorkflow: vi.fn(),
processImageScanningWorkflow: vi.fn(),
}));
vi.mock('~/server/utils/endpoint-helpers', () => ({
WebhookEndpoint: (handler: unknown) => handler,
}));
vi.mock('@civitai/client', () => ({ getWorkflow: mocks.getWorkflow }));
vi.mock('~/server/services/orchestrator/client', () => ({ internalOrchestratorClient: {} }));
vi.mock('~/server/services/feature-flags.service', () => ({
getFeatureFlagsLazy: () => ({ articleImageScanning: true }),
}));
vi.mock('~/server/prom/client', () => ({ imageScanWebhookCounter: { inc: vi.fn() } }));
// Both pipelines import the whole ingestion stack; this suite pins only the routing between them.
vi.mock('~/server/services/image-scan-pipeline', () => ({
captureJobFailureReason: mocks.captureJobFailureReason,
LEGACY_SCAN_LOG: { name: 'image-scan-result', source: 'image-scan-result.service' },
}));
vi.mock('~/server/services/image-scan-result.service', () => ({
processImageScanWorkflow: mocks.processImageScanWorkflow,
}));
// The real `isImageScanningWorkflow` makes the routing decision under test.
vi.mock('~/server/services/image-scanning-result.service', async (importOriginal) => ({
...(await importOriginal<typeof ImageScanningResultService>()),
processImageScanningWorkflow: mocks.processImageScanningWorkflow,
}));
vi.mock('~/server/services/scanner-audit.service', () => ({ recordImageScanningResult: vi.fn() }));
vi.mock('~/server/services/job-queue.service', () => ({ removeImageScanJobQueue: vi.fn() }));
vi.mock('~/server/services/orchestrator/orchestrator.service', () => ({
computePerceptualHash: vi.fn(),
}));
vi.mock('~/server/utils/webhook-debounce', () => ({ fanOutArticleImageUpdates: vi.fn() }));
import handler from '~/pages/api/webhooks/image-scan-result';
import { loggingMock } from '~/__tests__/mocks/logging.mock';
const call = async (body: unknown) => {
const res = {
status: vi.fn().mockReturnThis(),
json: vi.fn().mockReturnThis(),
send: vi.fn().mockReturnThis(),
};
await (handler as unknown as (req: NextApiRequest, res: NextApiResponse) => Promise<unknown>)(
{ method: 'POST', body } as NextApiRequest,
res as unknown as NextApiResponse
);
return res;
};
const deliver = (status: string, steps: unknown[]) => {
mocks.getWorkflow.mockResolvedValue({
data: { metadata: { imageId: 7 }, startedAt: 'start', completedAt: 'end', steps },
});
return call({ workflowId: 'wf', status });
};
const pipelineInput = (status: string, steps: unknown[]) => ({
workflowId: 'wf',
status,
steps,
imageId: 7,
articleImageScanning: true,
startedAt: 'start',
completedAt: 'end',
});
describe('image-scan-result webhook routing', () => {
beforeEach(() => vi.clearAllMocks());
it('stashes a job failure reason without fetching the workflow', async () => {
const event = { workflowId: 'wf', jobId: 'job', reason: 'timed out' };
const res = await call(event);
expect(mocks.captureJobFailureReason).toHaveBeenCalledWith(event);
expect(mocks.getWorkflow).not.toHaveBeenCalled();
expect(res.status).toHaveBeenCalledWith(200);
});
it('sends an imageScanning workflow to the imageScanning pipeline, failed or not', async () => {
const steps = [{ $type: 'imageScanning', status: 'failed' }];
await deliver('failed', steps);
expect(mocks.processImageScanningWorkflow).toHaveBeenCalledTimes(1);
expect(mocks.processImageScanningWorkflow).toHaveBeenCalledWith(pipelineInput('failed', steps));
expect(mocks.processImageScanWorkflow).not.toHaveBeenCalled();
});
it('sends a wdTagging + mediaRating workflow to the legacy pipeline', async () => {
const steps = [
{ $type: 'wdTagging', output: { tags: {} } },
{ $type: 'mediaRating', output: { nsfwLevel: 'pg' } },
];
await deliver('succeeded', steps);
expect(mocks.processImageScanWorkflow).toHaveBeenCalledTimes(1);
expect(mocks.processImageScanWorkflow).toHaveBeenCalledWith(pipelineInput('succeeded', steps));
expect(mocks.processImageScanningWorkflow).not.toHaveBeenCalled();
});
it('rejects a workflow it cannot load or that carries no imageId', async () => {
mocks.getWorkflow.mockResolvedValueOnce({ data: undefined });
const missing = await call({ workflowId: 'wf', status: 'succeeded' });
expect(missing.status).toHaveBeenCalledWith(400);
expect(missing.send).toHaveBeenCalledWith({ error: 'could not find workflow: wf' });
mocks.getWorkflow.mockResolvedValueOnce({ data: { metadata: {}, steps: [] } });
const noImage = await call({ workflowId: 'wf', status: 'succeeded' });
expect(noImage.send).toHaveBeenCalledWith({
error: 'missing workflow metadata.imageId - wf',
});
});
it.each([
['imageScanning', [{ $type: 'imageScanning' }], 'image-scanning-result'],
['legacy', [{ $type: 'wdTagging' }, { $type: 'mediaRating' }], 'image-scan-result'],
])('logs a %s pipeline failure under its own name', async (_, steps, name) => {
const pipeline =
name === 'image-scanning-result'
? mocks.processImageScanningWorkflow
: mocks.processImageScanWorkflow;
pipeline.mockRejectedValueOnce(new Error('boom'));
const res = await deliver('succeeded', steps);
expect(res.status).toHaveBeenCalledWith(400);
expect(loggingMock.logToAxiom).toHaveBeenCalledWith(
expect.objectContaining({ name, type: 'error', message: 'boom' })
);
});
it('acknowledges a result for an image deleted since the scan was submitted', async () => {
mocks.processImageScanningWorkflow.mockRejectedValueOnce(new Error('image not found: 7'));
const res = await deliver('succeeded', [{ $type: 'imageScanning', output: {} }]);
expect(res.status).toHaveBeenCalledWith(200);
expect(res.json).toHaveBeenCalledWith({ ok: true, skipped: 'deleted' });
});
});
@@ -1,11 +1,10 @@
import { describe, it, expect, vi, beforeEach } from 'vitest';
import type { NextApiRequest, NextApiResponse } from 'next';
import handler from '~/pages/api/webhooks/image-scan-result';
import { processImageScanWorkflow } from '~/server/services/image-scan-result.service';
import { processImageScanningWorkflow } from '~/server/services/image-scanning-result.service';
import { clickhouse } from '~/server/clickhouse/client';
import { signalClient } from '~/utils/signal-client';
import type * as ClickhouseClient from '~/server/clickhouse/client';
import { TagSource, ImageIngestionStatus } from '~/shared/utils/prisma/enums';
import { NsfwLevel } from '~/server/common/enums';
import type * as BlocklistService from '~/server/services/blocklist.service';
const {
@@ -271,7 +270,6 @@ vi.mock('~/server/services/image.service', () => ({
getImagesModRules: vi.fn().mockResolvedValue([]),
queueImageSearchIndexUpdate: vi.fn().mockResolvedValue(undefined),
enqueueImageIngestion: vi.fn().mockResolvedValue(undefined),
imageScanTypes: [3, 9], // ImageScanType.WD14, ImageScanType.SpineRating
}));
// Render a Prisma tagged-template call the way the driver does: nested `Prisma.sql` fragments
@@ -403,158 +401,6 @@ describe('image-scan-result webhook - pipeline tests', () => {
});
});
const runWebhook = (body: any) => {
const req = {
method: 'POST',
query: { token: 'mock-webhook-token' },
headers: { host: 'localhost:3000' },
body,
} as unknown as NextApiRequest;
const res = {
status: vi.fn().mockReturnThis(),
json: vi.fn().mockImplementation((val) => res),
send: vi.fn().mockImplementation((val) => res),
} as unknown as NextApiResponse;
return { promise: handler(req, res), res };
};
it('should prevent tag-bleed between Clavata (ignored) and WD14 (not ignored) concurrently', async () => {
const reqA = runWebhook({
id: 1,
status: 0,
source: TagSource.Clavata,
tags: [{ tag: 'hate symbols', confidence: 95 }],
});
const reqB = runWebhook({
id: 2,
status: 0,
source: TagSource.WD14,
tags: [{ tag: 'hate symbols', confidence: 95 }],
});
await Promise.all([reqA.promise, reqB.promise]);
expect(reqA.res.status).toHaveBeenCalledWith(200);
expect(reqB.res.status).toHaveBeenCalledWith(200);
const dbUpdates = mockDbWrite.image.update.mock.calls;
const updateForImage2 = dbUpdates.find((call: any) => call[0].where.id === 2);
expect(updateForImage2).toBeDefined();
expect(updateForImage2[0].data.nsfwLevel).toBe(32); // Blocked
expect(updateForImage2[0].data.ingestion).toBe(ImageIngestionStatus.Scanned);
const insertedTags = [
...mockInsertTagsOnImageNew.mock.calls.flatMap((call) => call[0]),
...mockUpsertTagsOnImageNew.mock.calls.flatMap((call) => call[0]),
];
const tagForImage1 = insertedTags.find((t) => t.imageId === 1 && t.tagId === 100);
const tagForImage2 = insertedTags.find((t) => t.imageId === 2 && t.tagId === 100);
expect(tagForImage1).toBeDefined();
expect(tagForImage1.disabled).toBe(true); // Clavata ignores hate symbols
expect(tagForImage2).toBeDefined();
expect(tagForImage2.disabled).toBe(false); // WD14 does not ignore hate symbols
});
it('should set ingestion to NotFound when status is NotFound', async () => {
const req = runWebhook({
id: 3,
status: 1, // NotFound
source: TagSource.WD14,
tags: [],
});
await req.promise;
expect(req.res.status).toHaveBeenCalledWith(200);
expect(mockDbWrite.image.updateMany).toHaveBeenCalledWith({
where: { id: 3, ingestion: { in: ['Pending', 'Error'] } },
data: { ingestion: ImageIngestionStatus.NotFound },
});
});
it('should increment retryCount when status is Unscannable', async () => {
const req = runWebhook({
id: 4,
status: 2, // Unscannable
source: TagSource.WD14,
tags: [],
});
await req.promise;
expect(req.res.status).toHaveBeenCalledWith(200);
const queryCall = imageUpdates.find(
(update) => update.text.includes('retryCount') && update.params.includes(4)
);
expect(queryCall).toBeDefined();
});
it('should set needsReview: minor when minor tag is present and image is NSFW', async () => {
const req = runWebhook({
id: 5,
status: 0,
source: TagSource.WD14,
tags: [
{ tag: 'teen', confidence: 95 },
{ tag: 'r', confidence: 95 },
],
});
await req.promise;
expect(req.res.status).toHaveBeenCalledWith(200);
const dbUpdates = mockDbWrite.image.update.mock.calls;
const updateForImage5 = dbUpdates.find((call: any) => call[0].where.id === 5);
expect(updateForImage5).toBeDefined();
expect(updateForImage5[0].data.needsReview).toBe('minor');
});
it('should set needsReview: poi when POI tag is present', async () => {
const req = runWebhook({
id: 6,
status: 0,
source: TagSource.WD14,
tags: [
{ tag: 'potential celebrity', confidence: 95 },
{ tag: 'pg', confidence: 95 },
],
});
await req.promise;
expect(req.res.status).toHaveBeenCalledWith(200);
const dbUpdates = mockDbWrite.image.update.mock.calls;
const updateForImage6 = dbUpdates.find((call: any) => call[0].where.id === 6);
expect(updateForImage6).toBeDefined();
expect(updateForImage6[0].data.needsReview).toBe('poi');
});
it('should not update nsfwLevel if nsfwLevelLocked is true', async () => {
const req = runWebhook({
id: 7,
status: 0,
source: TagSource.WD14,
tags: [
{ tag: 'some-tag', confidence: 95 },
{ tag: 'x', confidence: 95 },
],
});
await req.promise;
expect(req.res.status).toHaveBeenCalledWith(200);
const dbUpdates = mockDbWrite.image.update.mock.calls;
const updateForImage7 = dbUpdates.find((call: any) => call[0].where.id === 7);
expect(updateForImage7).toBeDefined();
expect(updateForImage7[0].data.nsfwLevel).toBeUndefined(); // locked, so undefined (not updated)
});
it('creates a never-before-seen tag with the columns the Tag table requires', async () => {
const passthrough = mockDbWrite.$queryRaw.getMockImplementation();
let tagInsert: { text: string; params: any[] } | undefined;
@@ -746,208 +592,202 @@ describe('image-scan-result webhook - pipeline tests', () => {
expect(errorFlips).toHaveLength(0);
});
describe('webhook body strings never reach SQL text', () => {
const INJECTED = `1 OR 1=1) < 5 AND disabled = false -- `;
it('rejects a non-numeric hash before it can reach the ClickHouse query', async () => {
envOverrides.BLOCKED_IMAGE_HASH_CHECK = true;
const req = runWebhook({
id: 8,
status: 0,
source: TagSource.ImageHash,
hash: INJECTED,
});
await req.promise;
expect(mockClickhouseQuery).not.toHaveBeenCalled();
expect(req.res.status).toHaveBeenCalledWith(400);
// The imageScanning pipeline through the real shared stages; only the database, ClickHouse
// and signals are faked. Scan outputs follow the shape the orchestrator returned on 2026-09-17.
describe('imageScanning workflows through the shared pipeline', () => {
const scanOutput = (overrides: Record<string, unknown> = {}) => ({
nsfwLevel: 'x',
score: 0.91,
topK: [{ label: 'X', score: 0.91 }],
aiRecognition: { label: 'AI', score: 0.86, topK: [] },
animeRecognition: { label: 'anime', score: 0.9, scores: {} },
humanRecognition: {
status: 'ok',
ran: true,
label: 'human',
score: 0.7,
humanScore: 0.7,
noHumanScore: 0.3,
scores: {},
evidence: [],
},
tagging: {
status: 'ran',
ran: true,
threshold: 0.55,
tagCount: 3,
totalAboveThreshold: 3,
truncated: false,
tags: [
{ tag: 'teen', category: 'general', score: 0.9 },
{ tag: 'hate symbols', category: 'copyright', score: 0.8 },
{ tag: 'some-tag', category: 'meta', score: 0.8 },
],
},
jointAgeClassification: {
status: 'ran',
ran: true,
detections: [{ ageBand: '21-24', under18Probability: 0.06, isMinor: false }],
minorDetected: false,
},
csam: false,
...overrides,
});
it('logs a match against the parsed hash, not the string the scanner sent', async () => {
envOverrides.BLOCKED_IMAGE_HASH_CHECK = true;
mockClickhouseQuery.mockResolvedValue([{ count: 1 }]);
const req = runWebhook({ id: 16, status: 0, source: TagSource.ImageHash, hash: ' 42 ' });
await req.promise;
const match = mockLogToAxiom.mock.calls.find(
(call) => call[0]?.message === 'Image pHash matched a blocked image'
);
expect(match).toBeDefined();
expect(match![0].pHash).toBe('42');
const scanStep = (output: unknown, status = 'succeeded') => ({
$type: 'imageScanning',
name: 'scan',
status,
output,
});
it('passes a valid hash to ClickHouse as digits only', async () => {
envOverrides.BLOCKED_IMAGE_HASH_CHECK = true;
const req = runWebhook({
id: 9,
status: 0,
source: TagSource.ImageHash,
hash: '-1234567890123456789',
});
await req.promise;
expect(mockClickhouseQuery).toHaveBeenCalled();
const [strings, ...values] = mockClickhouseQuery.mock.calls[0];
expect(strings.join('?')).toContain('bitXor(hash, ?)');
expect(values).toEqual([-1234567890123456789n]);
});
it('sends the scanner-supplied model id as a parameter, not as SQL text', async () => {
const req = runWebhook({
id: 10,
status: 0,
source: TagSource.WD14,
tags: [],
context: { movie_rating: 'PG', movie_rating_model_id: `x'; DROP TABLE "Image"; --` },
});
await req.promise;
const update = imageUpdates.find((u) => u.text.includes('"aiModel"'));
expect(update).toBeDefined();
expect(update!.text).not.toContain('DROP TABLE');
expect(update!.params).toContain(`x'; DROP TABLE "Image"; --`);
});
});
describe('perceptual hash handling', () => {
it('records the scan without a lookup when the hash is blank', async () => {
envOverrides.BLOCKED_IMAGE_HASH_CHECK = true;
const req = runWebhook({ id: 11, status: 0, source: TagSource.ImageHash, hash: ' ' });
await req.promise;
expect(mockClickhouseQuery).not.toHaveBeenCalled();
expect(req.res.status).toHaveBeenCalledWith(200);
const update = imageUpdates.find((u) => u.text.includes('jsonb_build_object'));
expect(update).toBeDefined();
expect(update!.text).not.toContain('"pHash"');
expect(
mockLogToAxiom.mock.calls.some(
(call) => call[0]?.message === 'blank hash from ImageHash scan'
)
).toBe(true);
});
it('stores a zero hash but does not look it up', async () => {
envOverrides.BLOCKED_IMAGE_HASH_CHECK = true;
const req = runWebhook({ id: 15, status: 0, source: TagSource.ImageHash, hash: '0' });
await req.promise;
const update = imageUpdates.find((u) => u.text.includes('"pHash"'));
expect(update).toBeDefined();
expect(update!.params).toContain(0n);
expect(mockClickhouseQuery).not.toHaveBeenCalled();
});
it('completes the scan when the blocklist lookup fails', async () => {
envOverrides.BLOCKED_IMAGE_HASH_CHECK = true;
mockClickhouseQuery.mockRejectedValue(new Error('socket hang up'));
const req = runWebhook({ id: 14, status: 0, source: TagSource.ImageHash, hash: '42' });
await req.promise;
expect(mockClickhouseQuery).toHaveBeenCalled();
expect(req.res.status).toHaveBeenCalledWith(200);
const update = imageUpdates.find((u) => u.params.includes(42n));
expect(update).toBeDefined();
});
it('rejects a hash wider than Int64 rather than letting ClickHouse reject it', async () => {
envOverrides.BLOCKED_IMAGE_HASH_CHECK = true;
const req = runWebhook({
id: 12,
status: 0,
source: TagSource.ImageHash,
hash: '99999999999999999999',
});
await req.promise;
expect(mockClickhouseQuery).not.toHaveBeenCalled();
expect(req.res.status).toHaveBeenCalledWith(400);
});
it('binds a hash as a bigint parameter, not as SQL text', async () => {
const req = runWebhook({
id: 13,
status: 0,
source: TagSource.ImageHash,
hash: '4611686018427387904',
});
await req.promise;
const update = imageUpdates.find((u) => u.text.includes('"pHash"'));
expect(update).toBeDefined();
expect(update!.params).toContain(4611686018427387904n);
expect(update!.text).not.toContain('4611686018427387904');
});
});
describe('moderator benign phrases reach the POI tagger', () => {
// The file's own `beforeEach` is `vi.clearAllMocks()`, which clears calls but NOT
// implementations — so the one a test below installs would leak into anything appended
// after this describe. Harmless while this is the last block, which is exactly the kind
// of "harmless" that stops being true without anyone noticing.
beforeEach(() => {
mockStripBenignPhrases.mockImplementation(async (text?: string) => text ?? '');
});
const seedImageWithPrompt = (id: number, prompt: string) => {
imageDbState.set(id, {
id,
createdAt: new Date(),
scannedAt: null,
type: 'image',
userId: 1,
meta: { prompt },
metadata: {},
postId: null,
nsfwLevelLocked: false,
nsfwLevel: null,
scanJobs: { scans: {} },
ingestion: ImageIngestionStatus.Pending,
});
const hashStep = {
$type: 'mediaHash',
name: 'hash',
status: 'succeeded',
output: { hashes: { perceptual: '6F51B11C49611E0E' } },
};
const deliver = (imageId: number, steps: unknown[], status = 'succeeded') =>
processImageScanningWorkflow({ workflowId: `wf-${imageId}`, status, steps, imageId });
const requestedTagNames = () =>
mockDbWrite.tag.findMany.mock.calls.flatMap((call: any) => call[0]?.where?.name?.in ?? []);
const sqlOf = (call: any[]) =>
renderSql(Array.isArray(call[0]) ? call[0] : call[0].strings, call.slice(1));
// The verdict row written by resolveScanOutcome.
const verdictFor = (imageId: number) => {
const update = mockDbWrite.$executeRaw.mock.calls
.map(sqlOf)
.find(
(sql: any) =>
sql.text.includes('UPDATE "Image"') &&
sql.text.includes('"needsReview"') &&
sql.params.at(-1) === imageId
);
if (!update) return undefined;
const param = (column: string) => {
const match = update.text.match(new RegExp(`"${column}" = \\$(\\d+)`));
return match ? update.params[Number(match[1]) - 1] : undefined;
};
return {
ingestion: param('ingestion'),
nsfwLevel: param('nsfwLevel'),
needsReview: param('needsReview'),
minor: param('minor'),
pHash: param('pHash'),
};
};
const tagsWritten = (imageId: number) =>
mockInsertTagsOnImageNew.mock.calls
.flatMap((call) => call[0])
.filter((tag: any) => tag.imageId === imageId);
const errorFlipsFor = (imageId: number) =>
mockDbWrite.$queryRaw.mock.calls
.map(sqlOf)
.filter((sql: any) => sql.text.includes("'{retryCount}'") && sql.params.includes(imageId));
const auditRows = () =>
vi
.mocked(clickhouse!.insert)
.mock.calls.filter((call: any) => call[0].table === 'scanner_label_results')
.flatMap((call: any) => call[0].values);
it('CONTROL: a POI name that is not whitelisted is still tagged', async () => {
seedImageWithPrompt(30, 'emma stone');
const req = runWebhook({ id: 30, status: 0, source: TagSource.WD14, tags: [] });
await req.promise;
// Deliberately asserts only the tag, with the strip passing the text through: it
// stays green with the fix reverted, so a failure below is the strip and not a
// broken fixture.
expect(requestedTagNames()).toContain('emma stone');
beforeEach(() => {
// An earlier test leaves $executeRaw throwing, which would skip the verdict write.
mockDbWrite.$executeRaw.mockResolvedValue(0);
vi.mocked(clickhouse!.insert).mockClear();
});
it('a whitelisted phrase is stripped before the POI check, so no POI tag is written', async () => {
mockStripBenignPhrases.mockImplementation(async (text?: string) =>
(text ?? '').replace('emma stone', '')
);
// Two real POI names, only one whitelisted. `tom hanks` survives the strip and must
// still be tagged, which proves this run reached the tag write at all — otherwise an
// early throw would satisfy the negative assertion below with an empty array.
seedImageWithPrompt(31, 'emma stone and tom hanks');
it('writes general tags, the rating and a Scanned verdict for an image', async () => {
await deliver(40, [scanStep(scanOutput()), hashStep]);
const req = runWebhook({ id: 31, status: 0, source: TagSource.WD14, tags: [] });
await req.promise;
// Pins WHICH list is consulted. Pointing the strip at ProfanityBenignWord instead
// leaves the whole fix inert in production and every other assertion here green.
expect(mockStripBenignPhrases).toHaveBeenCalledWith(
'emma stone and tom hanks',
'PromptBenignPhrase'
const written = tagsWritten(40);
expect(written).toContainEqual(
expect.objectContaining({ tagId: 200, source: TagSource.WD14, confidence: 90 })
);
expect(requestedTagNames()).not.toContain('emma stone');
expect(requestedTagNames()).toContain('tom hanks');
expect(written).toContainEqual(
expect.objectContaining({ tagId: 1004, source: TagSource.SpineRating })
);
// Other tagger categories are not ingested, even when the tag exists.
expect(written.some((tag: any) => tag.tagId === 100 || tag.tagId === 400)).toBe(false);
expect(verdictFor(40)).toMatchObject({
ingestion: ImageIngestionStatus.Scanned,
nsfwLevel: 8,
pHash: BigInt('0x6F51B11C49611E0E'),
});
expect(errorFlipsFor(40)).toHaveLength(0);
expect(vi.mocked(signalClient.send)).toHaveBeenCalledWith(
expect.objectContaining({
data: expect.objectContaining({ imageId: 40, ingestion: ImageIngestionStatus.Scanned }),
})
);
});
it('queues an NSFW image with a minor tag for review', async () => {
await deliver(41, [scanStep(scanOutput())]);
expect(verdictFor(41)).toMatchObject({ needsReview: 'minor', minor: true });
});
it('leaves a moderator-locked NSFW level alone', async () => {
await deliver(7, [scanStep(scanOutput())]);
expect(verdictFor(7)).toMatchObject({
ingestion: ImageIngestionStatus.Scanned,
nsfwLevel: 1,
});
});
it('writes version 2 scanner-audit rows, csam included', async () => {
await deliver(42, [scanStep(scanOutput({ csam: true })), hashStep]);
const rows = auditRows();
expect(rows.map((row: any) => [row.label, row.labelValue, row.triggered])).toEqual([
['x', 'nsfw_level', 1],
['csam', '', 1],
['minor', '21-24', 0],
['ai', 'ai_recognition', 1],
['anime', 'anime_recognition', 1],
]);
expect(rows.every((row: any) => row.version === '2' && row.entityIds[0] === '42')).toBe(true);
});
it('rates a video by its riskiest frame', async () => {
await deliver(43, [
{ $type: 'videoFrameExtraction', name: 'videoFrames', status: 'succeeded', output: {} },
{
$type: 'repeat',
status: 'succeeded',
input: { template: { $type: 'imageScanning' } },
output: {
steps: [
scanStep(scanOutput({ nsfwLevel: 'pg' })),
scanStep(scanOutput({ nsfwLevel: 'x' })),
scanStep(scanOutput({ nsfwLevel: 'pg13' })),
],
},
},
]);
expect(tagsWritten(43)).toContainEqual(
expect.objectContaining({ tagId: 1004, source: TagSource.SpineRating })
);
expect(verdictFor(43)).toMatchObject({
ingestion: ImageIngestionStatus.Scanned,
nsfwLevel: 8,
});
});
it.each([
['a failed workflow', [scanStep(undefined, 'failed')], 'failed', 'workflow-failed'],
[
'a scan whose tagging did not run',
[scanStep(scanOutput({ tagging: { status: 'skipped', ran: false, tags: [] } }))],
'succeeded',
'unusable-result',
],
])('marks %s as one Error without a verdict', async (_, steps, status, failureType) => {
await deliver(44, steps, status);
const flips = errorFlipsFor(44);
expect(flips).toHaveLength(1);
expect(flips[0].params.join(' ')).toContain(failureType);
expect(verdictFor(44)).toBeUndefined();
expect(tagsWritten(44)).toHaveLength(0);
});
});
});
-3
View File
@@ -389,10 +389,8 @@ export const serverSchema = z
UNAUTHENTICATED_DOWNLOAD: zc.booleanString,
UNAUTHENTICATED_LIST_NSFW: zc.booleanString,
LOGGING: commaDelimitedStringArray(),
IMAGE_SCANNING_ENDPOINT: isProd ? z.string() : z.string().optional(),
IMAGE_SCANNING_CALLBACK: z.string().optional(),
TEXT_MODERATION_CALLBACK: z.string().optional(),
IMAGE_SCANNING_MODEL: z.string().optional(),
IMAGE_SCANNING_RETRY_DELAY: z.coerce.number().default(5),
// Age-out threshold (minutes) for never-returning image scans. A scan verdict
// arrives via the fire-and-forget /image-scan-result webhook; a fraction never
@@ -419,7 +417,6 @@ export const serverSchema = z
// newest-first (starving the oldest backlog) — a bad hand-tune must fail
// loudly at boot instead.
IMAGE_SCANNING_MAX_PER_RUN: z.coerce.number().int().positive().default(1000),
IMAGE_SCANNER_NEW: zc.booleanString.default(false),
DELIVERY_WORKER_ENDPOINT: z.string().optional(),
DELIVERY_WORKER_TOKEN: z.string().optional(),
STORAGE_RESOLVER_ENDPOINT: z.string().optional(), // URL for storage-resolver microservice
@@ -1,78 +0,0 @@
import { chunk } from 'lodash-es';
import type { NextApiRequest, NextApiResponse } from 'next';
import * as z from 'zod';
import { ImageScanType } from '~/server/common/enums';
import { dbRead } from '~/server/db/client';
import { dataProcessor } from '~/server/db/db-helpers';
import type { IngestImageInput } from '~/server/schema/image.schema';
import { ingestImageBulk } from '~/server/services/image.service';
import { ModEndpoint } from '~/server/utils/endpoint-helpers';
const schema = z.object({
start: z.coerce.number().optional(),
end: z.coerce.number().optional(),
});
export default ModEndpoint(async function (req: NextApiRequest, res: NextApiResponse) {
const input = schema.parse(req.query);
const start = Date.now();
await dataProcessor({
params: { batchSize: 100000, concurrency: 10, start: 0 },
runContext: {
on: (event: 'close', listener: () => void) => {
// noop
},
},
rangeFetcher: async (ctx) => {
let [{ start, end }] = await dbRead.$queryRaw<{ start: number; end: number }[]>`
WITH dates AS (
SELECT
MIN("createdAt") as start,
MAX("createdAt") as end
FROM "Image"
)
SELECT MIN(id) as start, MAX(id) as end
FROM "Image" i
JOIN dates d ON d.start = i."createdAt" OR d.end = i."createdAt";
`;
if (input.start) start = input.start;
if (input.end) end = input.end;
return { start, end };
},
processor: async ({ start, end }) => {
const consoleFetchKey = `Fetch: ${start} - ${end}`;
console.log(consoleFetchKey);
console.time(consoleFetchKey);
const records = await dbRead.$queryRaw<IngestImageInput[]>`
SELECT
"id",
"url",
"type",
"height",
"width",
meta->>'prompt' as prompt
FROM "Image" i
WHERE i.id BETWEEN ${start} AND ${end}
AND i."pHash" IS NULL
`;
console.timeEnd(consoleFetchKey);
if (records.length === 0) return;
const consolePushKey = `Push: ${start} - ${end}: ${records.length}`;
console.log(consolePushKey);
console.time(consolePushKey);
for (const batch of chunk(records, 1000)) {
await ingestImageBulk({
images: batch,
scans: [ImageScanType.Hash],
lowPriority: true,
});
}
console.timeEnd(consolePushKey);
},
});
return res.status(200).json({ success: true, duration: Date.now() - start });
});
-89
View File
@@ -1,89 +0,0 @@
import type { NextApiRequest, NextApiResponse } from 'next';
import { dbWrite } from '~/server/db/client';
import * as z from 'zod';
import { ModEndpoint } from '~/server/utils/endpoint-helpers';
import type { Prisma } from '@prisma/client';
import { env } from '~/env/server';
import { chunk } from 'lodash-es';
import { getEdgeUrl } from '~/client-utils/edge-url';
const stringToNumberArraySchema = z
.string()
.transform((s) => s.split(',').map(Number))
.optional();
const importSchema = z.object({
imageCount: z.preprocess((x) => (x ? parseInt(String(x)) : undefined), z.number()).optional(),
imageIds: stringToNumberArraySchema,
wait: z.preprocess((val) => val === true || val === 'true', z.boolean()).optional(),
});
export default ModEndpoint(
async function scanImages(req: NextApiRequest, res: NextApiResponse) {
if (!env.IMAGE_SCANNING_ENDPOINT)
return res.status(400).json({ error: 'Image scanning is not enabled' });
const { imageCount, imageIds, wait } = importSchema.parse(req.query);
const where: Prisma.Enumerable<Prisma.ImageWhereInput> = {};
if (!!imageIds?.length) where.id = { in: imageIds };
else if (!!imageCount) {
where.scanRequestedAt = null;
where.scannedAt = null;
} else {
return res.status(400).json({
error: 'Must provide at least one of imageCount or imageIds',
});
}
const images = await dbWrite.image.findMany({
where,
take: imageCount,
select: { url: true, id: true, width: true, name: true, mimeType: true },
});
if (!wait) res.status(200).json({ images: images.length });
const batchSize = 100;
const batches = chunk(images, batchSize);
let i = 0;
for (const batch of batches) {
console.log(
`Sending batch ${i} to ${Math.min(i + batchSize, images.length)} of ${images.length} images`
);
const queued: number[] = [];
await Promise.all(
batch.map(async (image) => {
const width = Math.min(450, image.width ?? 450);
const anim =
image.name?.endsWith('.gif') || image.mimeType == 'image/gif' ? false : undefined;
const gamma = anim === false ? 0.99 : undefined;
const url = getEdgeUrl(image.url, { width, anim, gamma });
try {
await fetch(env.IMAGE_SCANNING_ENDPOINT as string, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ url, imageId: image.id }),
});
queued.push(image.id);
} catch (e: any) {
console.error('Failed to send image for scan', e.message);
}
})
);
if (!!queued.length) {
await dbWrite.image.updateMany({
where: { id: { in: queued } },
data: { scanRequestedAt: new Date() },
});
}
i += batchSize;
}
console.log('Done sending images for scan!');
if (wait) res.status(200).json({ images: images.length });
},
['GET']
);
File diff suppressed because it is too large Load Diff
-34
View File
@@ -1,34 +0,0 @@
import type { NextApiRequest, NextApiResponse } from 'next';
import * as z from 'zod';
import { Prisma } from '@prisma/client';
import { dbWrite } from '~/server/db/client';
import type { IngestImageInput } from '~/server/schema/image.schema';
import { ingestImageBulk } from '~/server/services/image.service';
import { WebhookEndpoint } from '~/server/utils/endpoint-helpers';
const schema = z.object({
imageIds: z.array(z.number()).min(1),
lowPriority: z.boolean().default(true),
});
export default WebhookEndpoint(async function (req: NextApiRequest, res: NextApiResponse) {
if (req.method !== 'POST') return res.status(405).json({ error: 'Method Not Allowed' });
const { imageIds, lowPriority } = schema.parse(req.body);
const images = await dbWrite.$queryRaw<IngestImageInput[]>`
SELECT id, url, type, width, height, meta->>'prompt' as prompt
FROM "Image"
WHERE id IN (${Prisma.join(imageIds)})
`;
if (!images.length) {
return res.status(404).json({ error: 'No images found for the provided IDs' });
}
const foundIds = images.map((i) => i.id);
const success = await ingestImageBulk({ images, lowPriority });
res.status(200).json({ success, count: images.length, imageIds: foundIds });
});
+5
View File
@@ -138,6 +138,11 @@ export enum FLIPT_FEATURE_FLAGS {
// OFF is the shipped default and means the pattern list is recorded but not enforced on these
// surfaces. The link-domain half throws either way — this flag has never governed it.
USER_CONTENT_PATTERN_ENFORCE = 'user-content-pattern-enforce',
// Submits image ingestion as one imageScanning step instead of wdTagging + mediaRating.
// DEFAULT-OFF — an unknown flag or unreachable Flipt keeps the two-step path. Evaluated
// with the imageId and no context, so ramp by percentage or boolean; a segment matches nothing.
IMAGE_INGESTION_IMAGE_SCANNING = 'image-ingestion-image-scanning',
}
// Flags exempt from caching: incident kill-switches where an operator expects a
+1 -1
View File
@@ -3,7 +3,7 @@ import { NewOrderRankType } from '~/shared/utils/prisma/enums';
/**
* Built-in fallback for `poolQuotas` when Redis config hasn't seeded one.
* Knight is the only rank that suffers from the SFW/NSFW priority misuse
* today (see `image-scan-result.ts` routing), so we default it to a 50/50
* today (see `addToNewOrderQueue` in `image-scan-pipeline.ts`), so we default it to a 50/50
* split between Knight1 (SFW) and Knight2 (NSFW). Acolyte/Templar are
* intentionally omitted to preserve their legacy sequential behavior.
*/
+6 -9
View File
@@ -15,12 +15,9 @@ import { decreaseDate } from '~/utils/date-helpers';
const IMAGE_SCANNING_ERROR_DELAY = 60 * 1; // 1 hour
const IMAGE_SCANNING_RETRY_LIMIT = 9;
// Hard per-image backstop for a single submit. The orchestrator submit is already
// bounded per-attempt (createImageIngestionRequest, ~15s AbortSignal), but this also
// covers any other external await inside ingestImage (Redis flag read, prompt lookup,
// the scanJobs UPDATE) so one hung submit ties up a single concurrency slot rather
// than the whole run. A fired timeout just fails that image → it stays queued and is
// retried on a later run.
// Hard per-image backstop. The orchestrator submit is bounded per attempt (~15s), but the
// Flipt flag read before it and the scanJobs UPDATE after it are not, so this keeps one hung
// image to one concurrency slot. A timed-out image stays queued for a later run.
const INGEST_IMAGE_TIMEOUT_MS = 60 * 1000;
// Per-run wall-clock budget. Once exceeded we stop STARTING new submits so the run
@@ -702,9 +699,9 @@ export const removeBlockedImages = createJob(
// DOES NOT RETRACT — the row is still hard-deleted here, exactly as before; only the shared
// object is left alone:
// • the scan pipeline's three block outcomes — the orchestrator content rating
// (`blockImageFromRating`), the prompt/text audit, and the moderation rule engine — in
// BOTH copies of that pipeline: `image-scan-result.service` and the legacy bodies in
// `api/webhooks/image-scan-result`. Automated, no moderator, no `ModActivity`.
// (`blockImageFromRating`), the prompt/text audit, and the moderation rule engine
// (`image-scan-result.service` / `image-scan-pipeline`). Automated, no moderator, no
// `ModActivity`.
// • the CSAM branch of `report.service` — reached from `report.create`, which is a
// `guardedProcedureAllowUnverifiedEmail`, so it is fired by ANY reporting user's report
// and not by a moderator reviewing one. This is the writer that would be most dangerous
+2 -2
View File
@@ -6,8 +6,8 @@
* a read back to a utf8 string before treating it as one. No-op when the value is
* already a string (single-node / dev) or null/undefined.
*
* Same coercion as the earlier per-site fixes (redis/queues.ts, image-scanner-flag.ts,
* metrics/base.metrics.ts). Prefer this helper for new call sites.
* Same coercion as the earlier per-site fixes (redis/queues.ts, metrics/base.metrics.ts).
* Prefer this helper for new call sites.
*/
export function decodeRedisString<T extends string | null | undefined>(value: T | Buffer): T {
return (Buffer.isBuffer(value) ? value.toString('utf8') : value) as T;
-5
View File
@@ -19,7 +19,6 @@ import {
getImagesByUserIdForModeration,
getImagesForModelVersionCache,
getMyImages,
ingestArticleCoverImages,
ingestImageById,
removeImageResource,
removeImageTechniques,
@@ -95,10 +94,6 @@ const isOwnerOrModerator = middleware(async ({ ctx, next, input = {} }) => {
// TODO.cleanup - remove unused router methods
export const imageRouter = router({
ingestArticleImages: protectedProcedure
.meta({ requiredScope: TokenScope.MediaWrite })
.input(z.array(z.object({ imageId: z.number(), articleId: z.number() })))
.mutation(({ input }) => ingestArticleCoverImages(input)),
moderate: moderatorProcedure.input(imageModerationSchema).mutation(moderateImageHandler),
delete: verifiedProcedure
.meta({ requiredScope: TokenScope.MediaDelete })
@@ -1,51 +0,0 @@
import { describe, expect, it } from 'vitest';
// Regression: `isImageScannerNewEnabled` reads a kill-switch from sysRedis and
// compared the reply against the literals '1'/'true'/'0'/'false'. The
// HA/Sentinel sysRedis returns a Buffer for BLOB_STRING replies, which matched
// none of the four literals → the function fell through and DESTRUCTIVELY
// overwrote the operator's '1' with 'false' (then returned false). The pure
// decision is extracted into its own `image-scanner-flag` module so the
// coercion + the seed-only-on-genuinely-unset behavior is testable without
// loading the 7.9K-line image.service module (which drags in Prisma/env/auth
// at import time and can't load under vitest). null return == "unset" == the
// only case the caller is allowed to default-seed.
import { parseScannerFlag } from '~/server/services/image-scanner-flag';
describe('parseScannerFlag — sysRedis Buffer-vs-string flag', () => {
it('Buffer("1") → true (was a destructive fall-through pre-fix)', () => {
expect(parseScannerFlag(Buffer.from('1', 'utf8'))).toBe(true);
});
it('Buffer("true") → true', () => {
expect(parseScannerFlag(Buffer.from('true', 'utf8'))).toBe(true);
});
it('Buffer("0") → false', () => {
expect(parseScannerFlag(Buffer.from('0', 'utf8'))).toBe(false);
});
it('Buffer("false") → false', () => {
expect(parseScannerFlag(Buffer.from('false', 'utf8'))).toBe(false);
});
it('string "1"/"true" → true (legacy single-pod, unchanged)', () => {
expect(parseScannerFlag('1')).toBe(true);
expect(parseScannerFlag('true')).toBe(true);
});
it('string "0"/"false" → false (legacy single-pod, unchanged)', () => {
expect(parseScannerFlag('0')).toBe(false);
expect(parseScannerFlag('false')).toBe(false);
});
it('null → null (genuinely unset → caller seeds the default)', () => {
expect(parseScannerFlag(null)).toBeNull();
});
it('an unrecognized value → null (treated as unset, not a destructive match)', () => {
expect(parseScannerFlag(Buffer.from('garbage', 'utf8'))).toBeNull();
expect(parseScannerFlag('garbage')).toBeNull();
});
});
@@ -0,0 +1,435 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
import {
classifyImageScanFailure,
ImageScanFailureClass,
} from '~/server/services/image-scan-failure';
const pipeline = vi.hoisted(() => ({
applyIngestionSideEffects: vi.fn(),
buildAndInsertScanTags: vi.fn(),
extractFailedSteps: vi.fn(),
loadImageForScan: vi.fn(),
logPerceptualHashMatch: vi.fn(),
markImageScanError: vi.fn(),
readJobFailureReason: vi.fn(),
resolveScanOutcome: vi.fn(),
sendIngestionSignal: vi.fn(),
}));
const { mockRecordAudit, mockRemoveJobQueue, mockFanOut } = vi.hoisted(() => ({
mockRecordAudit: vi.fn(),
mockRemoveJobQueue: vi.fn(),
mockFanOut: vi.fn(),
}));
// The pipeline imports the whole ingestion stack, and its stages run for real in
// image-scan-result.test.ts, so this suite pins only what this service passes them.
vi.mock('~/server/services/image-scan-pipeline', () => pipeline);
vi.mock('~/server/services/orchestrator/orchestrator.service', () => ({
computePerceptualHash: (hash?: string) => (hash ? BigInt(`0x${hash}`) : undefined),
}));
vi.mock('~/server/services/scanner-audit.service', () => ({
recordImageScanningResult: mockRecordAudit,
}));
vi.mock('~/server/services/job-queue.service', () => ({
removeImageScanJobQueue: mockRemoveJobQueue,
}));
vi.mock('~/server/utils/webhook-debounce', () => ({ fanOutArticleImageUpdates: mockFanOut }));
import {
IMAGE_SCANNING_LOG,
isImageScanningWorkflow,
parseImageScanningSteps,
processImageScanningWorkflow,
} from '~/server/services/image-scanning-result.service';
const scanStep = {
$type: 'imageScanning',
name: 'scan',
status: 'succeeded',
output: {
nsfwLevel: 'pg13',
tagging: { ran: true, tags: [{ tag: 'solo', category: 'general', score: 0.9 }] },
jointAgeClassification: { detections: [], minorDetected: false },
csam: null,
},
};
const failedScanStep = { ...scanStep, status: 'failed', output: undefined };
const hashStep = { $type: 'mediaHash', output: { hashes: { perceptual: '0A' } } };
const image = { id: 7, userId: 3, meta: { prompt: 'a prompt' } };
const deliver = (status: string, steps: unknown[]) =>
processImageScanningWorkflow({
workflowId: 'wf',
status,
steps,
imageId: 7,
articleImageScanning: true,
startedAt: 'start',
completedAt: 'end',
});
// Every failure must be recorded exactly once: a second write bumps retryCount again and
// overwrites the failure type the retry ceiling is chosen from.
const expectOneScanError = (failure: Record<string, unknown>) => {
expect(pipeline.markImageScanError).toHaveBeenCalledTimes(1);
expect(pipeline.markImageScanError).toHaveBeenCalledWith(expect.objectContaining(failure));
expect(pipeline.sendIngestionSignal).toHaveBeenCalledTimes(1);
expect(pipeline.sendIngestionSignal).toHaveBeenCalledWith({
imageId: 7,
userId: 3,
ingestion: 'Error',
log: IMAGE_SCANNING_LOG,
});
expect(mockFanOut).toHaveBeenCalledWith(7);
expect(pipeline.buildAndInsertScanTags).not.toHaveBeenCalled();
};
describe('processImageScanningWorkflow', () => {
beforeEach(() => {
vi.clearAllMocks();
pipeline.loadImageForScan.mockResolvedValue(image);
pipeline.resolveScanOutcome.mockResolvedValue({
ingestion: 'Scanned',
blockedFor: null,
reviewKey: null,
});
pipeline.markImageScanError.mockResolvedValue({ userId: 3, failureClass: 'unknown' });
pipeline.extractFailedSteps.mockReturnValue(['scan']);
pipeline.readJobFailureReason.mockResolvedValue('Failed to create container');
});
it('writes the scan through the shared pipeline without a hard block', async () => {
await deliver('succeeded', [scanStep, hashStep]);
expect(pipeline.logPerceptualHashMatch).toHaveBeenCalledWith({
imageId: 7,
pHash: 10n,
log: IMAGE_SCANNING_LOG,
});
expect(pipeline.buildAndInsertScanTags).toHaveBeenCalledWith({
imageId: 7,
wdTags: { solo: 0.9 },
ratingLevel: 'pg13',
prompt: 'a prompt',
});
expect(pipeline.resolveScanOutcome).toHaveBeenCalledWith({
image,
pHash: 10n,
workflowId: 'wf',
prompt: 'a prompt',
negativePrompt: undefined,
log: IMAGE_SCANNING_LOG,
});
expect(mockRemoveJobQueue).toHaveBeenCalledWith([7]);
expect(mockRecordAudit).toHaveBeenCalledWith(
expect.objectContaining({
workflowId: 'wf',
imageId: 7,
startedAt: 'start',
completedAt: 'end',
scan: expect.objectContaining({ nsfwLevel: 'pg13', csam: null }),
})
);
expect(pipeline.applyIngestionSideEffects).toHaveBeenCalledWith({
image,
outcome: expect.objectContaining({ ingestion: 'Scanned' }),
});
expect(mockFanOut).toHaveBeenCalledWith(7);
expect(pipeline.markImageScanError).not.toHaveBeenCalled();
expect(pipeline.sendIngestionSignal).toHaveBeenCalledTimes(1);
expect(pipeline.sendIngestionSignal).toHaveBeenCalledWith({
imageId: 7,
userId: 3,
ingestion: 'Scanned',
blockedFor: null,
log: IMAGE_SCANNING_LOG,
});
});
it('signals the outcome the pipeline resolved, blocked included', async () => {
pipeline.resolveScanOutcome.mockResolvedValue({
ingestion: 'Blocked',
blockedFor: 'moderated',
reviewKey: null,
});
await deliver('succeeded', [scanStep]);
expect(pipeline.sendIngestionSignal).toHaveBeenCalledWith({
imageId: 7,
userId: 3,
ingestion: 'Blocked',
blockedFor: 'moderated',
log: IMAGE_SCANNING_LOG,
});
});
it.each([
['failed', 'workflow-failed'],
['expired', 'expired'],
['canceled', 'canceled'],
])('marks a %s workflow as one %s error', async (status, failureType) => {
const steps = [failedScanStep];
await deliver(status, steps);
expect(pipeline.extractFailedSteps).toHaveBeenCalledWith(steps);
expect(pipeline.readJobFailureReason).toHaveBeenCalledWith('wf');
expect(pipeline.markImageScanError).toHaveBeenCalledWith({
workflowId: 'wf',
imageId: 7,
status,
failureType,
failedSteps: ['scan'],
reason: 'Failed to create container',
});
expectOneScanError({ failureType });
});
it('marks a succeeded workflow without tags as one unusable-result error', async () => {
await deliver('succeeded', [
{ ...scanStep, output: { ...scanStep.output, tagging: { status: 'skipped', ran: false } } },
]);
expectOneScanError({
failureType: 'unusable-result',
reason: expect.stringContaining('Tagging did not run'),
});
expect(pipeline.loadImageForScan).not.toHaveBeenCalled();
});
it('marks a processing failure as one processing-failed error', async () => {
pipeline.buildAndInsertScanTags.mockRejectedValueOnce(new Error('tag insert failed'));
await deliver('succeeded', [scanStep]);
expect(pipeline.markImageScanError).toHaveBeenCalledTimes(1);
expect(pipeline.markImageScanError).toHaveBeenCalledWith(
expect.objectContaining({ failureType: 'processing-failed', reason: 'tag insert failed' })
);
expect(pipeline.sendIngestionSignal).toHaveBeenCalledTimes(1);
expect(mockRecordAudit).not.toHaveBeenCalled();
});
it.each([
['succeeded', [scanStep]],
['failed', [failedScanStep]],
])('leaves articles alone when article scanning is off (%s)', async (status, steps) => {
await processImageScanningWorkflow({
workflowId: 'wf',
status,
steps,
imageId: 7,
articleImageScanning: false,
});
expect(mockFanOut).not.toHaveBeenCalled();
});
it('lets a deleted image through for the route to acknowledge', async () => {
pipeline.loadImageForScan.mockRejectedValueOnce(new Error('image not found: 7'));
await expect(deliver('succeeded', [scanStep])).rejects.toThrow('image not found');
expect(pipeline.markImageScanError).not.toHaveBeenCalled();
});
it('keeps the verdict when a side effect fails', async () => {
mockRemoveJobQueue.mockRejectedValueOnce(new Error('queue down'));
await deliver('succeeded', [scanStep]);
expect(pipeline.markImageScanError).not.toHaveBeenCalled();
expect(pipeline.sendIngestionSignal).toHaveBeenCalledWith(
expect.objectContaining({ ingestion: 'Scanned' })
);
});
});
const detection = (ageBand: string, isMinor: boolean) => ({
ageBand,
isMinor,
under18Probability: isMinor ? 0.8 : 0.1,
});
const scanOutput = (overrides: Record<string, unknown> = {}) => ({
nsfwLevel: 'r',
score: 0.8,
topK: [],
aiRecognition: { label: 'AI', score: 0.9 },
animeRecognition: { label: 'real', score: 0.7 },
tagging: {
status: 'ran',
ran: true,
tags: [
{ tag: 'solo', category: 'general', score: 0.9 },
{ tag: 'lowres', category: 'meta', score: 0.8 },
{ tag: 'worst quality', category: 'quality', score: 0.8 },
{ tag: 'original', category: 'copyright', score: 0.8 },
{ tag: 'hatsune miku', category: 'character', score: 0.8 },
{ tag: 'questionable', category: 'rating', score: 0.8 },
],
},
jointAgeClassification: {
status: 'ran',
ran: true,
detections: [detection('21-24', false)],
minorDetected: false,
},
csam: false,
...overrides,
});
const parserHashStep = {
$type: 'mediaHash',
output: { hashes: { perceptual: '6F51B11C49611E0E' } },
};
const parserScanStep = (output: unknown) => ({ $type: 'imageScanning', output });
const frames = (...outputs: unknown[]) => ({
$type: 'repeat',
output: { steps: outputs.map(parserScanStep) },
});
const untaggedFrame = scanOutput({ tagging: { status: 'skipped', ran: false, tags: [] } });
const unratedFrame = scanOutput({ nsfwLevel: 'na' });
describe('parseImageScanningSteps', () => {
it('reads an image scan, keeping only general tags', () => {
expect(parseImageScanningSteps([parserScanStep(scanOutput()), parserHashStep], 'wf')).toEqual({
nsfwLevel: 'r',
tags: { solo: 0.9 },
csam: false,
ageDetections: [detection('21-24', false)],
minorDetected: false,
aiRecognition: { label: 'AI', score: 0.9 },
animeRecognition: { label: 'real', score: 0.7 },
perceptualHash: '6F51B11C49611E0E',
});
});
const riskyFrame = scanOutput({
nsfwLevel: 'x',
csam: true,
aiRecognition: { label: 'AI', score: 0.9 },
animeRecognition: { label: 'anime', score: 0.95 },
tagging: { ran: true, tags: [{ tag: 'solo', category: 'general', score: 0.7 }] },
jointAgeClassification: {
ran: true,
detections: [detection('13-15', true)],
minorDetected: true,
},
});
const safeFrame = scanOutput({
nsfwLevel: 'pg',
csam: null,
aiRecognition: { label: 'Real', score: 0.6 },
animeRecognition: { label: 'real', score: 0.55 },
tagging: { ran: true, tags: [{ tag: 'solo', category: 'general', score: 0.6 }] },
jointAgeClassification: {
ran: true,
detections: [detection('21-24', false)],
minorDetected: false,
},
});
// Both orders, so a frame can only ever raise what the video reports.
it.each([
['riskiest frame first', [riskyFrame, safeFrame]],
['riskiest frame last', [safeFrame, riskyFrame]],
])('combines video frames by their highest values (%s)', (_, frameOutputs) => {
const scan = parseImageScanningSteps(
[{ $type: 'videoFrameExtraction', output: { frames: [] } }, frames(...frameOutputs)],
'wf'
);
expect(scan).toMatchObject({
nsfwLevel: 'x',
tags: { solo: 0.7 },
csam: true,
minorDetected: true,
aiRecognition: { label: 'AI', score: 0.9 },
animeRecognition: { label: 'anime', score: 0.95 },
});
expect(scan.ageDetections).toHaveLength(2);
expect(scan.ageDetections).toEqual(
expect.arrayContaining([detection('13-15', true), detection('21-24', false)])
);
});
it('reports csam as absent only when no frame reported it', () => {
const csamOf = (...values: unknown[]) =>
parseImageScanningSteps([frames(...values.map((csam) => scanOutput({ csam })))], 'wf').csam;
expect(csamOf(null, null)).toBeNull();
expect(csamOf(null, false)).toBe(false);
expect(csamOf(false, true, null)).toBe(true);
expect(csamOf(true, false)).toBe(true);
});
it.each([
['tagging did not run', [parserScanStep(untaggedFrame)], 'Tagging did not run'],
['the rating is unavailable', [parserScanStep(unratedFrame)], 'media rating unavailable'],
[
'the rating is one this app does not know',
[parserScanStep(scanOutput({ nsfwLevel: 'nc17' }))],
'media rating unavailable (nc17)',
],
['one frame was not tagged', [frames(scanOutput(), untaggedFrame)], 'Tagging did not run'],
['one frame has no rating', [frames(scanOutput(), unratedFrame)], 'media rating unavailable'],
[
'the scan step has no output',
[{ $type: 'imageScanning', status: 'failed' }],
'Missing imageScanning output',
],
[
'a frame has no output',
[
{
$type: 'repeat',
output: { steps: [parserScanStep(scanOutput()), { $type: 'imageScanning' }] },
},
],
'Missing imageScanning output',
],
['no frames were extracted', [frames()], 'Missing imageScanning output'],
])('throws when %s', (_, steps, message) => {
expect(() => parseImageScanningSteps(steps, 'wf')).toThrow(message);
});
// A permanent class stops retries after one attempt; none of these is a bad file.
it.each([[[parserScanStep(untaggedFrame)]], [[parserScanStep(unratedFrame)]], [[frames()]]])(
'throws an error the retry classifier keeps retrying',
(steps) => {
let reason = '';
try {
parseImageScanningSteps(steps, 'wf');
} catch (error) {
reason = (error as Error).message;
}
expect(reason).not.toBe('');
expect(
classifyImageScanFailure({ reason, failureType: 'unusable-result', failedSteps: [] })
).not.toBe(ImageScanFailureClass.Permanent);
}
);
});
describe('isImageScanningWorkflow', () => {
it.each([
['a finished image scan', [parserScanStep(scanOutput()), parserHashStep], true],
['a failed image scan with no output', [{ $type: 'imageScanning', status: 'failed' }], true],
['a finished video scan', [frames(scanOutput())], true],
[
'a video scan that never produced frames',
[{ $type: 'repeat', input: { template: { $type: 'imageScanning' } } }],
true,
],
[
'a legacy image scan',
[{ $type: 'wdTagging' }, { $type: 'mediaRating' }, parserHashStep],
false,
],
[
'a legacy video scan',
[
{ $type: 'repeat', input: { template: { $type: 'wdTagging' } } },
{ $type: 'repeat', input: { template: { $type: 'mediaRating' } } },
],
false,
],
])('%s → %s', (_, steps, expected) => {
expect(isImageScanningWorkflow(steps)).toBe(expected);
});
});
@@ -7,7 +7,7 @@ import path from 'path';
* must read the benign-stripped prompt.
*
* This exists because the fix for that bug was applied to one of two live copies. The image
* scan pipeline has two implementations `getTagsFromIncomingTags` in the webhook (legacy
* scan pipeline had two implementations `getTagsFromIncomingTags` in the webhook (legacy
* `TagSource.WD14`/`Clavata`/`Hive` bodies) and `processTags` in the service (bodies carrying
* `workflowId`, i.e. the orchestrator) and the webhook dispatches between them by body
* shape. Patching the one the audit named left the other writing a confidence-100 POI tag
@@ -33,8 +33,8 @@ import path from 'path';
* a new one written as an arrow needs the strip INLINE at the call to be checked properly.
* The escape also weakens as the enclosing function grows, since ANY `stripBenignPhrases`
* between the declaration and the call satisfies it. An entry whose call strips inline
* should therefore set `requireInline` and give up the escape entirely only the two scan
* files need it, because they shadow `prompt` with the stripped copy further up.
* should therefore set `requireInline` and give up the escape entirely only the scan
* pipeline needs it, because it shadows `prompt` with the stripped copy further up.
* 2. One other server-side `includesPoi` call reads a raw prompt and is deliberately NOT
* covered: `services/apps/shared-content-safety.ts` throws `SharedContentBlockedError('poi')`
* on shared app content, which is a hard legal reject rather than a moderation label.
@@ -53,13 +53,7 @@ const REPO_ROOT = path.resolve(__dirname, '../../../..');
*/
const GUARDED_FILES: { rel: string; consequence: string; requireInline?: boolean }[] = [
{
rel: 'src/pages/api/webhooks/image-scan-result.ts',
consequence:
'writes a confidence-100 POI tag that reaches the image search index, so a ' +
'moderator-whitelisted phrase would still flag the image',
},
{
rel: 'src/server/services/image-scan-result.service.ts',
rel: 'src/server/services/image-scan-pipeline.ts',
consequence:
'writes a confidence-100 POI tag that reaches the image search index, so a ' +
'moderator-whitelisted phrase would still flag the image',
@@ -68,7 +62,7 @@ const GUARDED_FILES: { rel: string; consequence: string; requireInline?: boolean
rel: 'src/server/services/orchestrator/orchestration-new.service.ts',
// The enclosing function is ~300 lines, so the scope escape below would be satisfied by ANY
// later `stripBenignPhrases` in it — negative-prompt stripping being the obvious one, since
// both scan files pair it with the prompt strip. This call needs no escape, so it gets none.
// the scan pipeline pairs it with the prompt strip. This call needs no escape, so it gets none.
requireInline: true,
consequence:
'refuses the generation outright for a viewer with `disablePoi`, so a ' +
@@ -0,0 +1,88 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
import type * as ClickhouseClient from '~/server/clickhouse/client';
const { mockInsert } = vi.hoisted(() => ({ mockInsert: vi.fn() }));
vi.mock('~/server/clickhouse/client', async (importOriginal) => ({
...(await importOriginal<typeof ClickhouseClient>()),
clickhouse: { insert: mockInsert },
}));
import { recordImageScanningResult } from '~/server/services/scanner-audit.service';
const detection = (ageBand: string, under18Probability: number, isMinor: boolean) =>
({ ageBand, under18Probability, isMinor } as never);
const scan = {
nsfwLevel: 'r' as const,
tags: {},
csam: false,
// The highest under-18 probability is not the first detection.
ageDetections: [detection('21-24', 0.1, false), detection('16-17', 0.7, true)],
minorDetected: true,
aiRecognition: { label: 'AI', score: 0.9 },
animeRecognition: { label: 'Real', score: 0.6 },
};
type Row = Record<string, unknown>;
const insertedRows = () => mockInsert.mock.calls[0][0].values as Row[];
const row = (label: string) => insertedRows().find((r) => r.label === label);
describe('recordImageScanningResult', () => {
beforeEach(() => mockInsert.mockReset().mockResolvedValue(undefined));
it('writes version 2 rows for the rating, csam, the joint age model and both recognizers', async () => {
await recordImageScanningResult({ workflowId: 'wf', imageId: 1, scan });
const rows = insertedRows();
expect(rows.map((r) => [r.label, r.labelValue, r.score, r.triggered])).toEqual([
['r', 'nsfw_level', 1, 1],
['csam', '', 0, 0],
['minor', '21-24, 16-17', 0.7, 1],
['ai', 'ai_recognition', 0.9, 1],
['real', 'anime_recognition', 0.6, 1],
]);
expect(rows.every((r) => r.version === '2' && r.modelVersion === '2')).toBe(true);
expect(rows.every((r) => r.scanner === 'image_ingestion')).toBe(true);
});
it('records a positive csam result as triggered', async () => {
await recordImageScanningResult({
workflowId: 'wf',
imageId: 1,
scan: { ...scan, csam: true },
});
expect(row('csam')).toMatchObject({ score: 1, triggered: 1 });
});
it('writes no csam row when the scanner did not report one', async () => {
await recordImageScanningResult({
workflowId: 'wf',
imageId: 1,
scan: { ...scan, csam: null },
});
expect(row('csam')).toBeUndefined();
});
it('does not trigger the minor row when only adults were detected', async () => {
await recordImageScanningResult({
workflowId: 'wf',
imageId: 1,
scan: {
...scan,
ageDetections: [detection('21-24', 0.1, false), detection('25+', 0.2, false)],
minorDetected: false,
},
});
expect(row('minor')).toMatchObject({ labelValue: '21-24, 25+', score: 0.2, triggered: 0 });
});
it('writes no minor row when nobody was detected', async () => {
await recordImageScanningResult({
workflowId: 'wf',
imageId: 1,
scan: { ...scan, ageDetections: [], minorDetected: false },
});
expect(row('minor')).toBeUndefined();
});
});
+1 -1
View File
@@ -1863,7 +1863,7 @@ export type ArticleTextModerationStatus = {
updatedAt: Date | null;
};
// `scanJobs.error` is stamped by `markImageScanError` (image-scan-result.service, scan
// `scanJobs.error` is stamped by `markImageScanError` (image-scan-pipeline, scan
// verdicts) and `markImageScanSubmitFailure` (image.service, submit rejections). Both
// carry the classifier's verdict (transient | permanent | unknown) plus the human reason,
// letting the scan-status UI render a class-aware cause.
@@ -1637,7 +1637,7 @@ export async function getImagesQueue({
// Per-rank pool weights drive a stratified fetch instead of the legacy
// strict-sequential drain. Knight1/Knight2/Knight3 are content-tier
// buckets in practice (see image-scan-result.ts), so reading Knight1
// buckets in practice (see addToNewOrderQueue in image-scan-pipeline.ts), so reading Knight1
// until full starved Knight2 (NSFW) entirely. Resolve weights from
// Redis (ops-tunable) with a built-in fallback; missing rank → legacy.
const rateLimitConfig = await getVotingRateLimitConfig();
+962
View File
@@ -0,0 +1,962 @@
import poiWords from '~/utils/metadata/lists/words-poi.json';
import { dbWrite } from '~/server/db/client';
import { clickhouse } from '~/server/clickhouse/client';
import { env } from '~/env/server';
import type { TagType } from '~/shared/utils/prisma/enums';
import {
ImageIngestionStatus,
ModerationRuleAction,
NewOrderRankType,
TagSource,
} from '~/shared/utils/prisma/enums';
import {
BlockedReason,
BlocklistType,
NotificationCategory,
NsfwLevel,
SearchIndexUpdateQueueAction,
SignalMessages,
} from '~/server/common/enums';
import { stripBenignPhrases } from '~/server/services/blocklist.service';
import {
auditMetaData,
getTagsFromPrompt,
includesInappropriate,
includesPoi,
} from '~/utils/metadata/audit';
import { getComputedTags, getConditionalTagsForReview } from '~/server/utils/tag-rules';
import { getTagRules } from '~/server/services/system-cache';
import { Prisma } from '@prisma/client';
import { insertTagsOnImageNew } from '~/server/services/tagsOnImageNew.service';
import { isDefined } from '~/utils/type-guards';
import { normalizeText } from '~/utils/normalize-text';
import { styleTags, tagsNeedingReview } from '~/libs/tags';
import { sfwBrowsingLevelsFlag } from '~/shared/constants/browsingLevel.constants';
import { createImageTagsForReview } from '~/server/services/image-review.service';
import {
tagIdsForImagesCache,
tagCacheByName,
userImageVideoCountCaches,
} from '~/server/redis/caches';
import type { RedisKeyTemplateSys } from '~/server/redis/client';
import { REDIS_SYS_KEYS, sysRedis } from '~/server/redis/client';
import { classifyImageScanFailure } from '~/server/services/image-scan-failure';
import type { MediaMetadata } from '~/server/schema/media.schema';
import { deleteUserProfilePictureCache } from '~/server/services/user.service';
import { bustCachesForPosts, updatePostNsfwLevel } from '~/server/services/post.service';
import {
queueComicsForPanelImage,
updateComicNsfwLevelsForImage,
updateModel3DNsfwLevelForThumbnailImage,
} from '~/server/services/nsfwLevels.service';
import { getImagesModRules, queueImageSearchIndexUpdate } from '~/server/services/image.service';
import { signalClient } from '~/utils/signal-client';
import { addImageToQueue } from '~/server/services/games/new-order.service';
import { logToAxiom } from '~/server/logging/client';
import { evaluateRules } from '~/server/utils/mod-rules';
import { createNotification } from '~/server/services/notification.service';
import { decreaseDate } from '~/utils/date-helpers';
import { withRetries } from '~/utils/errorHandling';
// Ingestion stages shared by the legacy (wdTagging + mediaRating) and imageScanning result handlers.
type NormalizedTag = {
name: string;
confidence: number;
source: TagSource;
};
type TagWithId = { id: number; name: string; nsfwLevel: number; type: TagType };
type ProcessedTag = {
source: TagSource;
confidence: number;
id: number;
name: string;
nsfwLevel: number;
type: TagType;
};
// TTL for a stashed job reason. Comfortably longer than the orchestrator's 10-min
// workflow expiry so the terminal workflow event can still read it.
const JOB_REASON_TTL_SECONDS = 15 * 60;
const jobReasonKey = (workflowId: string) =>
`${REDIS_SYS_KEYS.WEBHOOKS.IMAGE_SCAN_JOB_REASON}:${workflowId}` as RedisKeyTemplateSys;
/** Orchestrator job-level event shape we care about (a subset of WorkflowStepJobEvent). */
export type OrchestratorJobEvent = {
$type?: string;
workflowId?: string;
jobId?: string;
reason?: string | null;
};
// Stash the job's failure `reason` keyed by workflowId. No DB/orchestrator round-trip:
// just a short-lived Redis write so the terminal workflow event can classify.
//
// Correlation contract: job events fire before the terminal workflow event, so the
// reason is stashed by the time the workflow event reads it. Across a workflow's
// several jobs this is last-write-wins (fine — image scans typically have one failing
// job; the reason strings we classify on are equivalent). If the reason is missing
// (race, or a job event that never arrived) the failure classifies as Unknown, which
// retries conservatively under the bounded cap — safe by construction. The TTL bounds
// a stash that never gets a following workflow event so it can't linger.
export async function captureJobFailureReason(event: OrchestratorJobEvent) {
const workflowId = event.workflowId;
const reason = typeof event.reason === 'string' ? event.reason.trim() : '';
if (!workflowId || !reason) return;
await sysRedis
.set(jobReasonKey(workflowId), reason, { EX: JOB_REASON_TTL_SECONDS })
.catch(() => null);
}
export async function readJobFailureReason(workflowId: string): Promise<string | null> {
const reason = await sysRedis.get(jobReasonKey(workflowId)).catch(() => null);
if (reason) sysRedis.del(jobReasonKey(workflowId)).catch(() => null);
return reason ?? null;
}
/** Axiom `name`/`source` a pipeline's logs carry, so each pipeline can be queried alone. */
export type ScanLog = { name: string; source: string };
export const LEGACY_SCAN_LOG: ScanLog = {
name: 'image-scan-result',
source: 'image-scan-result.service',
};
/**
* Push the resolved ingestion state to the uploader's open editor. The editor renders
* Pending as an in-progress spinner, so any state that LEAVES Pending has to send
* Error included, retryable though it is or the card goes on claiming to analyze
* until the page is reloaded.
*/
export async function sendIngestionSignal({
imageId,
userId,
ingestion,
blockedFor,
log = LEGACY_SCAN_LOG,
}: {
imageId: number;
userId: number | null;
ingestion: ImageIngestionStatus;
blockedFor?: string | null;
log?: ScanLog;
}) {
if (!userId) return;
await signalClient
.send({
target: SignalMessages.ImageIngestionStatus,
data: { imageId, ingestion, blockedFor },
userId,
})
.catch((error) =>
logToAxiom(
{
name: log.name,
type: 'warning',
message: `signal send failed: ${
error instanceof Error ? error.message : 'Unknown error'
}`,
imageId,
source: log.source,
},
'webhooks'
).catch(() => null)
);
}
// Blocked-content detection
// --------------------------------------------------
async function getIsImageBlocked(hash: bigint) {
if (!env.BLOCKED_IMAGE_HASH_CHECK || !clickhouse) return false;
const client = clickhouse;
// $query has no retry of its own, and the observed failure is a dropped socket.
const rows = await withRetries(
() => client.$query<{ count: number }>`
SELECT cast(count() as int) as count
FROM blocked_images
WHERE bitCount(bitXor(hash, ${hash})) < 5 AND disabled = false
`,
2,
250
);
return (rows?.[0]?.count ?? 0) > 0;
}
export async function logPerceptualHashMatch({
imageId,
pHash,
log = LEGACY_SCAN_LOG,
}: {
imageId: number;
pHash: bigint;
log?: ScanLog;
}) {
// Nothing branches on this result, so an exhausted retry must not fail the webhook and
// discard a scan that already produced tags and a rating.
const pHashBlocked = await getIsImageBlocked(pHash).catch((error) => {
logToAxiom(
{
name: 'image-phash-match',
type: 'warning',
message: 'pHash blocklist check failed',
imageId,
error: error instanceof Error ? error.message : 'Unknown error',
source: log.source,
},
'webhooks'
).catch(() => null);
return false;
});
if (!pHashBlocked) return;
// blockedReason = 'Similar to blocked content';
logToAxiom(
{
name: 'image-phash-match',
type: 'info',
message: 'Image pHash matched a blocked image',
imageId,
pHash: pHash.toString(),
source: log.source,
},
'webhooks'
).catch(() => null);
}
/**
* Which orchestrator steps reported `failed`, by name (else `$type`). The workflow status
* carries no reason, so this is the only way to tell a scan-step failure from a `hash` failure.
*/
export function extractFailedSteps(steps: unknown[]): string[] {
return (steps as Array<{ name?: string; $type?: string; status?: string }>)
.filter((step) => step?.status === 'failed')
.map((step) => step.name ?? step.$type ?? 'unknown');
}
/**
* Flip an image to `Error`, increment its scan `retryCount`, and stamp a small
* `scanJobs.error = { status, failureType, failedSteps, reason, failureClass, at }`
* so a plain Postgres query can tell WHY a scan errored (and which step) without an
* orchestrator lookup and so the `ingest-images` cron can pick a retry ceiling
* from `failureClass`. retryCount ALWAYS increments: it's the absolute attempt
* count the per-class ceilings are applied against. Returns the new (post-increment)
* retryCount, the image's mediaType, and the computed failureClass; retryCount /
* mediaType are `null` when no row matched (e.g. the image was deleted between scan
* request and callback).
*/
export async function markImageScanError({
workflowId,
imageId,
status,
failureType,
failedSteps,
reason,
middleware,
}: {
workflowId: string;
imageId: number;
status: string;
failureType: string;
failedSteps: string[];
/** Human failure reason from the job-level callback, if captured. */
reason?: string | null;
/** Orchestrator middleware locus, if known. Not exposed on the v2 job event today. */
middleware?: string | null;
}): Promise<{
retryCount: number | null;
mediaType: string | null;
userId: number | null;
failureClass: string;
}> {
const failureClass = classifyImageScanFailure({ reason, failureType, middleware, failedSteps });
// undefined keys are dropped by JSON.stringify, so absent reason/middleware
// simply don't appear in the stored blob.
const errorJson = JSON.stringify({
status,
failureType,
failedSteps,
reason: reason ?? undefined,
middleware: middleware ?? undefined,
failureClass,
at: new Date().toISOString(),
});
const rows = await dbWrite.$queryRaw<
{ retryCount: number | null; mediaType: string | null; userId: number | null }[]
>`
UPDATE "Image"
SET
"ingestion" = ${ImageIngestionStatus.Error}::"ImageIngestionStatus",
"scanJobs" = jsonb_set(
jsonb_set(
jsonb_set(
COALESCE("scanJobs", '{}'),
'{retryCount}',
to_jsonb(COALESCE(("scanJobs"->>'retryCount')::int, 0) + 1)
),
'{workflowId}',
${JSON.stringify(workflowId)}::jsonb
),
'{error}',
${errorJson}::jsonb
)
WHERE id = ${imageId}
RETURNING ("scanJobs"->>'retryCount')::int as "retryCount", type as "mediaType",
"userId"
`;
return {
retryCount: rows[0]?.retryCount ?? null,
mediaType: rows[0]?.mediaType ?? null,
userId: rows[0]?.userId ?? null,
failureClass,
};
}
// Image loading
// --------------------------------------------------
export async function loadImageForScan(imageId: number) {
const image = await dbWrite.image.findUnique({
where: { id: imageId },
select: {
id: true,
createdAt: true,
scannedAt: true,
type: true,
userId: true,
meta: true,
metadata: true,
postId: true,
nsfwLevelLocked: true,
nsfwLevel: true,
ingestion: true,
},
});
if (!image) throw new Error(`image not found: ${imageId}`);
return image;
}
export type ScanImage = Awaited<ReturnType<typeof loadImageForScan>>;
// Tagging
// --------------------------------------------------
export async function buildAndInsertScanTags({
imageId,
wdTags,
ratingLevel,
prompt,
}: {
imageId: number;
wdTags: Record<string, number>;
ratingLevel: string;
prompt?: string;
}) {
const tagsWithSource = {
[TagSource.WD14]: wdTags,
[TagSource.SpineRating]: { [ratingLevel]: 100 },
};
const normalizedTags: NormalizedTag[] = Object.entries(tagsWithSource).flatMap(
([source, tagMap]) =>
Object.entries(tagMap).map(([name, confidence]) => {
if (source === TagSource.WD14) name = name.replace(/_/g, ' ');
return {
name,
confidence: Math.round(confidence * 100),
source: source as TagSource,
};
})
);
const tags = await processTags({ tags: normalizedTags, prompt });
await insertTagsOnImageNew(
tags.map((tag) => ({
imageId,
tagId: tag.id,
source: tag.source,
confidence: tag.confidence,
automated: true,
}))
);
}
async function processTags({
tags: normalized,
prompt,
}: {
tags: NormalizedTag[];
prompt?: string;
}): Promise<ProcessedTag[]> {
if (prompt) {
// Strip moderator-whitelisted phrases first, or a whitelisted proper noun is written as a
// confidence-100 POI tag that reaches the image search index.
const realPersonName = includesPoi(
// Normalized first, as both audit paths do. Stripping the raw text while the audit that
// runs next reads the normalized copy means two different alphabets decide what counts
// as whitelisted for the same prompt.
await stripBenignPhrases(normalizeText(prompt), BlocklistType.PromptBenignPhrase)
);
if (realPersonName) {
const tagName =
typeof realPersonName === 'object' ? realPersonName.matchedText : realPersonName;
normalized.push({
name: tagName.toLowerCase(),
confidence: 100,
source: TagSource.Computed,
});
}
// Detect tags from prompt
const promptTags = getTagsFromPrompt(prompt);
if (promptTags)
normalized.push(
...promptTags.map((name) => ({ name, confidence: 70, source: TagSource.Computed }))
);
}
// add computed tags
const computedTags = getComputedTags(
normalized.map((x) => x.name),
'WD14'
);
normalized.push(
...computedTags.map((name) => ({ name, confidence: 70, source: TagSource.Computed }))
);
// apply tag rules
const tagRules = await getTagRules();
for (const rule of tagRules) {
const match = normalized.find((x) => x.name === rule.toTag);
if (!match) continue;
if (rule.type === 'Replace') {
match.name = rule.fromTag;
} else if (rule.type === 'Append') {
normalized.push({ name: rule.fromTag, confidence: 70, source: TagSource.Computed });
}
}
// De-dupe incoming tags and keep tag with highest confidence
const tagMap: Record<string, NormalizedTag> = {};
for (const tag of normalized) {
if (!tagMap[tag.name] || tagMap[tag.name].confidence < tag.confidence) tagMap[tag.name] = tag;
}
const deduped: NormalizedTag[] = Object.values(tagMap);
const { found, missing } = await tagCacheByName.fetch(deduped.map((x) => x.name));
let queriedTags: TagWithId[] = [];
if (missing.length > 0) {
queriedTags = await dbWrite.tag.findMany({
where: { name: { in: missing } },
select: { id: true, name: true, nsfwLevel: true, type: true },
});
await tagCacheByName.setMany(queriedTags.map((data) => ({ key: data.name, data })));
}
const queriedNames = new Set(queriedTags.map((t) => t.name));
const tagsToCreate = missing.filter((name) => !queriedNames.has(name));
let createdTags: TagWithId[] = [];
if (tagsToCreate.length > 0) {
const tagsToInsert = deduped.filter((x) => tagsToCreate.includes(x.name));
// Raw SQL bypasses Prisma's @updatedAt stamp, and neither column has a DB
// default — both are NOT NULL, so omitting either raises 23502.
const now = new Date();
const values = tagsToInsert.map(
(tag) => Prisma.sql`(${tag.name}, ${now}, ARRAY['Image']::"TagTarget"[])`
);
createdTags = await dbWrite.$queryRaw<TagWithId[]>`
INSERT INTO "Tag" (name, "updatedAt", target)
VALUES ${Prisma.join(values)}
ON CONFLICT (name) DO UPDATE SET name = EXCLUDED.name
RETURNING id, name, "nsfwLevel", type
`;
await tagCacheByName.setMany(createdTags.map((data) => ({ key: data.name, data })));
}
const allTags = [...found.values(), ...queriedTags, ...createdTags]
.map((tag) => {
const match = normalized.find((x) => x.name === tag.name);
if (!match) return null;
return { ...tag, source: match.source, confidence: match.confidence };
})
.filter(isDefined);
return allTags;
}
// Outcome resolution (audit + moderation rules + persist)
// --------------------------------------------------
export type ScanOutcome = {
ingestion: ImageIngestionStatus;
blockedFor: string | null;
reviewKey: string | null;
/** Present only when the image was audited (i.e. not the `blocked` short-circuit). */
audit?: Awaited<ReturnType<typeof auditScanResults>>;
};
/**
* Resolve and persist the final state of the image row, returning the bits the
* post-update side effects need. Three terminal shapes:
* - `blocked` (legacy mediaRating hard block, already written by the caller): only
* stamp provenance re-auditing would recompute nsfwLevel/ingestion and silently
* un-block the image. A *prior* block is deliberately not sticky, so a moderator
* rescan can clear it. imageScanning never passes `blocked`, so on that path only
* `audit.blockedFor` and moderation rules re-block.
* - `audit.blockedFor` (prompt TOS/CSAM): block.
* - otherwise: Scanned, with moderation rules applied last so a rule Block/Hold
* takes precedence over the audit decision.
*/
export async function resolveScanOutcome({
image,
blocked,
pHash,
workflowId,
prompt,
negativePrompt,
log = LEGACY_SCAN_LOG,
}: {
image: ScanImage;
blocked?: { reason: string | null };
pHash?: bigint;
workflowId: string;
prompt?: string;
negativePrompt?: string;
log?: ScanLog;
}): Promise<ScanOutcome> {
const updatedAt = new Date();
if (blocked) {
await dbWrite.$executeRaw`
UPDATE "Image"
SET
"updatedAt" = ${updatedAt},
"pHash" = COALESCE(${pHash ?? null}, "pHash"),
"scanJobs" = jsonb_set(COALESCE("scanJobs", '{}'), '{workflowId}', ${JSON.stringify(
workflowId
)}::jsonb)
WHERE id = ${image.id}
`;
return {
ingestion: ImageIngestionStatus.Blocked,
blockedFor: blocked.reason,
reviewKey: null,
};
}
const audit = await auditScanResults({
imageId: image.id,
userId: image.userId,
prompt,
negativePrompt,
});
let reviewKey = audit.reviewKey ?? null;
const toUpdate: Prisma.ImageUpdateInput = { updatedAt, pHash };
// AI-generation verification is no longer a blocking gate (per operations
// 2026-05-11): nsfw images that we couldn't auto-verify as AI used to
// land in `Blocked + AiNotVerified`, but the false-positive rate didn't
// justify the friction. The remaining `audit.blockedFor` branch still
// catches hard violations (TOS / Moderated / CSAM) — everything else
// falls through to Scanned.
if (audit.blockedFor) {
toUpdate.ingestion = ImageIngestionStatus.Blocked;
toUpdate.blockedFor = audit.blockedFor;
toUpdate.nsfwLevel = NsfwLevel.Blocked;
} else {
toUpdate.ingestion = ImageIngestionStatus.Scanned;
toUpdate.needsReview = reviewKey;
toUpdate.minor = audit.minor;
toUpdate.poi = audit.poi;
toUpdate.blockedFor = null;
// Respect a manually-locked nsfw level — never overwrite it from the scan.
toUpdate.nsfwLevel = image.nsfwLevelLocked ? image.nsfwLevel : audit.nsfwLevel;
// scannedAt reassignment: always stamp the first scan; afterwards only
// re-stamp recent (<1 week old) non-Rescan images that haven't opted out
// via metadata.skipScannedAtReassignment. Older/rescanned images keep
// their original scannedAt.
const now = new Date();
if (!image.scannedAt) toUpdate.scannedAt = now;
else if (
!(image.metadata as any)?.skipScannedAtReassignment &&
image.ingestion !== 'Rescan' &&
new Date(image.createdAt).getTime() >= decreaseDate(now, 7, 'days').getTime()
)
toUpdate.scannedAt = now;
else toUpdate.scannedAt = image.scannedAt;
}
// Moderation rules can block the image, hold it for review, or annotate its
// metadata with the matched rule. Applied after the scan audit so a Block/Hold
// takes precedence over the audit's own decision.
const modRule = await evaluateImageModRules(image, audit.tags);
let metadataUpdate: Record<string, any> | undefined;
if (modRule) {
metadataUpdate = modRule.metadata;
if (modRule.ingestion) toUpdate.ingestion = modRule.ingestion;
if (modRule.nsfwLevel != null) toUpdate.nsfwLevel = modRule.nsfwLevel;
if (modRule.blockedFor) toUpdate.blockedFor = modRule.blockedFor;
if (modRule.needsReview !== undefined) {
toUpdate.needsReview = modRule.needsReview;
if (typeof modRule.needsReview === 'string') reviewKey = modRule.needsReview;
}
// Notify the user only when a rule auto-blocked the image (not on Hold).
if (modRule.ingestion === ImageIngestionStatus.Blocked) {
await notifyImageAutoBlocked({
imageId: image.id,
userId: image.userId,
reason: modRule.ruleReason,
log,
});
}
}
await dbWrite.$executeRaw`
UPDATE "Image"
SET
"updatedAt" = ${toUpdate.updatedAt},
"pHash" = ${pHash ?? null},
"ingestion" = ${toUpdate.ingestion as string}::"ImageIngestionStatus",
"blockedFor" = ${(toUpdate.blockedFor as string) ?? null},
"nsfwLevel" = ${toUpdate.nsfwLevel as number},
"needsReview" = ${(toUpdate.needsReview as string) ?? null},
"minor" = ${(toUpdate.minor as boolean) ?? false},
"poi" = ${(toUpdate.poi as boolean) ?? false},
"scannedAt" = ${(toUpdate.scannedAt as Date) ?? null},
"metadata" = COALESCE(${
metadataUpdate ? JSON.stringify(metadataUpdate) : null
}::jsonb, "metadata"),
"scanJobs" = jsonb_set(COALESCE("scanJobs", '{}'), '{workflowId}', ${JSON.stringify(
workflowId
)}::jsonb)
WHERE id = ${image.id}
`;
return {
ingestion: toUpdate.ingestion as ImageIngestionStatus,
blockedFor: (toUpdate.blockedFor as string) ?? null,
reviewKey,
audit,
};
}
async function auditScanResults(args: {
imageId: number;
userId: number;
prompt?: string;
negativePrompt?: string;
}) {
// Moderator-managed benign phrases (proper nouns / technical terms that coincidentally
// contain a detection token) are blanked up front so every downstream check — minor,
// poi, blockedFor — sees the same cleaned text. A benign phrase is innocent content, so
// it shouldn't feed any detector.
const [prompt, negativePrompt] = await Promise.all([
stripBenignPhrases(normalizeText(args.prompt), BlocklistType.PromptBenignPhrase),
stripBenignPhrases(normalizeText(args.negativePrompt), BlocklistType.NegativeBenignPhrase),
]);
const tags = await dbWrite.$queryRaw<
{ id: number; name: string; type: TagType; nsfwLevel: number; confidence: number }[]
>`
SELECT t.id, t.name, t."nsfwLevel", toi.confidence
FROM "TagsOnImageDetails" toi
JOIN "Tag" t ON t.id = toi."tagId"
WHERE toi."imageId" = ${args.imageId} AND toi.automated AND NOT toi.disabled
`;
const nsfwLevel = Math.max(...[...tags.map((x) => x.nsfwLevel), 0]);
const nsfw = nsfwLevel > sfwBrowsingLevelsFlag;
const minorTags = tags.filter((tag) => tagsNeedingReview.includes(tag.name.toLowerCase()));
const poiTags = tags.filter((tag) => poiWords.includes(tag.name.toLowerCase()));
const reviewTags = [
...tags.filter((tag) => tag.nsfwLevel === NsfwLevel.Blocked),
...getConditionalTagsForReview(tags, nsfwLevel),
];
const adultTags = tags.filter((tag) => tag.name === 'adult');
const cartoonTags = tags.filter((tag) => styleTags.includes(tag.name));
const tagReview = reviewTags.length > 0;
let poiReview = poiTags.length > 0;
let minorReview =
minorTags.length > 0 && adultTags.length === 0 && (cartoonTags.length === 0 || nsfw);
let newUserReview = false;
const inappropriate = includesInappropriate({ prompt, negativePrompt }, nsfw);
if (inappropriate === 'minor') minorReview = true;
if (inappropriate === 'poi') poiReview = true;
const associatedEntities = await getAssociatedEntities(args.imageId);
// Associated poi/minor resources only escalate to review when the image is
// nsfw — a sfw image with a poi/minor resource is still flagged (below) but
// not queued for moderator review.
if (associatedEntities.poi && nsfw) poiReview = true;
if (associatedEntities.minor && nsfw) minorReview = true;
if (!minorReview && !poiReview && !tagReview && nsfw) {
newUserReview = await getIsNewUser(args.userId);
}
const minor = minorTags.length > 0 || !!associatedEntities.minor;
const poi = poiTags.length > 0 || !!associatedEntities.poi || (!!prompt && !!includesPoi(prompt));
let reviewKey: string | undefined;
if (poiReview) reviewKey = 'poi';
else if (minorReview) reviewKey = 'minor';
else if (tagReview) reviewKey = 'tag';
else if (newUserReview) reviewKey = 'newUser';
let blockedFor: string | undefined;
if (nsfw && prompt) {
const auditResult = auditMetaData({ prompt }, nsfw);
if (!auditResult.success)
blockedFor = auditResult.blockedFor.join(',') ?? 'Failed audit, no explanation';
}
return {
tags,
nsfwLevel,
nsfw,
minorTags,
poiTags,
reviewTags,
tagReview,
poiReview,
minorReview,
newUserReview,
minor,
poi,
reviewKey,
blockedFor,
};
}
async function getAssociatedEntities(imageId: number) {
const [result] = await dbWrite.$queryRaw<
{ poi: boolean; minor: boolean; hasResource: boolean }[]
>`
WITH to_check AS (
-- Check based on associated resources
SELECT
SUM(IIF(m.poi, 1, 0)) > 0 "poi",
SUM(IIF(m.minor, 1, 0)) > 0 "minor",
true "hasResource"
FROM "ImageResourceNew" ir
JOIN "ModelVersion" mv ON ir."modelVersionId" = mv.id
JOIN "Model" m ON m.id = mv."modelId"
WHERE ir."imageId" = ${imageId}
UNION
-- Check based on associated bounties
SELECT
SUM(IIF(b.poi, 1, 0)) > 0 "poi",
false "minor",
false "hasResource"
FROM "Image" i
JOIN "ImageConnection" ic ON ic."imageId" = i.id
JOIN "Bounty" b ON ic."entityType" = 'Bounty' AND b.id = ic."entityId"
WHERE ic."imageId" = ${imageId}
UNION
-- Check based on associated bounty entries
SELECT
SUM(IIF(b.poi, 1, 0)) > 0 "poi",
false "minor",
false "hasResource"
FROM "Image" i
JOIN "ImageConnection" ic ON ic."imageId" = i.id
JOIN "BountyEntry" be ON ic."entityType" = 'BountyEntry' AND be.id = ic."entityId"
JOIN "Bounty" b ON b.id = be."bountyId"
WHERE ic."imageId" = ${imageId}
)
SELECT bool_or(poi) "poi", bool_or(minor) "minor", bool_or("hasResource") "hasResource" FROM to_check;
`;
return result;
}
async function getIsNewUser(userId: number) {
const [{ isNewUser }] =
(await dbWrite.$queryRaw<{ isNewUser: boolean }[]>`
SELECT is_new_user(CAST(${userId} AS INT)) "isNewUser";
`) ?? [];
return isNewUser;
}
// Moderation rules
// --------------------------------------------------
/**
* Evaluate the image moderation rules and return the row changes they imply
* (block / hold / metadata annotation). Pure of side effects the caller is
* responsible for persisting the changes and, on a Block, calling
* `notifyImageAutoBlocked`.
*/
async function evaluateImageModRules(
image: { meta: Prisma.JsonValue; metadata: Prisma.JsonValue },
tags: { name: string }[]
) {
const imageModRules = await getImagesModRules();
if (!imageModRules.length) return;
const tagNames = tags.map((x) => x.name);
const meta = (image.meta ?? {}) as Prisma.JsonObject;
const appliedRule = evaluateRules(imageModRules, { ...meta, tags: tagNames });
if (!appliedRule || appliedRule.action === ModerationRuleAction.Approve) return;
const result: {
metadata: Record<string, any>;
ruleReason?: string | null;
ingestion?: ImageIngestionStatus;
nsfwLevel?: NsfwLevel;
blockedFor?: string;
needsReview?: string | null;
} = {
metadata: {
...((image.metadata ?? {}) as Record<string, any>),
ruleId: appliedRule.id,
ruleReason: appliedRule.reason,
},
ruleReason: appliedRule.reason,
};
if (appliedRule.action === ModerationRuleAction.Block) {
result.ingestion = ImageIngestionStatus.Blocked;
result.nsfwLevel = NsfwLevel.Blocked;
result.blockedFor = BlockedReason.Moderated;
result.needsReview = null;
} else if (appliedRule.action === ModerationRuleAction.Hold) {
result.needsReview = 'modRule';
}
return result;
}
async function notifyImageAutoBlocked({
imageId,
userId,
reason,
log,
}: {
imageId: number;
userId: number;
reason?: string | null;
log: ScanLog;
}) {
await createNotification({
category: NotificationCategory.System,
key: `image-block:${imageId}`,
type: 'system-message',
userId,
details: {
message: `One of your images has been blocked due to a moderation rule violation${
reason ? ` by the following reason: ${reason}` : ''
}. If you believe this is a mistake, you can appeal this decision.`,
url: `/images/${imageId}`,
},
}).catch((error) =>
logToAxiom({
name: log.name,
type: 'error',
message: 'Could not create notification when blocking image',
data: {
imageId,
error: error.message,
cause: error.cause,
stack: error.stack,
},
})
);
}
// Post-update side effects
// --------------------------------------------------
export async function applyIngestionSideEffects({
image,
outcome,
}: {
image: ScanImage;
outcome: ScanOutcome;
}) {
// handle blocked image updates
if (outcome.ingestion === ImageIngestionStatus.Blocked) {
await queueImageSearchIndexUpdate({
ids: [image.id],
action: SearchIndexUpdateQueueAction.Delete,
});
// A previously-cached Blocked image can still satisfy the showcase query
// filters (needsReview IS NULL, nsfwLevel != 0) so drop it from the showcase.
if (image.postId) await bustCachesForPosts(image.postId);
await updateModel3DNsfwLevelForThumbnailImage({ imageId: image.id, postId: image.postId });
// If this image belongs to a comic panel, the parent project may
// have been search-indexed under the old (unblocked) state. Re-queue
// it so the next index pass re-evaluates visibility against the
// moderation gates in `comics.search-index.ts:WHERE`.
await queueComicsForPanelImage(image.id);
return;
}
// handle scanned image updates
if (outcome.ingestion === ImageIngestionStatus.Scanned) {
// Scanning is what makes an already-published image countable. Bust rather
// than refresh: this fires once per image, so a re-query here would be N
// identical counts for an N-image post.
await userImageVideoCountCaches.bust(image.userId);
await tagIdsForImagesCache.refresh(image.id);
if (
typeof image.metadata === 'object' &&
(image.metadata as MediaMetadata | undefined)?.profilePicture
) {
await deleteUserProfilePictureCache(image.userId);
}
if (image.postId) {
await updatePostNsfwLevel(image.postId);
// Without this, the showcase cache stays empty until its 24h TTL for any model version whose images hadn't scanned yet on first read.
await bustCachesForPosts(image.postId);
}
await updateModel3DNsfwLevelForThumbnailImage({ imageId: image.id, postId: image.postId });
await updateComicNsfwLevelsForImage(image.id);
// Refresh the comic project in the search index — even on a clean
// Scanned, `needsReview` may have been set, which the index treats
// as a visibility gate.
await queueComicsForPanelImage(image.id);
await queueImageSearchIndexUpdate({
ids: [image.id],
action: SearchIndexUpdateQueueAction.Update,
});
const { audit, reviewKey } = outcome;
if (audit) {
const tagsForReview = [...audit.poiTags, ...audit.minorTags, ...audit.reviewTags];
// Only persist review-tags when the image is actually queued for review
// (matches the legacy `if (reviewKey)` gate).
if (reviewKey && tagsForReview.length > 0) {
await createImageTagsForReview({
imageId: image.id,
tagIds: tagsForReview.map((x) => x.id),
});
}
if (!reviewKey && image.type === 'image') {
await addToNewOrderQueue({ imageId: image.id, nsfw: audit.nsfw });
}
}
}
}
const KONO_NSFW_SAMPLING_RATE = 0.3; // 30%
async function addToNewOrderQueue({ imageId, nsfw }: { imageId: number; nsfw: boolean }) {
let shouldAddToQueue = true;
let priority: 1 | 2 | 3 = 1;
const rankType = NewOrderRankType.Knight;
if (nsfw) {
priority = 2;
shouldAddToQueue = Math.random() < KONO_NSFW_SAMPLING_RATE;
}
if (shouldAddToQueue) {
await addImageToQueue({
imageIds: [imageId],
rankType,
priority,
});
}
}
File diff suppressed because it is too large Load Diff
-25
View File
@@ -1,25 +0,0 @@
/**
* Pure decision helper for the `image-scanner-new` sysRedis kill-switch.
*
* Lives in its own tiny module so the Buffer-coercion + seed-only-on-unset
* behavior is unit-testable without importing the ~7.9K-line image.service
* (which drags in Prisma/env/auth at load time and can't load under vitest).
*
* sysRedis.get is typed `string` but the HA/Sentinel client returns a Buffer
* for BLOB_STRING replies, matching none of the four literals pre-fix
* `isImageScannerNewEnabled` fell through and DESTRUCTIVELY overwrote the
* operator's '1' with 'false' (then returned false). Coerce once before
* comparing. See PR #2697/#2700 for the canonical Buffer-vs-string regression.
*
* Returns:
* - `true` for '1' / 'true'
* - `false` for '0' / 'false'
* - `null` for a genuinely-unset/unknown key the ONLY case in which the
* caller should default-seed 'false'.
*/
export function parseScannerFlag(raw: unknown): boolean | null {
const value = Buffer.isBuffer(raw) ? raw.toString('utf8') : raw;
if (value === '1' || value === 'true') return true;
if (value === '0' || value === 'false') return false;
return null;
}
@@ -0,0 +1,310 @@
import type {
ImageScanningJointAgeDetection,
ImageScanningOutput,
NsfwLevel,
} from '@civitai/orchestration-client';
import { orchestratorNsfwLevelMap } from '~/shared/constants/browsingLevel.constants';
import { logToAxiom } from '~/server/logging/client';
import {
applyIngestionSideEffects,
buildAndInsertScanTags,
extractFailedSteps,
loadImageForScan,
logPerceptualHashMatch,
markImageScanError,
readJobFailureReason,
resolveScanOutcome,
sendIngestionSignal,
type ScanLog,
} from '~/server/services/image-scan-pipeline';
import { removeImageScanJobQueue } from '~/server/services/job-queue.service';
import { computePerceptualHash } from '~/server/services/orchestrator/orchestrator.service';
import { recordImageScanningResult } from '~/server/services/scanner-audit.service';
import { fanOutArticleImageUpdates } from '~/server/utils/webhook-debounce';
import { ImageIngestionStatus } from '~/shared/utils/prisma/enums';
type RawRepeatStep = {
$type?: string;
input?: { template?: { $type?: string } };
output?: { steps?: Array<{ $type?: string }> };
};
/** Whether a workflow was submitted with imageScanning, whatever state its steps ended in. */
export function isImageScanningWorkflow(steps: unknown[]) {
return (steps as RawRepeatStep[]).some(
(step) =>
step.$type === 'imageScanning' ||
(step.$type === 'repeat' &&
(step.input?.template?.$type === 'imageScanning' ||
step.output?.steps?.[0]?.$type === 'imageScanning'))
);
}
type Recognition = { label: string; score: number };
/** One image's scan, or a video's frames combined into one. */
export type ImageScanningResult = {
nsfwLevel: NsfwLevel;
tags: Record<string, number>;
csam: boolean | null;
ageDetections: ImageScanningJointAgeDetection[];
minorDetected: boolean;
aiRecognition?: Recognition;
animeRecognition?: Recognition;
perceptualHash?: string;
};
type RawStep = { $type?: string; name?: string; status?: string; output?: unknown };
// wdTagging only ever returned general tags; the other categories would write tags no image had before.
const INGESTED_TAG_CATEGORIES = new Set(['general']);
function readScan(output: ImageScanningOutput, workflowId: string): ImageScanningResult {
const { tagging, jointAgeClassification } = output;
// An unmapped level ('na', or one added upstream) would write a rating with no NSFW level.
// Not "invalid media…": classifyImageScanFailure reads that as permanent, and this is recoverable.
if (orchestratorNsfwLevelMap[output.nsfwLevel] === undefined)
throw new Error(`media rating unavailable (${output.nsfwLevel}) for workflow: ${workflowId}`);
if (!tagging?.ran)
throw new Error(`Incomplete workflow: ${workflowId}. Tagging did not run (${tagging?.status})`);
const tags: Record<string, number> = {};
for (const { tag, category, score } of tagging.tags) {
if (INGESTED_TAG_CATEGORIES.has(category)) tags[tag] = Math.max(score, tags[tag] ?? 0);
}
return {
nsfwLevel: output.nsfwLevel,
tags,
csam: output.csam ?? null,
ageDetections: jointAgeClassification?.detections ?? [],
minorDetected: !!jointAgeClassification?.minorDetected,
aiRecognition: output.aiRecognition,
animeRecognition: output.animeRecognition,
};
}
const higher = (a?: Recognition, b?: Recognition) => (!a || (b && b.score > a.score) ? b : a);
// Every result is kept (a max or an OR per field), so a frame can only raise what the video reports.
function combineFrames(frames: ImageScanningResult[]): ImageScanningResult {
return frames.reduce((acc, frame) => {
const tags = { ...acc.tags };
for (const [tag, score] of Object.entries(frame.tags))
tags[tag] = Math.max(score, tags[tag] ?? 0);
return {
nsfwLevel:
orchestratorNsfwLevelMap[frame.nsfwLevel] > orchestratorNsfwLevelMap[acc.nsfwLevel]
? frame.nsfwLevel
: acc.nsfwLevel,
tags,
csam: acc.csam || frame.csam ? true : acc.csam ?? frame.csam,
ageDetections: [...acc.ageDetections, ...frame.ageDetections],
minorDetected: acc.minorDetected || frame.minorDetected,
aiRecognition: higher(acc.aiRecognition, frame.aiRecognition),
animeRecognition: higher(acc.animeRecognition, frame.animeRecognition),
};
});
}
/**
* Reads a succeeded imageScanning ingestion workflow: one `imageScanning` step for an image,
* or a `repeat` of them over extracted frames for a video, plus the `mediaHash` step.
* Throws when the workflow carries nothing usable.
*/
export function parseImageScanningSteps(workflowSteps: unknown[], workflowId: string) {
const steps = workflowSteps as RawStep[];
const perceptualHash = (
steps.find((step) => step.$type === 'mediaHash')?.output as
| { hashes?: { perceptual?: string } }
| undefined
)?.hashes?.perceptual;
const single = steps.find((step) => step.$type === 'imageScanning')?.output;
if (single) return { ...readScan(single as ImageScanningOutput, workflowId), perceptualHash };
const frames = (
steps.find(
(step) =>
step.$type === 'repeat' &&
(step.output as { steps?: RawStep[] } | undefined)?.steps?.[0]?.$type === 'imageScanning'
)?.output as { steps: RawStep[] } | undefined
)?.steps;
if (!frames?.length || !frames.every((frame) => frame.output))
throw new Error(`Incomplete workflow: ${workflowId}. Missing imageScanning output`);
return {
...combineFrames(
frames.map((frame) => readScan(frame.output as ImageScanningOutput, workflowId))
),
perceptualHash,
};
}
export const IMAGE_SCANNING_LOG: ScanLog = {
name: 'image-scanning-result',
source: 'image-scanning-result.service',
};
type ScanErrorInput = Parameters<typeof markImageScanError>[0];
async function failScan(
input: ScanErrorInput & {
articleImageScanning: boolean;
logType: 'warning' | 'error';
message: string;
stack?: string;
}
) {
const { articleImageScanning, logType, message, stack, ...error } = input;
const { retryCount, mediaType, userId, failureClass } = await markImageScanError(error);
logToAxiom(
{
name: IMAGE_SCANNING_LOG.name,
type: logType,
message,
source: IMAGE_SCANNING_LOG.source,
stack,
...error,
failureClass,
mediaType,
retryCount,
},
'webhooks'
).catch(() => null);
await sendIngestionSignal({
imageId: error.imageId,
userId,
ingestion: ImageIngestionStatus.Error,
log: IMAGE_SCANNING_LOG,
});
if (articleImageScanning) await fanOutArticleImageUpdates(error.imageId);
}
/** Core image scan processing for imageScanning workflows. */
export async function processImageScanningWorkflow({
workflowId,
status,
steps,
imageId,
articleImageScanning = false,
startedAt,
completedAt,
}: {
workflowId: string;
status: string;
steps: unknown[];
imageId: number;
articleImageScanning?: boolean;
startedAt?: Date | string | null;
completedAt?: Date | string | null;
}) {
if (status !== 'succeeded') {
await failScan({
workflowId,
imageId,
status,
failureType:
status === 'expired' ? 'expired' : status === 'canceled' ? 'canceled' : 'workflow-failed',
failedSteps: extractFailedSteps(steps),
reason: await readJobFailureReason(workflowId),
articleImageScanning,
logType: 'warning',
message: `workflow not succeeded: ${status}`,
});
return;
}
let scan: ReturnType<typeof parseImageScanningSteps>;
try {
scan = parseImageScanningSteps(steps, workflowId);
} catch (error) {
await failScan({
workflowId,
imageId,
status,
failureType: 'unusable-result',
failedSteps: extractFailedSteps(steps),
reason: error instanceof Error ? error.message : 'Unknown error',
articleImageScanning,
logType: 'warning',
message: 'succeeded workflow had no usable result',
});
return;
}
let image: Awaited<ReturnType<typeof loadImageForScan>>;
let outcome: Awaited<ReturnType<typeof resolveScanOutcome>>;
try {
const pHash = computePerceptualHash(scan.perceptualHash);
if (pHash) await logPerceptualHashMatch({ imageId, pHash, log: IMAGE_SCANNING_LOG });
image = await loadImageForScan(imageId);
const { prompt, negativePrompt } = (image.meta ?? {}) as {
prompt?: string;
negativePrompt?: string;
};
await buildAndInsertScanTags({
imageId,
wdTags: scan.tags,
ratingLevel: scan.nsfwLevel,
prompt,
});
outcome = await resolveScanOutcome({
image,
pHash,
workflowId,
prompt,
negativePrompt,
log: IMAGE_SCANNING_LOG,
});
} catch (error) {
// Deleted between submit and callback — no row to mark. The route ACKs it.
if (error instanceof Error && error.message.startsWith('image not found')) throw error;
await failScan({
workflowId,
imageId,
status,
failureType: 'processing-failed',
failedSteps: extractFailedSteps(steps),
reason: error instanceof Error ? error.message : 'Unknown error',
articleImageScanning,
logType: 'error',
message: 'failed to process a succeeded workflow',
stack: error instanceof Error ? error.stack : undefined,
});
return;
}
// The verdict is persisted; throwing now would 400 a finished webhook and make the
// orchestrator re-run it.
try {
await removeImageScanJobQueue([imageId]);
await recordImageScanningResult({ workflowId, imageId, scan, startedAt, completedAt });
if (articleImageScanning) await fanOutArticleImageUpdates(imageId);
await applyIngestionSideEffects({ image, outcome });
} catch (error) {
logToAxiom(
{
name: IMAGE_SCANNING_LOG.name,
type: 'error',
message: 'side effects failed after the scan verdict was persisted',
source: IMAGE_SCANNING_LOG.source,
stack: error instanceof Error ? error.stack : undefined,
reason: error instanceof Error ? error.message : 'Unknown error',
imageId,
workflowId,
ingestion: outcome.ingestion,
},
'webhooks'
).catch(() => null);
}
await sendIngestionSignal({
imageId,
userId: image.userId,
ingestion: outcome.ingestion,
blockedFor: outcome.blockedFor,
log: IMAGE_SCANNING_LOG,
});
}
+56 -275
View File
@@ -37,7 +37,6 @@ import {
import { imageReviewedSql } from '~/server/common/image-visibility';
import {
BlockedReason,
ImageScanType,
ImageSort,
NotificationCategory,
NsfwLevel,
@@ -140,11 +139,7 @@ import type {
} from '~/server/schema/image.schema';
import { imageMetaOutput, ingestImageSchema } from '~/server/schema/image.schema';
import type { ImageMetadata, VideoMetadata } from '~/server/schema/media.schema';
import {
articlesSearchIndex,
imagesMetricsSearchIndex,
imagesSearchIndex,
} from '~/server/search-index';
import { imagesMetricsSearchIndex, imagesSearchIndex } from '~/server/search-index';
import type {
ImageMetricsSearchIndexRecord,
MetricsImageFilterableAttribute,
@@ -172,7 +167,6 @@ import {
} from '~/server/services/model3d.service';
import { addImageToQueue } from '~/server/services/games/new-order.service';
import { upsertImageFlag } from '~/server/services/image-flag.service';
import { parseScannerFlag } from '~/server/services/image-scanner-flag';
import {
deleteImagTagsForReviewByImageIds,
getImagTagsForReviewByImageIds,
@@ -203,7 +197,6 @@ import {
throwInternalServerError,
throwNotFoundError,
} from '~/server/utils/errorHandling';
import { fetchTimeoutSignal } from '~/server/utils/fetch-timeout';
import type { RuleDefinition } from '~/server/utils/mod-rules';
import { getCursor } from '~/server/utils/pagination-helpers';
import {
@@ -267,7 +260,10 @@ import type {
} from '../../../event-engine-common/types/package-stubs';
import type { FeedQueryInput } from '../../../event-engine-common/feeds/types';
import type { ImageQueryInput } from '../../../event-engine-common/types/image-feed-types';
import { createImageIngestionRequest } from '~/server/services/orchestrator/orchestrator.service';
import {
createImageIngestionRequest,
imageIngestionLogName,
} from '~/server/services/orchestrator/orchestrator.service';
import { getGenerationDisplayKeys } from '~/server/services/orchestrator/legacy-metadata-mapper';
import {
sanitizeProvenance,
@@ -1154,29 +1150,6 @@ export const getImageById = async ({ id }: GetByIdInput) => {
});
};
/**
* Runtime toggle for the new image ingestion path (createImageIngestionRequest
* with the expanded mediaRating step). Reads from Redis so ops can flip
* without a deploy. Accepts '1'/'true' / '0'/'false' as string values.
*
* If the key doesn't exist (first request after deploy), seeds it to 'false'
* so the toggle is discoverable in Redis and explicitly off by default.
* Operators set the key to '1' to enable.
*/
async function isImageScannerNewEnabled(): Promise<boolean> {
// The HA/Sentinel sysRedis returns a Buffer for BLOB_STRING replies, which
// matched none of the literals pre-fix → fell through and destructively
// overwrote the operator's '1' with 'false'. parseScannerFlag coerces the
// Buffer first and returns null ONLY for a genuinely-unset/unknown key, so
// the seed below now fires only in its intended default-seeding case.
// See PR #2697/#2700 for the canonical Buffer-vs-string regression.
const raw = await sysRedis.get(REDIS_SYS_KEYS.SYSTEM.IMAGE_SCANNER_NEW);
const parsed = parseScannerFlag(raw);
if (parsed !== null) return parsed;
await sysRedis.set(REDIS_SYS_KEYS.SYSTEM.IMAGE_SCANNER_NEW, 'false');
return false;
}
export const ingestImageById = async ({ id }: GetByIdInput) => {
const images = await dbWrite.$queryRaw<IngestImageInput[]>`
SELECT id, url, type, width, height, meta->>'prompt' as prompt
@@ -1198,16 +1171,6 @@ export const ingestImageById = async ({ id }: GetByIdInput) => {
return await ingestImage({ image: images[0] });
};
// const scanner = env.EXTERNAL_IMAGE_SCANNER;
// const clavataScan = env.CLAVATA_SCAN;
export const imageScanTypes: ImageScanType[] = [
ImageScanType.WD14,
// ImageScanType.Hash,
// ImageScanType.Clavata,
// ImageScanType.Hive,
ImageScanType.SpineRating,
];
function extractSubmitErrorMessage(error: unknown): string | null {
if (!error) return null;
if (typeof error === 'string') return error;
@@ -1292,233 +1255,64 @@ export const ingestImage = async ({
const scanRequestedAt = new Date();
const dbClient = tx ?? dbWrite;
// if (!isProd || !env.IMAGE_SCANNING_ENDPOINT) {
// console.log('skipping image ingestion');
// const updated = await dbClient.image.update({
// where: { id: image.id },
// select: { postId: true },
// data: {
// scanRequestedAt,
// scannedAt: scanRequestedAt,
// ingestion: ImageIngestionStatus.Scanned,
// nsfwLevel: NsfwLevel.PG,
// },
// });
// // Update post NSFW level
// if (updated.postId) await updatePostNsfwLevel(updated.postId);
// return true;
// }
const parsedImage = ingestImageSchema.safeParse(image);
if (!parsedImage.success) throw new Error('Failed to parse image data');
const { url, id, type, width, height } = parsedImage.data;
const { url, id, type } = parsedImage.data;
const callbackUrl =
env.IMAGE_SCANNING_CALLBACK ??
`${env.NEXTAUTH_URL}/api/webhooks/image-scan-result?token=${env.WEBHOOK_TOKEN}`;
if (!image.prompt) {
const { prompt } = await dbClient.$queryRaw<{ prompt?: string }>`
SELECT meta->>'prompt' as prompt FROM "Image" WHERE id = ${id}
`;
image.prompt = prompt;
}
if (await isImageScannerNewEnabled()) {
const {
data: workflowResponse,
error: submitError,
status: submitStatus,
} = await createImageIngestionRequest({
imageId: id,
url,
type,
callbackUrl,
priority: lowPriority ? 'low' : undefined,
});
if (!workflowResponse) {
imageScanSubmittedCounter.inc({ lane: 'new', result: 'failed' });
const failureClass = await markImageScanSubmitFailure({
dbClient,
imageId: id,
status: submitStatus,
error: submitError,
});
// The orchestrator submit already logs the transient failure in
// createImageIngestionRequest, but from here it's otherwise a silent
// `return false` — surface it at the dispatch layer so the failure is
// attributable to a specific image + media type.
logToAxiom({
name: 'image-ingestion',
type: 'error',
reason: 'no-workflow-response',
failureType: 'send-fail',
failureClass,
responseStatus: submitStatus,
imageId: id,
mediaType: type,
}).catch(() => null);
return false;
}
const scanJobsJson = JSON.stringify({ workflowId: workflowResponse.id });
await dbClient.$executeRaw`
UPDATE "Image"
SET
"scanRequestedAt" = ${scanRequestedAt},
"scanJobs" = CASE
WHEN "scanJobs" IS NOT NULL AND "scanJobs" ? 'retryCount' THEN
${scanJobsJson}::jsonb || jsonb_build_object('retryCount', ("scanJobs"->'retryCount'))
ELSE
${scanJobsJson}::jsonb
END
WHERE id = ${id}
`;
imageScanSubmittedCounter.inc({ lane: 'new', result: 'success' });
return true;
}
let scanUrl = `${env.IMAGE_SCANNING_ENDPOINT}/enqueue`;
if (lowPriority) scanUrl += '?lowpri=true';
const response = await fetch(scanUrl, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
signal: fetchTimeoutSignal(60_000),
body: JSON.stringify({
imageId: id,
imageKey: url,
type,
width,
height,
prompt: image.prompt,
// wait: true,
scans: imageScanTypes,
callbackUrl,
movieRatingModel: env.IMAGE_SCANNING_MODEL,
}),
const {
data: workflowResponse,
error: submitError,
status: submitStatus,
useImageScanning,
} = await createImageIngestionRequest({
imageId: id,
url,
type,
callbackUrl,
priority: lowPriority ? 'low' : undefined,
});
if (response.status === 202) {
const scanJobs = (await response.json().catch(() => Prisma.JsonNull)) as
| { jobId: string }
| typeof Prisma.JsonNull;
// Convert scanJobs to JSON string for raw SQL, preserving existing retryCount if it exists
const scanJobsJson = scanJobs === Prisma.JsonNull ? null : JSON.stringify(scanJobs);
if (scanJobsJson) {
await dbClient.$executeRaw`
UPDATE "Image"
SET
"scanRequestedAt" = ${scanRequestedAt},
"scanJobs" = CASE
WHEN "scanJobs" IS NOT NULL AND "scanJobs" ? 'retryCount' THEN
${scanJobsJson}::jsonb || jsonb_build_object('retryCount', ("scanJobs"->'retryCount'))
ELSE
${scanJobsJson}::jsonb
END
WHERE id = ${id}
`;
} else {
await dbClient.$executeRaw`
UPDATE "Image"
SET "scanRequestedAt" = ${scanRequestedAt}
WHERE id = ${id}
`;
}
imageScanSubmittedCounter.inc({ lane: 'legacy', result: 'success' });
return true;
} else {
await logToAxiom({
name: 'image-ingestion',
type: 'error',
const lane = useImageScanning ? 'imageScanning' : 'new';
if (!workflowResponse) {
imageScanSubmittedCounter.inc({ lane, result: 'failed' });
const failureClass = await markImageScanSubmitFailure({
dbClient,
imageId: id,
url,
responseStatus: response.status,
status: submitStatus,
error: submitError,
});
imageScanSubmittedCounter.inc({ lane: 'legacy', result: 'failed' });
// createImageIngestionRequest's own log carries neither mediaType nor failureClass.
logToAxiom({
name: imageIngestionLogName(useImageScanning),
type: 'error',
reason: 'no-workflow-response',
failureType: 'send-fail',
failureClass,
responseStatus: submitStatus,
imageId: id,
mediaType: type,
}).catch(() => null);
return false;
}
};
export const ingestImageBulk = async ({
images,
tx,
lowPriority = true,
scans,
}: {
images: IngestImageInput[];
tx?: Prisma.TransactionClient;
lowPriority?: boolean;
scans?: ImageScanType[];
}): Promise<boolean> => {
if (!env.IMAGE_SCANNING_ENDPOINT)
throw new Error('missing IMAGE_SCANNING_ENDPOINT environment variable');
const callbackUrl = env.IMAGE_SCANNING_CALLBACK;
const scanRequestedAt = new Date();
const imageIds = images.map(({ id }) => id);
const dbClient = tx ?? dbWrite;
if (!imageIds.length) return false;
// TODO.articleImageScan: uncomment when ready to enable image scanning for articles
// if (!isProd || !callbackUrl) {
// console.log('skip ingest');
// await dbClient.image.updateMany({
// where: { id: { in: imageIds } },
// data: {
// scanRequestedAt,
// scannedAt: scanRequestedAt,
// ingestion: ImageIngestionStatus.Scanned,
// nsfwLevel: NsfwLevel.PG,
// },
// });
// return true;
// }
const needsPrompts = !images.some((x) => x.prompt);
if (needsPrompts) {
const prompts = await dbClient.$queryRaw<{ id: number; prompt?: string }[]>`
SELECT id, meta->>'prompt' as prompt FROM "Image" WHERE id IN (${Prisma.join(imageIds)})
const scanJobsJson = JSON.stringify({ workflowId: workflowResponse.id });
await dbClient.$executeRaw`
UPDATE "Image"
SET
"scanRequestedAt" = ${scanRequestedAt},
"scanJobs" = CASE
WHEN "scanJobs" IS NOT NULL AND "scanJobs" ? 'retryCount' THEN
${scanJobsJson}::jsonb || jsonb_build_object('retryCount', ("scanJobs"->'retryCount'))
ELSE
${scanJobsJson}::jsonb
END
WHERE id = ${id}
`;
const promptMap = Object.fromEntries(prompts.map((x) => [x.id, x.prompt]));
for (const image of images) image.prompt = promptMap[image.id];
}
const response = await fetch(
env.IMAGE_SCANNING_ENDPOINT + `/enqueue-bulk?lowpri=${lowPriority}`,
{
method: 'POST',
headers: { 'Content-Type': 'application/json' },
signal: fetchTimeoutSignal(60_000),
body: JSON.stringify(
images.map((image) => ({
imageId: image.id,
imageKey: image.url,
type: image.type,
width: image.width,
height: image.height,
prompt: image.prompt,
scans: scans ?? imageScanTypes,
callbackUrl,
}))
),
}
);
if (response.status === 202) {
await dbClient.image.updateMany({
where: { id: { in: imageIds } },
data: { scanRequestedAt },
});
return true;
}
return false;
imageScanSubmittedCounter.inc({ lane, result: 'success' });
return true;
};
export function enqueueImageIngestion({
@@ -6986,20 +6780,6 @@ export async function reportCsamImages({
await bulkSetReportStatus({ ids: reportIds, status: ReportStatus.Actioned, userId: user.id, ip });
}
export async function ingestArticleCoverImages(array: { imageId: number; articleId: number }[]) {
const imageIds = array.map((x) => x.imageId);
const images = await dbRead.image.findMany({
where: { id: { in: imageIds } },
select: { id: true, url: true, height: true, width: true },
});
await articlesSearchIndex.queueUpdate(
array.map((x) => ({ id: x.articleId, action: SearchIndexUpdateQueueAction.Update }))
);
await ingestImageBulk({ images, lowPriority: true });
}
export async function updateImageNsfwLevel({
id,
nsfwLevel,
@@ -7301,15 +7081,16 @@ export async function addImageTools({
select: {
id: true,
url: true,
type: true,
},
});
if (updated.length > 0) {
await ingestImageBulk({
images: updated,
lowPriority: true,
});
}
enqueueImageIngestion({
images: updated,
name: 'add-image-tools',
userId: user.id,
lowPriority: true,
});
for (const { imageId } of data) {
purgeImageGenerationDataCache(imageId);
@@ -1,4 +1,5 @@
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
import type * as FliptClient from '~/server/flipt/client';
const {
mockDbWrite,
@@ -7,6 +8,7 @@ const {
mockStringifyAIR,
mockResolveDownloadUrl,
mockIsProd,
mockIsFlipt,
} = vi.hoisted(() => ({
mockDbWrite: {
modelFile: { update: vi.fn() },
@@ -19,6 +21,7 @@ const {
// don't regress; failure-path tests reset to mockRejectedValue.
mockResolveDownloadUrl: vi.fn().mockResolvedValue({ url: 'https://cdn.example/file' }),
mockIsProd: { value: true },
mockIsFlipt: vi.fn(),
}));
vi.mock('~/server/db/client', () => ({ dbWrite: mockDbWrite }));
@@ -78,6 +81,11 @@ vi.mock('~/env/server', () => ({
},
}));
vi.mock('~/server/flipt/client', async (importOriginal) => ({
...(await importOriginal<typeof FliptClient>()),
isFlipt: mockIsFlipt,
}));
// orchestrator.service.ts pulls in edge-url which validates
// NEXT_PUBLIC_* env vars at import-time. Stub it to keep tests hermetic.
vi.mock('~/client-utils/edge-url', () => ({
@@ -85,6 +93,7 @@ vi.mock('~/client-utils/edge-url', () => ({
}));
import {
createImageIngestionRequest,
createModelFileScanRequest,
ModelFileScanSubmissionError,
} from '~/server/services/orchestrator/orchestrator.service';
@@ -609,3 +618,130 @@ describe('createModelFileScanRequest', () => {
});
});
});
describe('createImageIngestionRequest', () => {
const submittedSteps = () =>
mockSubmitWorkflow.mock.calls[0][0].body.steps as Array<{ $type: string; input: unknown }>;
beforeEach(() => {
mockIsFlipt.mockReset().mockResolvedValue(false);
mockSubmitWorkflow.mockResolvedValue({
data: { id: 'workflow-1' },
response: { status: 200, headers: new Headers() },
});
});
it('submits wdTagging + mediaRating while the imageScanning flag is off', async () => {
await createImageIngestionRequest({ imageId: 1, url: 'image-key' });
expect(mockIsFlipt).toHaveBeenCalledWith('image-ingestion-image-scanning', '1');
// The legacy submit is the production path; its inputs must stay exactly as they were.
const metadata = { imageId: 1 };
const priority = 'normal';
const mediaUrl = { $ref: '$arguments', path: 'mediaUrl' };
expect(submittedSteps()).toEqual([
{
$type: 'wdTagging',
name: 'tags',
metadata,
priority,
input: {
mediaUrl,
model:
'urn:air:siglip2:repository:huggingface:cella110n/cl_tagger_v2@b57909b8e9c63f71e208a26473e7aabdf45ed6b6.tar',
threshold: 0.55,
},
},
{
$type: 'mediaRating',
name: 'rating',
metadata,
priority,
input: {
mediaUrl,
engine: 'civitai',
includeAgeClassification: true,
includeAIRecognition: false,
includeFaceRecognition: false,
includeAnimeRecognition: false,
},
},
{
$type: 'mediaHash',
name: 'hash',
metadata,
priority,
input: { mediaUrl, hashTypes: ['perceptual'] },
},
]);
});
it.each([
[false, 'image-ingestion'],
[true, 'image-scanning-ingestion'],
])('logs a failed submit under its pipeline name (flag %s)', async (flagOn, name) => {
vi.useRealTimers();
mockIsFlipt.mockResolvedValue(flagOn);
mockSubmitWorkflow.mockResolvedValue({
data: undefined,
error: 'bad request',
response: { status: 400, headers: new Headers() },
});
const result = await createImageIngestionRequest({ imageId: 1, url: 'image-key' });
expect(result.useImageScanning).toBe(flagOn);
expect(mockLogToAxiom).toHaveBeenCalledWith(expect.objectContaining({ name, imageId: 1 }));
});
it('submits one imageScanning step when the flag is on', async () => {
mockIsFlipt.mockResolvedValue(true);
await createImageIngestionRequest({ imageId: 1, url: 'image-key' });
const steps = submittedSteps();
expect(steps.map((step) => step.$type)).toEqual(['imageScanning', 'mediaHash']);
expect(steps[0].input).toEqual({ image: { $ref: '$arguments', path: 'mediaUrl' } });
expect(steps[1].input).toEqual({
mediaUrl: { $ref: '$arguments', path: 'mediaUrl' },
hashTypes: ['perceptual'],
});
});
const frameUrl = { $ref: 'frame', path: 'url' };
it.each([
[
false,
[
['wdTagging', 'mediaUrl'],
['mediaRating', 'mediaUrl'],
],
],
[true, [['imageScanning', 'image']]],
])('repeats the per-frame steps over video frames (flag %s)', async (flagOn, templates) => {
mockIsFlipt.mockResolvedValue(flagOn);
await createImageIngestionRequest({ imageId: 1, url: 'video-key', type: 'video' });
const steps = submittedSteps() as Array<{
$type: string;
input: {
videoUrl?: unknown;
for?: unknown;
template?: { $type: string; input: Record<string, unknown> };
};
}>;
expect(steps[0]).toMatchObject({
$type: 'videoFrameExtraction',
input: {
videoUrl: { $ref: '$arguments', path: 'mediaUrl' },
frameRate: 1,
uniqueThreshold: 0.9,
maxFrames: 50,
},
});
expect(steps.slice(1)).toHaveLength(templates.length);
steps.slice(1).forEach((step, i) => {
const [type, urlField] = templates[i];
expect(step.$type).toBe('repeat');
expect(step.input.for).toEqual({ $ref: 'videoFrames', path: 'output.frames', as: 'frame' });
expect(step.input.template?.$type).toBe(type);
expect(step.input.template?.input[urlField]).toEqual(frameUrl);
});
});
});
@@ -18,6 +18,7 @@ import { logToAxiom } from '~/server/logging/client';
import { internalOrchestratorClient } from '~/server/services/orchestrator/client';
import { submitWorkflowWithRetry } from '~/server/services/orchestrator/workflows';
import { hashContent } from '~/server/services/entity-moderation.service';
import { FLIPT_FEATURE_FLAGS, isFlipt } from '~/server/flipt/client';
import type { MediaType, ModelType } from '~/shared/utils/prisma/enums';
import { EntityModerationStatus, ModelHashType, ScanResultCode } from '~/shared/utils/prisma/enums';
import { stringifyAIR } from '~/shared/utils/air';
@@ -47,6 +48,54 @@ const IMAGE_TAGGING_MODEL =
'urn:air:siglip2:repository:huggingface:cella110n/cl_tagger_v2@b57909b8e9c63f71e208a26473e7aabdf45ed6b6.tar';
const IMAGE_TAGGING_THRESHOLD = 0.55;
/** Axiom `name` for a failed ingestion submit, per scan pipeline. */
export function imageIngestionLogName(useImageScanning: boolean) {
return useImageScanning ? 'image-scanning-ingestion' : 'image-ingestion';
}
type MediaUrlRef = { $ref: string; path: string };
function imageScanSteps({
mediaUrl,
metadata,
priority,
useImageScanning,
}: {
mediaUrl: MediaUrlRef;
metadata: Record<string, unknown>;
priority: Priority;
useImageScanning: boolean;
}) {
if (useImageScanning)
return [
{ $type: 'imageScanning', name: 'scan', metadata, priority, input: { image: mediaUrl } },
];
return [
{
$type: 'wdTagging',
name: 'tags',
metadata,
priority,
input: { mediaUrl, model: IMAGE_TAGGING_MODEL, threshold: IMAGE_TAGGING_THRESHOLD },
},
{
$type: 'mediaRating',
name: 'rating',
metadata,
priority,
input: {
mediaUrl,
engine: 'civitai',
includeAgeClassification: true,
includeAIRecognition: false,
includeFaceRecognition: false,
includeAnimeRecognition: false,
},
},
];
}
export async function createImageIngestionRequest({
imageId,
url,
@@ -68,6 +117,45 @@ export async function createImageIngestionRequest({
// server-side, re-submitting with the same `externalId` returns the existing
// workflow instead of duplicating it (orchestrator dedupes on (userId, externalId)).
const externalId = randomUUID();
const useImageScanning = await isFlipt(
FLIPT_FEATURE_FLAGS.IMAGE_INGESTION_IMAGE_SCANNING,
String(imageId)
);
const mediaUrl = { $ref: '$arguments', path: 'mediaUrl' };
const steps =
type === 'image'
? [
...imageScanSteps({ mediaUrl, metadata, priority, useImageScanning }),
{
$type: 'mediaHash',
name: 'hash',
metadata,
priority,
input: { mediaUrl, hashTypes: ['perceptual'] },
},
]
: [
{
$type: 'videoFrameExtraction',
name: 'videoFrames',
metadata,
priority,
input: { videoUrl: mediaUrl, frameRate: 1, uniqueThreshold: 0.9, maxFrames: 50 },
},
...imageScanSteps({
mediaUrl: { $ref: 'frame', path: 'url' },
metadata,
priority,
useImageScanning,
}).map((template) => ({
$type: 'repeat',
input: {
for: { $ref: 'videoFrames', path: 'output.frames', as: 'frame' },
template,
},
})),
];
const body: WorkflowTemplate = {
externalId,
@@ -76,110 +164,8 @@ export async function createImageIngestionRequest({
mediaUrl: edgeUrl,
},
currencies: [],
steps:
type === 'image'
? [
{
$type: 'wdTagging',
name: 'tags',
metadata,
priority,
input: {
mediaUrl: { $ref: '$arguments', path: 'mediaUrl' },
model: IMAGE_TAGGING_MODEL,
threshold: IMAGE_TAGGING_THRESHOLD,
},
} as WorkflowStepTemplate,
{
$type: 'mediaRating',
name: 'rating',
metadata,
priority,
input: {
mediaUrl: { $ref: '$arguments', path: 'mediaUrl' },
engine: 'civitai',
includeAgeClassification: true,
includeAIRecognition: false,
includeFaceRecognition: false,
includeAnimeRecognition: false,
},
} as WorkflowStepTemplate,
{
$type: 'mediaHash',
name: 'hash',
metadata,
priority,
input: {
mediaUrl: { $ref: '$arguments', path: 'mediaUrl' },
hashTypes: ['perceptual'],
},
} as WorkflowStepTemplate,
]
: [
{
$type: 'videoFrameExtraction',
name: 'videoFrames',
metadata,
priority,
input: {
videoUrl: { $ref: '$arguments', path: 'mediaUrl' },
frameRate: 1,
uniqueThreshold: 0.9,
maxFrames: 50,
},
} as WorkflowStepTemplate,
{
$type: 'repeat',
input: {
for: {
$ref: 'videoFrames',
path: 'output.frames',
as: 'frame',
},
template: {
$type: 'wdTagging',
name: 'tags',
metadata,
priority,
input: {
mediaUrl: {
$ref: 'frame',
path: 'url',
},
model: IMAGE_TAGGING_MODEL,
threshold: IMAGE_TAGGING_THRESHOLD,
},
},
},
} as WorkflowStepTemplate,
{
$type: 'repeat',
input: {
for: {
$ref: 'videoFrames',
path: 'output.frames',
as: 'frame',
},
template: {
$type: 'mediaRating',
name: 'rating',
metadata,
priority,
input: {
mediaUrl: {
$ref: 'frame',
path: 'url',
},
engine: 'civitai',
includeAgeClassification: true,
includeAIRecognition: false,
includeFaceRecognition: false,
includeAnimeRecognition: false,
},
},
},
} as WorkflowStepTemplate,
],
// WorkflowTemplate is @civitai/client's, which predates the imageScanning step.
steps: steps as unknown as WorkflowStepTemplate[],
callbacks: callbackUrl
? [
{
@@ -229,7 +215,7 @@ export async function createImageIngestionRequest({
if (!data) {
logToAxiom({
type: 'error',
name: 'image-ingestion',
name: imageIngestionLogName(useImageScanning),
imageId,
url,
externalId,
@@ -240,7 +226,7 @@ export async function createImageIngestionRequest({
});
}
return { data, body, error, status: response?.status };
return { data, body, error, status: response?.status, useImageScanning };
}
const PERCEPTUAL_HASH_WAIT_SECONDS = 30;
@@ -258,7 +258,7 @@ export async function auditPromptServer(options: AuditPromptOptions): Promise<vo
// 🔴 Strip the NORMALIZED copy. `auditPromptEnriched` folds accents before the
// detector runs, so stripping raw text matches one alphabet while the detector
// reads another and a whitelisted `emma stone` still blocks `émma stone`. The
// scan paths in image-scan-result.service.ts already normalize first.
// scan paths in image-scan-pipeline.ts already normalize first.
const [auditedPrompt, auditedNegativePrompt] = await Promise.all([
stripBenignPhrases(normalizeText(prompt), BlocklistType.PromptBenignPhrase),
stripBenignPhrases(normalizeText(negativePrompt), BlocklistType.NegativeBenignPhrase),
@@ -14,6 +14,7 @@
import crypto from 'crypto';
import type { MediaRatingOutput, Workflow, XGuardModerationStep } from '@civitai/client';
import { clickhouse } from '~/server/clickhouse/client';
import type { ImageScanningResult } from '~/server/services/image-scanning-result.service';
import { logToAxiom } from '~/server/logging/client';
import {
applyDerivedLabels,
@@ -530,3 +531,91 @@ export async function recordImageScan({
labels,
});
}
// imageScanning's age and recognition models differ from mediaRating's, so its rows must not
// aggregate with version 1.
const IMAGE_SCANNING_AUDIT_VERSION = '2';
/** `recordImageScan` for the imageScanning step. */
export async function recordImageScanningResult({
workflowId,
imageId,
scan,
startedAt,
completedAt,
}: {
workflowId: string;
imageId: number;
scan: ImageScanningResult;
startedAt?: Date | string | null;
completedAt?: Date | string | null;
}) {
if (!clickhouse) return;
const baseRow = {
threshold: null,
version: IMAGE_SCANNING_AUDIT_VERSION,
matchedText: [] as string[],
matchedPositivePrompt: [] as string[],
matchedNegativePrompt: [] as string[],
};
const labels: LabelRowSeed[] = [
{
...baseRow,
label: normClassifierLabel(scan.nsfwLevel),
labelValue: 'nsfw_level',
score: 1,
triggered: 1,
},
];
// Nothing acts on this label yet.
if (scan.csam !== null) {
labels.push({
...baseRow,
label: 'csam',
labelValue: '',
score: scan.csam ? 1 : 0,
triggered: scan.csam ? 1 : 0,
});
}
if (scan.ageDetections.length > 0) {
labels.push({
...baseRow,
label: 'minor',
labelValue: scan.ageDetections
.map((d) => d.ageBand)
.filter((x): x is string => !!x)
.join(', '),
score: clamp01(Math.max(...scan.ageDetections.map((d) => d.under18Probability ?? 0))),
triggered: scan.minorDetected ? 1 : 0,
});
}
for (const [recognition, labelValue] of [
[scan.aiRecognition, 'ai_recognition'],
[scan.animeRecognition, 'anime_recognition'],
] as const) {
if (!recognition?.label) continue;
labels.push({
...baseRow,
label: normClassifierLabel(recognition.label),
labelValue,
score: clamp01(recognition.score),
triggered: 1,
});
}
await insertRows({
workflowId,
scanner: 'image_ingestion',
contentHash: computeContentHash(`image:${imageId}`),
entityType: 'image',
entityId: String(imageId),
modelVersion: IMAGE_SCANNING_AUDIT_VERSION,
startedAt,
completedAt,
labels,
});
}
+2 -3
View File
@@ -94,9 +94,8 @@ export async function debounceArticleUpdate(articleId: number): Promise<void> {
* advance article state, and `updateArticleNsfwLevels` picks up the new cover
* rating so `Article.nsfwLevel` no longer lags the cover.
*
* Callers are responsible for gating on the `articleImageScanning` feature flag
* because the two webhook entrypoints resolve the flag differently (one reads
* from a request context, the other receives it as a parameter).
* Callers gate on the `articleImageScanning` feature flag it needs a request context,
* which this helper does not have.
*/
export async function fanOutArticleImageUpdates(imageId: number): Promise<void> {
const [articleConnections, coverArticles] = await Promise.all([
+2 -3
View File
@@ -21,9 +21,8 @@ import { trpcMutation, trpcQuery, uniqueToken } from './preview-trpc';
* bucket (a tiny PNG is written each run harmless).
*
* Scope ceiling (honest): asserts the upload + attach + PUBLISH DB path. It does NOT
* assert public-feed visibility image ingestion/scan is unreachable in preview
* (placeholder IMAGE_SCANNING_ENDPOINT), so the row stays ingestion:Pending. That does
* NOT block row creation, attach, or publish.
* assert public-feed visibility image ingestion never completes in preview, so the row
* stays ingestion:Pending. That does NOT block row creation, attach, or publish.
*
* Role: tester (free member, passes the gate; guardedProcedure cleared via onboarding=15).
*/