// Package audit 提供上传/下载审计日志服务(需求 ③): // 记录操作时间/IP/UA/设备解析/动作/结果/字节数/耗时,落库 Postgres。 package audit import ( "context" "errors" "log" "strings" "sync" "time" "gorm.io/gorm" "filecodebox/internal/model" ) // Service 审计日志服务。 type Service struct { sink Sink } // Sink 审计落库抽象(生产为 Postgres,测试为内存实现)。 type Sink interface { // Save 批量落库。 Save(ctx context.Context, logs []model.AuditLog) error } // DBSink 基于 GORM 的落库实现。 type DBSink struct{ db *gorm.DB } // NewDBSink 构造数据库落库实现。 func NewDBSink(db *gorm.DB) *DBSink { return &DBSink{db: db} } // Save 批量插入审计记录。 func (s *DBSink) Save(ctx context.Context, logs []model.AuditLog) error { if len(logs) == 0 { return nil } return s.db.WithContext(ctx).CreateInBatches(&logs, 200).Error } // NewService 构造审计服务。 func NewService(sink Sink) *Service { return &Service{sink: sink} } // Entry 一次待落库的审计事件。 type Entry struct { Action string // upload | download FileCode string // 取件码 FileName string // 原始文件名 SizeBytes int64 // 文件总字节数 TransferredBytes int64 // 实际传输字节数 IP string // 客户端 IP UserAgent string // User-Agent DeviceOS string // 操作系统 DeviceBrowser string // 浏览器 DeviceType string // desktop/mobile/tablet/bot/other Actor string // admin | guest Result string // success | denied | failed ErrorMsg string // 失败原因 Duration time.Duration // 耗时 } // Record 异步写入一条审计日志:先尝试同步落库,失败时进入内存缓冲等待重试, // 避免审计失败影响主请求,也避免高峰期阻塞。 func (s *Service) Record(entry Entry) { record := model.AuditLog{ Action: entry.Action, FileCode: truncate(entry.FileCode, 64), FileName: truncate(entry.FileName, 255), SizeBytes: entry.SizeBytes, TransferredBytes: entry.TransferredBytes, IP: truncate(entry.IP, 64), UserAgent: truncate(entry.UserAgent, 512), DeviceOS: truncate(entry.DeviceOS, 64), DeviceBrowser: truncate(entry.DeviceBrowser, 64), DeviceType: truncate(entry.DeviceType, 32), Actor: truncate(entry.Actor, 64), Result: normalizeResult(entry.Result), ErrorMsg: truncate(entry.ErrorMsg, 512), DurationMs: entry.Duration.Milliseconds(), } go func() { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := s.sink.Save(ctx, []model.AuditLog{record}); err != nil { log.Printf("[audit] 审计日志落库失败,进入重试队列: %v", err) s.enqueue(record) } }() } // retryBuf 落库失败时的内存重试缓冲。 var retryBuf struct { sync.Mutex items []model.AuditLog } const maxRetryBuffer = 10000 // enqueue 入队;超出上限时丢弃最旧的,防止内存无限增长。 func (s *Service) enqueue(record model.AuditLog) { retryBuf.Lock() if len(retryBuf.items) >= maxRetryBuffer { retryBuf.items = retryBuf.items[1:] } retryBuf.items = append(retryBuf.items, record) retryBuf.Unlock() } // FlushRetry 将缓冲中的审计日志重新落库;由后台定时任务调用。 func (s *Service) FlushRetry() { retryBuf.Lock() if len(retryBuf.items) == 0 { retryBuf.Unlock() return } items := retryBuf.items retryBuf.items = nil retryBuf.Unlock() ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() if err := s.sink.Save(ctx, items); err != nil { log.Printf("[audit] 重试队列落库失败: %v", err) // 失败则放回队首 retryBuf.Lock() retryBuf.items = append(items, retryBuf.items...) if len(retryBuf.items) > maxRetryBuffer { retryBuf.items = retryBuf.items[:maxRetryBuffer] } retryBuf.Unlock() } } // StartRetryLoop 启动后台重试循环。 func (s *Service) StartRetryLoop(stop <-chan struct{}) { go func() { ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() for { select { case <-stop: s.FlushRetry() return case <-ticker.C: s.FlushRetry() } } }() } // Query 按条件分页查询审计日志(管理端使用)。 // action/ip/result 为可选过滤;begin/end 为创建时间范围(可选)。 func (s *Service) Query(page, pageSize int, action, ip, result string, begin, end *time.Time) ([]model.AuditLog, int64, error) { dbSink, ok := s.sink.(*DBSink) if !ok { return nil, 0, errors.New("audit: 当前 sink 不支持查询") } if page < 1 { page = 1 } if pageSize < 1 || pageSize > 200 { pageSize = 20 } q := dbSink.db.Model(&model.AuditLog{}) if action != "" { q = q.Where("action = ?", action) } if ip != "" { q = q.Where("ip = ?", ip) } if result != "" { q = q.Where("result = ?", result) } if begin != nil { q = q.Where("created_at >= ?", *begin) } if end != nil { q = q.Where("created_at <= ?", *end) } var total int64 if err := q.Count(&total).Error; err != nil { return nil, 0, err } var logs []model.AuditLog err := q.Order("id DESC"). Offset((page - 1) * pageSize). Limit(pageSize). Find(&logs).Error return logs, total, err } func truncate(s string, n int) string { if len(s) <= n { return s } return s[:n] } func normalizeResult(r string) string { switch strings.TrimSpace(r) { case model.AuditResultSuccess, model.AuditResultDenied, model.AuditResultFailed: return strings.TrimSpace(r) case "": return model.AuditResultFailed default: return model.AuditResultFailed } }