limiter.js
4.89 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
/**
* limiter 编排：规则匹配 → key 构造 → 调注入 store 的 take/peek → 成形 RateDecision。
*
* 本文件只做编排与判定成形，不碰桶数学（在 token-bucket.js）、不碰存储细节
* （在 store 实现里）、不读 env、不建连接——store 与 clock 全由工厂参数注入。
*
* 时钟可注入（`nowMs`），默认 `Date.now`，仅在此装配层引用一次，便于测试精确
* 推进时间；桶数学文件本身不调时钟。
*
* @typedef {import('./rules.js').RateRule} RateRule
*
* @typedef {Object} RateQuery
* @property {string} endpoint 端点类别
* @property {string} subject 主体（IP 或匿名 device-id），不透明字符串
*
* @typedef {Object} RateDecision
* @property {boolean} allowed
* @property {number} remaining 剩余配额（整数）
* @property {number} retryAfterMs 超限时建议等待毫秒；放行时为 0
* @property {string} ruleId 命中的规则 id（无匹配规则时为 ''）
*
* RateLimitStore（注入）：
* {
* take(key, cost, rule, nowMs): Promise<{ allowed, remaining, retryAfterMs }>,
* peek(key, rule, nowMs): Promise<{ remaining, retryAfterMs }>
* }
*/
import { RateLimitConfigError } from './errors.js'
import { validateRules, indexRules, selectRule, ruleIdOf } from './rules.js'
import { bucketKey } from './key.js'
/** 一次扣减的令牌数。一期固定为 1（每次 AI 调用扣 1）。 */
const COST = 1
/**
* 无匹配规则时的放行决策（fail-open）：没给某端点配规则即不限流，
* 而不是默默拒绝合法流量。规则缺失是配置问题，应在 review 时发现，不在运行时静默拒绝。
* @returns {RateDecision}
*/
function unlimited() {
return { allowed: true, remaining: Number.POSITIVE_INFINITY, retryAfterMs: 0, ruleId: '' }
}
/**
* 校验 query 的最小形态。limiter 是内部边界，仅做轻量自检（HTTP 入口的
* Valibot 校验在 @app/server 路由层）。
* @param {unknown} q
* @returns {{ endpoint: string, subject: string }}
*/
function requireQuery(q) {
if (!q || typeof q !== 'object') {
throw new RateLimitConfigError('query 必须是 { endpoint, subject } 对象')
}
const { endpoint, subject } = /** @type {Record<string, unknown>} */ (q)
if (typeof endpoint !== 'string' || endpoint.length === 0) {
throw new RateLimitConfigError('query.endpoint 必须是非空字符串')
}
if (typeof subject !== 'string' || subject.length === 0) {
throw new RateLimitConfigError('query.subject 必须是非空字符串')
}
return { endpoint, subject }
}
/**
* 限流器工厂。store 与规则由调用方注入；包内不读 env、不建连接。
* @param {Object} cfg
* @param {{ take: Function, peek: Function }} cfg.store 计数后端（注入）
* @param {RateRule[]} cfg.rules 规则表
* @param {() => number} [cfg.now] 时钟（注入，默认 Date.now）
* @returns {{
* check(ctx: unknown, q: RateQuery): Promise<RateDecision>,
* consume(ctx: unknown, q: RateQuery): Promise<RateDecision>
* }}
* @throws {RateLimitConfigError}
*/
export function createRateLimiter(cfg) {
if (!cfg || typeof cfg !== 'object') {
throw new RateLimitConfigError('createRateLimiter 需要配置对象 { store, rules }')
}
const { store, rules, now } = cfg
if (!store || typeof store.take !== 'function' || typeof store.peek !== 'function') {
throw new RateLimitConfigError('store 缺失或未实现 take/peek（RateLimitStore）')
}
const indexed = indexRules(validateRules(rules))
const clock = typeof now === 'function' ? now : Date.now
return {
/**
* 只读判定，不扣令牌。ctx 仅作日志/trace 透传（本包不直接依赖 observability）。
* @param {unknown} _ctx
* @param {RateQuery} q
* @returns {Promise<RateDecision>}
*/
async check(_ctx, q) {
const { endpoint, subject } = requireQuery(q)
const rule = selectRule(indexed, endpoint)
if (!rule) return unlimited()
const key = bucketKey(rule, subject)
const r = await store.peek(key, rule, clock())
return {
allowed: r.retryAfterMs === 0,
remaining: r.remaining,
retryAfterMs: r.retryAfterMs,
ruleId: ruleIdOf(rule),
}
},
/**
* 扣令牌并判定。超限返回 allowed:false + retryAfterMs，供路由层做 429 Retry-After。
* @param {unknown} _ctx
* @param {RateQuery} q
* @returns {Promise<RateDecision>}
*/
async consume(_ctx, q) {
const { endpoint, subject } = requireQuery(q)
const rule = selectRule(indexed, endpoint)
if (!rule) return unlimited()
const key = bucketKey(rule, subject)
const r = await store.take(key, COST, rule, clock())
return {
allowed: r.allowed,
remaining: r.remaining,
retryAfterMs: r.retryAfterMs,
ruleId: ruleIdOf(rule),
}
},
}
}