319 lines
9.1 KiB
Go
319 lines
9.1 KiB
Go
package store
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
)
|
||
|
||
// 包级错误，业务层可以 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")
|
||
)
|
||
|
||
const (
|
||
redisKeyPrefix = "dualcheck:session:"
|
||
fieldCreatedAt = "created_at"
|
||
fieldInitHash = "initiator_hash"
|
||
fieldInitNick = "initiator_nickname"
|
||
fieldAllowResub = "allow_resubmit"
|
||
// participant:{nick} 存该参与者的哈希值；participant:{nick}:match 存 "1"/"0"
|
||
partFieldPrefix = "participant:"
|
||
)
|
||
|
||
// ── 五个 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, "allow_resubmit", "0")
|
||
redis.call("EXPIRE", key, ARGV[2])
|
||
return 1
|
||
`)
|
||
|
||
// joinParticipantLuaScript 登记参与者昵称（空哈希占位，表示已加入未提交）
|
||
// KEYS[1] = session key
|
||
// ARGV[1] = nickname
|
||
var joinParticipantLuaScript = redis.NewScript(`
|
||
local key = KEYS[1]
|
||
local nick = ARGV[1]
|
||
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, "")
|
||
return 1
|
||
`)
|
||
|
||
// submitParticipantLuaScript 参与者提交自己的哈希
|
||
// 原子完成：检查 → 写入 → 与 initiator_hash 比对（若发起方已提交）
|
||
// 若已提交但 allow_resubmit=1 则覆盖（重算 match）
|
||
// KEYS[1] = session key
|
||
// ARGV[1] = nickname
|
||
// ARGV[2] = hash hex
|
||
// 返回 {initiatorSubmitted (0/1), match ("1"/"0"/"")}
|
||
var submitParticipantLuaScript = redis.NewScript(`
|
||
local key = KEYS[1]
|
||
local nick = ARGV[1]
|
||
local hash_val = ARGV[2]
|
||
|
||
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)
|
||
-- 清除旧的 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 后，遍历所有已提交的 participant，批量计算 match
|
||
// KEYS[1] = session key
|
||
// ARGV[1] = nickname
|
||
// ARGV[2] = initiator hash
|
||
var submitInitiatorLuaScript = redis.NewScript(`
|
||
local key = KEYS[1]
|
||
local nick = ARGV[1]
|
||
local init_hash = ARGV[2]
|
||
|
||
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_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 val ~= "" then
|
||
local match = (val == init_hash) and "1" or "0"
|
||
redis.call("HSET", key, field .. ":match", match)
|
||
end
|
||
end
|
||
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 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, 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 string) error {
|
||
_, err := joinParticipantLuaScript.Run(ctx, s.rdb, []string{key(sessionID)}, nickname).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 string) (bool, *bool, error) {
|
||
result, err := submitParticipantLuaScript.Run(ctx, s.rdb, []string{key(sessionID)}, nickname, hashHex).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)
|
||
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 string) error {
|
||
_, err := submitInitiatorLuaScript.Run(ctx, s.rdb, []string{key(sessionID)}, nickname, hashHex).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()
|
||
}
|
||
|
||
// 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],
|
||
InitiatorSubmitted: hash[fieldInitHash] != "",
|
||
AllowResubmit: hash[fieldAllowResub] == "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, ":match") {
|
||
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,
|
||
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
|
||
}
|