spark-mcp/internal/audit/repo.go

package audit

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"spark-mcp-go/internal/storage"
)

// Action enumerates admin write operations that are persisted to the audit log.
type Action string

const (
	ActionClusterCreate Action = "cluster.create"
	ActionClusterUpdate Action = "cluster.update"
	ActionClusterDelete Action = "cluster.delete"
	ActionUploadCreate  Action = "upload.create"
	ActionUploadDelete  Action = "upload.delete"
)

// Entry is one admin write operation recorded for accountability.
//
// Actor is stored as "admin:<token_name>" (the first 8 characters of the
// admin token when multiple tokens are in use).
// Details is a JSON-encoded string produced by MarshalDetails; callers that
// do not need structured details can leave it empty.
type Entry struct {
	ID        int64     `json:"id"`
	Timestamp time.Time `json:"timestamp"`
	Actor     string    `json:"actor"`
	Action    Action    `json:"action"`
	ClusterID string    `json:"cluster_id,omitempty"`
	Details   string    `json:"details,omitempty"`
}

// Repo writes and reads audit_log rows.
type Repo struct {
	db *storage.DB
}

// NewRepo returns a repository bound to the supplied DB handle.
func NewRepo(db *storage.DB) *Repo {
	return &Repo{db: db}
}

// Insert persists a single audit entry. A zero Timestamp is replaced with
// the current time before storage.
func (r *Repo) Insert(ctx context.Context, e *Entry) error {
	if e.Timestamp.IsZero() {
		e.Timestamp = time.Now()
	}
	var details any
	if e.Details != "" {
		details = e.Details
	}
	_, err := r.db.SQLDB().ExecContext(ctx, `
		INSERT INTO audit_log (timestamp, actor, action, cluster_id, details)
		VALUES (?, ?, ?, ?, ?)`,
		e.Timestamp.UnixNano(), e.Actor, string(e.Action), e.ClusterID, details,
	)
	if err != nil {
		return fmt.Errorf("audit: insert: %w", err)
	}
	return nil
}

// List returns the most recent audit entries ordered by timestamp descending.
// A limit of zero or less falls back to 100.
func (r *Repo) List(ctx context.Context, limit int) ([]*Entry, error) {
	if limit <= 0 {
		limit = 100
	}
	rows, err := r.db.SQLDB().QueryContext(ctx, `
		SELECT id, timestamp, actor, action, cluster_id, details
		FROM audit_log ORDER BY timestamp DESC LIMIT ?`, limit)
	if err != nil {
		return nil, fmt.Errorf("audit: list: %w", err)
	}
	defer rows.Close()

	var out []*Entry
	for rows.Next() {
		var e Entry
		var ts int64
		var clusterID, details sql.NullString
		var action string
		if err := rows.Scan(&e.ID, &ts, &e.Actor, &action, &clusterID, &details); err != nil {
			return nil, fmt.Errorf("audit: scan: %w", err)
		}
		e.Timestamp = time.Unix(0, ts)
		e.Action = Action(action)
		if clusterID.Valid {
			e.ClusterID = clusterID.String
		}
		if details.Valid {
			e.Details = details.String
		}
		out = append(out, &e)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("audit: rows: %w", err)
	}
	if out == nil {
		// Force empty array (not null) in JSON.
		out = []*Entry{}
	}
	return out, nil
}

// MarshalDetails encodes any value as a JSON string for Entry.Details.
func MarshalDetails(v any) (string, error) {
	if v == nil {
		return "", nil
	}
	b, err := json.Marshal(v)
	if err != nil {
		return "", err
	}
	return string(b), nil
}

// ErrNotFound is unused in this package (single-entry lookups are not
// supported), but keeps the errors import available for future use.
var _ = errors.New