app-api/internal/task/share_code_persist.go

package task

import (
	"context"
	"encoding/json"
	"time"

	"github.com/luxsin/app-api/internal/cache"
	"github.com/luxsin/app-api/internal/model"
	"github.com/luxsin/app-api/internal/repository"
	"go.uber.org/zap"
)

type ShareCodePersistTask struct {
	cache *cache.ShareCodeCache
	repo  *repository.ShareCodeRepository
	log   *zap.Logger
}

func NewShareCodePersistTask(shareCache *cache.ShareCodeCache, repo *repository.ShareCodeRepository, log *zap.Logger) *ShareCodePersistTask {
	return &ShareCodePersistTask{cache: shareCache, repo: repo, log: log}
}

func (t *ShareCodePersistTask) Start(interval time.Duration) {
	go func() {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()

		for range ticker.C {
			t.Persist()
		}
	}()
	t.log.Info("share code persist task started", zap.String("interval", interval.String()))
}

func (t *ShareCodePersistTask) Persist() {
	t.persistExports()
	t.persistImports()
}

func (t *ShareCodePersistTask) persistExports() {
	ctx := context.Background()

	codes, err := t.cache.ListPending(ctx)
	if err != nil {
		t.log.Error("list pending share codes failed", zap.Error(err))
		return
	}
	if len(codes) == 0 {
		return
	}

	t.log.Info("persisting share code export logs", zap.Int("count", len(codes)))

	var persisted, skipped, failed int

	for _, code := range codes {
		ok := t.persistExportOne(ctx, code)
		switch ok {
		case 1:
			persisted++
		case 0:
			skipped++
		default:
			failed++
		}
	}

	if persisted > 0 || failed > 0 {
		t.log.Info("share code export logs persist finished",
			zap.Int("persisted", persisted),
			zap.Int("skipped", skipped),
			zap.Int("failed", failed),
		)
	}
}

func (t *ShareCodePersistTask) persistImports() {
	ctx := context.Background()

	logs, err := t.cache.ListPendingImports(ctx)
	if err != nil {
		t.log.Error("list pending share import logs failed", zap.Error(err))
		return
	}
	if len(logs) == 0 {
		return
	}

	t.log.Info("persisting share code import logs", zap.Int("count", len(logs)))

	var persisted, skipped, failed int

	for field, payload := range logs {
		ok := t.persistImportOne(ctx, field, payload)
		switch ok {
		case 1:
			persisted++
		case 0:
			skipped++
		default:
			failed++
		}
	}

	if persisted > 0 || failed > 0 {
		t.log.Info("share code import logs persist finished",
			zap.Int("persisted", persisted),
			zap.Int("skipped", skipped),
			zap.Int("failed", failed),
		)
	}
}

// persistExportOne 返回值：1=成功刷入，0=跳过，-1=失败
func (t *ShareCodePersistTask) persistExportOne(ctx context.Context, shareCode string) int {
	acquired, err := t.cache.AcquireFlushLock(ctx, shareCode)
	if err != nil {
		t.log.Error("acquire share flush lock failed", zap.String("share_code", shareCode), zap.Error(err))
		return -1
	}
	if !acquired {
		return 0
	}
	defer func() {
		if err := t.cache.ReleaseFlushLock(ctx, shareCode); err != nil {
			t.log.Warn("release share flush lock failed", zap.String("share_code", shareCode), zap.Error(err))
		}
	}()

	data, err := t.cache.Get(ctx, shareCode)
	if err != nil {
		t.log.Error("get share code from redis failed", zap.String("share_code", shareCode), zap.Error(err))
		return -1
	}
	if data == nil {
		_ = t.cache.MarkPersisted(ctx, shareCode)
		return 0
	}
	if data.Persisted {
		return 0
	}

	exists, err := t.repo.HasExportLog(ctx, shareCode)
	if err != nil {
		t.log.Error("check share export log failed", zap.String("share_code", shareCode), zap.Error(err))
		return -1
	}
	if exists {
		if err := t.cache.MarkPersisted(ctx, shareCode); err != nil {
			t.log.Error("mark share code persisted failed", zap.String("share_code", shareCode), zap.Error(err))
			return -1
		}
		return 0
	}

	logEntry := model.ShareCodeLog{
		MacAddr:   data.MacAddr,
		ShareCode: data.ShareCode,
		Action:    "export",
		IpAddr:    data.IpAddr,
		EqData:    []byte(data.EqData),
		ExpireAt:  repository.ParseShareExpireAt(data.ExpireAt),
	}
	if err := t.repo.InsertLog(ctx, logEntry); err != nil {
		t.log.Error("insert share export log failed", zap.String("share_code", shareCode), zap.Error(err))
		return -1
	}

	if err := t.cache.MarkPersisted(ctx, shareCode); err != nil {
		t.log.Error("mark share code persisted failed", zap.String("share_code", shareCode), zap.Error(err))
		return -1
	}

	return 1
}

// persistImportOne 返回值：1=成功刷入，0=跳过，-1=失败
func (t *ShareCodePersistTask) persistImportOne(ctx context.Context, field, payload string) int {
	acquired, err := t.cache.AcquireImportFlushLock(ctx, field)
	if err != nil {
		t.log.Error("acquire share import flush lock failed", zap.String("field", field), zap.Error(err))
		return -1
	}
	if !acquired {
		return 0
	}
	defer func() {
		if err := t.cache.ReleaseImportFlushLock(ctx, field); err != nil {
			t.log.Warn("release share import flush lock failed", zap.String("field", field), zap.Error(err))
		}
	}()

	var pending cache.ShareImportPendingLog
	if err := json.Unmarshal([]byte(payload), &pending); err != nil {
		t.log.Error("unmarshal share import pending log failed", zap.String("field", field), zap.Error(err))
		return -1
	}

	expireAt, err := time.Parse(time.RFC3339, pending.ExpireAt)
	if err != nil {
		t.log.Error("parse share import expire_at failed", zap.String("field", field), zap.Error(err))
		return -1
	}

	logEntry := model.ShareCodeLog{
		MacAddr:   pending.MacAddr,
		ShareCode: pending.ShareCode,
		Action:    "import",
		IpAddr:    pending.IpAddr,
		EqData:    []byte(pending.EqData),
		ExpireAt:  repository.ParseShareExpireAt(expireAt),
	}
	if err := t.repo.InsertLog(ctx, logEntry); err != nil {
		t.log.Error("insert share import log failed",
			zap.String("share_code", pending.ShareCode),
			zap.String("field", field),
			zap.Error(err),
		)
		return -1
	}

	if err := t.cache.RemovePendingImport(ctx, field); err != nil {
		t.log.Error("remove pending share import log failed", zap.String("field", field), zap.Error(err))
		return -1
	}

	return 1
}