package citadel import ( "context" "log/slog" "sync" "time" "github.com/orchard9/rdev/internal/domain" "github.com/orchard9/rdev/internal/port" ) // AuditShipper wraps an existing AuditLogger and also ships audit events to Citadel. // It buffers events and flushes them in batches for efficiency. type AuditShipper struct { inner port.AuditLogger client port.CitadelClient tenantID string // rdev-platform tenant ID logger *slog.Logger mu sync.Mutex buffer []map[string]any done chan struct{} } // NewAuditShipper wraps an existing AuditLogger with Citadel shipping. // tenantID is the Citadel tenant ID for the rdev-platform environment. func NewAuditShipper(inner port.AuditLogger, client port.CitadelClient, tenantID string, logger *slog.Logger) *AuditShipper { if logger == nil { logger = slog.Default() } s := &AuditShipper{ inner: inner, client: client, tenantID: tenantID, logger: logger.With("component", "audit_shipper"), buffer: make([]map[string]any, 0, 64), done: make(chan struct{}), } go s.flushLoop() return s } // LogCommandStart records the start of a command and ships to Citadel. func (s *AuditShipper) LogCommandStart(ctx context.Context, entry *domain.AuditLogEntry) error { // Always write to primary store first if err := s.inner.LogCommandStart(ctx, entry); err != nil { return err } // Buffer for Citadel (best-effort, don't block) s.enqueue(map[string]any{ "message": "audit: command started", "level": "info", "service": "rdev-platform", "event_type": "audit", "audit_action": "command_start", "command_id": entry.CommandID, "command_type": string(entry.CommandType), "project_id": entry.ProjectID, "api_key_id": entry.APIKeyID, "client_ip": entry.ClientIP, "timestamp": entry.StartedAt.Format(time.RFC3339Nano), }) return nil } // LogCommandEnd records the completion of a command and ships to Citadel. func (s *AuditShipper) LogCommandEnd(ctx context.Context, commandID string, result *domain.AuditResult) error { // Always write to primary store first if err := s.inner.LogCommandEnd(ctx, commandID, result); err != nil { return err } // Buffer for Citadel (best-effort) s.enqueue(map[string]any{ "message": "audit: command completed", "level": auditStatusToLevel(result.Status), "service": "rdev-platform", "event_type": "audit", "audit_action": "command_end", "command_id": commandID, "status": string(result.Status), "exit_code": result.ExitCode, "duration_ms": result.DurationMs, "error_message": result.ErrorMessage, "output_size_bytes": result.OutputSizeBytes, "timestamp": time.Now().Format(time.RFC3339Nano), }) return nil } // List delegates to the inner AuditLogger. func (s *AuditShipper) List(ctx context.Context, filters domain.AuditFilters) ([]domain.AuditLogEntry, error) { return s.inner.List(ctx, filters) } // Get delegates to the inner AuditLogger. func (s *AuditShipper) Get(ctx context.Context, commandID string) (*domain.AuditLogEntry, error) { return s.inner.Get(ctx, commandID) } // Close flushes remaining events and stops the background goroutine. func (s *AuditShipper) Close() { close(s.done) s.flush() } func (s *AuditShipper) enqueue(event map[string]any) { s.mu.Lock() defer s.mu.Unlock() s.buffer = append(s.buffer, event) } func (s *AuditShipper) flushLoop() { ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() for { select { case <-ticker.C: s.flush() case <-s.done: return } } } func (s *AuditShipper) flush() { s.mu.Lock() if len(s.buffer) == 0 { s.mu.Unlock() return } events := s.buffer s.buffer = make([]map[string]any, 0, 64) s.mu.Unlock() ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() if err := s.client.IngestBatch(ctx, s.tenantID, events); err != nil { s.logger.Warn("failed to ship audit events to citadel", "error", err, "event_count", len(events), ) } } func auditStatusToLevel(status domain.AuditStatus) string { switch status { case domain.AuditStatusError: return "error" case domain.AuditStatusCancelled: return "warn" default: return "info" } }