goclaw/internal/workstation/activity_sink.go

// Package workstation contains the activity sink that subscribes to domain events
// and persists exec audit rows to WorkstationActivityStore.
package workstation

import (
	"context"
	"crypto/sha256"
	"fmt"
	"log/slog"
	"regexp"
	"strings"
	"time"

	"github.com/google/uuid"

	"github.com/nextlevelbuilder/goclaw/internal/eventbus"
	"github.com/nextlevelbuilder/goclaw/internal/store"
	"github.com/nextlevelbuilder/goclaw/pkg/protocol"
)

// sensitivePatterns is a list of compiled regexes that redact secret-bearing fragments.
// Applied to cmd_preview before storage; raw command is never persisted.
var sensitivePatterns = []*regexp.Regexp{
	regexp.MustCompile(`(?i)(api[_-]?key|password|secret|token|auth)[=:]\S+`),
	regexp.MustCompile(`-H\s+"Authorization:[^"]*"`),
	regexp.MustCompile(`Bearer\s+[A-Za-z0-9\-_\.]+`),
	regexp.MustCompile(`eyJ[A-Za-z0-9\-_]+\.[A-Za-z0-9\-_]+\.[A-Za-z0-9\-_]+`), // JWT
}

// WireActivitySink subscribes to EventWorkstationExecDone on domainBus and writes
// audit rows to activityStore. The subscription is fire-and-forget (Insert is buffered).
// Also starts a nightly retention goroutine that prunes rows older than 30 days.
// Returns a cleanup function that stops the retention goroutine.
func WireActivitySink(bus eventbus.DomainEventBus, activityStore store.WorkstationActivityStore) func() {
	if bus == nil || activityStore == nil {
		return func() {}
	}

	// Subscribe to exec done events (emitted by WorkstationExecTool.streamAndCollect).
	// The payload is map[string]any (see internal/tools/workstation_exec.go).
	bus.Subscribe(eventbus.EventType(protocol.EventWorkstationExecDone), func(ctx context.Context, ev eventbus.DomainEvent) error {
		payload, ok := ev.Payload.(map[string]any)
		if !ok {
			return nil
		}

		wsIDStr, _ := payload["workstation_id"].(string)
		wsID, err := uuid.Parse(wsIDStr)
		if err != nil {
			return nil
		}
		tenantID, _ := uuid.Parse(ev.TenantID)
		agentID := ev.AgentID
		sessionKey, _ := payload["session_key"].(string)

		// I3 fix: use the "command" field from the done event payload for meaningful
		// cmd_hash and cmd_preview. Falls back to sessionKey if command is absent
		// (e.g. events from older tool versions).
		cmdRaw, _ := payload["command"].(string)
		if cmdRaw == "" {
			// Fallback for events without the command field.
			cmdRaw = "session:" + sessionKey
		}
		cmdPreview := redactSensitive(cmdRaw)

		exitCodeF, _ := payload["exit_code"].(int)
		durationF, _ := payload["duration_ms"].(int64)
		// JSON numbers decode as float64 from map[string]any.
		if ef, ok := payload["exit_code"].(float64); ok {
			exitCodeF = int(ef)
		}
		if df, ok := payload["duration_ms"].(float64); ok {
			durationF = int64(df)
		}

		cmdHash := fmt.Sprintf("%x", sha256.Sum256([]byte(cmdRaw)))[:16]

		exitCodeVal := exitCodeF
		durationVal := durationF

		row := &store.WorkstationActivity{
			ID:            uuid.New(),
			TenantID:      tenantID,
			WorkstationID: wsID,
			AgentID:       agentID,
			Action:        "exec",
			CmdHash:       cmdHash,
			CmdPreview:    cmdPreview,
			ExitCode:      &exitCodeVal,
			DurationMS:    &durationVal,
			CreatedAt:     time.Now().UTC(),
		}

		if err := activityStore.Insert(ctx, row); err != nil {
			slog.Warn("workstation.activity.insert_error", "error", err)
		}

		slog.Info("workstation.exec.completed",
			"workstation_id", wsIDStr,
			"tenant_id", ev.TenantID,
			"agent_id", agentID,
			"cmd_hash", cmdHash,
			"exit_code", exitCodeVal,
			"duration_ms", durationVal,
		)
		return nil
	})

	// Start nightly retention goroutine.
	stopCh := make(chan struct{})
	go func() {
		ticker := time.NewTicker(24 * time.Hour)
		defer ticker.Stop()
		for {
			select {
			case <-ticker.C:
				before := time.Now().Add(-30 * 24 * time.Hour)
				n, err := activityStore.Prune(context.Background(), before)
				if err != nil {
					slog.Warn("workstation.activity.prune_error", "error", err)
				} else if n > 0 {
					slog.Info("workstation.activity.pruned", "rows", n, "before", before.Format(time.RFC3339))
				}
			case <-stopCh:
				return
			}
		}
	}()

	return func() { close(stopCh) }
}

// redactSensitive strips lines or fragments matching known secret patterns from cmd.
// Returns a truncated, redacted string safe for tenant-admin display.
func redactSensitive(cmd string) string {
	result := cmd
	for _, re := range sensitivePatterns {
		result = re.ReplaceAllString(result, "[REDACTED]")
	}
	// Truncate to 200 chars.
	if len(result) > 200 {
		result = result[:200]
	}
	return strings.TrimSpace(result)
}