backend/internal/service/async_log_manager.go

package service

import (
	"cmp"
	"context"
	"os"
	"sync"
	"time"

	"carrot_bbs/internal/model"
	"carrot_bbs/internal/repository"

	"go.uber.org/zap"
	"gorm.io/gorm"
)

// AsyncLogConfig 异步日志配置
type AsyncLogConfig struct {
	BufferSize     int
	BatchSize      int
	FlushInterval  time.Duration
	WorkerCount    int
	EnableFallback bool
	FallbackPath   string
}

// AsyncLogManager 异步日志管理器
type AsyncLogManager struct {
	operationChan  chan *model.OperationLog
	loginChan      chan *model.LoginLog
	dataChangeChan chan *model.DataChangeLog

	cfg    *AsyncLogConfig
	db     *gorm.DB
	wg     sync.WaitGroup
	ctx    context.Context
	cancel context.CancelFunc
	logger *zap.Logger

	fallbackWriter *os.File
}

// NewAsyncLogManager 创建异步日志管理器
func NewAsyncLogManager(
	db *gorm.DB,
	operationRepo *repository.OperationLogRepository,
	loginRepo *repository.LoginLogRepository,
	dataChangeRepo *repository.DataChangeLogRepository,
	logger *zap.Logger,
	cfg *AsyncLogConfig,
) *AsyncLogManager {
	ctx, cancel := context.WithCancel(context.Background())

	m := &AsyncLogManager{
		operationChan:  make(chan *model.OperationLog, cfg.BufferSize),
		loginChan:      make(chan *model.LoginLog, cfg.BufferSize),
		dataChangeChan: make(chan *model.DataChangeLog, cfg.BufferSize),
		cfg:            cfg,
		db:             db,
		ctx:            ctx,
		cancel:         cancel,
		logger:         logger,
	}

	if cfg.EnableFallback {
		if err := m.initFallbackWriter(); err != nil {
			logger.Warn("failed to init fallback writer", zap.Error(err))
		}
	}

	for i := range cfg.WorkerCount {
		workerID := i
		m.wg.Go(func() {
			m.worker(workerID, operationRepo, loginRepo, dataChangeRepo)
		})
	}

	go m.flushScheduler()

	logger.Info("async log manager started",
		zap.Int("buffer_size", cfg.BufferSize),
		zap.Int("batch_size", cfg.BatchSize),
		zap.Duration("flush_interval", cfg.FlushInterval),
		zap.Int("worker_count", cfg.WorkerCount),
	)

	return m
}

// worker 处理协程
func (m *AsyncLogManager) worker(
	id int,
	operationRepo *repository.OperationLogRepository,
	loginRepo *repository.LoginLogRepository,
	dataChangeRepo *repository.DataChangeLogRepository,
) {
	opBatch := make([]*model.OperationLog, 0, m.cfg.BatchSize)
	loginBatch := make([]*model.LoginLog, 0, m.cfg.BatchSize)
	changeBatch := make([]*model.DataChangeLog, 0, m.cfg.BatchSize)

	ticker := time.NewTicker(m.cfg.FlushInterval)
	defer ticker.Stop()

	for {
		select {
		case <-m.ctx.Done():
			m.flushBatches(opBatch, loginBatch, changeBatch, operationRepo, loginRepo, dataChangeRepo)
			return

		case op := <-m.operationChan:
			opBatch = append(opBatch, op)
			if len(opBatch) >= m.cfg.BatchSize {
				if err := m.flushOperationLogs(opBatch, operationRepo); err != nil {
					m.handleError("operation_logs", opBatch, err)
				}
				opBatch = opBatch[:0]
			}

		case login := <-m.loginChan:
			loginBatch = append(loginBatch, login)
			if len(loginBatch) >= m.cfg.BatchSize {
				if err := m.flushLoginLogs(loginBatch, loginRepo); err != nil {
					m.handleError("login_logs", loginBatch, err)
				}
				loginBatch = loginBatch[:0]
			}

		case change := <-m.dataChangeChan:
			changeBatch = append(changeBatch, change)
			if len(changeBatch) >= m.cfg.BatchSize {
				if err := m.flushDataChangeLogs(changeBatch, dataChangeRepo); err != nil {
					m.handleError("data_change_logs", changeBatch, err)
				}
				changeBatch = changeBatch[:0]
			}

		case <-ticker.C:
			if len(opBatch) > 0 || len(loginBatch) > 0 || len(changeBatch) > 0 {
				m.flushBatches(opBatch, loginBatch, changeBatch, operationRepo, loginRepo, dataChangeRepo)
				opBatch = opBatch[:0]
				loginBatch = loginBatch[:0]
				changeBatch = changeBatch[:0]
			}
		}
	}
}

// flushScheduler 定时刷新
func (m *AsyncLogManager) flushScheduler() {
	ticker := time.NewTicker(m.cfg.FlushInterval)
	defer ticker.Stop()

	for {
		select {
		case <-m.ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

// flushBatches 批量写入
func (m *AsyncLogManager) flushBatches(
	ops []*model.OperationLog,
	logins []*model.LoginLog,
	changes []*model.DataChangeLog,
	operationRepo *repository.OperationLogRepository,
	loginRepo *repository.LoginLogRepository,
	dataChangeRepo *repository.DataChangeLogRepository,
) {
	if len(ops) == 0 && len(logins) == 0 && len(changes) == 0 {
		return
	}

	ctx := context.Background()

	if len(ops) > 0 {
		if err := operationRepo.BatchCreateOperationLogs(ctx, ops); err != nil {
			m.handleError("operation_logs", ops, err)
		}
	}

	if len(logins) > 0 {
		if err := loginRepo.BatchCreateLoginLogs(ctx, logins); err != nil {
			m.handleError("login_logs", logins, err)
		}
	}

	if len(changes) > 0 {
		if err := dataChangeRepo.BatchCreateDataChangeLogs(ctx, changes); err != nil {
			m.handleError("data_change_logs", changes, err)
		}
	}
}

// flushOperationLogs 刷新操作日志
func (m *AsyncLogManager) flushOperationLogs(logs []*model.OperationLog, repo *repository.OperationLogRepository) error {
	return repo.BatchCreateOperationLogs(context.Background(), logs)
}

// flushLoginLogs 刷新登录日志
func (m *AsyncLogManager) flushLoginLogs(logs []*model.LoginLog, repo *repository.LoginLogRepository) error {
	return repo.BatchCreateLoginLogs(context.Background(), logs)
}

// flushDataChangeLogs 刷新数据变更日志
func (m *AsyncLogManager) flushDataChangeLogs(logs []*model.DataChangeLog, repo *repository.DataChangeLogRepository) error {
	return repo.BatchCreateDataChangeLogs(context.Background(), logs)
}

// handleError 处理写入错误
func (m *AsyncLogManager) handleError(logType string, logs interface{}, err error) {
	m.logger.Error("failed to batch write logs",
		zap.String("log_type", logType),
		zap.Error(err),
	)

	if m.fallbackWriter != nil {
		if err := m.writeToFallback(logType, logs); err != nil {
			m.logger.Error("failed to write to fallback", zap.Error(err))
		}
	}
}

// writeToFallback 写入降级文件
func (m *AsyncLogManager) writeToFallback(logType string, logs interface{}) error {
	return nil
}

// initFallbackWriter 初始化降级写入器
func (m *AsyncLogManager) initFallbackWriter() error {
	m.cfg.FallbackPath = cmp.Or(m.cfg.FallbackPath, "./logs/fallback.log")

	dir := "./logs"
	if _, err := os.Stat(dir); os.IsNotExist(err) {
		os.MkdirAll(dir, 0755)
	}

	file, err := os.OpenFile(m.cfg.FallbackPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
	if err != nil {
		return err
	}
	m.fallbackWriter = file
	return nil
}

// RecordOperation 记录操作日志（非阻塞）
func (m *AsyncLogManager) RecordOperation(log *model.OperationLog) {
	select {
	case m.operationChan <- log:
	default:
		m.logger.Warn("operation log channel full, dropping log",
			zap.String("user_id", log.UserID),
			zap.String("operation", log.Operation),
		)
	}
}

// RecordLogin 记录登录日志（非阻塞）
func (m *AsyncLogManager) RecordLogin(log *model.LoginLog) {
	select {
	case m.loginChan <- log:
	default:
		m.logger.Warn("login log channel full, dropping log",
			zap.String("user_id", log.UserID),
			zap.String("event", log.Event),
		)
	}
}

// RecordDataChange 记录数据变更日志（非阻塞）
func (m *AsyncLogManager) RecordDataChange(log *model.DataChangeLog) {
	select {
	case m.dataChangeChan <- log:
	default:
		m.logger.Warn("data change log channel full, dropping log",
			zap.String("user_id", log.UserID),
			zap.String("change_type", log.ChangeType),
		)
	}
}

// Stop 停止异步日志管理器
func (m *AsyncLogManager) Stop() {
	m.cancel()

	done := make(chan struct{})
	go func() {
		m.wg.Wait()
		close(done)
	}()

	select {
	case <-done:
		m.logger.Info("async log manager stopped gracefully")
	case <-time.After(10 * time.Second):
		m.logger.Warn("async log manager stop timeout, force closing")
	}

	if m.fallbackWriter != nil {
		m.fallbackWriter.Close()
	}
}