545 lines
17 KiB
Go
545 lines
17 KiB
Go
package store
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
|
||
"cancerai.top/toolbox-server/internal/pkg/hash"
|
||
)
|
||
|
||
// 包级错误，业务层可以 errors.Is 检查
|
||
var (
|
||
ErrSessionNotFound = errors.New("session not found or expired")
|
||
ErrDuplicateJoin = errors.New("nickname already joined this session")
|
||
ErrDuplicateSubmit = errors.New("this nickname has already submitted")
|
||
ErrNicknameConflict = errors.New("nickname conflict (exists or matches initiator)")
|
||
ErrOldNotFound = errors.New("old nickname not found in session")
|
||
)
|
||
|
||
const (
|
||
redisKeyPrefix = "dualcheck:session:"
|
||
fieldCreatedAt = "created_at"
|
||
fieldInitHash = "initiator_hash"
|
||
fieldInitText = "initiator_text" // 发起方原始文本（trim 前）
|
||
fieldInitNick = "initiator_nickname"
|
||
fieldInitBrowser = "initiator_browser_id"
|
||
fieldAllowResub = "allow_resubmit"
|
||
fieldIgnoreWs = "ignore_ws"
|
||
// participant:{nick} 存该参与者的哈希值；participant:{nick}:text 存原始文本；participant:{nick}:match 存 "1"/"0"
|
||
// participant:{nick}:browser_id 存该参与者的 browserId
|
||
partFieldPrefix = "participant:"
|
||
partTextSuffix = ":text"
|
||
partMatchSuffix = ":match"
|
||
partBrowserSuffix = ":browser_id"
|
||
)
|
||
|
||
// ── 五个 Lua 脚本 ──
|
||
|
||
// createLuaScript 原子创建 Hash 类型的会话 key
|
||
var createLuaScript = redis.NewScript(`
|
||
local key = KEYS[1]
|
||
if redis.call("EXISTS", key) ~= 0 then
|
||
return 0
|
||
end
|
||
redis.call("HSET", key, "created_at", ARGV[1])
|
||
redis.call("HSET", key, "initiator_nickname", ARGV[2])
|
||
redis.call("HSET", key, "initiator_browser_id", ARGV[3])
|
||
redis.call("HSET", key, "allow_resubmit", "0")
|
||
redis.call("HSET", key, "ignore_ws", "0")
|
||
redis.call("EXPIRE", key, ARGV[4])
|
||
return 1
|
||
`)
|
||
|
||
// joinParticipantLuaScript 登记参与者昵称（空哈希占位，表示已加入未提交）
|
||
// KEYS[1] = session key
|
||
// ARGV[1] = nickname
|
||
// ARGV[2] = browserId
|
||
var joinParticipantLuaScript = redis.NewScript(`
|
||
local key = KEYS[1]
|
||
local nick = ARGV[1]
|
||
local bid = ARGV[2] or ""
|
||
local exists = redis.call("EXISTS", key)
|
||
if exists == 0 then
|
||
return redis.error_reply("SESSION_NOT_FOUND")
|
||
end
|
||
local field = "participant:" .. nick
|
||
-- 昵称已存在视为已加入（幂等），直接返回 1
|
||
if redis.call("HEXISTS", key, field) == 1 then
|
||
return 1
|
||
end
|
||
redis.call("HSET", key, field, "")
|
||
if bid ~= "" then
|
||
redis.call("HSET", key, field .. ":browser_id", bid)
|
||
end
|
||
return 1
|
||
`)
|
||
|
||
// submitParticipantLuaScript 参与者提交自己的哈希
|
||
// 原子完成：检查 → 写入哈希+原始文本 → 与 initiator_hash 比对（若发起方已提交）
|
||
// 若已提交但 allow_resubmit=1 则覆盖（重算 match）
|
||
// KEYS[1] = session key
|
||
// ARGV[1] = nickname
|
||
// ARGV[2] = hash hex（已按当前 ignore_ws 规则 trim + hash）
|
||
// ARGV[3] = 原始文本（trim 前）
|
||
// 返回 {initiatorSubmitted (0/1), match ("1"/"0"/"")}
|
||
var submitParticipantLuaScript = redis.NewScript(`
|
||
local key = KEYS[1]
|
||
local nick = ARGV[1]
|
||
local hash_val = ARGV[2]
|
||
local orig_text = ARGV[3]
|
||
|
||
if redis.call("EXISTS", key) == 0 then
|
||
return redis.error_reply("SESSION_NOT_FOUND")
|
||
end
|
||
|
||
local field = "participant:" .. nick
|
||
if redis.call("HEXISTS", key, field) == 0 then
|
||
return redis.error_reply("NICKNAME_NOT_JOINED")
|
||
end
|
||
local existing = redis.call("HGET", key, field)
|
||
if existing ~= "" then
|
||
-- 已提交过：看是否允许重提交
|
||
local allow = redis.call("HGET", key, "allow_resubmit")
|
||
if allow ~= "1" then
|
||
return redis.error_reply("DUPLICATE_SUBMIT")
|
||
end
|
||
end
|
||
|
||
-- 写入（覆盖）哈希 + 原始文本
|
||
redis.call("HSET", key, field, hash_val)
|
||
redis.call("HSET", key, field .. ":text", orig_text)
|
||
-- 清除旧的 match 字段，稍后按需重算
|
||
redis.call("HDEL", key, field .. ":match")
|
||
|
||
local init_hash = redis.call("HGET", key, "initiator_hash")
|
||
if init_hash and init_hash ~= "" then
|
||
local match = (hash_val == init_hash) and "1" or "0"
|
||
redis.call("HSET", key, field .. ":match", match)
|
||
return {1, match}
|
||
else
|
||
return {0, ""}
|
||
end
|
||
`)
|
||
|
||
// submitInitiatorLuaScript 发起方提交标准答案
|
||
// 写入 initiator_hash + initiator_text 后，遍历所有已提交的 participant，批量计算 match
|
||
// KEYS[1] = session key
|
||
// ARGV[1] = nickname
|
||
// ARGV[2] = initiator hash（已按当前 ignore_ws 规则 trim + hash）
|
||
// ARGV[3] = 原始文本（trim 前）
|
||
var submitInitiatorLuaScript = redis.NewScript(`
|
||
local key = KEYS[1]
|
||
local nick = ARGV[1]
|
||
local init_hash = ARGV[2]
|
||
local orig_text = ARGV[3]
|
||
|
||
if redis.call("EXISTS", key) == 0 then
|
||
return redis.error_reply("SESSION_NOT_FOUND")
|
||
end
|
||
|
||
-- 发起方只允许提交一次（幂等：已提交过则跳过）
|
||
if redis.call("HEXISTS", key, "initiator_hash") == 1 then
|
||
return 1
|
||
end
|
||
|
||
redis.call("HSET", key, "initiator_hash", init_hash)
|
||
redis.call("HSET", key, "initiator_text", orig_text)
|
||
redis.call("HSET", key, "initiator_nickname", nick)
|
||
|
||
-- 遍历所有 participant:{nick} 且哈希非空的，计算 match
|
||
local all = redis.call("HGETALL", key)
|
||
for i = 1, #all, 2 do
|
||
local field = all[i]
|
||
local val = all[i + 1]
|
||
if string.sub(field, 1, #("participant:")) == "participant:"
|
||
and not string.find(field, ":match$")
|
||
and not string.find(field, ":text$")
|
||
and val ~= "" then
|
||
local match = (val == init_hash) and "1" or "0"
|
||
redis.call("HSET", key, field .. ":match", match)
|
||
end
|
||
end
|
||
return 1
|
||
`)
|
||
|
||
// renameParticipantLuaScript 原子重命名参与者
|
||
// 检查：session 存在、old 存在、new 不与 initiator_nickname/已有 participant 冲突
|
||
// （但如果 oldNick 的 browserId 和冲突方相同，视为同一个人，跳过冲突）
|
||
// 迁移 participant:{old}、participant:{old}:text、participant:{old}:match、participant:{old}:browser_id → new
|
||
// KEYS[1] = session key
|
||
// ARGV[1] = old nickname
|
||
// ARGV[2] = new nickname
|
||
// ARGV[3] = old participant's browserId
|
||
var renameParticipantLuaScript = redis.NewScript(`
|
||
local key = KEYS[1]
|
||
local oldNick = ARGV[1]
|
||
local newNick = ARGV[2]
|
||
local oldBid = ARGV[3] or ""
|
||
|
||
if redis.call("EXISTS", key) == 0 then
|
||
return redis.error_reply("SESSION_NOT_FOUND")
|
||
end
|
||
|
||
-- oldNick 必须存在
|
||
local oldField = "participant:" .. oldNick
|
||
if redis.call("HEXISTS", key, oldField) == 0 then
|
||
return redis.error_reply("OLD_NOT_FOUND")
|
||
end
|
||
|
||
-- newNick 不能等于发起方昵称（除非是同一个浏览器）
|
||
local initNick = redis.call("HGET", key, "initiator_nickname")
|
||
local initBid = redis.call("HGET", key, "initiator_browser_id") or ""
|
||
if initNick and initNick == newNick then
|
||
if oldBid == "" or initBid ~= oldBid then
|
||
return redis.error_reply("NICKNAME_CONFLICT")
|
||
end
|
||
end
|
||
|
||
-- newNick 不能已被其他 participant 占用（除非是同一个浏览器）
|
||
local newField = "participant:" .. newNick
|
||
if redis.call("HEXISTS", key, newField) == 1 then
|
||
local newBid = redis.call("HGET", key, newField .. ":browser_id") or ""
|
||
if oldBid == "" or newBid ~= oldBid then
|
||
return redis.error_reply("NICKNAME_CONFLICT")
|
||
end
|
||
end
|
||
|
||
-- 迁移四个 field
|
||
local hashVal = redis.call("HGET", key, oldField)
|
||
local textVal = redis.call("HGET", key, oldField .. ":text")
|
||
local matchVal = redis.call("HGET", key, oldField .. ":match")
|
||
local bidVal = redis.call("HGET", key, oldField .. ":browser_id")
|
||
|
||
redis.call("HSET", key, newField, hashVal)
|
||
if textVal ~= false then
|
||
redis.call("HSET", key, newField .. ":text", textVal)
|
||
end
|
||
if matchVal ~= false then
|
||
redis.call("HSET", key, newField .. ":match", matchVal)
|
||
end
|
||
if bidVal ~= false then
|
||
redis.call("HSET", key, newField .. ":browser_id", bidVal)
|
||
end
|
||
redis.call("HDEL", key, oldField)
|
||
redis.call("HDEL", key, oldField .. ":text")
|
||
redis.call("HDEL", key, oldField .. ":match")
|
||
redis.call("HDEL", key, oldField .. ":browser_id")
|
||
return 1
|
||
`)
|
||
|
||
// ── RedisSessionStore 实现 ──
|
||
|
||
type RedisSessionStore struct {
|
||
rdb *redis.Client
|
||
ttl time.Duration
|
||
}
|
||
|
||
func NewRedisSessionStore(rdb *redis.Client, defaultTTL time.Duration) *RedisSessionStore {
|
||
return &RedisSessionStore{rdb: rdb, ttl: defaultTTL}
|
||
}
|
||
|
||
func key(sessionID string) string {
|
||
return redisKeyPrefix + sessionID
|
||
}
|
||
|
||
// Create 原子创建会话 Hash key
|
||
func (s *RedisSessionStore) Create(ctx context.Context, sessionID, initiatorNickname, initiatorBrowserID string, ttl time.Duration) error {
|
||
if ttl <= 0 {
|
||
ttl = s.ttl
|
||
}
|
||
now := time.Now().Unix()
|
||
ttlSeconds := int64(ttl.Seconds())
|
||
|
||
result, err := createLuaScript.Run(ctx, s.rdb, []string{key(sessionID)},
|
||
now, initiatorNickname, initiatorBrowserID, ttlSeconds).Int64()
|
||
if err != nil {
|
||
return fmt.Errorf("redis create session: %w", err)
|
||
}
|
||
if result == 0 {
|
||
return fmt.Errorf("session id collision: %s", sessionID)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// JoinParticipant 登记参与者（空哈希占位）
|
||
func (s *RedisSessionStore) JoinParticipant(ctx context.Context, sessionID, nickname, browserID string) error {
|
||
_, err := joinParticipantLuaScript.Run(ctx, s.rdb, []string{key(sessionID)}, nickname, browserID).Int64()
|
||
if err != nil {
|
||
msg := err.Error()
|
||
if strings.Contains(msg, "SESSION_NOT_FOUND") {
|
||
return ErrSessionNotFound
|
||
}
|
||
return fmt.Errorf("redis join participant: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// SubmitParticipant 参与者提交自己的文本哈希（同时存原始文本）
|
||
func (s *RedisSessionStore) SubmitParticipant(ctx context.Context, sessionID, nickname, hashHex, originalText string) (bool, *bool, error) {
|
||
result, err := submitParticipantLuaScript.Run(ctx, s.rdb, []string{key(sessionID)}, nickname, hashHex, originalText).Result()
|
||
if err != nil {
|
||
msg := err.Error()
|
||
switch {
|
||
case strings.Contains(msg, "SESSION_NOT_FOUND"):
|
||
return false, nil, ErrSessionNotFound
|
||
case strings.Contains(msg, "DUPLICATE_SUBMIT"):
|
||
return false, nil, ErrDuplicateSubmit
|
||
case strings.Contains(msg, "NICKNAME_NOT_JOINED"):
|
||
// 参与者没走 join 直接 submit，自动补登记
|
||
if err := s.JoinParticipant(ctx, sessionID, nickname, ""); err != nil {
|
||
return false, nil, err
|
||
}
|
||
return s.SubmitParticipant(ctx, sessionID, nickname, hashHex, originalText)
|
||
default:
|
||
return false, nil, fmt.Errorf("redis submit participant: %w", err)
|
||
}
|
||
}
|
||
|
||
values := result.([]interface{})
|
||
initSubmitted := values[0].(int64) == 1
|
||
var match *bool
|
||
if initSubmitted {
|
||
b := values[1].(string) == "1"
|
||
match = &b
|
||
}
|
||
return initSubmitted, match, nil
|
||
}
|
||
|
||
// SubmitInitiator 发起方提交标准答案哈希（同时存原始文本）
|
||
func (s *RedisSessionStore) SubmitInitiator(ctx context.Context, sessionID, nickname, hashHex, originalText string) error {
|
||
_, err := submitInitiatorLuaScript.Run(ctx, s.rdb, []string{key(sessionID)}, nickname, hashHex, originalText).Int64()
|
||
if err != nil {
|
||
if strings.Contains(err.Error(), "SESSION_NOT_FOUND") {
|
||
return ErrSessionNotFound
|
||
}
|
||
return fmt.Errorf("redis submit initiator: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// SetResubmit 设置会话是否允许参与者重复提交
|
||
func (s *RedisSessionStore) SetResubmit(ctx context.Context, sessionID string, allow bool) error {
|
||
key := key(sessionID)
|
||
exists, err := s.rdb.Exists(ctx, key).Result()
|
||
if err != nil {
|
||
return fmt.Errorf("redis exists: %w", err)
|
||
}
|
||
if exists == 0 {
|
||
return ErrSessionNotFound
|
||
}
|
||
val := "0"
|
||
if allow {
|
||
val = "1"
|
||
}
|
||
return s.rdb.HSet(ctx, key, fieldAllowResub, val).Err()
|
||
}
|
||
|
||
func (s *RedisSessionStore) SetIgnoreWs(ctx context.Context, sessionID string, ignore bool) error {
|
||
key := key(sessionID)
|
||
exists, err := s.rdb.Exists(ctx, key).Result()
|
||
if err != nil {
|
||
return fmt.Errorf("redis exists: %w", err)
|
||
}
|
||
if exists == 0 {
|
||
return ErrSessionNotFound
|
||
}
|
||
val := "0"
|
||
if ignore {
|
||
val = "1"
|
||
}
|
||
return s.rdb.HSet(ctx, key, fieldIgnoreWs, val).Err()
|
||
}
|
||
|
||
func (s *RedisSessionStore) GetIgnoreWs(ctx context.Context, sessionID string) (bool, error) {
|
||
v, err := s.rdb.HGet(ctx, key(sessionID), fieldIgnoreWs).Result()
|
||
if err != nil {
|
||
if err == redis.Nil {
|
||
return false, nil
|
||
}
|
||
return false, fmt.Errorf("redis hget ignore_ws: %w", err)
|
||
}
|
||
return v == "1", nil
|
||
}
|
||
|
||
// RecalcMatches 按新的 ignoreWs 规则重新 trim + hash 所有已提交文本并重算 match
|
||
func (s *RedisSessionStore) RecalcMatches(ctx context.Context, sessionID string, ignoreWs bool) error {
|
||
k := key(sessionID)
|
||
all, err := s.rdb.HGetAll(ctx, k).Result()
|
||
if err != nil {
|
||
return fmt.Errorf("redis hgetall: %w", err)
|
||
}
|
||
if len(all) == 0 {
|
||
return ErrSessionNotFound
|
||
}
|
||
|
||
// 读发起方原始文本
|
||
initText := all[fieldInitText]
|
||
if initText == "" {
|
||
return nil // 发起方还没提交，不需要重算
|
||
}
|
||
initProcessed := initText
|
||
if ignoreWs {
|
||
initProcessed = strings.TrimSpace(initText)
|
||
}
|
||
initHash := hash.SHA256(initProcessed)
|
||
|
||
pipe := s.rdb.TxPipeline()
|
||
pipe.HSet(ctx, k, fieldInitHash, initHash)
|
||
|
||
// 遍历所有已提交参与者，重算 hash + match
|
||
for field, val := range all {
|
||
if !strings.HasPrefix(field, partFieldPrefix) {
|
||
continue
|
||
}
|
||
if strings.HasSuffix(field, partMatchSuffix) || strings.HasSuffix(field, partTextSuffix) {
|
||
continue
|
||
}
|
||
if val == "" {
|
||
continue
|
||
}
|
||
textField := field + partTextSuffix
|
||
origText := all[textField]
|
||
if origText == "" {
|
||
continue // 旧数据（没存过原始文本），跳过
|
||
}
|
||
processed := origText
|
||
if ignoreWs {
|
||
processed = strings.TrimSpace(origText)
|
||
}
|
||
newHash := hash.SHA256(processed)
|
||
match := "0"
|
||
if newHash == initHash {
|
||
match = "1"
|
||
}
|
||
pipe.HSet(ctx, k, field, newHash)
|
||
pipe.HSet(ctx, k, field+partMatchSuffix, match)
|
||
}
|
||
|
||
_, err = pipe.Exec(ctx)
|
||
if err != nil {
|
||
return fmt.Errorf("redis recalc pipeline: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// UpdateParticipantNickname 原子重命名参与者
|
||
func (s *RedisSessionStore) UpdateParticipantNickname(ctx context.Context, sessionID, oldNick, newNick, oldBrowserID string) error {
|
||
_, err := renameParticipantLuaScript.Run(ctx, s.rdb, []string{key(sessionID)}, oldNick, newNick, oldBrowserID).Int64()
|
||
if err != nil {
|
||
msg := err.Error()
|
||
switch {
|
||
case strings.Contains(msg, "SESSION_NOT_FOUND"):
|
||
return ErrSessionNotFound
|
||
case strings.Contains(msg, "OLD_NOT_FOUND"):
|
||
return ErrOldNotFound
|
||
case strings.Contains(msg, "NICKNAME_CONFLICT"):
|
||
return ErrNicknameConflict
|
||
default:
|
||
return fmt.Errorf("redis rename participant: %w", err)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// RemoveParticipant 参与者离开时从 Redis 删除其所有字段
|
||
func (s *RedisSessionStore) RemoveParticipant(ctx context.Context, sessionID, nickname string) error {
|
||
k := key(sessionID)
|
||
exists, err := s.rdb.Exists(ctx, k).Result()
|
||
if err != nil {
|
||
return fmt.Errorf("redis exists: %w", err)
|
||
}
|
||
if exists == 0 {
|
||
return ErrSessionNotFound
|
||
}
|
||
pipe := s.rdb.TxPipeline()
|
||
pipe.HDel(ctx, k, partFieldPrefix+nickname)
|
||
pipe.HDel(ctx, k, partFieldPrefix+nickname+partTextSuffix)
|
||
pipe.HDel(ctx, k, partFieldPrefix+nickname+partMatchSuffix)
|
||
_, err = pipe.Exec(ctx)
|
||
if err != nil {
|
||
return fmt.Errorf("redis remove participant: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// Get 获取会话当前状态全量快照
|
||
func (s *RedisSessionStore) Get(ctx context.Context, sessionID string) (*SessionSnapshot, error) {
|
||
exists, err := s.rdb.Exists(ctx, key(sessionID)).Result()
|
||
if err != nil {
|
||
return nil, fmt.Errorf("redis exists: %w", err)
|
||
}
|
||
if exists == 0 {
|
||
return nil, ErrSessionNotFound
|
||
}
|
||
|
||
hash, err := s.rdb.HGetAll(ctx, key(sessionID)).Result()
|
||
if err != nil {
|
||
return nil, fmt.Errorf("redis hgetall: %w", err)
|
||
}
|
||
|
||
ttl, err := s.rdb.TTL(ctx, key(sessionID)).Result()
|
||
if err != nil {
|
||
return nil, fmt.Errorf("redis ttl: %w", err)
|
||
}
|
||
|
||
snap := &SessionSnapshot{
|
||
SessionID: sessionID,
|
||
InitiatorHash: hash[fieldInitHash],
|
||
InitiatorNickname: hash[fieldInitNick],
|
||
InitiatorBrowserID: hash[fieldInitBrowser],
|
||
InitiatorSubmitted: hash[fieldInitHash] != "",
|
||
AllowResubmit: hash[fieldAllowResub] == "1",
|
||
IgnoreWs: hash[fieldIgnoreWs] == "1",
|
||
ExpiresAt: time.Now().Add(ttl),
|
||
}
|
||
|
||
// 收集所有 participant 条目，按 nickname 排序
|
||
var nicks []string
|
||
partMap := map[string]string{} // nick → hash
|
||
for k, v := range hash {
|
||
if strings.HasPrefix(k, partFieldPrefix) &&
|
||
!strings.HasSuffix(k, partMatchSuffix) &&
|
||
!strings.HasSuffix(k, partTextSuffix) &&
|
||
!strings.HasSuffix(k, partBrowserSuffix) {
|
||
nick := strings.TrimPrefix(k, partFieldPrefix)
|
||
nicks = append(nicks, nick)
|
||
partMap[nick] = v
|
||
}
|
||
}
|
||
sort.Strings(nicks)
|
||
|
||
initiatorHash := hash[fieldInitHash]
|
||
for _, nick := range nicks {
|
||
h := partMap[nick]
|
||
p := ParticipantInfo{
|
||
Nickname: nick,
|
||
BrowserID: hash[partFieldPrefix+nick+partBrowserSuffix],
|
||
Submitted: h != "",
|
||
}
|
||
if h != "" && initiatorHash != "" {
|
||
m := h == initiatorHash
|
||
p.Match = &m
|
||
} else if mStr, ok := hash[partFieldPrefix+nick+":match"]; ok && mStr != "" {
|
||
m := mStr == "1"
|
||
p.Match = &m
|
||
}
|
||
snap.Participants = append(snap.Participants, p)
|
||
}
|
||
|
||
return snap, nil
|
||
}
|
||
|
||
// Delete 删除会话
|
||
func (s *RedisSessionStore) Delete(ctx context.Context, sessionID string) error {
|
||
if err := s.rdb.Del(ctx, key(sessionID)).Err(); err != nil {
|
||
return fmt.Errorf("redis del: %w", err)
|
||
}
|
||
return nil
|
||
}
|