diff --git a/apps/backend/src/db/migrations/110_order_items_fulfillment_product_index.sql b/apps/backend/src/db/migrations/110_order_items_fulfillment_product_index.sql new file mode 100644 index 00000000..db668b2c --- /dev/null +++ b/apps/backend/src/db/migrations/110_order_items_fulfillment_product_index.sql @@ -0,0 +1,32 @@ +-- 110_order_items_fulfillment_product_index.sql +-- +-- 背景(2026-10-30 生产实测): +-- cost-finance-sync 每 15 分钟执行一次「人工单补商品维度」的更新,原写法是无条件: +-- UPDATE cost_snapshots cs SET canonical_product_id = oi.canonical_product_id +-- FROM order_items oi +-- WHERE cs.order_item_id = oi.id +-- AND cs.canonical_product_id IS NULL +-- AND oi.fulfillment_mode = 'manual' +-- AND oi.canonical_product_id IS NOT NULL +-- 规划器只能走 BitmapAnd(idx_order_items_canonical_product, idx_order_items_fulfillment_mode) +-- 再回表,而 order_items 堆已经 899MB(43 万行),每次要在其中随机读上百 MB: +-- 连续两次实测 15.2s / 16.5s,直接撞 statement_timeout,后面的销售事实重算与 +-- 影子总账同步整条被跳过。 +-- +-- 两条修复并行: +-- 1. 语句本身收口到「本批刚归一过的 order_item_id」(见 +-- manual-product-normalization-service.ts),彻底不再全表扫; +-- 2. 本迁移补一个覆盖索引,让所有「按履约方式 + 有无商品」筛选的查询都能 index-only scan。 +-- order_items 的其它读路径(人工渠道回填、成本中心商品维度核对)同样吃这几个条件。 +-- +-- 迁移在 singleTransaction 里执行(见 migrate.ts),因此这里用普通 CREATE INDEX; +-- 生产按 086/107/108 的惯例先手工执行 CREATE INDEX CONCURRENTLY 预建, +-- 部署时 IF NOT EXISTS 直接跳过,不锁表。 + +CREATE INDEX IF NOT EXISTS idx_order_items_fulfillment_product + ON order_items (fulfillment_mode, canonical_product_id) + INCLUDE (id) + WHERE canonical_product_id IS NOT NULL; + +COMMENT ON INDEX idx_order_items_fulfillment_product IS + '按履约方式 + 商品维度筛选的覆盖索引(2026-10-30 财务同步 15s 超时修复)'; diff --git a/apps/backend/src/db/migrations/111_worker_risk_bans.sql b/apps/backend/src/db/migrations/111_worker_risk_bans.sql new file mode 100644 index 00000000..634140a1 --- /dev/null +++ b/apps/backend/src/db/migrations/111_worker_risk_bans.sql @@ -0,0 +1,51 @@ +-- 111_worker_risk_bans.sql +-- +-- 后台「接单平台 → 风控」新增**手动封禁**能力(IP / 账号)。 +-- +-- 背景(2026-10-30):大厅被一个无 UA、无 referer 的脚本按 pageSize=100 轮询, +-- 10 分钟 238 次请求占全站 21%,全部 401 + 324 次 404 探测。原有风控只会按阈值 +-- 自动升级(二级惩罚 → 三级封号),运营没有任何"现在就把它挡住"的手段; +-- 而内存里的惩罚状态进程重启即清零,关不住跨容器重建的刷子。 +-- +-- 设计取舍: +-- · 一行 = 一条生效中的封禁;撤销走 revoked_at,不物理删除,便于事后追溯。 +-- · expires_at 为 NULL 表示永久;中间件每次只加载「未撤销且未过期」的条目。 +-- · target_type='ip' 时 target_value 就是 IP;'worker' 时 target_value 是账号 id 字符串, +-- worker_id 冗余存引用,账号被删时整条封禁一起删除(ON DELETE CASCADE)。 +-- · 同类型同目标只允许一条生效中的记录(部分唯一索引),避免列表里出现重复行。 +-- · 被拒请求不落库,拦截只读进程内缓存;写入时才落这张表(低频)。 + +CREATE TABLE IF NOT EXISTS worker_risk_bans ( + id BIGSERIAL PRIMARY KEY, + target_type TEXT NOT NULL CHECK (target_type IN ('ip', 'worker')), + target_value TEXT NOT NULL, + worker_id BIGINT REFERENCES worker_users(id) ON DELETE CASCADE, + worker_username TEXT NOT NULL DEFAULT '', + reason TEXT NOT NULL DEFAULT '', + -- NULL = 永久封禁 + expires_at TIMESTAMPTZ, + created_by BIGINT, + created_by_username TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + revoked_at TIMESTAMPTZ, + revoked_by_username TEXT NOT NULL DEFAULT '', + revoke_reason TEXT NOT NULL DEFAULT '', + CONSTRAINT worker_risk_bans_target_check CHECK ( + (target_type = 'worker' AND worker_id IS NOT NULL) + OR (target_type = 'ip' AND worker_id IS NULL) + ) +); + +-- 同一目标只允许一条生效中的封禁 +CREATE UNIQUE INDEX IF NOT EXISTS uq_worker_risk_bans_active + ON worker_risk_bans (target_type, target_value) + WHERE revoked_at IS NULL; + +-- 列表按时间倒序分页 +CREATE INDEX IF NOT EXISTS idx_worker_risk_bans_created + ON worker_risk_bans (created_at DESC, id DESC); + +-- 中间件冷启动时只捞「未撤销」的条目,再过一遍过期时间 +CREATE INDEX IF NOT EXISTS idx_worker_risk_bans_active_expiry + ON worker_risk_bans (expires_at) + WHERE revoked_at IS NULL; diff --git a/apps/backend/src/middleware/worker-ban-guard.test.ts b/apps/backend/src/middleware/worker-ban-guard.test.ts new file mode 100644 index 00000000..f9ab7812 --- /dev/null +++ b/apps/backend/src/middleware/worker-ban-guard.test.ts @@ -0,0 +1,138 @@ +import assert from 'node:assert/strict' +import test from 'node:test' +import type { Request, Response } from 'express' + +import { + createWorkerBanGuardMiddleware, + normalizeWorkerIp, + resetWorkerBanGuardForTest, + type WorkerBanLoader, +} from './worker-ban-guard.js' + +function buildMockReq(options: { ip?: string; workerId?: number }): Request { + const headers: Record = {} + if (options.workerId) { + const payload = Buffer.from( + JSON.stringify({ typ: 'worker', uid: options.workerId }), + 'utf8', + ).toString('base64url') + headers.authorization = `Bearer ${payload}.mockSignature` + } + return { + ip: options.ip || '127.0.0.1', + headers, + method: 'GET', + originalUrl: '/api/v1/worker/hall/orders', + socket: { remoteAddress: options.ip || '127.0.0.1' }, + } as unknown as Request +} + +function buildMockRes() { + const state = { status: 200, payload: null as unknown } + const res = { + status(code: number) { + state.status = code + return res + }, + setHeader() { + return res + }, + json(body: unknown) { + state.payload = body + return res + }, + } as unknown as Response + return { + get status() { + return state.status + }, + get payload() { + return state.payload + }, + res, + } +} + +function errorCodeOf(payload: unknown): string { + const body = payload as { error?: { errorCode?: string }; errorCode?: string } | null + return body?.error?.errorCode || body?.errorCode || '' +} + +test('IP 封禁:命中即 403,且不放行到后续中间件', async () => { + resetWorkerBanGuardForTest() + const loadBans: WorkerBanLoader = async () => [ + { target_type: 'ip', target_value: '::ffff:203.0.113.9' }, + ] + const middleware = createWorkerBanGuardMiddleware({ loadBans }) + const req = buildMockReq({ ip: '203.0.113.9' }) + const res = buildMockRes() + let nextCalled = false + + await middleware(req, res.res, () => { + nextCalled = true + }) + + assert.equal(nextCalled, false, '被封 IP 不能放行') + assert.equal(res.status, 403) + assert.equal(errorCodeOf(res.payload), 'worker_ip_banned') +}) + +test('账号封禁:按 token 里的 workerId 拦截(即使换了 IP)', async () => { + resetWorkerBanGuardForTest() + const loadBans: WorkerBanLoader = async () => [ + { target_type: 'worker', target_value: '42', worker_id: '42' }, + ] + const middleware = createWorkerBanGuardMiddleware({ loadBans }) + const req = buildMockReq({ ip: '198.51.100.7', workerId: 42 }) + const res = buildMockRes() + let nextCalled = false + + await middleware(req, res.res, () => { + nextCalled = true + }) + + assert.equal(nextCalled, false, '被封账号不能放行') + assert.equal(res.status, 403) + assert.equal(errorCodeOf(res.payload), 'worker_account_banned') +}) + +test('未命中封禁时正常放行', async () => { + resetWorkerBanGuardForTest() + const middleware = createWorkerBanGuardMiddleware({ + loadBans: async () => [{ target_type: 'ip', target_value: '203.0.113.9' }], + }) + const req = buildMockReq({ ip: '198.51.100.7', workerId: 7 }) + const res = buildMockRes() + let nextCalled = false + + await middleware(req, res.res, () => { + nextCalled = true + }) + + assert.equal(nextCalled, true) + assert.equal(res.status, 200) +}) + +test('加载封禁列表失败时 fail-open(不能因为数据库抖动挡死整个接单端)', async () => { + resetWorkerBanGuardForTest() + const middleware = createWorkerBanGuardMiddleware({ + loadBans: async () => { + throw new Error('db down') + }, + }) + const req = buildMockReq({ ip: '203.0.113.9' }) + const res = buildMockRes() + let nextCalled = false + + await middleware(req, res.res, () => { + nextCalled = true + }) + + assert.equal(nextCalled, true) +}) + +test('normalizeWorkerIp 归一化 IPv6-mapped 与方括号写法', () => { + assert.equal(normalizeWorkerIp('::ffff:203.0.113.9'), '203.0.113.9') + assert.equal(normalizeWorkerIp('[2001:DB8::1]'), '2001:db8::1') + assert.equal(normalizeWorkerIp(' 203.0.113.9 '), '203.0.113.9') +}) diff --git a/apps/backend/src/middleware/worker-ban-guard.ts b/apps/backend/src/middleware/worker-ban-guard.ts new file mode 100644 index 00000000..dd892896 --- /dev/null +++ b/apps/backend/src/middleware/worker-ban-guard.ts @@ -0,0 +1,170 @@ +import type { NextFunction, Request, Response } from 'express' + +import { listActiveWorkerRiskBans } from '../repositories/worker-platform/worker-risk-ban-repo.js' +import { buildErrorPayload, createHttpError } from '../utils/http.js' +import { logError, logWarn } from '../utils/logger.js' +import { extractFastWorkerId, resolveClientIp } from './worker-risk-control.js' + +/** + * 手动封禁(IP / 账号)拦截中间件。 + * + * 与 worker-risk-control 的分工: + * · worker-risk-control:按**行为**自动升级(限流 → 惩罚 → 自动封号),状态在进程内存; + * · 本中间件:按**人工决定**拦截,事实存在 worker_risk_bans 表,进程重启不丢。 + * + * 性能约束(这是被刷场景下最热的代码): + * · 每个请求只做两次 Set/Map 命中,绝不查库; + * · 全量封禁列表进程内缓存 30 秒,后台增删封禁时立即失效(同进程); + * · 加载失败 **fail-open**:宁可短暂漏挡,也不能因为一次数据库抖动把整个接单端挡死。 + * + * 拦截范围:挂在整个 /api/v1/worker 路由之前,因此登录、大厅、订单、上传都会被挡; + * 被封 IP 连登录都进不来——这正是运营想要的"现在就挡住它"。 + */ + +type WorkerBanSnapshot = { + ips: Set + workers: Set + expiresAt: number +} + +const CACHE_TTL_MS = 30_000 + +/** + * 归一化 IP:去掉 IPv6-mapped 前缀与方括号、统一小写。 + * 反向代理(Caddy)对 IPv4 请求可能给出 `::ffff:1.2.3.4`,不归一化会导致封了却挡不住。 + */ +export function normalizeWorkerIp(value: unknown): string { + return String(value || '') + .trim() + .replace(/^\[|\]$/g, '') + .replace(/^::ffff:/i, '') + .toLowerCase() +} + +let snapshot: WorkerBanSnapshot | null = null +let inflightLoad: Promise | null = null + +/** 后台增删封禁后调用,让本进程立即生效,不必等 TTL。 */ +export function invalidateWorkerBanCache(): void { + snapshot = null +} + +async function loadSnapshot(): Promise { + const rows = await listActiveWorkerRiskBans() + const ips = new Set() + const workers = new Set() + for (const row of rows) { + if (row.target_type === 'ip') { + const ip = normalizeWorkerIp(row.target_value) + if (ip) ips.add(ip) + continue + } + const workerId = Number(row.worker_id || row.target_value) + if (Number.isInteger(workerId) && workerId > 0) workers.add(workerId) + } + snapshot = { ips, workers, expiresAt: Date.now() + CACHE_TTL_MS } +} + +async function ensureSnapshot(): Promise { + const now = Date.now() + if (snapshot && snapshot.expiresAt > now) return snapshot + if (!inflightLoad) { + inflightLoad = loadSnapshot() + .catch((error) => { + // fail-open:记录一次,保留旧的快照(如果有),不要阻塞请求 + logError('[worker/ban-guard]', '加载手动封禁列表失败,本进程暂时放行', { + error: error instanceof Error ? error.message : String(error), + }) + if (snapshot) snapshot.expiresAt = Date.now() + CACHE_TTL_MS + }) + .finally(() => { + inflightLoad = null + }) + } + await inflightLoad + return snapshot +} + +/** 同一目标 60 秒内只打一条日志,避免被封的脚本刷爆日志盘。 */ +const lastLogAt = new Map() +const LOG_INTERVAL_MS = 60_000 + +function shouldLog(key: string): boolean { + const now = Date.now() + const last = lastLogAt.get(key) || 0 + if (now - last < LOG_INTERVAL_MS) return false + lastLogAt.set(key, now) + if (lastLogAt.size > 5_000) lastLogAt.clear() + return true +} + +export type WorkerBanLoader = () => Promise< + Array<{ target_type: 'ip' | 'worker'; target_value: string; worker_id?: string | null }> +> + +export function createWorkerBanGuardMiddleware(deps: { loadBans?: WorkerBanLoader } = {}) { + // 测试注入用:不传则走真实仓储 + 30 秒缓存 + const injected = deps.loadBans + return async (req: Request, res: Response, next: NextFunction): Promise => { + const ip = normalizeWorkerIp(resolveClientIp(req)) + const workerId = extractFastWorkerId(req) + + let bannedIp = false + let bannedWorker = false + + if (injected) { + try { + const rows = await injected() + bannedIp = rows.some( + (row) => row.target_type === 'ip' && normalizeWorkerIp(row.target_value) === ip, + ) + bannedWorker = + workerId !== null && + rows.some( + (row) => + row.target_type === 'worker' && + Number(row.worker_id || row.target_value) === workerId, + ) + } catch { + bannedIp = false + bannedWorker = false + } + } else { + const current = await ensureSnapshot() + bannedIp = Boolean(current?.ips.has(ip)) + bannedWorker = workerId !== null && Boolean(current?.workers.has(workerId)) + } + + if (!bannedIp && !bannedWorker) { + next() + return + } + + const targetKey = bannedIp ? `ip:${ip}` : `worker:${workerId}` + if (shouldLog(targetKey)) { + logWarn('[worker/ban-guard]', '命中手动封禁,拒绝请求', { + targetKey, + ip, + workerId, + method: req.method, + originalUrl: req.originalUrl, + }) + } + + const error = createHttpError( + bannedIp ? '该网络出口已被后台封禁,无法访问接单端' : '该接单账号已被后台手动封禁', + { + statusCode: 403, + errorCode: bannedIp ? 'worker_ip_banned' : 'worker_account_banned', + }, + ) + res.status(403).json(buildErrorPayload(error, '已被后台手动封禁')) + } +} + +/** 测试辅助:清空缓存与日志节流表。 */ +export function resetWorkerBanGuardForTest(): void { + snapshot = null + inflightLoad = null + lastLogAt.clear() +} diff --git a/apps/backend/src/repositories/worker-platform/worker-risk-ban-repo.ts b/apps/backend/src/repositories/worker-platform/worker-risk-ban-repo.ts new file mode 100644 index 00000000..56e9e5d0 --- /dev/null +++ b/apps/backend/src/repositories/worker-platform/worker-risk-ban-repo.ts @@ -0,0 +1,186 @@ +import { query } from '../../db/client.js' + +/** + * 风控「手动封禁」(IP / 账号)的数据访问层。 + * + * 与 worker_risk_events 的分工: + * · worker_risk_bans = **生效中的封禁事实**,中间件据此拦截请求; + * · worker_risk_events = **留痕**,每次人工封禁/解封都追加一条,便于事后追溯。 + * + * 读取侧只有「加载全部生效封禁」是热路径(进程内缓存 30 秒兜底),其余都是后台低频分页查询。 + */ + +export type WorkerRiskBanTargetType = 'ip' | 'worker' + +export type WorkerRiskBanRow = { + id: string + target_type: WorkerRiskBanTargetType + target_value: string + worker_id: string | null + worker_username: string + reason: string + expires_at: Date | null + created_by: string | null + created_by_username: string + created_at: Date + revoked_at: Date | null + revoked_by_username: string + revoke_reason: string +} + +/** 中间件冷启动/缓存过期时加载:只返回「未撤销且未过期」的封禁。 */ +export async function listActiveWorkerRiskBans(): Promise { + const result = await query( + ` + SELECT * + FROM worker_risk_bans + WHERE revoked_at IS NULL + AND (expires_at IS NULL OR expires_at > NOW()) + ORDER BY id DESC + LIMIT 5000 + `, + ) + return result.rows +} + +export type ListWorkerRiskBansInput = { + limit: number + offset?: number | undefined + targetType?: string | undefined + keyword?: string | undefined + /** true = 只看生效中;false = 只看已撤销/已过期;不传 = 全部 */ + activeOnly?: boolean | undefined +} + +export async function listWorkerRiskBans( + input: ListWorkerRiskBansInput, +): Promise { + const { where, params } = buildBanFilters(input) + const limit = Math.min(200, Math.max(1, Math.trunc(Number(input.limit) || 20))) + const offset = Math.max(0, Math.trunc(Number(input.offset) || 0)) + const result = await query( + ` + SELECT * + FROM worker_risk_bans + ${where} + ORDER BY revoked_at NULLS FIRST, id DESC + LIMIT $${params.length + 1} OFFSET $${params.length + 2} + `, + [...params, limit, offset], + ) + return result.rows +} + +export async function countWorkerRiskBans( + input: Omit, +): Promise { + const { where, params } = buildBanFilters(input) + const result = await query<{ total: string }>( + `SELECT COUNT(*)::text AS total FROM worker_risk_bans ${where}`, + params, + ) + return Number(result.rows[0]?.total || 0) +} + +/** + * 新增一条封禁。 + * + * 同一目标已有生效中的封禁时返回 null(依赖部分唯一索引 uq_worker_risk_bans_active), + * 由服务层翻译成 409,避免同一 IP 在列表里重复出现多条。 + */ +export async function insertWorkerRiskBan(input: { + targetType: WorkerRiskBanTargetType + targetValue: string + workerId?: number | null + workerUsername?: string + reason?: string + expiresAt?: string | null + createdBy?: number | null + createdByUsername?: string +}): Promise { + const result = await query( + ` + INSERT INTO worker_risk_bans ( + target_type, target_value, worker_id, worker_username, + reason, expires_at, created_by, created_by_username, created_at + ) VALUES ($1, $2, $3, $4, $5, $6::timestamptz, $7, $8, NOW()) + ON CONFLICT (target_type, target_value) WHERE revoked_at IS NULL DO NOTHING + RETURNING * + `, + [ + input.targetType, + String(input.targetValue || '').slice(0, 200), + input.workerId ?? null, + String(input.workerUsername || '').slice(0, 100), + String(input.reason || '').slice(0, 500), + input.expiresAt ?? null, + input.createdBy ?? null, + String(input.createdByUsername || '').slice(0, 100), + ], + ) + return result.rows[0] ?? null +} + +/** 撤销一条封禁;已撤销/不存在返回 null。 */ +export async function revokeWorkerRiskBanById( + id: number | string, + input: { revokedByUsername?: string; reason?: string } = {}, +): Promise { + const result = await query( + ` + UPDATE worker_risk_bans + SET revoked_at = NOW(), + revoked_by_username = $2, + revoke_reason = $3 + WHERE id = $1 AND revoked_at IS NULL + RETURNING * + `, + [ + Number(id), + String(input.revokedByUsername || '').slice(0, 100), + String(input.reason || '').slice(0, 500), + ], + ) + return result.rows[0] ?? null +} + +export async function getWorkerRiskBanById(id: number | string): Promise { + const result = await query(`SELECT * FROM worker_risk_bans WHERE id = $1`, [ + Number(id), + ]) + return result.rows[0] ?? null +} + +function buildBanFilters(input: { + targetType?: string | undefined + keyword?: string | undefined + activeOnly?: boolean | undefined +}): { where: string; params: unknown[] } { + const conditions: string[] = [] + const params: unknown[] = [] + + const targetType = String(input.targetType || '').trim() + if (targetType === 'ip' || targetType === 'worker') { + params.push(targetType) + conditions.push(`target_type = $${params.length}`) + } + + const keyword = String(input.keyword || '').trim() + if (keyword) { + params.push(`%${keyword}%`) + conditions.push( + `(target_value ILIKE $${params.length} OR worker_username ILIKE $${params.length})`, + ) + } + + if (input.activeOnly === true) { + conditions.push(`revoked_at IS NULL AND (expires_at IS NULL OR expires_at > NOW())`) + } else if (input.activeOnly === false) { + conditions.push(`(revoked_at IS NOT NULL OR (expires_at IS NOT NULL AND expires_at <= NOW()))`) + } + + return { + where: conditions.length ? `WHERE ${conditions.join(' AND ')}` : '', + params, + } +} diff --git a/apps/backend/src/routes/admin/worker-platform/risk.ts b/apps/backend/src/routes/admin/worker-platform/risk.ts index 6413c389..aad33e95 100644 --- a/apps/backend/src/routes/admin/worker-platform/risk.ts +++ b/apps/backend/src/routes/admin/worker-platform/risk.ts @@ -3,9 +3,12 @@ import type { Request } from 'express' import { banAdminWorkerRisk, + banAdminWorkerRiskTarget, getAdminWorkerRiskOverview, + listAdminWorkerRiskBans, listAdminWorkerRiskEvents, releaseAdminWorkerRiskTracker, + revokeAdminWorkerRiskBan, saveAdminWorkerRiskConfig, unbanAdminWorkerRisk, } from '../../../services/worker-platform/index.js' @@ -63,6 +66,70 @@ router.post( }), ) +/** + * 手动封禁(IP / 账号)的持久化列表。 + * + * 与「实时风控」里那批内存 tracker 的区别:这里是落库的事实,进程重启不丢, + * 中间件按它拦截 /api/v1/worker 下的所有请求(含登录)。 + */ +router.get( + '/worker-platform/risk-control/bans', + requireAdminRoles(['admin', 'operator', 'support']), + createJsonHandler((req) => listAdminWorkerRiskBans((req.query || {}) as JsonRecord), { + successMessage: 'ok', + errorMessage: '读取封禁列表失败', + scope: '[admin/worker-platform/risk-control/bans]', + }), +) + +router.post( + '/worker-platform/risk-control/bans', + requireAdminRoles(['admin']), + createJsonHandler( + (req) => banAdminWorkerRiskTarget(actorOf(req), (req.body || {}) as JsonRecord), + { + successMessage: '封禁已生效', + errorMessage: '封禁失败', + scope: '[admin/worker-platform/risk-control/bans]', + audit: (req, data) => { + const result = data as JsonRecord + const ban = (result.ban || {}) as JsonRecord + return { + action: 'worker_risk_ban_created', + targetType: String( + ban.targetType || (req.body as JsonRecord | undefined)?.targetType || '', + ), + targetId: String(ban.target || ''), + data: { ban }, + } + }, + }, + ), +) + +router.post( + '/worker-platform/risk-control/bans/:banId/revoke', + requireAdminRoles(['admin']), + createJsonHandler( + (req) => + revokeAdminWorkerRiskBan( + paramOf(req.params.banId), + actorOf(req), + (req.body || {}) as JsonRecord, + ), + { + successMessage: '封禁已解除', + errorMessage: '解除封禁失败', + scope: '[admin/worker-platform/risk-control/bans/:banId/revoke]', + audit: (req) => ({ + action: 'worker_risk_ban_revoked', + targetType: 'worker_risk_ban', + targetId: String(paramOf(req.params.banId)), + }), + }, + ), +) + router.get( '/worker-platform/risk-control/events', requireAdminRoles(['admin', 'operator', 'support']), diff --git a/apps/backend/src/routes/worker.ts b/apps/backend/src/routes/worker.ts index 0a1ed8ce..33cb7532 100644 --- a/apps/backend/src/routes/worker.ts +++ b/apps/backend/src/routes/worker.ts @@ -7,11 +7,16 @@ import workerHallRoutes from './worker/hall.js' import workerOrdersRoutes from './worker/orders.js' import workerConsignmentRoutes from './worker/consignment.js' import { requireWorkerSession } from './worker/session.js' +import { createWorkerBanGuardMiddleware } from '../middleware/worker-ban-guard.js' import { createWorkerHallRiskMiddleware } from '../middleware/worker-risk-control.js' import { buildNotFoundPayload } from '../utils/http.js' const router = Router() +// 手动封禁(IP / 账号)最先判:命中直接 403,连登录都不放行。 +// 整个过程只查进程内缓存,不碰数据库,被刷时也不会放大压力。 +router.use(createWorkerBanGuardMiddleware()) + router.use(workerAuthPublicRoutes) // 风控限流前置:在校验数据库 session 之前先行拦截超频/风控请求,避免外挂脚本打满数据库连接 diff --git a/apps/backend/src/routes/worker/orders.ts b/apps/backend/src/routes/worker/orders.ts index d9928bef..7a46de55 100644 --- a/apps/backend/src/routes/worker/orders.ts +++ b/apps/backend/src/routes/worker/orders.ts @@ -22,9 +22,23 @@ import { } from '../../services/worker-platform/index.js' import { getRequiredWorkerSession, requireActiveWorker } from './session.js' import { createRouteHandler } from '../../utils/http.js' +import { createRateLimitMiddleware } from '../../middleware/rate-limit.js' const router = Router() +/** + * 「标记已读」的限流。 + * + * 打手每打开一张单就会 POST 一次,正常操作远低于 2 次/秒; + * 脚本轮询同一个订单号时(生产实测 10 分钟 337 次、且集中在几个固定 id)会先被挡在这里, + * 不必等它打到数据库。 + */ +const markReadRateLimit = createRateLimitMiddleware({ + scope: 'worker:order-read', + windowMs: 60_000, + max: 120, +}) + router.get( '/orders', requireActiveWorker, @@ -167,6 +181,7 @@ router.get( router.post( '/orders/:workOrderId/read', requireActiveWorker, + markReadRateLimit, createRouteHandler( (req) => markWorkerOrderRead( diff --git a/apps/backend/src/routes/worker/profile.ts b/apps/backend/src/routes/worker/profile.ts index 4edc94d4..cfe84697 100644 --- a/apps/backend/src/routes/worker/profile.ts +++ b/apps/backend/src/routes/worker/profile.ts @@ -15,9 +15,20 @@ import { } from '../../services/worker-platform/index.js' import { getRequiredWorkerSession, requireActiveWorker } from './session.js' import { createRouteHandler } from '../../utils/http.js' +import { createRateLimitMiddleware } from '../../middleware/rate-limit.js' const router = Router() +/** + * 附件上传限流:每次上传都要落 OSS + 写 file_assets,是最贵的写接口之一。 + * 正常交单不会超过 30 次/分钟(连拍多张图也只是十几次),脚本批量传图会先被挡。 + */ +const uploadRateLimit = createRateLimitMiddleware({ + scope: 'worker:files-upload', + windowMs: 60_000, + max: 30, +}) + router.get( '/profile', createRouteHandler((req) => getWorkerProfile(getRequiredWorkerSession(req)), { @@ -125,6 +136,7 @@ router.post( router.post( '/files/upload', requireActiveWorker, + uploadRateLimit, uploadSingleFile, createRouteHandler( (req) => { diff --git a/apps/backend/src/services/cost-center/manual-product-normalization-service.ts b/apps/backend/src/services/cost-center/manual-product-normalization-service.ts index a0226492..7212ef01 100644 --- a/apps/backend/src/services/cost-center/manual-product-normalization-service.ts +++ b/apps/backend/src/services/cost-center/manual-product-normalization-service.ts @@ -164,6 +164,7 @@ export async function normalizeManualProducts( // 回填订单项:按 id 分批,每批重新取日期参数,避免单条语句超过 15s。 let updatedItemCount = 0 + let updatedSnapshotCount = 0 let lastId = 0 for (let batch = 0; batch < 40; batch += 1) { const range = buildRangeClause(3) @@ -197,26 +198,29 @@ export async function normalizeManualProducts( updatedItemCount += rowCount const ids = updated.rows as Array<{ id: number }> lastId = Number(ids[ids.length - 1]?.id || lastId) + // 已有成本快照补商品维度:**只处理本批刚归一过的订单项**。 + // 原来是无条件全表条件更新,要扫 900MB 的 order_items(堆 + 两个位图索引回表), + // 生产实测每次 15~16 秒、必然撞 statement_timeout,而且失败会把后半段 + // (销售事实重算、影子总账同步)整条链跳过。收口到本批 id 后走 idx_cost_snapshots_item, + // 毫秒级完成,且结果完全一样(唯一会变的行就是这些刚补上商品的订单项)。 + if (ids.length > 0) { + const updatedSnapshots = await query( + ` + UPDATE cost_snapshots cs + SET canonical_product_id = oi.canonical_product_id + FROM order_items oi + WHERE oi.id = ANY($1::bigint[]) + AND cs.order_item_id = oi.id + AND cs.canonical_product_id IS NULL + AND oi.canonical_product_id IS NOT NULL + `, + [ids.map((row) => Number(row.id))], + ) + updatedSnapshotCount += updatedSnapshots.rowCount || 0 + } if (rowCount < NORMALIZE_BATCH_SIZE) break } - // 已有实际成本快照补上商品维度,让毛利能按商品归集。 - let updatedSnapshotCount = 0 - if (updatedItemCount > 0) { - const updatedSnapshots = await query( - ` - UPDATE cost_snapshots cs - SET canonical_product_id = oi.canonical_product_id - FROM order_items oi - WHERE cs.order_item_id = oi.id - AND cs.canonical_product_id IS NULL - AND oi.fulfillment_mode = 'manual' - AND oi.canonical_product_id IS NOT NULL - `, - ) - updatedSnapshotCount = updatedSnapshots.rowCount || 0 - } - let resolvedExceptionCount = 0 if (input.resolveExceptions !== false && updatedItemCount > 0) { const resolved = await query( diff --git a/apps/backend/src/services/cost-center/manual-product-normalization.test.ts b/apps/backend/src/services/cost-center/manual-product-normalization.test.ts new file mode 100644 index 00000000..d1c54dd7 --- /dev/null +++ b/apps/backend/src/services/cost-center/manual-product-normalization.test.ts @@ -0,0 +1,60 @@ +import assert from 'node:assert/strict' +import test from 'node:test' + +/** + * 2026-10-30 生产事故复盘:cost-finance-sync 每 15 分钟的「成本快照补商品维度」 + * 原先是一条无条件全表更新,要扫 899MB 的 order_items,实测 15.2s / 16.5s 撞 + * statement_timeout,并且失败会把后面的「销售事实重算 + 影子总账同步」整条跳过。 + * + * 修复口径:只更新本批刚归一过的 order_item_id(走 idx_cost_snapshots_item), + * 结果完全一样但毫秒级完成。这里用源码断言把它钉住。 + */ +test('成本快照补商品维度必须按本批 order_item_id 收口,不能再全表更新', async () => { + const { readFile } = await import('node:fs/promises') + const src = await readFile( + new URL('./manual-product-normalization-service.ts', import.meta.url), + 'utf8', + ) + + assert.ok( + src.includes('oi.id = ANY($1::bigint[])'), + '成本快照更新缺少本批 id 收口:会退化成扫 900MB 的 order_items(实测 15 秒超时)', + ) + assert.ok( + src.includes('AND cs.order_item_id = oi.id'), + '收口后仍要按 order_item_id 关联,不能改成按 subject_id 猜', + ) + + // 更新必须发生在分批循环内部(本批 ids 还在作用域里),不能在循环之后 + const loopStart = src.indexOf('for (let batch = 0; batch < 40; batch += 1) {') + const loopEnd = src.indexOf('let resolvedExceptionCount = 0') + const snapshotUpdateAt = src.indexOf('UPDATE cost_snapshots cs') + assert.ok(loopStart > 0 && loopEnd > loopStart, '找不到归一的分批循环') + assert.ok( + snapshotUpdateAt > loopStart && snapshotUpdateAt < loopEnd, + '成本快照更新必须留在分批循环内:循环外拿不到本批 ids,就只能退回全表条件', + ) + + // 兜底:不允许再出现「无 id 限制 + fulfillment_mode 条件」的老写法 + const legacyPattern = /UPDATE cost_snapshots cs[\s\S]{0,400}?oi\.fulfillment_mode = 'manual'/ + assert.equal( + legacyPattern.test(src), + false, + '不能再按 fulfillment_mode 全表更新(会整表 bitmap 回表)', + ) +}) + +test('迁移 110 提供 order_items(fulfillment_mode, canonical_product_id) 覆盖索引', async () => { + const { readFile } = await import('node:fs/promises') + const migration = await readFile( + new URL('../../db/migrations/110_order_items_fulfillment_product_index.sql', import.meta.url), + 'utf8', + ) + assert.ok(migration.includes('idx_order_items_fulfillment_product')) + assert.ok(migration.includes('(fulfillment_mode, canonical_product_id)')) + assert.ok(migration.includes('INCLUDE (id)')) + assert.ok( + !/^\s*CREATE INDEX CONCURRENTLY/m.test(migration), + '迁移在事务内执行,不能使用 CONCURRENTLY', + ) +}) diff --git a/apps/backend/src/services/worker-platform/admin-worker-risk-service.ts b/apps/backend/src/services/worker-platform/admin-worker-risk-service.ts index a8eb42a6..781d245e 100644 --- a/apps/backend/src/services/worker-platform/admin-worker-risk-service.ts +++ b/apps/backend/src/services/worker-platform/admin-worker-risk-service.ts @@ -1,11 +1,20 @@ import type { JsonObject } from '../../types/json.js' import { createHttpError } from '../../utils/http.js' import { nowIso } from '../../utils/time.js' +import { isIP } from 'node:net' +import { invalidateWorkerBanCache, normalizeWorkerIp } from '../../middleware/worker-ban-guard.js' import { listActiveWorkerRiskTrackers, markWorkerRiskBanned, releaseWorkerRiskTracker, } from '../../middleware/worker-risk-control.js' +import { + countWorkerRiskBans, + insertWorkerRiskBan, + listWorkerRiskBans, + revokeWorkerRiskBanById, + type WorkerRiskBanRow, +} from '../../repositories/worker-platform/worker-risk-ban-repo.js' import { countWorkerRiskEvents, insertWorkerRiskEvent, @@ -15,6 +24,7 @@ import { } from '../../repositories/worker-platform/worker-risk-repo.js' import { getWorkerUserById, + getWorkerUserByUsername, updateWorkerUser, } from '../../repositories/worker-platform/worker-repo.js' import { revokeAllWorkerSessions } from '../../repositories/worker-platform/worker-session-repo.js' @@ -199,6 +209,253 @@ function normalizeWorkerId(value: number | string): number { return id } +/** + * 手动封禁列表(IP / 账号)。 + * + * 与「实时风控」的区别:这里读的是持久表 worker_risk_bans,进程重启不丢; + * 列表默认把生效中的排在最前(ORDER BY revoked_at NULLS FIRST)。 + */ +export async function listAdminWorkerRiskBans(query: JsonObject = {}) { + const page = normalizePage(query.page) + const pageSize = normalizePageSize(query.pageSize) + const activeRaw = query.activeOnly ?? query.active_only + const activeOnly = + activeRaw === undefined || activeRaw === null || activeRaw === '' + ? undefined + : String(activeRaw) === 'true' || activeRaw === true + + const filters = { + targetType: String(query.targetType ?? query.target_type ?? '').trim(), + keyword: String(query.keyword || '').trim(), + activeOnly, + } + const [rows, total] = await Promise.all([ + listWorkerRiskBans({ ...filters, limit: pageSize, offset: (page - 1) * pageSize }), + countWorkerRiskBans(filters), + ]) + return { items: rows.map(mapWorkerRiskBan), total, page, pageSize } +} + +/** 最多封 30 天;0 / 不传 = 永久(运营处理外挂时一般选永久,再由人工解封)。 */ +const MAX_BAN_DURATION_MINUTES = 30 * 24 * 60 + +/** + * 手动封禁一个 IP 或一个接单账号。 + * + * · IP:写封禁表即可,中间件按请求 IP 拦截(登录也算,连门都进不来); + * · 账号:写封禁表 + 冻结账号 + 踢掉全部登录设备 + 置位内存风控标记, + * 与「后台封号」等价,但多了一条可追溯的封禁记录。 + */ +export async function banAdminWorkerRiskTarget( + actor: WorkerRiskActor = {}, + payload: JsonObject = {}, +) { + const targetType = String(payload.targetType ?? payload.target_type ?? '') + .trim() + .toLowerCase() + if (targetType !== 'ip' && targetType !== 'worker') { + throw createHttpError('封禁类型只支持 IP 或账号', { + statusCode: 400, + errorCode: 'worker_risk_ban_type_invalid', + }) + } + const rawTarget = String( + payload.target ?? payload.targetValue ?? payload.target_value ?? '', + ).trim() + if (!rawTarget) { + throw createHttpError('请填写要封禁的 IP 或账号', { + statusCode: 400, + errorCode: 'worker_risk_ban_target_invalid', + }) + } + + const reason = + String(payload.reason || '').trim() || + (targetType === 'ip' ? '后台人工封禁 IP' : '后台人工封禁账号') + const durationMinutes = normalizeBanDuration(payload.durationMinutes ?? payload.duration_minutes) + const expiresAt = + durationMinutes > 0 ? new Date(Date.now() + durationMinutes * 60_000).toISOString() : null + const actorName = actor.displayName || actor.username || '' + + if (targetType === 'ip') { + const ip = normalizeWorkerIp(rawTarget) + if (!isIP(ip)) { + throw createHttpError('IP 格式不正确(示例:203.0.113.9)', { + statusCode: 400, + errorCode: 'worker_risk_ban_target_invalid', + }) + } + const row = await insertWorkerRiskBan({ + targetType: 'ip', + targetValue: ip, + reason, + expiresAt, + createdBy: actor.userId ?? null, + createdByUsername: actorName, + }) + if (!row) { + throw createHttpError('该 IP 已在封禁列表中', { + statusCode: 409, + errorCode: 'worker_risk_ban_exists', + }) + } + invalidateWorkerBanCache() + await insertWorkerRiskEvent({ + eventType: 'manual_banned', + targetKey: `ip:${ip}`, + ip, + reason, + action: 'manual', + actorUserId: actor.userId ?? null, + actorUsername: actorName, + detail: { banId: row.id, targetType: 'ip', expiresAt }, + }) + return { ban: mapWorkerRiskBan(row) } + } + + const worker = await resolveWorkerForBan(rawTarget) + if (!worker) { + throw createHttpError('接单账号不存在', { statusCode: 404, errorCode: 'worker_not_found' }) + } + const row = await insertWorkerRiskBan({ + targetType: 'worker', + targetValue: String(worker.id), + workerId: Number(worker.id), + workerUsername: worker.username, + reason, + expiresAt, + createdBy: actor.userId ?? null, + createdByUsername: actorName, + }) + if (!row) { + throw createHttpError('该账号已在封禁列表中', { + statusCode: 409, + errorCode: 'worker_risk_ban_exists', + }) + } + + const now = nowIso() + await updateWorkerUser(worker.id, { + status: 'frozen', + frozen_reason: reason, + frozen_at: now, + session_version: Number(worker.session_version || 1) + 1, + updated_at: now, + }) + await revokeAllWorkerSessions(worker.id, now) + markWorkerRiskBanned(Number(worker.id)) + invalidateWorkerBanCache() + + await insertWorkerRiskEvent({ + eventType: 'manual_banned', + targetKey: `worker:${worker.id}`, + workerId: Number(worker.id), + ip: String(payload.ip || '').trim(), + reason, + action: 'manual', + actorUserId: actor.userId ?? null, + actorUsername: actorName, + detail: { banId: row.id, targetType: 'worker', expiresAt }, + }) + + return { ban: mapWorkerRiskBan(row), workerId: Number(worker.id), status: 'frozen' } +} + +/** + * 解除一条手动封禁。 + * + * 账号类型的封禁会连同解冻账号 + 释放内存风控标记——否则封禁记录没了、账号还是 frozen, + * 运营会以为没生效。IP 类型只改封禁表,不影响任何账号状态。 + */ +export async function revokeAdminWorkerRiskBan( + banId: number | string, + actor: WorkerRiskActor = {}, + payload: JsonObject = {}, +) { + const id = Number(banId) + if (!Number.isInteger(id) || id <= 0) { + throw createHttpError('封禁记录 ID 无效', { + statusCode: 400, + errorCode: 'worker_risk_ban_id_invalid', + }) + } + const reason = String(payload.reason || '').trim() || '后台人工解封' + const actorName = actor.displayName || actor.username || '' + const row = await revokeWorkerRiskBanById(id, { revokedByUsername: actorName, reason }) + if (!row) { + throw createHttpError('封禁记录不存在或已解除', { + statusCode: 404, + errorCode: 'worker_risk_ban_not_found', + }) + } + invalidateWorkerBanCache() + + const workerId = row.worker_id === null ? null : Number(row.worker_id) + if (row.target_type === 'worker' && workerId) { + const worker = await getWorkerUserById(workerId) + if (worker && worker.status === 'frozen') { + const now = nowIso() + await updateWorkerUser(workerId, { + status: 'active', + frozen_reason: '', + frozen_at: null, + session_version: Number(worker.session_version || 1) + 1, + updated_at: now, + }) + await revokeAllWorkerSessions(workerId, now) + } + releaseWorkerRiskTracker({ workerId }) + } + + await insertWorkerRiskEvent({ + eventType: 'manual_unbanned', + targetKey: row.target_type === 'ip' ? `ip:${row.target_value}` : `worker:${workerId}`, + workerId, + ip: row.target_type === 'ip' ? row.target_value : '', + reason, + action: 'manual', + actorUserId: actor.userId ?? null, + actorUsername: actorName, + detail: { banId: id, targetType: row.target_type }, + }) + + return { ban: mapWorkerRiskBan(row) } +} + +function normalizeBanDuration(value: unknown): number { + const parsed = Number(value) + if (!Number.isFinite(parsed) || parsed <= 0) return 0 + return Math.min(MAX_BAN_DURATION_MINUTES, Math.trunc(parsed)) +} + +async function resolveWorkerForBan(target: string) { + const numeric = Number(target) + if (Number.isInteger(numeric) && numeric > 0) { + return getWorkerUserById(numeric) + } + return getWorkerUserByUsername(target) +} + +function mapWorkerRiskBan(row: WorkerRiskBanRow) { + const expiresAt = row.expires_at ? new Date(row.expires_at) : null + const revokedAt = row.revoked_at ? new Date(row.revoked_at) : null + return { + id: String(row.id), + targetType: row.target_type, + target: row.target_value, + workerId: row.worker_id === null ? null : Number(row.worker_id), + workerUsername: row.worker_username, + reason: row.reason, + expiresAtIso: expiresAt ? expiresAt.toISOString() : null, + active: !revokedAt && (!expiresAt || expiresAt.getTime() > Date.now()), + createdByUsername: row.created_by_username, + createdAtIso: new Date(row.created_at).toISOString(), + revokedAtIso: revokedAt ? revokedAt.toISOString() : null, + revokedByUsername: row.revoked_by_username, + revokeReason: row.revoke_reason, + } +} + function mapWorkerRiskEvent(row: WorkerRiskEventRow) { return { id: String(row.id), diff --git a/apps/backend/src/services/worker-platform/worker-hall-service.ts b/apps/backend/src/services/worker-platform/worker-hall-service.ts index 328e5912..5366bc04 100644 --- a/apps/backend/src/services/worker-platform/worker-hall-service.ts +++ b/apps/backend/src/services/worker-platform/worker-hall-service.ts @@ -26,6 +26,15 @@ import { WorkerListCache } from './worker-list-cache.js' const HALL_LIST_CACHE_TTL_MS = 2_000 const hallListCache = new WorkerListCache(HALL_LIST_CACHE_TTL_MS) +/** + * 大厅列表单页硬上限。 + * + * 每张卡片都要拼单汇总 + 权限判定,pageSize=100 在生产实测单次接近 2 秒(DB 执行时间), + * 而 2026-10-30 那个脚本固定请求 pageSize=100 轮询,是唯一能低成本放大的入口。 + * 前端本来就按 page 翻页,收到更小的 pageSize 也只是多翻几页。 + */ +const HALL_MAX_PAGE_SIZE = 20 + // 排行榜每次都要对全量 work_orders + work_order_shares 做去重聚合(生产实测 // 742ms / 35,483 块 ≈ 277MB),而结果只在工单验收完成时才变化。 // 大厅每次刷新/窗口聚焦都会触发前端重新拉取,不缓存时该查询独占全库约 8% 的缓冲读。 @@ -38,7 +47,7 @@ const hallLeaderboardCache = new WorkerListCache(HALL_LEADERBOARD_CACHE_TTL_MS) export async function listWorkerHallOrders(query: JsonObject = {}, session: WorkerSession) { requireActiveWorkerSession(session) const page = normalizePage(query.page) - const pageSize = normalizePageSize(query.pageSize) + const pageSize = Math.min(HALL_MAX_PAGE_SIZE, normalizePageSize(query.pageSize)) const categoryId = normalizeOptionalId(query.categoryId) const worker = await getRequiredWorker(session.workerId) const permissions = resolveWorkerPermissions(worker) diff --git a/apps/backend/src/services/worker-platform/worker-platform-service.test.ts b/apps/backend/src/services/worker-platform/worker-platform-service.test.ts index 97af762e..3df73929 100644 --- a/apps/backend/src/services/worker-platform/worker-platform-service.test.ts +++ b/apps/backend/src/services/worker-platform/worker-platform-service.test.ts @@ -2990,6 +2990,37 @@ test('后台「店铺」筛选下拉必须缓存 + 写路径失效(单次全 ) }) +test('接单端入口必须有手动封禁守卫,大厅单页上限 20,高频写接口有独立限流', async () => { + const { readFile } = await import('node:fs/promises') + + // 2026-10-30:脚本固定 pageSize=100 轮询大厅,单次查询近 2 秒(每张卡都要拼单汇总) + const hallSrc = await readFile(new URL('./worker-hall-service.ts', import.meta.url), 'utf8') + assert.ok(hallSrc.includes('HALL_MAX_PAGE_SIZE = 20'), '大厅缺少单页上限常量') + assert.ok( + hallSrc.includes('Math.min(HALL_MAX_PAGE_SIZE, normalizePageSize(query.pageSize))'), + '大厅 pageSize 必须被服务端夹到 20,不能信任客户端', + ) + + // 手动封禁守卫挂在整个 /worker 路由之前(登录也算),否则封了 IP 还能重新登录 + const routerSrc = await readFile(new URL('../../routes/worker.ts', import.meta.url), 'utf8') + const guardAt = routerSrc.indexOf('createWorkerBanGuardMiddleware()') + const authAt = routerSrc.indexOf('router.use(workerAuthPublicRoutes)') + assert.ok(guardAt > 0, '缺少手动封禁守卫') + assert.ok(guardAt < authAt, '封禁守卫必须在鉴权路由之前挂载:被封 IP 连登录都不能放') + + // 两个最贵的写接口各自限流(读接口 337 次/10 分钟、上传 95 次/10 分钟都是生产实测) + const ordersSrc = await readFile( + new URL('../../routes/worker/orders.ts', import.meta.url), + 'utf8', + ) + assert.ok(ordersSrc.includes("scope: 'worker:order-read'"), '标记已读缺少限流') + const profileSrc = await readFile( + new URL('../../routes/worker/profile.ts', import.meta.url), + 'utf8', + ) + assert.ok(profileSrc.includes("scope: 'worker:files-upload'"), '附件上传缺少限流') +}) + test('残留字段 work_orders.task_id 不得再被当成关联键 JOIN(迁移 091)', async () => { const { readFile } = await import('node:fs/promises') diff --git a/apps/frontend/src/pages/admin/panels/RiskControlPanel.tsx b/apps/frontend/src/pages/admin/panels/RiskControlPanel.tsx index 0641f849..97b1ceaf 100644 --- a/apps/frontend/src/pages/admin/panels/RiskControlPanel.tsx +++ b/apps/frontend/src/pages/admin/panels/RiskControlPanel.tsx @@ -19,14 +19,19 @@ import { useEffect, useState } from 'react' import { banAdminWorkerRisk, + createAdminWorkerRiskBan, + fetchAdminWorkerRiskBans, fetchAdminWorkerRiskEvents, fetchAdminWorkerRiskOverview, releaseAdminWorkerRiskTracker, + revokeAdminWorkerRiskBan, saveAdminWorkerRiskConfig, unbanAdminWorkerRisk, } from '@/services/admin' import type { WorkerRiskActiveTracker, + WorkerRiskBan, + WorkerRiskBanTargetType, WorkerRiskConfig, WorkerRiskEvent, WorkerRiskEventType, @@ -84,6 +89,29 @@ export default function RiskControlPanel() { }) const events = eventsQuery.data?.data + // 手动封禁:落库的持久事实(进程重启不丢),中间件按它拦截整个接单端 + const [banForm] = Form.useForm<{ + targetType: WorkerRiskBanTargetType + target: string + durationMinutes?: number + reason?: string + }>() + const [banPage, setBanPage] = useState(1) + const [banPageSize, setBanPageSize] = useState(20) + const [banKeyword, setBanKeyword] = useState('') + const [banStatusFilter, setBanStatusFilter] = useState<'active' | 'inactive' | 'all'>('active') + const bansQuery = useQuery({ + queryKey: ['admin-worker-risk-bans', banPage, banPageSize, banKeyword, banStatusFilter], + queryFn: () => + fetchAdminWorkerRiskBans({ + page: banPage, + pageSize: banPageSize, + keyword: banKeyword.trim() || undefined, + activeOnly: banStatusFilter === 'all' ? undefined : banStatusFilter === 'active', + }), + }) + const bans = bansQuery.data?.data + useEffect(() => { if (overview?.config) { form.setFieldsValue(overview.config) @@ -94,6 +122,7 @@ export default function RiskControlPanel() { await Promise.all([ queryClient.invalidateQueries({ queryKey: ['admin-worker-risk-overview'] }), queryClient.invalidateQueries({ queryKey: ['admin-worker-risk-events'] }), + queryClient.invalidateQueries({ queryKey: ['admin-worker-risk-bans'] }), ]) } @@ -110,6 +139,58 @@ export default function RiskControlPanel() { } } + /** 新增一条手动封禁(IP 或账号)。账号类型会同时冻结账号并踢掉登录设备。 */ + async function submitBan(values: { + targetType: WorkerRiskBanTargetType + target: string + durationMinutes?: number + reason?: string + }) { + setActing(true) + try { + await createAdminWorkerRiskBan({ + targetType: values.targetType, + target: String(values.target || '').trim(), + durationMinutes: Number(values.durationMinutes || 0), + reason: String(values.reason || '').trim(), + }) + banForm.resetFields(['target', 'reason']) + await refreshAll() + message.success(values.targetType === 'ip' ? 'IP 已封禁' : '账号已封禁') + } catch (error) { + message.error(error instanceof Error ? error.message : '封禁失败') + } finally { + setActing(false) + } + } + + /** 实时风控表格里的一键封 IP:脚本最常换账号,但出口 IP 换起来更贵。 */ + function quickBanIp(ip: string) { + confirmAction( + '封禁 IP', + `确认封禁 IP ${ip}?该出口下的所有接单请求(含登录)都会立即被拒绝,可在下方封禁列表解除。`, + async () => { + await createAdminWorkerRiskBan({ + targetType: 'ip', + target: ip, + reason: '后台风控页一键封禁', + }) + }, + ) + } + + function confirmRevokeBan(ban: WorkerRiskBan) { + const label = + ban.targetType === 'ip' ? `IP ${ban.target}` : `账号 #${ban.workerId} ${ban.workerUsername}` + confirmAction( + '解除封禁', + `确认解除对 ${label} 的封禁?账号类型会同时解冻并释放内存风控状态。`, + async () => { + await revokeAdminWorkerRiskBan(ban.id, '后台风控页人工解封') + }, + ) + } + function confirmAction(title: string, content: string, run: () => Promise) { modal.confirm({ title, @@ -171,6 +252,16 @@ export default function RiskControlPanel() { width: 240, render: (_value, row) => ( + {row.ip && row.ip !== 'unknown' ? ( + + ) : null} {row.workerId ? ( <> + ), + }, + ] + const summary = overview?.summary24h || [] return ( @@ -391,6 +547,107 @@ export default function RiskControlPanel() { /> + + 落库持久化,命中后直接 403,连登录一并拦截 + + } + > +
+ + + + + + + + + + + + +
+ + + setBanKeyword(event.target.value)} + onPressEnter={() => { + setBanPage(1) + void bansQuery.refetch() + }} + /> + + ('/api/v1/admin/worker-platform/risk-control/bans', params) +} + +export function createAdminWorkerRiskBan(payload: { + targetType: WorkerRiskBanTargetType + target: string + /** 0 / 不传 = 永久 */ + durationMinutes?: number + reason?: string +}) { + return apiPost<{ ban: WorkerRiskBan; workerId?: number; status?: string }>( + '/api/v1/admin/worker-platform/risk-control/bans', + payload, + ) +} + +export function revokeAdminWorkerRiskBan(banId: string, reason?: string) { + return apiPost<{ ban: WorkerRiskBan }>( + `/api/v1/admin/worker-platform/risk-control/bans/${banId}/revoke`, + { reason }, + ) +} diff --git a/apps/frontend/src/types/worker-platform.ts b/apps/frontend/src/types/worker-platform.ts index 8f208087..8fcbea8a 100644 --- a/apps/frontend/src/types/worker-platform.ts +++ b/apps/frontend/src/types/worker-platform.ts @@ -927,3 +927,30 @@ export type WorkerRiskEventList = { page: number pageSize: number } + +/** 手动封禁目标:IP 或接单账号 */ +export type WorkerRiskBanTargetType = 'ip' | 'worker' + +/** 落库的手动封禁记录(进程重启不丢,中间件按它拦截) */ +export type WorkerRiskBan = { + id: string + targetType: WorkerRiskBanTargetType + target: string + workerId: number | null + workerUsername: string + reason: string + expiresAtIso: string | null + active: boolean + createdByUsername: string + createdAtIso: string + revokedAtIso: string | null + revokedByUsername: string + revokeReason: string +} + +export type WorkerRiskBanList = { + items: WorkerRiskBan[] + total: number + page: number + pageSize: number +}