Add curated Infrastructure view with repeater suggestions

Dockerfile

@@ -27,6 +27,7 @@ COPY internal/mbcapqueue/ ../../internal/mbcapqueue/
 COPY internal/lora/ ../../internal/lora/
 COPY internal/regions/ ../../internal/regions/
 COPY internal/channelregistry/ ../../internal/channelregistry/
+COPY internal/infrastructure/ ../../internal/infrastructure/
 RUN go mod download
 COPY cmd/server/ ./
 RUN CGO_ENABLED=0 GOOS=${TARGETOS} GOARCH=${TARGETARCH} \
@@ -46,6 +47,7 @@ COPY internal/brokerurl/ ../../internal/brokerurl/
 COPY internal/mbcapqueue/ ../../internal/mbcapqueue/
 COPY internal/regions/ ../../internal/regions/
 COPY internal/channelregistry/ ../../internal/channelregistry/
+COPY internal/infrastructure/ ../../internal/infrastructure/
 RUN go mod download
 COPY cmd/ingestor/ ./
 RUN CGO_ENABLED=0 GOOS=${TARGETOS} GOARCH=${TARGETARCH} \

cmd/ingestor/go.mod

@@ -60,6 +60,10 @@ require github.com/meshcore-analyzer/channelregistry v0.0.0
 
 replace github.com/meshcore-analyzer/channelregistry => ../../internal/channelregistry
 
+require github.com/meshcore-analyzer/infrastructure v0.0.0
+
+replace github.com/meshcore-analyzer/infrastructure => ../../internal/infrastructure
+
 require github.com/meshcore-analyzer/brokerurl v0.0.0
 
 replace github.com/meshcore-analyzer/brokerurl => ../../internal/brokerurl

cmd/ingestor/infrastructure.go

@@ -0,0 +1,79 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/meshcore-analyzer/channelregistry"
+	"github.com/meshcore-analyzer/infrastructure"
+)
+
+// runInfrastructureQueueOnce is called by the ingestor only. The server never
+// writes the selected snapshot; it writes atomic queue commands only.
+func runInfrastructureQueueOnce(store *Store) error {
+	q := channelregistry.NewQueue(infrastructure.QueuePath(store.path))
+	commands, err := q.Pending()
+	if err != nil {
+		return err
+	}
+	state, err := infrastructure.LoadDB(store.db)
+	if err != nil {
+		return err
+	}
+	now := time.Now().UnixMilli()
+	for _, queued := range commands {
+		if queued.Err != nil {
+			log.Printf("[infrastructure] unreadable command: %v", queued.Err)
+			if err := q.Discard(queued.Path); err != nil {
+				return err
+			}
+			continue
+		}
+		cmd := queued.Command
+		res := channelregistry.Result{RequestID: cmd.RequestID, CompletedAt: now}
+		op, key := "", ""
+		switch cmd.Op {
+		case channelregistry.OpSubmit:
+			op, key = infrastructure.OpSelect, cmd.Name
+		case channelregistry.OpRevoke:
+			op, key = infrastructure.OpRemove, cmd.ProposalID
+		}
+		if cmd.CreatedAt <= 0 || now-cmd.CreatedAt > 5*60*1000 || cmd.CreatedAt > now+60*1000 {
+			res.Status, res.Error = channelregistry.RequestError, "request expired"
+		} else if op == "" {
+			res.Status, res.Error = channelregistry.RequestError, "invalid operation"
+		} else if op == infrastructure.OpSelect && !knownInfrastructureRepeater(store, key) {
+			res.Status, res.Error = channelregistry.RequestError, "node is not a known repeater"
+		} else if err := state.Apply(cmd.RequestID, op, key, now); err != nil {
+			res.Status, res.Error = channelregistry.RequestError, err.Error()
+		} else {
+			// The SQLite row is the durable idempotence record as well as
+			// the selected set. Commit it before writing the result file.
+			if err := infrastructure.SaveDB(store.db, state); err != nil {
+				return fmt.Errorf("save infrastructure state: %w", err)
+			}
+			if op == infrastructure.OpSelect {
+				res.Status = channelregistry.RequestApproved
+			} else {
+				res.Status = channelregistry.RequestRevoked
+			}
+		}
+		if err := q.Complete(res); err != nil {
+			return fmt.Errorf("complete infrastructure request: %w", err)
+		}
+	}
+	if _, err := q.Prune(time.Now(), 10*time.Minute, infrastructure.MaxRecent); err != nil && !errors.Is(err, channelregistry.ErrUnknownRequest) {
+		return err
+	}
+	return nil
+}
+
+func knownInfrastructureRepeater(store *Store, key string) bool {
+	known, err := infrastructure.KnownRepeater(store.db, key)
+	if err != nil {
+		log.Printf("[infrastructure] repeater lookup failed: %v", err)
+	}
+	return err == nil && known
+}

cmd/ingestor/infrastructure_test.go

@@ -0,0 +1,89 @@
+package main
+
+import (
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/meshcore-analyzer/channelregistry"
+	"github.com/meshcore-analyzer/infrastructure"
+)
+
+func TestInfrastructureQueueRoundTripAndReplay(t *testing.T) {
+	dbPath := filepath.Join(t.TempDir(), "mesh.db")
+	store, err := OpenStore(dbPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer store.Close()
+	q := channelregistry.NewQueue(infrastructure.QueuePath(dbPath))
+	key := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+	if _, err := store.db.Exec(`INSERT INTO nodes(public_key, role) VALUES(?, 'repeater')`, key); err != nil {
+		t.Fatal(err)
+	}
+	phantom := channelregistry.Command{RequestID: channelregistry.NewID(), Op: channelregistry.OpSubmit, Name: "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc", CreatedAt: time.Now().UnixMilli()}
+	if err := q.Enqueue(phantom, 256); err != nil {
+		t.Fatal(err)
+	}
+	if err := runInfrastructureQueueOnce(store); err != nil {
+		t.Fatal(err)
+	}
+	bad, err := q.Lookup(phantom.RequestID)
+	if err != nil || bad.Status != channelregistry.RequestError {
+		t.Fatalf("phantom accepted: %+v, %v", bad, err)
+	}
+	a := channelregistry.Command{RequestID: channelregistry.NewID(), Op: channelregistry.OpSubmit, Name: key, CreatedAt: time.Now().UnixMilli()}
+	if err := q.Enqueue(a, 256); err != nil {
+		t.Fatal(err)
+	}
+	if err := runInfrastructureQueueOnce(store); err != nil {
+		t.Fatal(err)
+	}
+	s, err := infrastructure.LoadDB(store.db)
+	if err != nil || len(s.Selected) != 1 {
+		t.Fatalf("selected = %+v, %v", s.Selected, err)
+	}
+	b := channelregistry.Command{RequestID: channelregistry.NewID(), Op: channelregistry.OpRevoke, ProposalID: key, CreatedAt: time.Now().UnixMilli()}
+	if err := q.Enqueue(b, 256); err != nil {
+		t.Fatal(err)
+	}
+	if err := runInfrastructureQueueOnce(store); err != nil {
+		t.Fatal(err)
+	}
+	// A stale, already-applied command may reappear after a crash. Its durable
+	// request ID must not undo the later remove.
+	if err := q.Enqueue(a, 256); err != nil {
+		t.Fatal(err)
+	}
+	if err := runInfrastructureQueueOnce(store); err != nil {
+		t.Fatal(err)
+	}
+	s, err = infrastructure.LoadDB(store.db)
+	if err != nil || len(s.Selected) != 0 {
+		t.Fatalf("replayed old select: %+v, %v", s, err)
+	}
+}
+
+func TestInfrastructureSelectionSurvivesDatabaseRestart(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "mesh.db")
+	store, err := OpenStore(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	key := "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+	if err := infrastructure.SaveDB(store.db, infrastructure.State{Selected: []infrastructure.Entry{{PublicKey: key, AddedAt: 42}}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.Close(); err != nil {
+		t.Fatal(err)
+	}
+	reopened, err := OpenStore(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer reopened.Close()
+	state, err := infrastructure.LoadDB(reopened.db)
+	if err != nil || len(state.Selected) != 1 || state.Selected[0].PublicKey != key {
+		t.Fatalf("restart lost curation: %+v, %v", state, err)
+	}
+}

cmd/ingestor/main.go

@@ -445,11 +445,17 @@ func main() {
 	// SQLITE_BUSY.
 	store.RunPendingPruneRequests()
 	store.RunPendingClientRxDeletes()
+	if err := runInfrastructureQueueOnce(store); err != nil {
+		log.Printf("[infrastructure] startup queue: %v", err)
+	}
 	pruneQueueTicker := time.NewTicker(15 * time.Second)
 	go func() {
 		for range pruneQueueTicker.C {
 			store.RunPendingPruneRequests()
 			store.RunPendingClientRxDeletes()
+			if err := runInfrastructureQueueOnce(store); err != nil {
+				log.Printf("[infrastructure] queue: %v", err)
+			}
 		}
 	}()
 

cmd/server/go.mod

@@ -65,6 +65,10 @@ require github.com/meshcore-analyzer/channelregistry v0.0.0
 
 replace github.com/meshcore-analyzer/channelregistry => ../../internal/channelregistry
 
+require github.com/meshcore-analyzer/infrastructure v0.0.0
+
+replace github.com/meshcore-analyzer/infrastructure => ../../internal/infrastructure
+
 require github.com/meshcore-analyzer/brokerurl v0.0.0
 
 replace github.com/meshcore-analyzer/brokerurl => ../../internal/brokerurl

cmd/server/infrastructure.go

@@ -0,0 +1,105 @@
+package main
+
+import (
+	"errors"
+	"net/http"
+	"time"
+
+	"github.com/gorilla/mux"
+	"github.com/meshcore-analyzer/channelregistry"
+	"github.com/meshcore-analyzer/infrastructure"
+)
+
+// InfrastructureResponse deliberately contains only curation, not a second
+// copy of /api/nodes. The browser joins this bounded set to its bulk node
+// snapshot and calculates candidate suggestions from existing metrics.
+type InfrastructureResponse struct {
+	Selected []infrastructure.Entry `json:"selected"`
+}
+
+type InfrastructureAuthResponse struct {
+	OK bool `json:"ok"`
+}
+
+func (s *Server) handleInfrastructureAuth(w http.ResponseWriter, r *http.Request) {
+	writeJSON(w, InfrastructureAuthResponse{OK: true})
+}
+
+func (s *Server) infrastructureQueue() *channelregistry.Queue {
+	// Reuse the existing atomic, bounded file transport in a separate
+	// directory: OpSubmit maps to select and OpRevoke to remove. No channel
+	// proposal consumer ever sees these command files.
+	return channelregistry.NewQueue(infrastructure.QueuePath(s.db.path))
+}
+
+func (s *Server) handleInfrastructure(w http.ResponseWriter, r *http.Request) {
+	if s.db == nil || s.db.path == "" {
+		writeError(w, http.StatusServiceUnavailable, "database unavailable")
+		return
+	}
+	state, err := infrastructure.LoadDB(s.db.conn)
+	if err != nil {
+		writeError(w, http.StatusServiceUnavailable, "infrastructure snapshot unavailable")
+		return
+	}
+	if state.Selected == nil {
+		state.Selected = []infrastructure.Entry{}
+	}
+	writeJSON(w, InfrastructureResponse{Selected: state.Selected})
+}
+
+func (s *Server) handleInfrastructureDecision(op string) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		if s.db == nil || s.db.path == "" {
+			writeError(w, http.StatusServiceUnavailable, "database unavailable")
+			return
+		}
+		key := mux.Vars(r)["pubkey"]
+		if !infrastructure.ValidPublicKey(key) {
+			writeError(w, http.StatusBadRequest, "invalid public key")
+			return
+		}
+		if op == infrastructure.OpSelect {
+			known, err := infrastructure.KnownRepeater(s.db.conn, key)
+			if err != nil {
+				writeError(w, http.StatusServiceUnavailable, "node lookup unavailable")
+				return
+			}
+			if !known {
+				writeError(w, http.StatusConflict, "node is not a known repeater")
+				return
+			}
+		}
+		id := channelregistry.NewID()
+		cmd := channelregistry.Command{RequestID: id, CreatedAt: time.Now().UnixMilli()}
+		if op == infrastructure.OpSelect {
+			cmd.Op, cmd.Name = channelregistry.OpSubmit, key
+		} else {
+			cmd.Op, cmd.ProposalID = channelregistry.OpRevoke, key
+		}
+		if err := s.infrastructureQueue().Enqueue(cmd, 256); err != nil {
+			if errors.Is(err, channelregistry.ErrQueueFull) {
+				writeError(w, http.StatusTooManyRequests, "infrastructure queue full")
+				return
+			}
+			writeError(w, http.StatusServiceUnavailable, "infrastructure queue unavailable")
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusAccepted)
+		writeJSON(w, ChannelProposalAcceptedResponse{RequestID: id})
+	}
+}
+
+func (s *Server) handleInfrastructureRequest(w http.ResponseWriter, r *http.Request) {
+	if s.db == nil || s.db.path == "" {
+		writeError(w, http.StatusServiceUnavailable, "database unavailable")
+		return
+	}
+	status, err := s.infrastructureQueue().Lookup(mux.Vars(r)["id"])
+	if err != nil {
+		writeError(w, http.StatusNotFound, "unknown or expired request")
+		return
+	}
+	writeJSON(w, status)
+}