implemented NATS event bus, websocket hub upgrade, and audit log

internal/api/handlers/audit.go

@@ -0,0 +1,64 @@
+package handlers
+
+import (
+	"net/http"
+	"strconv"
+	"time"
+
+	"xorm.io/xorm"
+
+	"github.com/forgeo/forgebucket/internal/api/middleware"
+	"github.com/forgeo/forgebucket/internal/models"
+)
+
+type AuditHandler struct{ db *xorm.Engine }
+
+func NewAuditHandler(db *xorm.Engine) *AuditHandler { return &AuditHandler{db: db} }
+
+// List returns recent audit log entries. Admin-only.
+// Query params: limit (default 50, max 200), before (ID cursor for pagination),
+// actor_id, method, since (RFC3339).
+func (h *AuditHandler) List(w http.ResponseWriter, r *http.Request) {
+	isAdmin, _ := r.Context().Value(middleware.ContextKeyIsAdmin).(bool)
+	if !isAdmin {
+		jsonError(w, "admin access required", http.StatusForbidden)
+		return
+	}
+
+	q := r.URL.Query()
+
+	limit := 50
+	if l, err := strconv.Atoi(q.Get("limit")); err == nil && l > 0 {
+		if l > 200 {
+			l = 200
+		}
+		limit = l
+	}
+
+	sess := h.db.Desc("id").Limit(limit)
+
+	if before, err := strconv.ParseInt(q.Get("before"), 10, 64); err == nil && before > 0 {
+		sess = sess.And("id < ?", before)
+	}
+	if actorID, err := strconv.ParseInt(q.Get("actor_id"), 10, 64); err == nil && actorID > 0 {
+		sess = sess.And("actor_id = ?", actorID)
+	}
+	if method := q.Get("method"); method != "" {
+		sess = sess.And("method = ?", method)
+	}
+	if since := q.Get("since"); since != "" {
+		if t, err := time.Parse(time.RFC3339, since); err == nil {
+			sess = sess.And("occurred_at >= ?", t.UTC())
+		}
+	}
+
+	var entries []models.AuditLog
+	if err := sess.Find(&entries); err != nil {
+		jsonError(w, "could not fetch audit log", http.StatusInternalServerError)
+		return
+	}
+	if entries == nil {
+		entries = []models.AuditLog{}
+	}
+	jsonOK(w, entries)
+}

internal/api/handlers/ws.go

@@ -1,18 +1,74 @@
 package handlers
 
 import (
+	"encoding/json"
+	"log"
 	"net/http"
+	"sync"
 
 	"nhooyr.io/websocket"
 	"nhooyr.io/websocket/wsjson"
+
+	"github.com/forgeo/forgebucket/internal/events"
 )
 
-type WSHandler struct{}
-
-func NewWSHandler() *WSHandler {
-	return &WSHandler{}
+// WSHandler manages WebSocket connections and bridges NATS events to clients.
+// All connected clients receive every platform event (subject + payload).
+// Per-user filtering is added in Phase 2B when CI events carry access control.
+type WSHandler struct {
+	bus     events.EventBus
+	mu      sync.RWMutex
+	clients map[string]chan events.WSEnvelope // connID → send channel
 }
 
+func NewWSHandler(bus events.EventBus) *WSHandler {
+	h := &WSHandler{
+		bus:     bus,
+		clients: make(map[string]chan events.WSEnvelope),
+	}
+	h.subscribeAll()
+	return h
+}
+
+// subscribeAll wires the NATS wildcard subscription that fans events to all clients.
+func (h *WSHandler) subscribeAll() {
+	unsub, err := h.bus.Subscribe(">", func(subject string, data []byte) {
+		env := events.WSEnvelope{Subject: subject, Payload: data}
+		h.mu.RLock()
+		defer h.mu.RUnlock()
+		for id, ch := range h.clients {
+			select {
+			case ch <- env:
+			default:
+				// Client send buffer full — drop event rather than block.
+				log.Printf("ws: dropped event %s for slow client %s", subject, id)
+			}
+		}
+	})
+	if err != nil {
+		log.Printf("ws: nats subscribe failed: %v", err)
+		return
+	}
+	// The subscription lives for the process lifetime; no cleanup needed.
+	_ = unsub
+}
+
+func (h *WSHandler) register(id string) chan events.WSEnvelope {
+	ch := make(chan events.WSEnvelope, 64)
+	h.mu.Lock()
+	h.clients[id] = ch
+	h.mu.Unlock()
+	return ch
+}
+
+func (h *WSHandler) unregister(id string) {
+	h.mu.Lock()
+	delete(h.clients, id)
+	h.mu.Unlock()
+}
+
+// Hub upgrades the HTTP connection to WebSocket, registers the client, and
+// pumps NATS-sourced events to it until the connection closes.
 func (h *WSHandler) Hub(w http.ResponseWriter, r *http.Request) {
 	conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
 		OriginPatterns: []string{"localhost:*"},
@@ -22,13 +78,38 @@ func (h *WSHandler) Hub(w http.ResponseWriter, r *http.Request) {
 	}
 	defer conn.CloseNow()
 
+	// Use remote address + a monotonic component as a simple unique ID.
+	connID := r.RemoteAddr + "|" + r.Header.Get("Sec-Websocket-Key")
+	send := h.register(connID)
+	defer h.unregister(connID)
+
 	ctx := r.Context()
-	for {
-		var msg map[string]any
-		if err := wsjson.Read(ctx, conn, &msg); err != nil {
-			break
+
+	// Write pump: drain the send channel and write to the WebSocket.
+	go func() {
+		for {
+			select {
+			case env, ok := <-send:
+				if !ok {
+					return
+				}
+				msg := map[string]any{
+					"subject": env.Subject,
+					"payload": json.RawMessage(env.Payload),
+				}
+				if err := wsjson.Write(ctx, conn, msg); err != nil {
+					return
+				}
+			case <-ctx.Done():
+				return
+			}
 		}
-		if err := wsjson.Write(ctx, conn, msg); err != nil {
+	}()
+
+	// Read pump: discard incoming messages (clients send no data for now),
+	// keeping the connection alive and detecting closes.
+	for {
+		if _, _, err := conn.Read(ctx); err != nil {
 			break
 		}
 	}

internal/api/middleware/audit.go

@@ -0,0 +1,67 @@
+package middleware
+
+import (
+	"net/http"
+	"time"
+
+	"xorm.io/xorm"
+
+	"github.com/forgeo/forgebucket/internal/events"
+	"github.com/forgeo/forgebucket/internal/models"
+)
+
+// statusRecorder wraps ResponseWriter to capture the written status code.
+type statusRecorder struct {
+	http.ResponseWriter
+	status int
+}
+
+func (r *statusRecorder) WriteHeader(code int) {
+	r.status = code
+	r.ResponseWriter.WriteHeader(code)
+}
+
+// AuditLog middleware records every state-mutating request (POST/PUT/PATCH/DELETE)
+// to the audit_log table and publishes an audit.event to the event bus.
+// It runs after auth middleware so the actor is available in context.
+func AuditLog(db *xorm.Engine, bus events.EventBus) func(http.Handler) http.Handler {
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			switch r.Method {
+			case http.MethodGet, http.MethodHead, http.MethodOptions, http.MethodTrace:
+				next.ServeHTTP(w, r)
+				return
+			}
+
+			rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK}
+			next.ServeHTTP(rec, r)
+
+			actorID, _ := UserIDFromContext(r.Context())
+			actorName, _ := r.Context().Value(ContextKeyUsername).(string)
+
+			entry := &models.AuditLog{
+				ActorID:    actorID,
+				ActorName:  actorName,
+				Method:     r.Method,
+				Path:       r.URL.Path,
+				StatusCode: rec.status,
+				IPAddress:  r.RemoteAddr,
+				UserAgent:  r.UserAgent(),
+				OccurredAt: time.Now().UTC(),
+			}
+
+			go func() {
+				db.Insert(entry) //nolint:errcheck
+				bus.Publish(events.SubjectAuditEvent, events.AuditEvent{ //nolint:errcheck
+					ActorID:      entry.ActorID,
+					ActorName:    entry.ActorName,
+					Action:       entry.Method + " " + entry.Path,
+					ResourcePath: entry.Path,
+					IPAddress:    entry.IPAddress,
+					StatusCode:   entry.StatusCode,
+					At:           entry.OccurredAt,
+				})
+			}()
+		})
+	}
+}

internal/api/router.go

@@ -17,9 +17,10 @@ import (
 	"github.com/forgeo/forgebucket/internal/api/handlers"
 	"github.com/forgeo/forgebucket/internal/api/middleware"
 	"github.com/forgeo/forgebucket/internal/config"
+	"github.com/forgeo/forgebucket/internal/events"
 )
 
-func New(cfg *config.Config, engine *xorm.Engine, store sessions.Store, staticFiles fs.FS) http.Handler {
+func New(cfg *config.Config, engine *xorm.Engine, store sessions.Store, bus events.EventBus, staticFiles fs.FS) http.Handler {
 	r := chi.NewRouter()
 
 	r.Use(chimiddleware.Logger)
@@ -33,14 +34,15 @@ func New(cfg *config.Config, engine *xorm.Engine, store sessions.Store, staticFi
 		MaxAge:           300,
 	}))
 
-	csrf := middleware.CSRF(!cfg.Debug)
-	auth      := middleware.NewAuth(store, engine, handlers.LookupAccessToken)
+	csrf  := middleware.CSRF(!cfg.Debug)
+	auth  := middleware.NewAuth(store, engine, handlers.LookupAccessToken)
+	audit := middleware.AuditLog(engine, bus)
 
 	repoH    := handlers.NewRepoHandler(engine, cfg)
 	userH    := handlers.NewUserHandler(engine, store)
 	prH      := handlers.NewPRHandler(engine)
 	pipeH    := handlers.NewPipelineHandler(engine)
-	wsH      := handlers.NewWSHandler()
+	wsH      := handlers.NewWSHandler(bus)
 	gitH     := handlers.NewGitHTTPHandler(engine, cfg)
 	issueH   := handlers.NewIssueHandler(engine)
 	sshKeyH  := handlers.NewSSHKeyHandler(engine)
@@ -53,6 +55,7 @@ func New(cfg *config.Config, engine *xorm.Engine, store sessions.Store, staticFi
 	lfsH        := handlers.NewLFSHandler(engine)
 	exploreH    := handlers.NewExploreHandler(engine)
 	dashH       := handlers.NewDashboardHandler(engine)
+	auditH      := handlers.NewAuditHandler(engine)
 
 	// ── Git smart-HTTP transport ───────────────────────────────────────────────
 	// Regex constraint ensures only *.git paths match, so asset/SPA URLs
@@ -94,9 +97,11 @@ func New(cfg *config.Config, engine *xorm.Engine, store sessions.Store, staticFi
 		// ── Protected (session + CSRF for mutations) ──────────────────────────
 		r.Group(func(r chi.Router) {
 			r.Use(auth.Require)
+			r.Use(audit)
 
 			r.Get("/me", userH.Me)
 			r.Get("/dashboard", dashH.Get)
+			r.Get("/audit", auditH.List)
 
 			// SSH key management
 			r.Get("/user/keys", sshKeyH.List)