Record and expose per-resource history

internal/app/app.go

@@ -23,9 +23,25 @@ import (
 const shutdownTimeout = 10 * time.Second
 
 type App struct {
-	cfg config.Config
-	log *slog.Logger
-	set *settings.Service
+	cfg    config.Config
+	log    *slog.Logger
+	set    *settings.Service
+	events reconcile.EventSink
+}
+
+// eventSink adapts the store to reconcile.EventSink, recording resource history.
+type eventSink struct {
+	store *store.Store
+	log   *slog.Logger
+}
+
+func (s *eventSink) Record(ctx context.Context, e reconcile.Event) {
+	err := s.store.AppendEvent(ctx, store.ResourceEvent{
+		Kind: string(e.Kind), Name: e.Name, Type: e.Type, Reason: e.Reason, Commit: e.Commit,
+	})
+	if err != nil {
+		s.log.Error("record event", "err", err)
+	}
 }
 
 func New(cfg config.Config, log *slog.Logger) *App {
@@ -99,6 +115,7 @@ func (a *App) Run(ctx context.Context, addr string) error {
 		return fmt.Errorf("load encryption key: %w", err)
 	}
 	a.set = settings.New(st, key)
+	a.events = &eventSink{store: st, log: a.log}
 
 	authSvc := auth.New(st, a.log)
 	if err := authSvc.Init(ctx); err != nil {
@@ -112,6 +129,7 @@ func (a *App) Run(ctx context.Context, addr string) error {
 		Status:       statusStore,
 		Settings:     a.set,
 		Deleter:      a,
+		Events:       st,
 		CookieSecure: a.cfg.Server.CookieSecure,
 	}, a.log)
 
@@ -180,8 +198,10 @@ func (a *App) reconcileLoop(ctx context.Context, statusStore *status.Store) erro
 		if complete {
 			if dispatcher == nil {
 				dispatcher = reconcile.NewDispatcher(cfg.Reconcile.Concurrency, a.log)
+				dispatcher.SetEventSink(a.events)
 			}
 			engine := a.newEngine(cfg, statusStore)
+			engine.SetEventSink(a.events)
 			if err := engine.Scan(ctx, dispatcher); err != nil && !errors.Is(err, context.Canceled) {
 				a.log.Error("reconcile failed", "err", err)
 			}

internal/reconcile/dispatcher.go

@@ -14,6 +14,7 @@ import (
 // dropped and picked up on a later scan.
 type Dispatcher struct {
 	log      *slog.Logger
+	events   EventSink
 	sem      chan struct{}
 	mu       sync.Mutex
 	inflight map[actionKey]struct{}
@@ -37,6 +38,10 @@ func NewDispatcher(limit int, log *slog.Logger) *Dispatcher {
 	}
 }
 
+// SetEventSink attaches a sink so applied/failed actions are recorded in the
+// resource history. Optional; nil means no history is recorded.
+func (d *Dispatcher) SetEventSink(s EventSink) { d.events = s }
+
 // Submit starts action a in the background unless it is already running or the
 // pool is full. It never blocks; a dropped action is retried on the next scan.
 func (d *Dispatcher) Submit(ctx context.Context, a Action) {
@@ -66,12 +71,21 @@ func (d *Dispatcher) Submit(ctx context.Context, a Action) {
 		d.log.Info("applying", "action", a.String(), "reason", a.Reason)
 		if err := a.Apply(ctx); err != nil {
 			d.log.Error("apply failed", "action", a.String(), "err", err)
+			d.record(Event{Kind: a.Kind, Name: a.Name, Type: EventFailed, Reason: err.Error(), Commit: a.Commit})
 			return
 		}
 		d.log.Info("applied", "action", a.String())
+		d.record(Event{Kind: a.Kind, Name: a.Name, Type: EventApplied, Reason: string(a.Type), Commit: a.Commit})
 	})
 }
 
+// record forwards an event to the sink when one is attached.
+func (d *Dispatcher) record(ev Event) {
+	if d.events != nil {
+		d.events.Record(context.Background(), ev)
+	}
+}
+
 // Wait blocks until every in-flight action has finished.
 func (d *Dispatcher) Wait() {
 	d.wg.Wait()

internal/reconcile/engine.go

@@ -44,6 +44,7 @@ type Engine struct {
 	opts        Options
 	log         *slog.Logger
 	status      *status.Store
+	events      EventSink
 }
 
 // NewEngine wires an Engine from its collaborators. status may be nil (e.g. for
@@ -52,6 +53,10 @@ func NewEngine(source Source, reconcilers []Reconciler, opts Options, log *slog.
 	return &Engine{source: source, reconcilers: reconcilers, opts: opts, log: log, status: status}
 }
 
+// SetEventSink attaches a sink for resource history events. Optional; nil means
+// no history is recorded.
+func (e *Engine) SetEventSink(s EventSink) { e.events = s }
+
 // computePlan loads the desired state and aggregates every reconciler's plan.
 func (e *Engine) computePlan(ctx context.Context) ([]manifest.Resource, Plan, error) {
 	desired, err := e.source.Desired(ctx)
@@ -90,6 +95,7 @@ func (e *Engine) Scan(ctx context.Context, d *Dispatcher) error {
 	if plan.Empty() || !e.opts.AutoSync {
 		return nil
 	}
+	commit := e.commit()
 	for _, a := range plan.Actions {
 		switch {
 		case a.Informational:
@@ -99,6 +105,7 @@ func (e *Engine) Scan(ctx context.Context, d *Dispatcher) error {
 		case e.opts.DryRun:
 			e.log.Info("dry-run", "action", a.String(), "reason", a.Reason)
 		default:
+			a.Commit = commit
 			d.Submit(ctx, a)
 		}
 	}
@@ -179,13 +186,45 @@ func (e *Engine) recordStatus(desired []manifest.Resource, plan Plan) {
 		resources = append(resources, res)
 	}
 
+	commit := e.commit()
 	e.status.Set(status.Snapshot{
 		UpdatedAt:  time.Now(),
 		InSync:     plan.Empty(),
 		Configured: true,
-		Commit:     e.commit(),
+		Commit:     commit,
 		Resources:  resources,
 	})
+
+	// History: record state transitions against the previous snapshot. A brand
+	// new resource is only logged when it starts OutOfSync, to avoid a burst of
+	// "synced" events for everything already in sync on boot.
+	if e.events == nil {
+		return
+	}
+	current := make(map[key]bool, len(resources))
+	for _, r := range resources {
+		k := key{manifest.Kind(r.Kind), r.Name}
+		current[k] = true
+		prev, existed := prevState[k]
+		switch {
+		case existed && prev != r.State && r.State == status.StateSynced:
+			e.record(Event{Kind: k.kind, Name: k.name, Type: EventSynced, Commit: commit})
+		case prev != r.State && r.State == status.StateOutOfSync:
+			e.record(Event{Kind: k.kind, Name: k.name, Type: EventDrifted, Reason: r.Reason, Commit: commit})
+		}
+	}
+	for k, prev := range prevState {
+		if _, ok := current[k]; !ok && prev != "" {
+			e.record(Event{Kind: k.kind, Name: k.name, Type: EventRemoved, Commit: commit})
+		}
+	}
+}
+
+// record forwards an event to the sink when one is attached.
+func (e *Engine) record(ev Event) {
+	if e.events != nil {
+		e.events.Record(context.Background(), ev)
+	}
 }
 
 // committer is implemented by sources that track a Git commit (the Git source);

internal/reconcile/events.go

@@ -0,0 +1,31 @@
+package reconcile
+
+import (
+	"context"
+
+	"github.com/prop4n/proxmops/internal/manifest"
+)
+
+// Event types recorded in a resource's history.
+const (
+	EventDrifted = "drifted" // went OutOfSync
+	EventSynced  = "synced"  // returned to Synced
+	EventRemoved = "removed" // disappeared from the desired/observed set
+	EventApplied = "applied" // an action ran successfully
+	EventFailed  = "failed"  // an action failed
+)
+
+// Event is one entry in a resource's history.
+type Event struct {
+	Kind   manifest.Kind
+	Name   string
+	Type   string
+	Reason string
+	Commit string
+}
+
+// EventSink records resource events. Implementations must be safe for
+// concurrent use, since the dispatcher records from background goroutines.
+type EventSink interface {
+	Record(ctx context.Context, e Event)
+}

internal/reconcile/events_test.go

@@ -0,0 +1,88 @@
+package reconcile
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"github.com/prop4n/proxmops/internal/manifest"
+	"github.com/prop4n/proxmops/internal/status"
+)
+
+// recordingSink collects events for assertions, safe for concurrent use.
+type recordingSink struct {
+	mu     sync.Mutex
+	events []Event
+}
+
+func (s *recordingSink) Record(_ context.Context, e Event) {
+	s.mu.Lock()
+	s.events = append(s.events, e)
+	s.mu.Unlock()
+}
+
+func (s *recordingSink) types() []string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := make([]string, len(s.events))
+	for i, e := range s.events {
+		out[i] = e.Type
+	}
+	return out
+}
+
+func TestEngineRecordsDriftThenSynced(t *testing.T) {
+	st := status.NewStore()
+	rec := &staticReconciler{}
+	sink := &recordingSink{}
+	src := staticSource{res: []manifest.Resource{vmResource("web-01")}}
+	eng := NewEngine(src, []Reconciler{rec}, Options{AutoSync: false}, testLogger(), st)
+	eng.SetEventSink(sink)
+	d := NewDispatcher(1, testLogger())
+
+	// Pass 1: a create is planned -> the resource is OutOfSync -> drifted event.
+	rec.plan = Plan{Actions: []Action{{Type: ActionCreate, Kind: manifest.KindVirtualMachine, Name: "web-01"}}}
+	if err := eng.Scan(context.Background(), d); err != nil {
+		t.Fatal(err)
+	}
+	// Pass 2: plan empty -> resource returns to Synced -> synced event.
+	rec.plan = Plan{}
+	if err := eng.Scan(context.Background(), d); err != nil {
+		t.Fatal(err)
+	}
+
+	got := sink.types()
+	if len(got) != 2 || got[0] != EventDrifted || got[1] != EventSynced {
+		t.Fatalf("events = %v, want [drifted synced]", got)
+	}
+}
+
+func TestDispatcherRecordsAppliedAndFailed(t *testing.T) {
+	sink := &recordingSink{}
+	d := NewDispatcher(2, testLogger())
+	d.SetEventSink(sink)
+
+	ok := Action{Type: ActionCreate, Kind: manifest.KindIso, Name: "ok", Commit: "c1",
+		Apply: func(context.Context) error { return nil }}
+	bad := Action{Type: ActionCreate, Kind: manifest.KindIso, Name: "bad", Commit: "c1",
+		Apply: func(context.Context) error { return context.DeadlineExceeded }}
+	d.Submit(context.Background(), ok)
+	d.Submit(context.Background(), bad)
+	d.Wait()
+
+	var applied, failed int
+	for _, e := range sink.events {
+		switch e.Type {
+		case EventApplied:
+			applied++
+			if e.Commit != "c1" || e.Reason != string(ActionCreate) {
+				t.Errorf("applied event fields: %+v", e)
+			}
+		case EventFailed:
+			failed++
+		}
+	}
+	if applied != 1 || failed != 1 {
+		t.Fatalf("applied=%d failed=%d, want 1/1", applied, failed)
+	}
+}

internal/reconcile/plan.go

@@ -37,6 +37,9 @@ type Action struct {
 	// Informational actions report a condition (e.g. "reboot required") in the
 	// status without being applied. The engine never dispatches them.
 	Informational bool
+	// Commit is the Git commit this action was planned from, set by the engine
+	// before dispatch so applied/failed events are attributed to it.
+	Commit string
 }
 
 // String renders an action for display, e.g. "create VirtualMachine/web-01".

internal/server/resources.go

@@ -6,8 +6,15 @@ import (
 	"net/http"
 
 	"github.com/go-chi/chi/v5"
+
+	"github.com/prop4n/proxmops/internal/store"
 )
 
+// EventsReader reads a resource's recorded history.
+type EventsReader interface {
+	EventsFor(ctx context.Context, kind, name string, limit int) ([]store.ResourceEvent, error)
+}
+
 // ResourceDeleter removes a managed resource from the cluster. It is implemented
 // outside the transport layer and injected through Options.
 type ResourceDeleter interface {
@@ -43,3 +50,23 @@ func (s *Server) handleDeleteResource(w http.ResponseWriter, r *http.Request) {
 		writeError(w, http.StatusInternalServerError, "internal error")
 	}
 }
+
+// handleResourceEvents returns a resource's recorded history, newest first.
+func (s *Server) handleResourceEvents(w http.ResponseWriter, r *http.Request) {
+	if s.events == nil {
+		writeJSON(w, http.StatusOK, []store.ResourceEvent{})
+		return
+	}
+	kind := chi.URLParam(r, "kind")
+	name := chi.URLParam(r, "name")
+	events, err := s.events.EventsFor(r.Context(), kind, name, 200)
+	if err != nil {
+		s.log.Error("read events", "kind", kind, "name", name, "err", err)
+		writeError(w, http.StatusInternalServerError, "internal error")
+		return
+	}
+	if events == nil {
+		events = []store.ResourceEvent{}
+	}
+	writeJSON(w, http.StatusOK, events)
+}

internal/server/routes.go

@@ -31,6 +31,7 @@ func (s *Server) routes() http.Handler {
 			r.Get("/me", s.handleMe)
 			r.Get("/resources", s.handleResources)
 			r.Delete("/resources/{kind}/{name}", s.handleDeleteResource)
+			r.Get("/resources/{kind}/{name}/events", s.handleResourceEvents)
 			r.Get("/events", s.handleEvents)
 			r.Get("/settings", s.handleGetSettings)
 			r.Put("/settings", s.handlePutSettings)

internal/server/server.go

@@ -22,6 +22,7 @@ type Server struct {
 	status       *status.Store
 	settings     *settings.Service
 	deleter      ResourceDeleter
+	events       EventsReader
 	cookieSecure bool
 }
 
@@ -32,6 +33,7 @@ type Options struct {
 	Status       *status.Store
 	Settings     *settings.Service
 	Deleter      ResourceDeleter
+	Events       EventsReader
 	CookieSecure bool
 }
 
@@ -43,6 +45,7 @@ func New(opts Options, log *slog.Logger) *Server {
 		status:       opts.Status,
 		settings:     opts.Settings,
 		deleter:      opts.Deleter,
+		events:       opts.Events,
 		cookieSecure: opts.CookieSecure,
 	}
 	s.http = &http.Server{