drjones/AetherForge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Close DV-01–04,07,09,13,14: unify operator-deck UX in server/web.

Browse Source 
Standardize neon cyan on #00e8f5, wire SacredPageHeader across Builds/Path Tracer/LOTL, dedupe Pages.css, align Path Tracer chrome, drop dead wealth-deck CSS, add prefers-color-scheme shell + HelpTip, sidebar version from package.json build inject, and Vitest CSS regression guards.
This commit is contained in:
AetherForge
2026-06-07 06:33:51 -07:00
parent 85d55df37c
commit 74c006a04b
35 changed files with 1144 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
server/internal/api/architecture_deferred_test.go Normal file
View File

@@ -0,0 +1,50 @@
package api
import (
"os"
"strings"
"testing"
)
func TestArchitectureDeferredHonestStubs(t *testing.T) {
t.Run("tunnel_stream WS stub", func(t *testing.T) {
src, err := os.ReadFile("websocket.go")
if err != nil {
t.Fatal(err)
}
body := string(src)
if !strings.Contains(body, `case "tunnel_stream":`) || !strings.Contains(body, `"implemented": false`) {
t.Fatal("expected dashboard tunnel_stream not-implemented stub")
}
})
t.Run("WS init pagination option", func(t *testing.T) {
src, err := os.ReadFile("dashboard_ws_init.go")
if err != nil {
t.Fatal(err)
}
if !strings.Contains(string(src), "init_limit") {
t.Fatal("expected init_limit query param parser")
}
})
t.Run("Path Tracer SQLite persistence", func(t *testing.T) {
src, err := os.ReadFile("pathtracer_handler.go")
if err != nil {
t.Fatal(err)
}
if !strings.Contains(string(src), "loadPersistedSessions") {
t.Fatal("expected Path Tracer startup restore from SQLite")
}
})
t.Run("SQLite single-writer ceiling documented", func(t *testing.T) {
src, err := os.ReadFile("../db/sqlite.go")
if err != nil {
t.Fatal(err)
}
if !strings.Contains(string(src), "SetMaxOpenConns(1)") {
t.Fatal("expected SQLite single-writer guard")
}
})
}

35
server/internal/api/dashboard_ws_init.go Normal file
View File

@@ -0,0 +1,35 @@
package api
import (
"net/http"
"strconv"
"crypto-miner-server/internal/db"
)
const (
dashboardInitDefaultLimit = 100
dashboardInitMaxLimit = 2000
)
// parseDashboardInitFilter reads optional init_limit / init_offset on /ws/dashboard.
func parseDashboardInitFilter(r *http.Request) (db.AgentListFilter, bool) {
limitStr := r.URL.Query().Get("init_limit")
if limitStr == "" {
return db.AgentListFilter{}, false
}
limit, err := strconv.Atoi(limitStr)
if err != nil || limit <= 0 {
limit = dashboardInitDefaultLimit
}
if limit > dashboardInitMaxLimit {
limit = dashboardInitMaxLimit
}
offset := 0
if offStr := r.URL.Query().Get("init_offset"); offStr != "" {
if o, err := strconv.Atoi(offStr); err == nil && o >= 0 {
offset = o
}
}
return db.AgentListFilter{Limit: limit, Offset: offset}, true
}

106
server/internal/api/pathtracer_handler.go
View File

@@ -102,10 +102,112 @@ func NewPathTracerHandler(hub *WSHub) *PathTracerHandler {
sessions: make(map[string]*TraceSession),
stopCh: make(chan struct{}),
}
h.loadPersistedSessions()
go h.sessionCleanupLoop()
return h
}
// pathTraceSessionPersist is the SQLite JSON shape (includes WG client keys for QR after restart).
type pathTraceSessionPersist struct {
ID string `json:"id"`
AgentIDs []string `json:"agent_ids"`
Hops []*HopInfo `json:"hops"`
Ready bool `json:"ready"`
Error string `json:"error,omitempty"`
CreatedAt time.Time `json:"created_at"`
ServiceGraph map[string]ServiceGraphHost `json:"service_graph,omitempty"`
DiscoverInProgress bool `json:"discover_in_progress,omitempty"`
DiscoverError string `json:"discover_error,omitempty"`
DiscoveredAt *time.Time `json:"discovered_at,omitempty"`
NetworkHints json.RawMessage `json:"network_hints,omitempty"`
ClientPrivKey string `json:"client_priv_key,omitempty"`
ClientPubKey string `json:"client_pub_key,omitempty"`
}
func (h *PathTracerHandler) loadPersistedSessions() {
if h.hub == nil || h.hub.db == nil {
return
}
cutoff := time.Now().Add(-pathTraceSessionTTL)
if n, err := h.hub.db.DeleteExpiredPathTraceSessions(cutoff); err == nil && n > 0 {
log.Printf("[pathtrace] startup sweep removed %d expired session(s)", n)
}
rows, err := h.hub.db.ListPathTraceSessions()
if err != nil {
log.Printf("[pathtrace] load sessions failed: %v", err)
return
}
h.mu.Lock()
defer h.mu.Unlock()
for _, row := range rows {
if time.Since(row.CreatedAt) > pathTraceSessionTTL {
_ = h.hub.db.DeletePathTraceSession(row.ID)
continue
}
var rec pathTraceSessionPersist
if err := json.Unmarshal(row.Payload, &rec); err != nil {
_ = h.hub.db.DeletePathTraceSession(row.ID)
continue
}
h.sessions[rec.ID] = &TraceSession{
ID: rec.ID,
AgentIDs: rec.AgentIDs,
Hops: rec.Hops,
Ready: rec.Ready,
Error: rec.Error,
CreatedAt: rec.CreatedAt,
ServiceGraph: rec.ServiceGraph,
DiscoverInProgress: rec.DiscoverInProgress,
DiscoverError: rec.DiscoverError,
DiscoveredAt: rec.DiscoveredAt,
NetworkHints: rec.NetworkHints,
clientPrivKey: rec.ClientPrivKey,
clientPubKey: rec.ClientPubKey,
}
}
if len(rows) > 0 {
log.Printf("[pathtrace] restored %d session(s) from SQLite", len(h.sessions))
}
}
func (h *PathTracerHandler) persistSession(sess *TraceSession) {
if h.hub == nil || h.hub.db == nil || sess == nil {
return
}
h.mu.Lock()
rec := pathTraceSessionPersist{
ID: sess.ID,
AgentIDs: sess.AgentIDs,
Hops: sess.Hops,
Ready: sess.Ready,
Error: sess.Error,
CreatedAt: sess.CreatedAt,
ServiceGraph: sess.ServiceGraph,
DiscoverInProgress: sess.DiscoverInProgress,
DiscoverError: sess.DiscoverError,
DiscoveredAt: sess.DiscoveredAt,
NetworkHints: sess.NetworkHints,
ClientPrivKey: sess.clientPrivKey,
ClientPubKey: sess.clientPubKey,
}
h.mu.Unlock()
raw, err := json.Marshal(rec)
if err != nil {
log.Printf("[pathtrace] marshal session %s: %v", sess.ID[:min(8, len(sess.ID))], err)
return
}
if err := h.hub.db.UpsertPathTraceSession(sess.ID, sess.CreatedAt, raw); err != nil {
log.Printf("[pathtrace] persist session %s: %v", sess.ID[:min(8, len(sess.ID))], err)
}
}
func (h *PathTracerHandler) deletePersistedSession(id string) {
if h.hub == nil || h.hub.db == nil || id == "" {
return
}
_ = h.hub.db.DeletePathTraceSession(id)
}
func (h *PathTracerHandler) sessionCleanupLoop() {
ticker := time.NewTicker(pathTraceCleanupInterval)
defer ticker.Stop()
@@ -133,6 +235,7 @@ func (h *PathTracerHandler) expireSessions() {
for _, sess := range expired {
log.Printf("[pathtrace] session %s expired after %s", sess.ID[:8], pathTraceSessionTTL)
h.teardownHops(sess.Hops)
h.deletePersistedSession(sess.ID)
}
}
@@ -201,6 +304,7 @@ func (h *PathTracerHandler) Start(w http.ResponseWriter, r *http.Request) {
h.mu.Lock()
h.sessions[sess.ID] = sess
h.mu.Unlock()
h.persistSession(sess)
// Orchestrate asynchronously so the HTTP response returns quickly.
go h.orchestrate(sess)
@@ -301,6 +405,7 @@ func (h *PathTracerHandler) Delete(w http.ResponseWriter, r *http.Request) {
h.mu.Lock()
delete(h.sessions, id)
h.mu.Unlock()
h.deletePersistedSession(id)
writeJSON(w, map[string]interface{}{"ok": true})
}
@@ -777,6 +882,7 @@ func (h *PathTracerHandler) orchestrate(sess *TraceSession) {
}
log.Printf("[pathtrace] session %s: orchestration complete, ready=%v", sess.ID[:8], allReady)
h.persistSession(sess)
}
func (h *PathTracerHandler) teardownHops(hops []*HopInfo) {

48
server/internal/api/pathtracer_persist_test.go Normal file
View File

@@ -0,0 +1,48 @@
package api
import (
"testing"
"time"
"crypto-miner-server/internal/db"
)
func TestPathTracerSessionSurvivesHandlerRestart(t *testing.T) {
database, err := db.New(t.TempDir())
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = database.Close() })
hub := NewWSHub(database)
sess := &TraceSession{
ID: "sess-restart-abcdef12",
AgentIDs: []string{"agent-1"},
CreatedAt: time.Now().Add(-10 * time.Minute),
Ready: true,
clientPrivKey: "PRIV",
clientPubKey: "PUB",
Hops: []*HopInfo{{
AgentID: "agent-1",
AgentName: "Hop One",
ExternalIP: "203.0.113.10",
Port: 51820,
Status: HopReady,
}},
}
h1 := NewPathTracerHandler(hub)
h1.mu.Lock()
h1.sessions[sess.ID] = sess
h1.mu.Unlock()
h1.persistSession(sess)
h2 := NewPathTracerHandler(hub)
restored := h2.getSession(sess.ID)
if restored == nil {
t.Fatal("session missing after handler restart")
}
if !restored.Ready || restored.clientPubKey != "PUB" {
t.Fatalf("restored session incomplete: ready=%v pub=%q", restored.Ready, restored.clientPubKey)
}
}

5
server/internal/api/testdata/ws_types_fixture.json vendored
View File

@@ -1,6 +1,9 @@
{
"WSDashboardInit": [
"agents"
"agents",
"limit",
"offset",
"total"
],
"WSAgentOffline": [
"agent_id"

41
server/internal/api/websocket.go
View File

@@ -241,8 +241,7 @@ func (h *WSHub) runStaleAgentSweep() {
if h.db == nil {
return
}
const staleness = 3 * time.Minute
ticker := time.NewTicker(45 * time.Second)
ticker := time.NewTicker(StaleAgentSweepInterval)
defer ticker.Stop()
for range ticker.C {
// Only sweep agents that are NOT currently connected in memory.
@@ -254,15 +253,12 @@ func (h *WSHub) runStaleAgentSweep() {
}
h.mu.RUnlock()
agents, err := h.db.ListAgents()
agents, err := h.db.ListStaleOnlineAgents(StaleAgentThreshold)
if err != nil {
continue
}
for _, a := range agents {
if a == nil || a.Status != "online" || liveIDs[a.ID] {
continue
}
if time.Since(a.LastSeen) < staleness {
if a == nil || liveIDs[a.ID] {
continue
}
// Row claims online, no live socket, last_seen is stale — fix it.
@@ -1672,7 +1668,15 @@ func (h *WSHub) HandleDashboardWS(w http.ResponseWriter, r *http.Request) {
// Send initial data — reconcile DB status against live hub state so a
// freshly loaded dashboard never shows stale "online" phantoms.
agents, _ := h.db.ListAgents()
initFilter, paginated := parseDashboardInitFilter(r)
var agents []*models.Agent
var initTotal int
if paginated {
agents, _ = h.db.ListAgentsFiltered(initFilter)
initTotal, _ = h.db.CountAgentsFiltered(initFilter)
} else {
agents, _ = h.db.ListAgents()
}
h.enrichAgentsCapabilities(agents)
for _, a := range agents {
if a == nil {
@@ -1687,10 +1691,16 @@ func (h *WSHub) HandleDashboardWS(w http.ResponseWriter, r *http.Request) {
}
stats, _ := h.db.GetFleetStats()
_ = dc.WriteJSON(Message{Type: "init", Payload: mustMarshal(map[string]interface{}{
initPayload := map[string]interface{}{
"agents": agents,
"stats": stats,
})})
}
if paginated {
initPayload["total"] = initTotal
initPayload["limit"] = initFilter.Limit
initPayload["offset"] = initFilter.Offset
}
_ = dc.WriteJSON(Message{Type: "init", Payload: mustMarshal(initPayload)})
_ = dc.WriteJSON(Message{Type: "presence_snapshot", Payload: mustMarshal(map[string]interface{}{
"comrades": h.presenceSnapshotLocked(),
})})
@@ -1733,14 +1743,17 @@ func (h *WSHub) HandleDashboardWS(w http.ResponseWriter, r *http.Request) {
continue
}
h.broadcastNotesTyping(username, body.Active)
case "tunnel_stream":
_ = dc.WriteJSON(Message{Type: "tunnel_stream", Payload: mustMarshal(map[string]interface{}{
"implemented": false,
"error": "tunnel_stream TCP reverse relay is not implemented; use tunnel_cloudflared or tunnel_ssh_forward on fleet agents",
})})
}
}
}
const statsBatchInterval = 250 * time.Millisecond
// mergeStatsPayload shallow-merges two stats maps so stats + mining_status in the
// same 250ms window both land in one stats_batch update for dashboards.
// same coalesce window both land in one stats_batch update for dashboards.
func mergeStatsPayload(existing, incoming json.RawMessage) json.RawMessage {
var base, patch map[string]interface{}
if json.Unmarshal(existing, &base) != nil || base == nil {
@@ -1795,7 +1808,7 @@ func (h *WSHub) queueStatsBroadcast(payload map[string]interface{}) {
h.statsBatch[agentID] = data
h.cacheAgentTelemetry(agentID, payload)
if h.statsBatchTimer == nil {
h.statsBatchTimer = time.AfterFunc(statsBatchInterval, h.flushStatsBatch)
h.statsBatchTimer = time.AfterFunc(StatsBatchCoalesceInterval, h.flushStatsBatch)
}
h.statsBatchMu.Unlock()
}

209
server/internal/api/websocket_test.go
View File

@@ -186,6 +186,116 @@ func TestHandleDashboardWSAuthorizedInit(t *testing.T) {
}
}
func TestHandleDashboardWSInitLimit(t *testing.T) {
resetWSAuthUsers(t, testAuthUser, testAuthPass)
database, err := db.New(t.TempDir())
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = database.Close() })
hub := NewWSHub(database)
for i := 0; i < 5; i++ {
if err := database.UpsertAgent(&models.Agent{
ID: fmt.Sprintf("agent-%d", i),
Name: fmt.Sprintf("Worker %d", i),
Status: "offline",
LastSeen: time.Now(),
}); err != nil {
t.Fatal(err)
}
}
srv := httptest.NewServer(http.HandlerFunc(hub.HandleDashboardWS))
t.Cleanup(srv.Close)
wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") +
"?token=" + wsDashboardToken(testAuthUser, testAuthPass) + "&init_limit=2&init_offset=1"
conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
if err != nil {
t.Fatalf("dial: %v", err)
}
t.Cleanup(func() { _ = conn.Close() })
var msg Message
if err := conn.ReadJSON(&msg); err != nil {
t.Fatalf("read init: %v", err)
}
if msg.Type != "init" {
t.Fatalf("expected init, got %q", msg.Type)
}
var body map[string]interface{}
if err := json.Unmarshal(msg.Payload, &body); err != nil {
t.Fatal(err)
}
agents, _ := body["agents"].([]interface{})
if len(agents) != 2 {
t.Fatalf("expected 2 agents in init page, got %d", len(agents))
}
if int(body["total"].(float64)) != 5 {
t.Fatalf("total=%v want 5", body["total"])
}
if int(body["limit"].(float64)) != 2 || int(body["offset"].(float64)) != 1 {
t.Fatalf("limit/offset = %v/%v", body["limit"], body["offset"])
}
}
func TestHandleDashboardWSTunnelStreamStub(t *testing.T) {
resetWSAuthUsers(t, testAuthUser, testAuthPass)
database, err := db.New(t.TempDir())
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = database.Close() })
hub := NewWSHub(database)
srv := httptest.NewServer(http.HandlerFunc(hub.HandleDashboardWS))
t.Cleanup(srv.Close)
wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + "?token=" + wsDashboardToken(testAuthUser, testAuthPass)
conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
if err != nil {
t.Fatalf("dial: %v", err)
}
t.Cleanup(func() { _ = conn.Close() })
// Drain init + presence_snapshot.
for i := 0; i < 2; i++ {
var msg Message
if err := conn.ReadJSON(&msg); err != nil {
t.Fatalf("read bootstrap %d: %v", i, err)
}
}
if err := conn.WriteJSON(Message{Type: "tunnel_stream", Payload: json.RawMessage(`{"port":8989}`)}); err != nil {
t.Fatal(err)
}
deadline := time.Now().Add(2 * time.Second)
for {
if time.Now().After(deadline) {
t.Fatal("timed out waiting for tunnel_stream response")
}
var resp Message
if err := conn.ReadJSON(&resp); err != nil {
t.Fatal(err)
}
if resp.Type != "tunnel_stream" {
continue
}
var body map[string]interface{}
if err := json.Unmarshal(resp.Payload, &body); err != nil {
t.Fatal(err)
}
if body["implemented"] != false {
t.Fatalf("expected implemented=false, got %+v", body)
}
if _, ok := body["error"].(string); !ok {
t.Fatalf("expected error string, got %+v", body)
}
break
}
}
func resetWSAuthUsersMulti(t *testing.T, creds map[string]string) {
t.Helper()
users := make(map[string]string, len(creds))
@@ -795,6 +905,105 @@ func TestMiningStatusRelayCoalescedToStatsBatch(t *testing.T) {
}
}
func TestStatsBatchRelaysFailedMethods(t *testing.T) {
resetWSAuthUsers(t, testAuthUser, testAuthPass)
database, err := db.New(t.TempDir())
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = database.Close() })
hub := NewWSHub(database)
dashSrv := httptest.NewServer(http.HandlerFunc(hub.HandleDashboardWS))
t.Cleanup(dashSrv.Close)
dashURL := "ws" + strings.TrimPrefix(dashSrv.URL, "http") + "?token=" + wsDashboardToken(testAuthUser, testAuthPass)
dashConn, _, err := websocket.DefaultDialer.Dial(dashURL, nil)
if err != nil {
t.Fatalf("dial dashboard: %v", err)
}
t.Cleanup(func() { _ = dashConn.Close() })
type batchResult struct {
updates []map[string]interface{}
err string
}
batchCh := make(chan batchResult, 1)
go func() {
_ = dashConn.SetReadDeadline(time.Now().Add(5 * time.Second))
for {
var msg Message
if err := dashConn.ReadJSON(&msg); err != nil {
batchCh <- batchResult{err: err.Error()}
return
}
if msg.Type != "stats_batch" {
continue
}
var body struct {
Updates []json.RawMessage `json:"updates"`
}
if parseErr := json.Unmarshal(msg.Payload, &body); parseErr != nil {
batchCh <- batchResult{err: parseErr.Error()}
return
}
updates := make([]map[string]interface{}, 0, len(body.Updates))
for _, raw := range body.Updates {
var u map[string]interface{}
if json.Unmarshal(raw, &u) != nil {
continue
}
updates = append(updates, u)
}
if len(updates) < 1 {
continue
}
batchCh <- batchResult{updates: updates}
return
}
}()
agentID := "failed-methods-agent"
conn := connectTestAgent(t, hub, agentID)
statsPayload, _ := json.Marshal(map[string]interface{}{
"hashrate_15s": 0.0,
"active_method": "inprocess",
"chain_exhausted": false,
"failed_methods": []map[string]interface{}{
{"method": "container", "reason": "image not found", "at": "2026-06-07T12:00:00Z"},
{"method": "docker_load", "reason": "no tar", "at": "2026-06-07T12:00:01Z"},
},
})
if err := conn.WriteJSON(Message{Type: "stats", Payload: statsPayload}); err != nil {
t.Fatal(err)
}
stopStatsBatchTimer(hub)
hub.flushStatsBatch()
select {
case r := <-batchCh:
if r.err != "" {
t.Fatalf("dashboard did not receive stats_batch: %s", r.err)
}
if len(r.updates) != 1 {
t.Fatalf("expected 1 update, got %d: %+v", len(r.updates), r.updates)
}
u := r.updates[0]
if u["agent_id"] != agentID {
t.Errorf("agent_id = %v", u["agent_id"])
}
failed, ok := u["failed_methods"].([]interface{})
if !ok || len(failed) != 2 {
t.Fatalf("failed_methods = %T %v", u["failed_methods"], u["failed_methods"])
}
first, _ := failed[0].(map[string]interface{})
if first["method"] != "container" || first["reason"] != "image not found" {
t.Errorf("first failure = %v", first)
}
case <-time.After(2 * time.Second):
t.Fatal("timed out waiting for stats_batch with failed_methods")
}
}
// stopStatsBatchTimer cancels the 250ms flush timer so tests can read statsBatch
// without racing flushStatsBatch clearing the pending map.
func stopStatsBatchTimer(hub *WSHub) {

3
server/internal/api/ws_types.go
View File

@@ -6,6 +6,9 @@ package api
type WSDashboardInit struct {
Agents []interface{} `json:"agents"`
Total int `json:"total,omitempty"`
Limit int `json:"limit,omitempty"`
Offset int `json:"offset,omitempty"`
}
type WSAgentOffline struct {