diff --git a/server/plugins/warehouse/queries/ClickhouseActionExecutionsAdapter.test.ts b/server/plugins/warehouse/queries/ClickhouseActionExecutionsAdapter.test.ts index 9d3a73404..12c824220 100644 --- a/server/plugins/warehouse/queries/ClickhouseActionExecutionsAdapter.test.ts +++ b/server/plugins/warehouse/queries/ClickhouseActionExecutionsAdapter.test.ts @@ -142,6 +142,468 @@ describe('ClickhouseActionExecutionsAdapter.findInferredUserIdentity', () => { }); }); +describe('ClickhouseActionExecutionsAdapter.getRecentModeratorActions', () => { + const groupRow = (overrides: Record = {}) => ({ + correlation_id: 'manual-action-run:abc', + last_ts: '2026-08-05 12:00:00.000', + actor_id: 'user-7', + item_type_id: 'post', + actor_note: 'spam sweep', + policies: '[{"id":"pol-1","name":"Spam"}]', + action_ids: ['act-1', 'act-2'], + item_count: '3', + failed_count: '0', + ...overrides, + }); + + it('collapses one bulk operation into a single record', async () => { + const { adapter } = makeAdapter([groupRow()]); + + const result = await adapter.getRecentModeratorActions({ + orgId: 'org-1', + limit: 100, + }); + + expect(result).toEqual([ + { + correlationId: 'manual-action-run:abc', + actorId: 'user-7', + itemTypeId: 'post', + actionIds: ['act-1', 'act-2'], + policyIds: ['pol-1'], + actorNote: 'spam sweep', + itemCount: 3, + failedCount: 0, + occurredAt: new Date('2026-08-05T12:00:00.000Z'), + }, + ]); + }); + + it('reports how many executions failed', async () => { + const { adapter } = makeAdapter([ + groupRow({ item_count: '500', failed_count: '3' }), + ]); + + const [group] = await adapter.getRecentModeratorActions({ + orgId: 'org-1', + limit: 100, + }); + + expect(group.itemCount).toBe(500); + expect(group.failedCount).toBe(3); + }); + + it('counts items exactly rather than estimating', async () => { + // uniq() is HyperLogLog. An "84 items" label that reads 83 is a bug the + // reader cannot detect. + const { adapter, query } = makeAdapter([]); + + await adapter.getRecentModeratorActions({ orgId: 'org-1', limit: 100 }); + + const sentSql = query.mock.calls[0][0]; + expect(sentSql).toContain('uniqExact(item_id)'); + expect(sentSql).not.toMatch(/[^a-zA-Z]uniq\(/); + }); + + it('counts failures in items, not executions', async () => { + // countIf counts (item, action) rows, so a 2-item x 2-action run that fails + // completely would report "4 of 2 failed". + const { adapter, query } = makeAdapter([]); + + await adapter.getRecentModeratorActions({ orgId: 'org-1', limit: 100 }); + + const sentSql = query.mock.calls[0][0]; + expect(sentSql).toContain('uniqExactIf(item_id, failed = 1)'); + expect(sentSql).not.toContain('countIf('); + }); + + it('scopes to moderator action sources rather than a null job id', async () => { + // Rule-driven and user-strike executions also carry a null job_id, so + // filtering on that would pull automation into the feed. + const { adapter, query } = makeAdapter([]); + + await adapter.getRecentModeratorActions({ orgId: 'org-1', limit: 100 }); + + const sentSql = query.mock.calls[0][0]; + expect(sentSql).toContain("action_source IN ('manual-action-run')"); + expect(sentSql).not.toContain('job_id'); + expect(sentSql).toContain('GROUP BY correlation_id'); + }); + + it('bounds the partition scan with a ds floor derived from the cursor', async () => { + const { adapter, query } = makeAdapter([]); + + await adapter.getRecentModeratorActions({ + orgId: 'org-1', + limit: 100, + cursor: { + ts: new Date('2026-08-05T12:00:00.000Z'), + correlationId: 'manual-action-run:abc', + }, + lookbackWindowMs: 24 * 60 * 60 * 1000, + }); + + const sentSql = query.mock.calls[0][0]; + expect(sentSql).toContain("toDate('2026-08-04')"); + }); + + it('reaches forward to an endTime later than now, so a future-dated row is recoverable', async () => { + // Deriving the ds ceiling from the cursor alone made a row dated ahead of + // the server clock permanently unreachable — page 1's ceiling is `now`, + // paging only moves backwards, and no filter could widen it. Clock skew on + // a writer host is enough to produce one. + const { adapter, query } = makeAdapter([]); + const future = new Date(Date.now() + 5 * 24 * 60 * 60 * 1000); + + await adapter.getRecentModeratorActions({ + orgId: 'org-1', + limit: 100, + before: future, + }); + + const sentSql = query.mock.calls[0][0]; + expect(sentSql).toContain( + `ds <= toDate('${future.toISOString().slice(0, 10)}') + 1`, + ); + expect(sentSql).toContain( + `max(ts) <= '${future.toISOString().replace('T', ' ').replace('Z', '')}'`, + ); + }); + + it('anchors the lookback window to endTime, not to now', async () => { + // With an endTime and no startTime, anchoring the floor at `now` put the + // partitions at [now-window, now] while HAVING max(ts) <= endTime excluded + // every one of them — the feed returned nothing, with no error. + const { adapter, query } = makeAdapter([]); + + await adapter.getRecentModeratorActions({ + orgId: 'org-1', + limit: 100, + before: new Date('2026-01-31T23:59:59.000Z'), + lookbackWindowMs: 30 * 24 * 60 * 60 * 1000, + }); + + const sentSql = query.mock.calls[0][0]; + expect(sentSql).toContain("ds >= toDate('2026-01-01') - 1"); + expect(sentSql).toContain("ds <= toDate('2026-01-31') + 1"); + }); + + it('keeps a run straddling the start date whole', async () => { + // A bulk run beginning just before the start date and finishing just after + // must not lose its earlier rows — that would group into a partial record + // with a wrong item count, the same defect the cursor fix addresses. + const { adapter, query } = makeAdapter([]); + + await adapter.getRecentModeratorActions({ + orgId: 'org-1', + limit: 100, + after: new Date('2026-08-01T00:00:00.000Z'), + }); + + expect(query.mock.calls[0][0]).toContain("toDate('2026-08-01') - 1"); + }); + + it('filters to operations that touched a given item', async () => { + const { adapter, query } = makeAdapter([]); + + await adapter.getRecentModeratorActions({ + orgId: 'org-1', + limit: 100, + itemId: 'usr_8813', + }); + + expect(query.mock.calls[0][0]).toContain( + "has(groupUniqArray(item_id), 'usr_8813')", + ); + }); + + it('does not alias the aggregate to `ts`', async () => { + // ClickHouse resolves `ts < ?` in WHERE against a `max(ts) AS ts` alias and + // rejects the query with ILLEGAL_AGGREGATION. Mocked rows can't catch this, + // so assert the alias directly. + const { adapter, query } = makeAdapter([]); + + await adapter.getRecentModeratorActions({ orgId: 'org-1', limit: 100 }); + + const sentSql = query.mock.calls[0][0]; + expect(sentSql).toContain('max(ts) AS last_ts'); + expect(sentSql).not.toContain('max(ts) AS ts'); + expect(sentSql).toContain('ORDER BY last_ts DESC'); + }); + + it('reads timestamps as UTC even though ClickHouse omits the zone', async () => { + const { adapter } = makeAdapter([ + groupRow({ last_ts: '2026-08-05 12:00:00.000' }), + ]); + + const [group] = await adapter.getRecentModeratorActions({ + orgId: 'org-1', + limit: 100, + }); + + expect(group.occurredAt.toISOString()).toBe('2026-08-05T12:00:00.000Z'); + }); + + it('scopes to the requested moderators', async () => { + // Without this the reviewer filter would narrow decisions while still + // showing every other moderator's actions. + const { adapter, query } = makeAdapter([]); + + await adapter.getRecentModeratorActions({ + orgId: 'org-1', + limit: 100, + actorIds: ['user-7', 'user-8'], + }); + + expect(query.mock.calls[0][0]).toContain( + "actor_id IN ('user-7', 'user-8')", + ); + }); + + it('matches policies through the policies JSON, not the unpopulated column', async () => { + const { adapter, query } = makeAdapter([]); + + await adapter.getRecentModeratorActions({ + orgId: 'org-1', + limit: 100, + policyIds: ['pol-1'], + }); + + const sentSql = query.mock.calls[0][0]; + expect(sentSql).toContain('JSONExtractArrayRaw(policies)'); + expect(sentSql).toContain("['pol-1']"); + expect(sentSql).not.toContain('policy_ids'); + }); + + it('omits optional filters entirely when not supplied', async () => { + const { adapter, query } = makeAdapter([]); + + await adapter.getRecentModeratorActions({ + orgId: 'org-1', + limit: 100, + actorIds: [], + policyIds: [], + }); + + const sentSql = query.mock.calls[0][0]; + expect(sentSql).not.toContain('actor_id IN'); + expect(sentSql).not.toContain('JSONExtractArrayRaw'); + }); + + it('tightens the ds floor to a user-set start date', async () => { + const { adapter, query } = makeAdapter([]); + + await adapter.getRecentModeratorActions({ + orgId: 'org-1', + limit: 100, + cursor: { + ts: new Date('2026-08-05T12:00:00.000Z'), + correlationId: 'manual-action-run:abc', + }, + after: new Date('2026-08-01T00:00:00.000Z'), + }); + + const sentSql = query.mock.calls[0][0]; + // The rolling window would reach back months; the start date wins, minus a + // day of slack so a run straddling midnight is not truncated. + expect(sentSql).toContain("toDate('2026-08-01') - 1"); + // `after` is an exact bound on the complete group, so it lives in HAVING. + expect(sentSql).toContain("max(ts) >= '2026-08-01 00:00:00.000'"); + }); + + it('honours a start date older than the rolling window', async () => { + const { adapter, query } = makeAdapter([]); + + await adapter.getRecentModeratorActions({ + orgId: 'org-1', + limit: 100, + cursor: { + ts: new Date('2026-08-05T12:00:00.000Z'), + correlationId: 'manual-action-run:abc', + }, + after: new Date('2026-01-01T00:00:00.000Z'), + lookbackWindowMs: 30 * 24 * 60 * 60 * 1000, + }); + + const sentSql = query.mock.calls[0][0]; + // The 30-day window would floor at 2026-07-06 and silently hide January. + expect(sentSql).toContain("toDate('2026-01-01') - 1"); + expect(sentSql).not.toContain("toDate('2026-07-06')"); + }); + + it('tolerates null aggregate columns', async () => { + const { adapter } = makeAdapter([ + groupRow({ + actor_id: null, + item_type_id: null, + actor_note: null, + policies: null, + action_ids: null, + item_count: null, + failed_count: null, + }), + ]); + + const [group] = await adapter.getRecentModeratorActions({ + orgId: 'org-1', + limit: 100, + }); + + expect(group.actionIds).toEqual([]); + expect(group.policyIds).toEqual([]); + expect(group.itemCount).toBe(0); + expect(group.failedCount).toBe(0); + }); + + it('applies the cursor to the complete group, not to raw rows', async () => { + // A bulk run trickles in over seconds (pLimit(10) in ActionApi). Bounding + // raw `ts` before GROUP BY truncates a run straddling the cursor into a + // partial group with a lower max(ts) and itemCount, so the same run lands + // on two consecutive pages with wrong counts on both. + const { adapter, query } = makeAdapter([]); + + await adapter.getRecentModeratorActions({ + orgId: 'org-1', + limit: 100, + cursor: { + ts: new Date('2026-08-05T12:00:00.000Z'), + correlationId: 'manual-action-run:abc', + }, + }); + + const sentSql = query.mock.calls[0][0]; + expect(sentSql).toContain('HAVING'); + expect(sentSql).toContain( + "(max(ts), correlation_id) < ('2026-08-05 12:00:00.000', 'manual-action-run:abc')", + ); + // The cursor must not appear as a raw-row predicate. + expect(sentSql).not.toMatch(/WHERE[\s\S]*\bts\s*<\s*'/); + }); + + it('orders by the composite key so ties cannot repeat or vanish', async () => { + const { adapter, query } = makeAdapter([]); + + await adapter.getRecentModeratorActions({ orgId: 'org-1', limit: 100 }); + + expect(query.mock.calls[0][0]).toContain( + 'ORDER BY last_ts DESC, correlation_id DESC', + ); + }); + + it('bounds the upper end of a date range on the complete group', async () => { + // Without this, a January filter renders today's bulk runs beside January + // decisions: endTime reaches Postgres but never reaches this store. + const { adapter, query } = makeAdapter([]); + + await adapter.getRecentModeratorActions({ + orgId: 'org-1', + limit: 100, + after: new Date('2026-01-01T00:00:00.000Z'), + before: new Date('2026-01-31T23:59:59.000Z'), + }); + + const sentSql = query.mock.calls[0][0]; + expect(sentSql).toContain("max(ts) <= '2026-01-31 23:59:59.000'"); + expect(sentSql).toContain("max(ts) >= '2026-01-01 00:00:00.000'"); + }); +}); + +describe('ClickhouseActionExecutionsAdapter.getManualActionItems', () => { + it('returns one record per item with its failure state', async () => { + const { adapter } = makeAdapter([ + { item_id: 'i-1', item_type_id: 'post', failed: 0, total_count: '3' }, + { item_id: 'i-2', item_type_id: 'post', failed: 1, total_count: '3' }, + ]); + + const result = await adapter.getManualActionItems({ + orgId: 'org-1', + correlationId: 'manual-action-run:abc', + occurredAt: new Date('2026-08-05T12:00:00.000Z'), + limit: 100, + offset: 0, + }); + + expect(result).toEqual({ + items: [ + { itemId: 'i-1', itemTypeId: 'post', failed: false }, + { itemId: 'i-2', itemTypeId: 'post', failed: true }, + ], + totalCount: 3, + }); + }); + + it('scans back at least as far as the feed groups, so the two agree', async () => { + // The feed groups a run over the whole lookback window. Bounding this to + // `occurredAt ± 1 day` made a row and its own detail panel disagree: the + // row read "8 items / 1 of 8 failed" while the panel showed 3 items and no + // failures at all — the audit log dropping the failures it exists to + // record. `correlation_id` is not in the sort key, so a ds bound is still + // required; it just has to be the feed's, not a narrower one. + const { adapter, query } = makeAdapter([]); + + await adapter.getManualActionItems({ + orgId: 'org-1', + correlationId: 'manual-action-run:abc', + occurredAt: new Date('2026-08-05T12:00:00.000Z'), + limit: 100, + offset: 0, + lookbackWindowMs: 30 * 24 * 60 * 60 * 1000, + }); + + const sentSql = query.mock.calls[0][0]; + expect(sentSql).toContain("ds >= toDate('2026-07-06')"); + expect(sentSql).toContain("ds <= toDate('2026-08-05') + 1"); + expect(sentSql).toContain("correlation_id = 'manual-action-run:abc'"); + }); + + it('defaults its window to the feed lookback rather than a single day', async () => { + const { adapter, query } = makeAdapter([]); + + await adapter.getManualActionItems({ + orgId: 'org-1', + correlationId: 'manual-action-run:abc', + occurredAt: new Date('2026-08-05T12:00:00.000Z'), + limit: 100, + offset: 0, + }); + + // 3 months back, matching DEFAULT_ACTION_FEED_LOOKBACK_MS — not 08-04. + expect(query.mock.calls[0][0]).toContain("ds >= toDate('2026-05-07')"); + }); + + it('reads only moderator actions, not any correlation id', async () => { + // mrt-decision / rule-driven correlation ids must not resolve here — their + // item lists are what the NCMEC permission gate exists to hide. + const { adapter, query } = makeAdapter([]); + + await adapter.getManualActionItems({ + orgId: 'org-1', + correlationId: 'mrt-decision:abc', + occurredAt: new Date('2026-08-05T12:00:00.000Z'), + limit: 100, + offset: 0, + }); + + expect(query.mock.calls[0][0]).toContain( + "action_source IN ('manual-action-run')", + ); + }); + + it('reports zero total when the operation has no rows', async () => { + const { adapter } = makeAdapter([]); + + const result = await adapter.getManualActionItems({ + orgId: 'org-1', + correlationId: 'manual-action-run:missing', + occurredAt: new Date('2026-08-05T12:00:00.000Z'), + limit: 100, + offset: 0, + }); + + expect(result).toEqual({ items: [], totalCount: 0 }); + }); +}); + describe('ClickhouseActionExecutionsAdapter.findContentCreatorIdentity', () => { it('returns null when no rows match', async () => { const { adapter } = makeAdapter([]); diff --git a/server/plugins/warehouse/queries/ClickhouseActionExecutionsAdapter.ts b/server/plugins/warehouse/queries/ClickhouseActionExecutionsAdapter.ts index d14924d0b..2741cf1a2 100644 --- a/server/plugins/warehouse/queries/ClickhouseActionExecutionsAdapter.ts +++ b/server/plugins/warehouse/queries/ClickhouseActionExecutionsAdapter.ts @@ -1,8 +1,19 @@ import type { IDataWarehouse } from '../../../storage/dataWarehouse/IDataWarehouse.js'; -import { jsonParse, type JsonOf } from '../../../utils/encoding.js'; import type SafeTracer from '../../../utils/SafeTracer.js'; -import { SIX_MONTHS_MS } from '../../../utils/time.js'; +import { MONTH_MS, SIX_MONTHS_MS } from '../../../utils/time.js'; import { formatClickhouseQuery } from '../utils/clickhouseSql.js'; +import { + deriveActionScanWindow, + formatWarehouseDateTime, + parseWarehouseDateTime, + toDsString, +} from '../utils/warehouseDateTime.js'; +import { + type ClickhouseActionExecutionRow, + type ClickhouseManualActionItemRow, + type ClickhouseModeratorActionGroupRow, +} from './clickhouseActionExecutionRows.js'; +import { extractIds, parseJsonIdArray } from './clickhouseJsonIdArray.js'; import { type ContentCreatorIdentityInput, type ContentCreatorIdentityRecord, @@ -11,24 +22,28 @@ import { type InferredUserIdentityRecord, type ItemActionHistoryInput, type ItemActionHistoryRecord, + type ManualActionItemsInput, + type ManualActionItemsResult, + type ModeratorActionGroupRecord, + type RecentModeratorActionsInput, type UserStrikeActionRecord, type UserStrikeActionsInput, } from './IActionExecutionsAdapter.js'; -interface ClickhouseActionExecutionRow { - ts: string; - item_id: string | null; - item_type_id: string | null; - item_type_kind: string; - item_creator_id: string | null; - item_creator_type_id: string | null; - actor_id: string | null; - job_id: string | null; - policies?: string | null; - rules?: string | null; - action_id: string; - action_source?: string; -} +/** + * `action_source` values written by the moderator-facing action UIs. Bulk + * Actioning and Investigation both publish under `manual-action-run` (see + * `ActionApi.bulkExecuteActions`). + * + * This allowlist — not `job_id IS NULL` — is what separates moderator work from + * automation. Rule-driven and user-strike executions also carry a null + * `job_id`, and `actor_id IS NOT NULL` would still admit the programmatic + * `/action` API route, which sets an actor whenever a user is attached. + */ +const MODERATOR_ACTION_SOURCES = ['manual-action-run']; + +/** How far back a single page of the moderator action feed may scan. */ +const DEFAULT_ACTION_FEED_LOOKBACK_MS = 3 * MONTH_MS; export class ClickhouseActionExecutionsAdapter implements IActionExecutionsAdapter { constructor( @@ -84,12 +99,204 @@ export class ClickhouseActionExecutionsAdapter implements IActionExecutionsAdapt jobId: row.job_id ?? null, userId: row.item_creator_id ?? null, userTypeId: row.item_creator_type_id ?? null, - policies: this.extractIds(this.parseJsonArray(row.policies)), - ruleIds: this.extractIds(this.parseJsonArray(row.rules)), + policies: extractIds(parseJsonIdArray(row.policies)), + ruleIds: extractIds(parseJsonIdArray(row.rules)), occurredAt: new Date(row.ts), })); } + async getRecentModeratorActions( + input: RecentModeratorActionsInput, + ): Promise> { + const { + orgId, + cursor, + after, + before, + limit, + actorIds, + policyIds, + itemId, + lookbackWindowMs = DEFAULT_ACTION_FEED_LOOKBACK_MS, + } = input; + + const { start: lookbackStart, end: upperBound } = deriveActionScanWindow({ + cursorTs: cursor?.ts, + after, + before, + lookbackWindowMs, + }); + + // `ds` bounds the partition scan only. The cursor and `after` are applied + // in HAVING, against the complete group. + // + // Each bound carries a day of slack so it never truncates a group we + // intend to keep whole. A group whose max(ts) sits just past the cursor + // day may still have earlier rows filed under the cursor's own ds day, so + // the upper bound reaches one day past it. Symmetrically, a group whose + // rows start just before the lookback/`after` floor may have later rows + // on the floor's own ds day, so the lower bound reaches one day before it. + const conditions = [ + 'org_id = ?', + 'ds >= toDate(?) - 1', + 'ds <= toDate(?) + 1', + `action_source IN (${MODERATOR_ACTION_SOURCES.map(() => '?').join(', ')})`, + ]; + const params: unknown[] = [ + orgId, + toDsString(lookbackStart), + toDsString(upperBound), + ...MODERATOR_ACTION_SOURCES, + ]; + + if (actorIds && actorIds.length > 0) { + conditions.push(`actor_id IN (${actorIds.map(() => '?').join(', ')})`); + params.push(...actorIds); + } + if (policyIds && policyIds.length > 0) { + // `policies` is a JSON array of objects; `policy_ids` exists on the table + // but ActionExecutionLogger never populates it, so extract from the JSON. + conditions.push( + `hasAny(arrayMap(p -> JSONExtractString(p, 'id'), JSONExtractArrayRaw(policies)), ?)`, + ); + params.push([...policyIds]); + } + + const having: string[] = []; + if (cursor) { + having.push('(max(ts), correlation_id) < (?, ?)'); + params.push(formatWarehouseDateTime(cursor.ts), cursor.correlationId); + } + if (after) { + having.push('max(ts) >= ?'); + params.push(formatWarehouseDateTime(after)); + } + if (before) { + having.push('max(ts) <= ?'); + params.push(formatWarehouseDateTime(before)); + } + if (itemId) { + having.push('has(groupUniqArray(item_id), ?)'); + params.push(itemId); + } + + // Grouping by correlation_id collapses the one-row-per-(item, action) fan + // out into a single record per moderator operation. Without it, a bulk + // submit of 500 ids with 2 actions contributes 1,000 rows and buries every + // other entry in the merged feed. + const sql = ` + SELECT + correlation_id, + -- Deliberately not aliased to "ts". ClickHouse would resolve a ts + -- bound against this aggregate and reject the query with + -- ILLEGAL_AGGREGATION. (Placeholders are substituted across the whole + -- string, so never put one in a comment either.) + max(ts) AS last_ts, + any(actor_id) AS actor_id, + any(item_type_id) AS item_type_id, + any(actor_note) AS actor_note, + any(policies) AS policies, + groupUniqArray(action_id) AS action_ids, + uniqExact(item_id) AS item_count, + uniqExactIf(item_id, failed = 1) AS failed_count + FROM analytics.ACTION_EXECUTIONS + WHERE ${conditions.join('\n AND ')} + GROUP BY correlation_id + ${having.length > 0 ? `HAVING ${having.join('\n AND ')}` : ''} + ORDER BY last_ts DESC, correlation_id DESC + LIMIT ${Number(limit)} + `; + + const rows = await this.query( + sql, + params, + ); + + return rows.map((row) => ({ + correlationId: row.correlation_id, + actorId: row.actor_id ?? null, + itemTypeId: row.item_type_id ?? null, + actionIds: row.action_ids ?? [], + policyIds: extractIds(parseJsonIdArray(row.policies)), + actorNote: row.actor_note ?? null, + itemCount: Number(row.item_count) || 0, + failedCount: Number(row.failed_count) || 0, + occurredAt: parseWarehouseDateTime(row.last_ts), + })); + } + + async getManualActionItems( + input: ManualActionItemsInput, + ): Promise { + const { + orgId, + correlationId, + occurredAt, + limit, + offset, + lookbackWindowMs = DEFAULT_ACTION_FEED_LOOKBACK_MS, + } = input; + + // The `ds` bounds must span at least as much as the feed query's window, + // or a row and its own detail panel disagree. The feed groups a run over + // the whole lookback window; bounding this to `occurredAt ± 1 day` made an + // 8-item run render as "8 items / 1 of 8 failed" in the row and "3 items, + // no failures" in the panel — the audit log silently dropping the failures + // it exists to record. `occurredAt` is `max(ts)`, so reach backwards by + // the window and forwards by a day. + const windowStart = new Date( + occurredAt.valueOf() - Math.max(1, lookbackWindowMs), + ); + + // One row per (item, action); an item appears once per action applied. + // Group so the caller sees items, and treat an item as failed if any of + // its executions failed. + // + // `correlation_id` alone is not enough: mrt-decision, post-items, + // submit-report, and user-strike-action-execution runs are also grouped + // under a correlation id, and their item lists must not be resolvable + // here (mrt-decision's in particular is what the NCMEC permission gate + // exists to hide). The same allowlist the feed query uses keeps this to + // moderator-initiated runs only. + const sql = ` + SELECT + item_id, + any(item_type_id) AS item_type_id, + max(failed) AS failed, + count() OVER () AS total_count + FROM ( + SELECT item_id, item_type_id, failed + FROM analytics.ACTION_EXECUTIONS + WHERE org_id = ? + AND ds >= toDate(?) + AND ds <= toDate(?) + 1 + AND correlation_id = ? + AND item_id IS NOT NULL + AND action_source IN (${MODERATOR_ACTION_SOURCES.map(() => '?').join(', ')}) + ) + GROUP BY item_id + ORDER BY item_id ASC + LIMIT ${Number(limit)} OFFSET ${Number(offset)} + `; + + const rows = await this.query(sql, [ + orgId, + toDsString(windowStart), + toDsString(occurredAt), + correlationId, + ...MODERATOR_ACTION_SOURCES, + ]); + + return { + items: rows.map((row) => ({ + itemId: row.item_id, + itemTypeId: row.item_type_id ?? null, + failed: Number(row.failed) === 1, + })), + totalCount: Number(rows[0]?.total_count ?? 0) || 0, + }; + } + async getRecentUserStrikeActions( input: UserStrikeActionsInput, ): Promise> { @@ -267,40 +474,6 @@ export class ClickhouseActionExecutionsAdapter implements IActionExecutionsAdapt }; } - private parseJsonArray( - jsonString: string | null | undefined, - ): Array<{ id: string }> | null { - if (!jsonString || jsonString === '[]') { - return null; - } - try { - const parsed = jsonParse(jsonString as JsonOf); - if (Array.isArray(parsed)) { - return parsed.filter( - (item): item is { id: string } => - typeof item === 'object' && - item !== null && - 'id' in item && - typeof item.id === 'string', - ); - } - return null; - } catch { - return null; - } - } - - private extractIds( - values: Array<{ id: string }> | null | undefined, - ): readonly string[] { - if (!values) { - return []; - } - return values - .map((entry) => entry.id) - .filter((id): id is string => typeof id === 'string' && id.length > 0); - } - private async query( statement: string, params: readonly unknown[], diff --git a/server/plugins/warehouse/queries/IActionExecutionsAdapter.ts b/server/plugins/warehouse/queries/IActionExecutionsAdapter.ts index ce8365736..599efd908 100644 --- a/server/plugins/warehouse/queries/IActionExecutionsAdapter.ts +++ b/server/plugins/warehouse/queries/IActionExecutionsAdapter.ts @@ -26,6 +26,90 @@ export interface ItemActionHistoryInput { itemSubmissionTime?: Date; } +/** + * Position in the moderator action feed. Composite because timestamps collide — + * a `ts`-only cursor either skips rows sharing the boundary instant or repeats + * them, and the exclusion set that patches around that grows without bound. + */ +export interface ModeratorActionCursor { + /** `max(ts)` of the last group already returned. */ + ts: Date; + correlationId: string; +} + +export interface RecentModeratorActionsInput { + orgId: string; + /** Absent for the newest page. */ + cursor?: ModeratorActionCursor; + /** Inclusive lower bound on `max(ts)`, from a user-set date range. */ + after?: Date; + /** Inclusive upper bound on `max(ts)`, from a user-set date range. */ + before?: Date; + limit: number; + /** Restrict to actions taken by these moderators. */ + actorIds?: readonly string[]; + /** Restrict to actions carrying at least one of these policies. */ + policyIds?: readonly string[]; + /** Restrict to operations that touched this item. */ + itemId?: string; + /** + * How far back a single fetch may scan, relative to the cursor. + * `ACTION_EXECUTIONS` is partitioned by `ds`, so without a floor every page + * reads all partitions older than the cursor. + */ + lookbackWindowMs?: number; +} + +/** + * One moderator operation, collapsed from the many `(item, action)` rows it + * wrote. A single bulk submit of 500 ids with 2 actions selected produces 1,000 + * rows sharing one `correlation_id`; this is that operation as one record. + */ +export interface ModeratorActionGroupRecord { + correlationId: string; + actorId: string | null; + itemTypeId: string | null; + actionIds: readonly string[]; + policyIds: readonly string[]; + actorNote: string | null; + /** Distinct items the operation touched. Exact, not estimated. */ + itemCount: number; + /** `(item, action)` executions that failed after retries. */ + failedCount: number; + occurredAt: Date; +} + +export interface ManualActionItemsInput { + orgId: string; + correlationId: string; + /** + * `max(ts)` of the operation, from the feed row. Bounds the partition scan — + * `correlation_id` is not in the table's sort key, so without this every + * lookup scans the whole retention window. + */ + occurredAt: Date; + limit: number; + offset: number; + /** + * How far back of `occurredAt` to scan. Must be at least the feed query's + * lookback, or a run's row and its own item list disagree — the feed groups + * over the whole window while this would group over a narrower one, silently + * dropping items and their failures. Defaults to the feed's window. + */ + lookbackWindowMs?: number; +} + +export interface ManualActionItemRecord { + itemId: string; + itemTypeId: string | null; + failed: boolean; +} + +export interface ManualActionItemsResult { + items: readonly ManualActionItemRecord[]; + totalCount: number; +} + export interface UserStrikeActionsInput { orgId: string; filterBy?: { @@ -66,6 +150,20 @@ export interface IActionExecutionsAdapter { input: ItemActionHistoryInput, ): Promise>; + /** + * Feed of actions a moderator took outside a review job — from Bulk + * Actioning or Investigation. These never produce a `manual_review_decisions` + * row, so this is the only record of them. + */ + getRecentModeratorActions( + input: RecentModeratorActionsInput, + ): Promise>; + + /** Every item one moderator operation touched, paged. */ + getManualActionItems( + input: ManualActionItemsInput, + ): Promise; + getRecentUserStrikeActions( input: UserStrikeActionsInput, ): Promise>; diff --git a/server/plugins/warehouse/queries/clickhouseActionExecutionRows.ts b/server/plugins/warehouse/queries/clickhouseActionExecutionRows.ts new file mode 100644 index 000000000..560ed5fae --- /dev/null +++ b/server/plugins/warehouse/queries/clickhouseActionExecutionRows.ts @@ -0,0 +1,34 @@ +export interface ClickhouseActionExecutionRow { + ts: string; + item_id: string | null; + item_type_id: string | null; + item_type_kind: string; + item_creator_id: string | null; + item_creator_type_id: string | null; + actor_id: string | null; + job_id: string | null; + policies?: string | null; + rules?: string | null; + action_id: string; + action_source?: string; +} + +export interface ClickhouseModeratorActionGroupRow { + correlation_id: string; + last_ts: string; + actor_id: string | null; + item_type_id: string | null; + actor_note: string | null; + policies?: string | null; + action_ids: string[] | null; + // ClickHouse returns UInt64 aggregates as strings over the JSON interface. + item_count: string | number; + failed_count: string | number; +} + +export interface ClickhouseManualActionItemRow { + item_id: string; + item_type_id: string | null; + failed: string | number; + total_count: string | number; +} diff --git a/server/plugins/warehouse/queries/clickhouseJsonIdArray.ts b/server/plugins/warehouse/queries/clickhouseJsonIdArray.ts new file mode 100644 index 000000000..3738f012c --- /dev/null +++ b/server/plugins/warehouse/queries/clickhouseJsonIdArray.ts @@ -0,0 +1,41 @@ +import { jsonParse, type JsonOf } from '../../../utils/encoding.js'; + +/** + * Parses a ClickHouse column holding a JSON array of `{ id, ... }` objects + * (e.g. `policies`, `rules`). Returns `null` for empty/absent/malformed input + * so callers can distinguish "nothing there" from "zero ids after filtering". + */ +export function parseJsonIdArray( + jsonString: string | null | undefined, +): Array<{ id: string }> | null { + if (!jsonString || jsonString === '[]') { + return null; + } + try { + const parsed = jsonParse(jsonString as JsonOf); + if (Array.isArray(parsed)) { + return parsed.filter( + (item): item is { id: string } => + typeof item === 'object' && + item !== null && + 'id' in item && + typeof item.id === 'string', + ); + } + return null; + } catch { + return null; + } +} + +/** Projects the `id` field out of {@link parseJsonIdArray}'s result. */ +export function extractIds( + values: Array<{ id: string }> | null | undefined, +): readonly string[] { + if (!values) { + return []; + } + return values + .map((entry) => entry.id) + .filter((id): id is string => typeof id === 'string' && id.length > 0); +} diff --git a/server/plugins/warehouse/utils/warehouseDateTime.ts b/server/plugins/warehouse/utils/warehouseDateTime.ts new file mode 100644 index 000000000..3090fd9a9 --- /dev/null +++ b/server/plugins/warehouse/utils/warehouseDateTime.ts @@ -0,0 +1,65 @@ +/** + * Render a `Date` the way `ts` was written, so comparisons are like-for-like. + * Writes go through `ClickhouseAnalyticsAdapter.formatDate`, which stores UTC + * wall-clock with no zone suffix (`YYYY-MM-DD HH:MM:SS.mmm`). + */ +export function formatWarehouseDateTime(date: Date): string { + return date.toISOString().replace('T', ' ').replace('Z', ''); +} + +/** `YYYY-MM-DD`, for comparison against the `ds` partition column. */ +export function toDsString(date: Date): string { + return date.toISOString().slice(0, 10); +} + +/** + * Parse a warehouse timestamp as UTC. ClickHouse returns `DateTime64` without a + * zone suffix, and `new Date('2026-08-05 12:00:00.000')` reads that shape as + * *local* time — which would shift every row on a non-UTC host and interleave + * this feed incorrectly against the Postgres decisions it merges with. + */ +export function parseWarehouseDateTime(value: string): Date { + const isoish = value.includes('T') ? value : value.replace(' ', 'T'); + const hasZone = /(?:Z|[+-]\d{2}:?\d{2})$/.test(isoish); + return new Date(hasZone ? isoish : `${isoish}Z`); +} + +/** + * The `ds` partition range a moderator-action feed page must scan. + * + * `ACTION_EXECUTIONS` is partitioned by `ds`, so every query needs a bounded + * range — but deriving it from the wrong anchor silently loses rows, and a + * feed that lost rows looks exactly like one that has none. + * + * - `end` is the newest row the call can return. `HAVING` enforces both + * `< cursor` and `<= before`, so it is the tighter of the two. On the first + * page there is no cursor, and defaulting to `now` was itself an upper + * bound: it made a row dated ahead of the server clock permanently + * unreachable, since paging only moves backwards and `before` never widened + * the ceiling. Clock skew on a writer host is enough to produce one. + * - `start` reaches back from `end`, not from `now`. With an `endTime` and no + * `startTime` the two differ: anchoring at `now` put the partitions at + * `[now - window, now]` while `HAVING max(ts) <= before` excluded every one + * of them, so the feed returned nothing at all. + * - A user-set `after` replaces the rolling window rather than competing with + * it. Taking whichever is later silently drops a start date older than the + * window: `HAVING` passes but the partitions are already gone. + */ +export function deriveActionScanWindow(opts: { + cursorTs: Date | undefined; + after: Date | undefined; + before: Date | undefined; + lookbackWindowMs: number; +}): { start: Date; end: Date } { + const { cursorTs, after, before, lookbackWindowMs } = opts; + + const end = cursorTs + ? before && before.valueOf() < cursorTs.valueOf() + ? before + : cursorTs + : (before ?? new Date()); + + const windowStart = new Date(end.valueOf() - Math.max(1, lookbackWindowMs)); + + return { start: after ?? windowStart, end }; +} diff --git a/server/services/manualReviewToolService/manualReviewToolService.ts b/server/services/manualReviewToolService/manualReviewToolService.ts index d8a733abd..83904e25b 100644 --- a/server/services/manualReviewToolService/manualReviewToolService.ts +++ b/server/services/manualReviewToolService/manualReviewToolService.ts @@ -1183,6 +1183,16 @@ export class ManualReviewToolService { return this.decisionAnalytics.getRecentDecisions(opts); } + async getDecisionsForActivityFeed(opts: { + userPermissions: UserPermission[]; + orgId: string; + input: RecentDecisionsFilterInput; + cursor?: { ts: Date; id: string }; + limit: number; + }) { + return this.decisionAnalytics.getDecisionsForActivityFeed(opts); + } + async getSkippedJobsForRecentDecisions(opts: { orgId: string; input: Omit; diff --git a/server/services/manualReviewToolService/modules/DecisionAnalytics.test.ts b/server/services/manualReviewToolService/modules/DecisionAnalytics.test.ts new file mode 100644 index 000000000..20c13b767 --- /dev/null +++ b/server/services/manualReviewToolService/modules/DecisionAnalytics.test.ts @@ -0,0 +1,235 @@ +import { sql } from 'kysely'; +import { uid } from 'uid'; +import { v1 as uuidv1 } from 'uuid'; + +import { type Dependencies } from '../../../iocContainer/index.js'; +import createMrtQueue from '../../../test/fixtureHelpers/createMrtQueue.js'; +import createOrg from '../../../test/fixtureHelpers/createOrg.js'; +import createUser from '../../../test/fixtureHelpers/createUser.js'; +import makeDummyMrtJobPayload from '../../../test/fixtureHelpers/makeDummyMrtJobPayload.js'; +import { makeTransactionalTestWithFixture } from '../../../test/harness/transactionalTest.js'; +import { + type JobId, + type ManualReviewJob, +} from '../manualReviewToolService.js'; + +describe('DecisionAnalytics', () => { + const testWithDecisions = () => + makeTransactionalTestWithFixture(async ({ deps }) => { + const { org } = await createOrg( + { + KyselyPg: deps.KyselyPg, + ModerationConfigService: deps.ModerationConfigService, + ApiKeyService: deps.ApiKeyService, + }, + uid(), + ); + const { user } = await createUser(deps.KyselyPg, org.id); + const { queue } = await createMrtQueue({ + orgId: org.id, + mrtService: deps.ManualReviewToolService, + userId: user.id, + }); + + // `manual_review_decisions.created_at` is `GeneratedAlways` in the + // Kysely schema (DB default now()), so a typed `.values()` insert + // can't set it. `insertDecision` inserts the row, then overwrites + // `created_at` with a raw update so tests can control ordering + // deterministically. + const insertDecision = async (createdAt: Date) => { + const id = uuidv1(); + const jobPayload: ManualReviewJob = { + ...makeDummyMrtJobPayload(), + id: uuidv1() as JobId, + orgId: org.id, + }; + await deps.KyselyPg.insertInto( + 'manual_review_tool.manual_review_decisions', + ) + .values({ + id, + job_payload: jobPayload, + queue_id: queue.id, + reviewer_id: user.id, + org_id: org.id, + decision_components: [{ type: 'IGNORE' }], + related_actions: [], + }) + .execute(); + await sql` + update manual_review_tool.manual_review_decisions + set created_at = ${createdAt} + where id = ${id} + `.execute(deps.KyselyPg); + return id; + }; + + return { + org, + user, + queue, + mrtService: deps.ManualReviewToolService, + insertDecision, + }; + }); + + const baseTime = new Date('2026-01-01T00:00:00.000Z'); + const minutesAfterBase = (minutes: number) => + new Date(baseTime.getTime() + minutes * 60_000); + + /** + * Pages through the entire activity feed for `orgId` via + * `getDecisionsForActivityFeed`, following each page's last row into the + * next cursor, and returns every decision id encountered in page order. + */ + const collectAllPages = async (opts: { + mrtService: Dependencies['ManualReviewToolService']; + orgId: string; + limit: number; + }) => { + const { mrtService, orgId, limit } = opts; + let ids: string[] = []; + let cursor: { ts: Date; id: string } | undefined; + for (;;) { + const page = await mrtService.getDecisionsForActivityFeed({ + userPermissions: [], + orgId, + input: { page: 0 }, + cursor, + limit, + }); + ids = [...ids, ...page.map((decision) => decision.id)]; + if (page.length < limit) { + break; + } + const last = page[page.length - 1]; + cursor = { ts: last.createdAt, id: last.id }; + } + return ids; + }; + + testWithDecisions()( + 'pages a real cursor across the id-cast boundary without erroring', + async ({ org, mrtService, insertDecision }) => { + const ids = await Promise.all([ + insertDecision(minutesAfterBase(0)), + insertDecision(minutesAfterBase(1)), + insertDecision(minutesAfterBase(2)), + ]); + + const firstPage = await mrtService.getDecisionsForActivityFeed({ + userPermissions: [], + orgId: org.id, + input: { page: 0 }, + limit: 2, + }); + expect(firstPage.map((d) => d.id)).toEqual([ids[2], ids[1]]); + + const last = firstPage[firstPage.length - 1]; + // The cursor's `id` is a real uuid pulled from a previous row, exactly + // as a real caller would pass it. If the `::uuid` cast on the cursor + // predicate were missing or wrong, this call raises + // `22P02 invalid input syntax for type uuid` instead of returning. + const secondPage = await mrtService.getDecisionsForActivityFeed({ + userPermissions: [], + orgId: org.id, + input: { page: 0 }, + cursor: { ts: last.createdAt, id: last.id }, + limit: 2, + }); + expect(secondPage.map((d) => d.id)).toEqual([ids[0]]); + }, + ); + + testWithDecisions()( + 'returns every seeded decision exactly once across pages, with no gaps', + async ({ org, mrtService, insertDecision }) => { + const ids = await Promise.all([ + insertDecision(minutesAfterBase(0)), + insertDecision(minutesAfterBase(1)), + insertDecision(minutesAfterBase(2)), + insertDecision(minutesAfterBase(3)), + insertDecision(minutesAfterBase(4)), + ]); + + const collected = await collectAllPages({ + mrtService, + orgId: org.id, + limit: 2, + }); + + expect(collected.length).toEqual(ids.length); + expect(new Set(collected).size).toEqual(ids.length); + expect(new Set(collected)).toEqual(new Set(ids)); + }, + ); + + testWithDecisions()( + 'neither loses nor repeats rows that share a created_at, across a page boundary that splits them', + async ({ org, mrtService, insertDecision }) => { + const tiedTimestamp = minutesAfterBase(5); + const earlier = await insertDecision(minutesAfterBase(0)); + const tiedIds = await Promise.all([ + insertDecision(tiedTimestamp), + insertDecision(tiedTimestamp), + ]); + const allIds = [earlier, ...tiedIds]; + + // limit: 1 forces the two tied rows onto separate pages, exercising + // the `id` tie-break leg of the `(created_at, id)` sort key. + const collected = await collectAllPages({ + mrtService, + orgId: org.id, + limit: 1, + }); + + expect(collected.length).toEqual(allIds.length); + expect(new Set(collected)).toEqual(new Set(allIds)); + }, + ); + + testWithDecisions()( + 'returns the newest page in created_at DESC order when there is no cursor', + async ({ org, mrtService, insertDecision }) => { + const oldest = await insertDecision(minutesAfterBase(0)); + const middle = await insertDecision(minutesAfterBase(1)); + const newest = await insertDecision(minutesAfterBase(2)); + + const page = await mrtService.getDecisionsForActivityFeed({ + userPermissions: [], + orgId: org.id, + input: { page: 0 }, + limit: 10, + }); + + expect(page.map((d) => d.id)).toEqual([newest, middle, oldest]); + const timestamps = page.map((d) => d.createdAt.getTime()); + expect(timestamps).toEqual([...timestamps].sort((a, b) => b - a)); + }, + ); + + testWithDecisions()( + 'getRecentDecisions still returns offset-paged decisions in created_at DESC order after the extraction', + async ({ org, mrtService, insertDecision }) => { + const oldest = await insertDecision(minutesAfterBase(0)); + const middle = await insertDecision(minutesAfterBase(1)); + const newest = await insertDecision(minutesAfterBase(2)); + + const firstPage = await mrtService.getRecentDecisions({ + userPermissions: [], + orgId: org.id, + input: { page: 0 }, + }); + expect(firstPage.map((d) => d.id)).toEqual([newest, middle, oldest]); + + // Offset paging: page 100 rows at a time, so page 1 is past the end of + // a 3-row result set and must come back empty. + const secondPage = await mrtService.getRecentDecisions({ + userPermissions: [], + orgId: org.id, + input: { page: 1 }, + }); + expect(secondPage).toEqual([]); + }, + ); +}); diff --git a/server/services/manualReviewToolService/modules/DecisionAnalytics.ts b/server/services/manualReviewToolService/modules/DecisionAnalytics.ts index bcf319d6c..afe942528 100644 --- a/server/services/manualReviewToolService/modules/DecisionAnalytics.ts +++ b/server/services/manualReviewToolService/modules/DecisionAnalytics.ts @@ -9,7 +9,11 @@ import { type ManualReviewJob, type ManualReviewJobEnqueueSource, } from '../manualReviewToolService.js'; -import { type ManualReviewDecisionType } from './JobDecisioning.js'; +import { + type ManualReviewDecisionComponent, + type ManualReviewDecisionRelatedAction, + type ManualReviewDecisionType, +} from './JobDecisioning.js'; export type RecentDecisionsFilterInput = { userSearchString?: string; @@ -233,7 +237,14 @@ export default class DecisionAnalytics { .execute(); } - async getRecentDecisions(opts: { + /** + * Shared query body for `getRecentDecisions` and + * `getDecisionsForActivityFeed`: the select list, every filter predicate, + * the `userSearchString` branch, and the NCMEC permission gate. Stops + * short of `orderBy`/`limit`/`offset` so each caller applies its own + * paging. + */ + private buildRecentDecisionsQuery(opts: { userPermissions: UserPermission[]; orgId: string; input: RecentDecisionsFilterInput; @@ -247,156 +258,192 @@ export default class DecisionAnalytics { queueIds, startTime, endTime, - page, } = input; - const limit = 100; - const decisions = await this.pgQuery - .selectFrom('manual_review_tool.manual_review_decisions') - .select([ - 'id', - 'queue_id', - 'reviewer_id', - 'decision_components', - 'related_actions', - 'created_at', - sql`((job_payload->'payload'::text)->'item'::text) -> 'itemId'::text`.as( - 'item_id', - ), - sql`(((job_payload->'payload'::text)->'item'::text) -> 'itemTypeIdentifier'::text) ->> 'id'::text`.as( - 'item_type_id', - ), - 'decision_reason', - sql`(job_payload->>'id')::text`.as('job_id'), - ]) - .where('org_id', '=', orgId) - .where(({ eb, selectFrom }) => { - return eb.and([ - ...(startTime ? [eb('created_at', '>=', new Date(startTime))] : []), - ...(endTime ? [eb('created_at', '<=', new Date(endTime))] : []), - ...(queueIds && queueIds.length > 0 - ? [eb('queue_id', 'in', queueIds)] - : []), - ...(reviewerIds && reviewerIds.length > 0 - ? [eb('reviewer_id', 'in', reviewerIds)] - : []), - ...(policyIds - ? [ - eb.exists( - selectFrom( - sql`unnest(manual_review_tool.manual_review_decisions.decision_components)`.as( - 'decision_component', - ), - ) - .selectAll() - .where( - eb.or( - policyIds.map((policyId) => - eb( - sql`decision_component->>'policies'`, - 'like', - `%"${policyId}"%`, + return ( + this.pgQuery + .selectFrom('manual_review_tool.manual_review_decisions') + .select([ + 'id', + 'queue_id', + 'reviewer_id', + 'decision_components', + 'related_actions', + 'created_at', + sql`((job_payload->'payload'::text)->'item'::text) -> 'itemId'::text`.as( + 'item_id', + ), + sql`(((job_payload->'payload'::text)->'item'::text) -> 'itemTypeIdentifier'::text) ->> 'id'::text`.as( + 'item_type_id', + ), + 'decision_reason', + sql`(job_payload->>'id')::text`.as('job_id'), + ]) + .where('org_id', '=', orgId) + .where(({ eb, selectFrom }) => { + return eb.and([ + ...(startTime ? [eb('created_at', '>=', new Date(startTime))] : []), + ...(endTime ? [eb('created_at', '<=', new Date(endTime))] : []), + ...(queueIds && queueIds.length > 0 + ? [eb('queue_id', 'in', queueIds)] + : []), + ...(reviewerIds && reviewerIds.length > 0 + ? [eb('reviewer_id', 'in', reviewerIds)] + : []), + ...(policyIds + ? [ + eb.exists( + selectFrom( + sql`unnest(manual_review_tool.manual_review_decisions.decision_components)`.as( + 'decision_component', + ), + ) + .selectAll() + .where( + eb.or( + policyIds.map((policyId) => + eb( + sql`decision_component->>'policies'`, + 'like', + `%"${policyId}"%`, + ), ), ), ), - ), - ), - ] - : []), - ...(decisionsFilter - ? [ - eb.or( - decisionsFilter.flatMap((it) => [ - eb.exists( - selectFrom( - sql`unnest(manual_review_tool.manual_review_decisions.decision_components)`.as( - 'decision_component', - ), - ) - .selectAll() - .where( - sql`decision_component->>'type'`, - '=', - it.type, + ), + ] + : []), + ...(decisionsFilter + ? [ + eb.or( + decisionsFilter.flatMap((it) => [ + eb.exists( + selectFrom( + sql`unnest(manual_review_tool.manual_review_decisions.decision_components)`.as( + 'decision_component', + ), ) - .$if(it.actionIds !== undefined, (qb) => - qb.where( - eb.or( - it.actionIds!.map((actionId) => - eb( - sql`decision_component->>'actions'`, - 'like', - `%"${actionId}"%`, + .selectAll() + .where( + sql`decision_component->>'type'`, + '=', + it.type, + ) + .$if(it.actionIds !== undefined, (qb) => + qb.where( + eb.or( + it.actionIds!.map((actionId) => + eb( + sql`decision_component->>'actions'`, + 'like', + `%"${actionId}"%`, + ), ), ), ), ), - ), - ), - ]), - ), - ] - : []), - ]); - }) - .$if(userSearchString !== undefined, (qb) => - // See https://stackoverflow.com/a/55607847 - qb.where(({ and, eb, val }) => - and([ - eb('created_at', '>', val(new Date(Date.now() - 3 * MONTH_MS))), - eb( - sql`(manual_review_tool.manual_review_decisions.job_payload->'payload'->'item'->>'itemId')`, - '=', - // Above, the 'itemId' field is of type jsonb, so we cast it to a string using ::text, but that - // cast will leave quotes around the resulting string because it's just stringifying what it thinks - // is a jsonb object. The easiest way to handle this is to just add quotes around the userSearchString - // to match the quotes in the value above. - val(`${userSearchString}`), + ), + ]), + ), + ] + : []), + ]); + }) + .$if(userSearchString !== undefined, (qb) => + // See https://stackoverflow.com/a/55607847 + qb.where(({ and, eb, val }) => + and([ + eb('created_at', '>', val(new Date(Date.now() - 3 * MONTH_MS))), + eb( + sql`(manual_review_tool.manual_review_decisions.job_payload->'payload'->'item'->>'itemId')`, + '=', + // Above, the 'itemId' field is of type jsonb, so we cast it to a string using ::text, but that + // cast will leave quotes around the resulting string because it's just stringifying what it thinks + // is a jsonb object. The easiest way to handle this is to just add quotes around the userSearchString + // to match the quotes in the value above. + val(`${userSearchString}`), + ), + ]), + ), + ) + // If the user doesn't have the VIEW_CHILD_SAFETY_DATA permission, filter out decisions on + // all NCMEC jobs + .$if( + !userPermissions.includes(UserPermission.VIEW_CHILD_SAFETY_DATA), + (qb) => + qb.where(({ eb, val }) => + eb( + sql`(job_payload->'payload'->'kind')::text`, + '!=', + val('"NCMEC"'), + ), ), - ]), + ) + ); + } + + async getRecentDecisions(opts: { + userPermissions: UserPermission[]; + orgId: string; + input: RecentDecisionsFilterInput; + }) { + const { userPermissions, orgId, input } = opts; + const { page } = input; + const limit = 100; + const decisions = await this.buildRecentDecisionsQuery({ + userPermissions, + orgId, + input, + }) + .orderBy('created_at', 'desc') + .limit(limit) + .offset(page * limit) + .execute(); + return decisions.map(mapDecisionRow); + } + + /** + * Cursor-paged twin of `getRecentDecisions`, for the merged activity feed. + * + * Kept separate rather than folded into `getRecentDecisions`: that query is + * offset-paged and consumed elsewhere, and changing its contract would + * break callers this feature has no reason to touch. + * + * Ordering is `(created_at, id)` descending. The id leg makes each row's + * position unique so the merged feed's cursor is exact — a + * `created_at`-only bound would drop or repeat decisions sharing an + * instant. + */ + async getDecisionsForActivityFeed(opts: { + userPermissions: UserPermission[]; + orgId: string; + input: RecentDecisionsFilterInput; + // The DECISIONS side of a per-store cursor. The caller must never pass + // the actions side's id here: `id` is uuid, and a non-uuid string raises + // 22P02 invalid input syntax for type uuid. + cursor?: { ts: Date; id: string }; + limit: number; + }) { + const { userPermissions, orgId, input, cursor, limit } = opts; + const decisions = await this.buildRecentDecisionsQuery({ + userPermissions, + orgId, + input, + }) + .$if(cursor !== undefined, (qb) => + qb.where( + sql`(created_at, id)`, + '<', + // `id` is uuid. The cast is load-bearing: without it Postgres + // infers the bind type from the column and a non-uuid string + // raises 22P02. + sql`(${cursor!.ts}, ${cursor!.id}::uuid)`, ), ) - // If the user doesn't have the VIEW_CHILD_SAFETY_DATA permission, filter out decisions on - // all NCMEC jobs - .$if( - !userPermissions.includes(UserPermission.VIEW_CHILD_SAFETY_DATA), - (qb) => - qb.where(({ eb, val }) => - eb( - sql`(job_payload->'payload'->'kind')::text`, - '!=', - val('"NCMEC"'), - ), - ), - ) .orderBy('created_at', 'desc') + .orderBy('id', 'desc') .limit(limit) - .offset(page * limit) .execute(); - return decisions.map((decision) => ({ - id: decision.id, - itemId: decision.item_id, - itemTypeId: decision.item_type_id, - queueId: decision.queue_id, - reviewerId: decision.reviewer_id, - decisions: decision.decision_components.map((it) => { - if (it.type !== 'CUSTOM_ACTION') { - return it; - } - return { - ...it, - actionIds: it.actions.map((it) => it.id), - policyIds: it.policies.map((it) => it.id), - itemTypeId: it.itemTypeId, - }; - }), - relatedActions: decision.related_actions.map((action) => ({ - ...action, - type: 'RELATED_ACTION' as const, - })), - createdAt: decision.created_at, - decisionReason: decision.decision_reason, - jobId: decision.job_id, - })); + return decisions.map(mapDecisionRow); } async getResolvedJobCounts(input: JobCountsInput) { @@ -526,6 +573,57 @@ export default class DecisionAnalytics { } } +/** + * Row shape shared by `getRecentDecisions` and `getDecisionsForActivityFeed`, + * as selected by `DecisionAnalytics['buildRecentDecisionsQuery']`. + */ +type RecentDecisionRow = { + id: string; + item_id: string; + item_type_id: string; + queue_id: string; + reviewer_id: string | null; + decision_components: ManualReviewDecisionComponent[]; + related_actions: ManualReviewDecisionRelatedAction[]; + created_at: Date; + decision_reason: string | null; + job_id: string; +}; + +/** + * Projects a raw `manual_review_decisions` row into the shape both + * `getRecentDecisions` and `getDecisionsForActivityFeed` return. Moved + * verbatim out of `getRecentDecisions` — do not alter any field here without + * updating both callers' consumers. + */ +function mapDecisionRow(decision: RecentDecisionRow) { + return { + id: decision.id, + itemId: decision.item_id, + itemTypeId: decision.item_type_id, + queueId: decision.queue_id, + reviewerId: decision.reviewer_id, + decisions: decision.decision_components.map((it) => { + if (it.type !== 'CUSTOM_ACTION') { + return it; + } + return { + ...it, + actionIds: it.actions.map((it) => it.id), + policyIds: it.policies.map((it) => it.id), + itemTypeId: it.itemTypeId, + }; + }), + relatedActions: decision.related_actions.map((action) => ({ + ...action, + type: 'RELATED_ACTION' as const, + })), + createdAt: decision.created_at, + decisionReason: decision.decision_reason, + jobId: decision.job_id, + }; +} + /** * These options are meant to be passed to the psql `DATE_TRUNC()` * function, so the strings must conform to valid `field`