wangjia
b5ab92a57e
来自 xhigh code review 的正确性/健壮性修复，覆盖全部五端： - server：鉴权 fail-closed、计量交叉校验与配额扣穿处理、WS 网关并发与关闭顺序、 billing 行锁、redis Lua 过期与设备槽刷新、config 解析 - desktop：会话 epoch 防串话、WS 重连与 401 处理、api 客户端复用、统一 usePoll 轮询 - android：握手时序、请求头封装、账户状态派生、按需重组 - ios：finalize 宽限、串行采集、错误文案服务端优先、删除死代码 CommitController Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
119 lines
5.4 KiB
Go
119 lines
5.4 KiB
Go
package store
|
||
|
||
import (
|
||
"context"
|
||
"time"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
)
|
||
|
||
// Redis Key 设计见 doc/backend-architecture.html 第七章。
|
||
func KeyQuotaBalance(uid string) string { return "quota:" + uid + ":balance" }
|
||
func KeyQuotaTrial(uid string, day string) string {
|
||
return "quota:" + uid + ":trial:" + day
|
||
}
|
||
func KeyAuthQr(state string) string { return "authqr:" + state }
|
||
func KeyJwtBlock(jti string) string { return "jwt:block:" + jti }
|
||
func KeyRateCnt(did string) string { return "rate:" + did + ":asr:cnt" }
|
||
func KeyRateSecs(did string) string { return "rate:" + did + ":asr:secs" }
|
||
func KeyRateFb(uid, day string) string { return "rate:" + uid + ":fb:" + day }
|
||
func KeyActiveSession(did string) string { return "asr:active:" + did }
|
||
|
||
func OpenRedis(addr string, db int) *redis.Client {
|
||
return redis.NewClient(&redis.Options{Addr: addr, DB: db})
|
||
}
|
||
|
||
// Day 返回服务端时区（Asia/Shanghai）的自然日，作为试用与反馈限频的键。
|
||
var cst = time.FixedZone("CST", 8*3600)
|
||
|
||
func Day(t time.Time) string { return t.In(cst).Format("2006-01-02") }
|
||
|
||
// ─── 设备 30 分钟滑动窗口限制（ZSET，member 唯一、score 为时间戳秒）───────────────
|
||
|
||
// slideWindow 原子地：清理过期成员 → 检查阈值 → 通过则记录本次。
|
||
// cnt 窗口按"次"记 1；secs 窗口按本次秒数记。
|
||
var slideScript = redis.NewScript(`
|
||
local key, now, window, limit, val = KEYS[1], tonumber(ARGV[1]), tonumber(ARGV[2]), tonumber(ARGV[3]), tonumber(ARGV[4])
|
||
redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
|
||
local sum = 0
|
||
if ARGV[5] == 'count' then
|
||
sum = redis.call('ZCARD', key)
|
||
else
|
||
local members = redis.call('ZRANGE', key, 0, -1)
|
||
for _, m in ipairs(members) do
|
||
local v = string.match(m, ':(%d+)$')
|
||
if v then sum = sum + tonumber(v) end
|
||
end
|
||
end
|
||
if sum + val > limit then return 0 end
|
||
local seq = redis.call('INCR', key .. ':seq')
|
||
redis.call('EXPIRE', key .. ':seq', window + 60) -- 17C：seq 计数器与 ZSET 同寿命，避免按设备永久泄漏
|
||
redis.call('ZADD', key, now, now .. '-' .. seq .. ':' .. val)
|
||
redis.call('EXPIRE', key, window + 60)
|
||
return 1
|
||
`)
|
||
|
||
// AllowSession 设备维度新会话准入：30 分钟内 ≤30 次。
|
||
func AllowSession(ctx context.Context, rdb *redis.Client, deviceID string, now time.Time) (bool, error) {
|
||
ok, err := slideScript.Run(ctx, rdb, []string{KeyRateCnt(deviceID)},
|
||
now.Unix(), 30*60, 30, 1, "count").Int()
|
||
return ok == 1, err
|
||
}
|
||
|
||
// AudioWindowExhausted 会话 start 时检查时长窗口是否已满（只查不记；本次秒数在结束时
|
||
// 经 RecordAudioSeconds 记录——音频已实际消耗，结束时无条件记账）。
|
||
func AudioWindowExhausted(ctx context.Context, rdb *redis.Client, deviceID string, now time.Time) (bool, error) {
|
||
ok, err := slideScript.Run(ctx, rdb, []string{KeyRateSecs(deviceID)},
|
||
now.Unix(), 30*60, 30*60, 0, "sum").Int()
|
||
return ok == 0, err
|
||
}
|
||
|
||
// RecordAudioSeconds 会话结束记录本次识别秒数（无条件，limit 取大数）。
|
||
func RecordAudioSeconds(ctx context.Context, rdb *redis.Client, deviceID string, seconds int, now time.Time) error {
|
||
return slideScript.Run(ctx, rdb, []string{KeyRateSecs(deviceID)},
|
||
now.Unix(), 30*60, 1<<30, seconds, "sum").Err()
|
||
}
|
||
|
||
const deviceSlotTTL = 4 * time.Minute
|
||
|
||
// AcquireDeviceSlot 单设备同时仅 1 路识别会话（SET NX + TTL 兜底防泄漏）。
|
||
//
|
||
// 续期契约（17D）：槽位 TTL 仅 deviceSlotTTL（4min）作为崩溃/泄漏兜底，并非会话上限。
|
||
// 长会话期间持有者必须周期性调用 RefreshDeviceSlot 续期（间隔需 < TTL），
|
||
// 否则 TTL 到期后槽位被释放、并发设备可抢占。网关 usageLoop 每个 tick（2s）
|
||
// 应顺带调用 RefreshDeviceSlot 续命；会话正常/异常结束时由 ReleaseDeviceSlot 主动释放。
|
||
func AcquireDeviceSlot(ctx context.Context, rdb *redis.Client, deviceID, sessionID string) (bool, error) {
|
||
return rdb.SetNX(ctx, KeyActiveSession(deviceID), sessionID, deviceSlotTTL).Result()
|
||
}
|
||
|
||
// RefreshDeviceSlot 仅当槽位仍由本会话持有时续期 TTL（17D）。
|
||
// 返回 (true,nil) 表示续期成功；(false,nil) 表示槽位已不属于自己（被抢占/已释放），
|
||
// 调用方应据此判定会话是否仍合法持有槽位。供 gateway usageLoop 每 tick 调用。
|
||
func RefreshDeviceSlot(ctx context.Context, rdb *redis.Client, deviceID, sessionID string) (bool, error) {
|
||
// 持有者校验 + EXPIRE 原子化，避免续到别人刚抢占的槽位。
|
||
script := redis.NewScript(`
|
||
if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('EXPIRE', KEYS[1], ARGV[2]) end
|
||
return 0`)
|
||
n, err := script.Run(ctx, rdb, []string{KeyActiveSession(deviceID)}, sessionID, int(deviceSlotTTL.Seconds())).Int()
|
||
return n == 1, err
|
||
}
|
||
|
||
func ReleaseDeviceSlot(ctx context.Context, rdb *redis.Client, deviceID, sessionID string) error {
|
||
// 仅当持有者是自己时释放
|
||
script := redis.NewScript(`
|
||
if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) end
|
||
return 0`)
|
||
return script.Run(ctx, rdb, []string{KeyActiveSession(deviceID)}, sessionID).Err()
|
||
}
|
||
|
||
// IncrDailyCounter 自然日计数器（反馈限频等），返回自增后的值。
|
||
func IncrDailyCounter(ctx context.Context, rdb *redis.Client, key string) (int64, error) {
|
||
pipe := rdb.TxPipeline()
|
||
incr := pipe.Incr(ctx, key)
|
||
pipe.Expire(ctx, key, 48*time.Hour)
|
||
if _, err := pipe.Exec(ctx); err != nil {
|
||
return 0, err
|
||
}
|
||
return incr.Val(), nil
|
||
}
|