app-api/internal/cache/share_code_cache.go

// Package cache — 分享码 Redis 缓存层
//
// # Redis 存储结构
//
//	share:{code}              Hash, TTL = SHARE_CODE_TTL_MIN — 分享码主数据
//	  mac_addr                创建者 MAC 地址
//	  ip_addr                 创建者 IP
//	  model                   操作设备型号
//	  eq_data                 EQ 参数 JSON
//	  expire_at               过期时间 (RFC3339)
//	  persisted               是否已刷入 DB ("0"/"1")
//
//	share:mac:{mac}           ZSET, TTL = SHARE_CODE_TTL_MIN — MAC 二级索引 (查询时主动清理过期成员)
//	  member = share_code
//	  score  = expire_at unix timestamp
//
//	share:pending             SET — 导出持久化队列 (无 TTL, hash 过期后自动清理)
//
//	share:import:pending      Hash, TTL 12h (兜底) — 导入持久化队列
//	  field = {mac}:{code}    幂等: 同 MAC+code 只保留一条
//	  value = JSON payload
//
//	share:flush:lock:{code}         String, TTL 30s — 导出刷入分布式锁
//	share:import:flush:lock:{field} String, TTL 30s — 导入刷入分布式锁
//
// # 数据流
//
//	创建: Lua 原子写入 hash + SADD pending + ZADD mac 索引
//	查询: ZREMRANGEBYSCORE 清过期索引 → ZRANGE 取剩余 → HGETALL 兆底校验
//	导入: HGETALL hash → HSET import:pending (幂等)
//	删除: 校验 mac+code → DEL hash + SREM pending + ZREM mac 索引
//	持久化: 每 5 分钟任务 SMEMBERS/HGETALL pending → INSERT DB → 移除 pending 条目
package cache

import (
	"context"
	"crypto/rand"
	"encoding/json"
	"errors"
	"fmt"
	"math/big"
	"strings"
	"time"

	"github.com/redis/go-redis/v9"
)

const (
	shareCodeKeyPrefix       = "share:"
	sharePendingSet          = "share:pending"
	shareImportPendingHash   = "share:import:pending"
	shareFlushLockPref       = "share:flush:lock:"
	shareImportFlushLockPref = "share:import:flush:lock:"
	shareMacIndexPrefix      = "share:mac:"
	shareCodeLength          = 5
	shareImportPendingTTL    = 12 * time.Hour
	shareCodeCharset         = "23456789ABCDEFGHJKLMNPQRSTUVWXYZ"
	shareCodeMaxRetries      = 20

	fieldMacAddr   = "mac_addr"
	fieldIPAddr    = "ip_addr"
	fieldModel     = "model"
	fieldEqData    = "eq_data"
	fieldExpireAt  = "expire_at"
	fieldPersisted = "persisted"
)

var shareCreateScript = redis.NewScript(`
if redis.call('EXISTS', KEYS[1]) == 1 then
  return 0
end
redis.call('HSET', KEYS[1],
  'mac_addr', ARGV[1],
  'ip_addr', ARGV[2],
  'model', ARGV[3],
  'eq_data', ARGV[4],
  'expire_at', ARGV[5],
  'persisted', '0'
)
redis.call('EXPIRE', KEYS[1], tonumber(ARGV[6]))
redis.call('SADD', KEYS[2], ARGV[7])
-- ZSET: score = expire_at unix timestamp, member = share code
redis.call('ZADD', KEYS[3], tonumber(ARGV[8]), ARGV[7])
redis.call('EXPIRE', KEYS[3], tonumber(ARGV[9]))
return 1
`)

var shareDeleteScript = redis.NewScript(`
if redis.call('EXISTS', KEYS[1]) == 0 then
  return 0
end
if redis.call('HGET', KEYS[1], 'mac_addr') ~= ARGV[1] then
  return -1
end
redis.call('DEL', KEYS[1])
redis.call('SREM', KEYS[2], ARGV[2])
redis.call('ZREM', KEYS[3], ARGV[2])
return 1
`)

type ShareDeleteResult int

const (
	ShareDeleteNotFound ShareDeleteResult = 0
	ShareDeleteForbidden ShareDeleteResult = -1
	ShareDeleteOK ShareDeleteResult = 1
)

type ShareImportPendingLog struct {
	MacAddr   string `json:"mac_addr"`
	ShareCode string `json:"share_code"`
	Model     string `json:"model"`
	IpAddr    string `json:"ip_addr"`
	EqData    string `json:"eq_data"`
	ExpireAt  string `json:"expire_at"`
}

type ShareCodeData struct {
	ShareCode string
	MacAddr   string
	IpAddr    string
	Model     string
	EqData    string
	ExpireAt  time.Time
	Persisted bool
}

type ShareCodeCache struct {
	rdb     *redis.Client
	codeTTL time.Duration
}

func NewShareCodeCache(rdb *redis.Client, codeTTL time.Duration) *ShareCodeCache {
	return &ShareCodeCache{rdb: rdb, codeTTL: codeTTL}
}

// ShareCodeTTL returns the configured share code TTL.
func (c *ShareCodeCache) ShareCodeTTL() time.Duration {
	return c.codeTTL
}

func (c *ShareCodeCache) Create(ctx context.Context, macAddr, ipAddr, model string, eqData []byte) (*ShareCodeData, error) {
	expireAt := time.Now().Add(c.codeTTL)
	eqJSON := string(eqData)

	for i := 0; i < shareCodeMaxRetries; i++ {
		code, err := randomShareCode()
		if err != nil {
			return nil, err
		}

		key := shareCodeKey(code)
		macIdxKey := shareMacIndexKey(macAddr)
		ok, err := shareCreateScript.Run(ctx, c.rdb, []string{key, sharePendingSet, macIdxKey},
			macAddr, ipAddr, model, eqJSON, expireAt.Format(time.RFC3339), int(c.codeTTL.Seconds()), code, expireAt.Unix(), int(c.codeTTL.Seconds()),
		).Int()
		if err != nil {
			return nil, fmt.Errorf("create share code in redis: %w", err)
		}
		if ok == 1 {
			return &ShareCodeData{
				ShareCode: code,
				MacAddr:   macAddr,
				IpAddr:    ipAddr,
				Model:     model,
				EqData:    eqJSON,
				ExpireAt:  expireAt,
				Persisted: false,
			}, nil
		}
	}

	return nil, errors.New("failed to allocate unique share code")
}

func (c *ShareCodeCache) Get(ctx context.Context, shareCode string) (*ShareCodeData, error) {
	key := shareCodeKey(shareCode)
	values, err := c.rdb.HGetAll(ctx, key).Result()
	if err != nil {
		return nil, fmt.Errorf("hgetall share code: %w", err)
	}
	if len(values) == 0 {
		return nil, nil
	}

	expireAt, err := time.Parse(time.RFC3339, values[fieldExpireAt])
	if err != nil {
		return nil, fmt.Errorf("parse share code expire_at: %w", err)
	}

	return &ShareCodeData{
		ShareCode: shareCode,
		MacAddr:   values[fieldMacAddr],
		IpAddr:    values[fieldIPAddr],
		Model:     values[fieldModel],
		EqData:    values[fieldEqData],
		ExpireAt:  expireAt,
		Persisted: values[fieldPersisted] == "1",
	}, nil
}

func (c *ShareCodeCache) ListPending(ctx context.Context) ([]string, error) {
	codes, err := c.rdb.SMembers(ctx, sharePendingSet).Result()
	if err != nil {
		return nil, fmt.Errorf("smembers share pending: %w", err)
	}
	return codes, nil
}

func (c *ShareCodeCache) AcquireFlushLock(ctx context.Context, shareCode string) (bool, error) {
	lockKey := shareFlushLockPref + shareCode
	return c.rdb.SetNX(ctx, lockKey, "1", 30*time.Second).Result()
}

func (c *ShareCodeCache) ReleaseFlushLock(ctx context.Context, shareCode string) error {
	return c.rdb.Del(ctx, shareFlushLockPref+shareCode).Err()
}

func (c *ShareCodeCache) MarkPersisted(ctx context.Context, shareCode string) error {
	key := shareCodeKey(shareCode)

	// preserve remaining TTL before HSET (HSET clears TTL in some Redis versions)
	remainTTL, err := c.rdb.PTTL(ctx, key).Result()
	if err != nil {
		return fmt.Errorf("get share code ttl: %w", err)
	}

	if err := c.rdb.HSet(ctx, key, fieldPersisted, "1").Err(); err != nil {
		return fmt.Errorf("mark share code persisted: %w", err)
	}

	// restore TTL
	if remainTTL > 0 {
		_ = c.rdb.PExpire(ctx, key, remainTTL).Err()
	}

	return c.rdb.SRem(ctx, sharePendingSet, shareCode).Err()
}

func (c *ShareCodeCache) EnqueueImportLog(ctx context.Context, macAddr, shareCode, model, ipAddr, eqData string, expireAt time.Time) error {
	field := fmt.Sprintf("%s:%s", macAddr, shareCode) // 同 MAC+code 幂等，避免重复导入产生多条记录
	payload, err := json.Marshal(ShareImportPendingLog{
		MacAddr:   macAddr,
		ShareCode: shareCode,
		Model:     model,
		IpAddr:    ipAddr,
		EqData:    eqData,
		ExpireAt:  expireAt.Format(time.RFC3339),
	})
	if err != nil {
		return fmt.Errorf("marshal import pending log: %w", err)
	}
	if err := c.rdb.HSet(ctx, shareImportPendingHash, field, payload).Err(); err != nil {
		return fmt.Errorf("enqueue import log: %w", err)
	}
	// Fallback TTL: prevent unbounded growth if persist task is disabled or DB is down
	c.rdb.Expire(ctx, shareImportPendingHash, shareImportPendingTTL)
	return nil
}

func (c *ShareCodeCache) ListPendingImports(ctx context.Context) (map[string]string, error) {
	logs, err := c.rdb.HGetAll(ctx, shareImportPendingHash).Result()
	if err != nil {
		return nil, fmt.Errorf("hgetall import pending: %w", err)
	}
	return logs, nil
}

func (c *ShareCodeCache) RemovePendingImport(ctx context.Context, field string) error {
	return c.rdb.HDel(ctx, shareImportPendingHash, field).Err()
}

func (c *ShareCodeCache) AcquireImportFlushLock(ctx context.Context, field string) (bool, error) {
	return c.rdb.SetNX(ctx, shareImportFlushLockPref+field, "1", 30*time.Second).Result()
}

func (c *ShareCodeCache) ReleaseImportFlushLock(ctx context.Context, field string) error {
	return c.rdb.Del(ctx, shareImportFlushLockPref+field).Err()
}

func (c *ShareCodeCache) ListByMac(ctx context.Context, macAddr string) ([]*ShareCodeData, error) {
	key := shareMacIndexKey(macAddr)
	now := time.Now()

	// Remove expired entries from ZSET (score = expire_at unix timestamp)
	_, err := c.rdb.ZRemRangeByScore(ctx, key, "-inf", fmt.Sprintf("%d", now.Unix())).Result()
	if err != nil && strings.Contains(err.Error(), "WRONGTYPE") {
		// Old SET-type key from previous version, delete it
		_ = c.rdb.Del(ctx, key).Err()
		return nil, nil
	}

	// Get remaining (unexpired) codes
	codes, err := c.rdb.ZRange(ctx, key, 0, -1).Result()
	if err != nil {
		return nil, fmt.Errorf("zrange mac index: %w", err)
	}
	if len(codes) == 0 {
		// ZSET is empty, remove the key to avoid accumulating empty keys
		_ = c.rdb.Del(ctx, key).Err()
		return nil, nil
	}

	var result []*ShareCodeData
	for _, code := range codes {
		data, err := c.Get(ctx, code)
		if err != nil {
			return nil, err
		}
		if data == nil {
			// hash already expired/removed, clean from ZSET
			_ = c.rdb.ZRem(ctx, key, code).Err()
			continue
		}
		result = append(result, data)
	}

	return result, nil
}

func (c *ShareCodeCache) Delete(ctx context.Context, macAddr, shareCode string) (ShareDeleteResult, error) {
	key := shareCodeKey(shareCode)
	macIdxKey := shareMacIndexKey(macAddr)
	ok, err := shareDeleteScript.Run(ctx, c.rdb, []string{key, sharePendingSet, macIdxKey},
		macAddr, shareCode,
	).Int()
	if err != nil {
		return ShareDeleteNotFound, fmt.Errorf("delete share code in redis: %w", err)
	}
	return ShareDeleteResult(ok), nil
}

func shareMacIndexKey(macAddr string) string {
	return shareMacIndexPrefix + macAddr
}

func shareCodeKey(code string) string {
	return shareCodeKeyPrefix + code
}

func randomShareCode() (string, error) {
	b := make([]byte, shareCodeLength)
	max := big.NewInt(int64(len(shareCodeCharset)))
	for i := range b {
		n, err := rand.Int(rand.Reader, max)
		if err != nil {
			return "", err
		}
		b[i] = shareCodeCharset[n.Int64()]
	}
	return string(b), nil
}