Week 3 Day 15: Implement P2P GossipSub layer and configure regional OSRM routing backend

Author: Anbu-00001

.gitignore

@@ -72,3 +72,8 @@ docker/osrm/data/
 # Artifacts
 artifacts/*
 *.tar.gz
+
+# P2P and OSRM Local Data
+backend/p2p-node/p2p-daemon
+backend/p2p-node/test_ws.js
+osrm-data/

backend/p2p-node/api.go

@@ -0,0 +1,145 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+	"sync"
+
+	"github.com/gorilla/websocket"
+)
+
+var upgrader = websocket.Upgrader{
+	ReadBufferSize:  1024,
+	WriteBufferSize: 1024,
+	CheckOrigin: func(r *http.Request) bool {
+		return true // Allow all cross-origin connections for local testing
+	},
+}
+
+// APIServer handles HTTP requests and WebSocket connections
+type APIServer struct {
+	port          int
+	pubSubManager *PubSubManager
+	msgChan       chan IncidentMessage
+	clients       map[*websocket.Conn]bool
+	clientsMux    sync.RWMutex
+}
+
+// NewAPIServer creates a new API Server instance
+func NewAPIServer(port int, psm *PubSubManager, msgChan chan IncidentMessage) *APIServer {
+	return &APIServer{
+		port:          port,
+		pubSubManager: psm,
+		msgChan:       msgChan,
+		clients:       make(map[*websocket.Conn]bool),
+	}
+}
+
+// Start boots the HTTP server in a goroutine
+func (s *APIServer) Start() error {
+	mux := http.NewServeMux()
+
+	// Setup routes
+	mux.HandleFunc("/broadcast", s.handleBroadcast)
+	mux.HandleFunc("/events", s.handleEvents)
+
+	go s.broadcastLoop() // Loop to read from msgChan and send to all WS clients
+
+	serverAddr := fmt.Sprintf(":%d", s.port)
+	log.Printf("Starting P2P API Server at HTTP/WS %s", serverAddr)
+	
+	go func() {
+		if err := http.ListenAndServe(serverAddr, mux); err != nil {
+			log.Fatalf("API server failed: %v", err)
+		}
+	}()
+
+	return nil
+}
+
+// handleBroadcast is the HTTP handler for POST /broadcast
+func (s *APIServer) handleBroadcast(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	var inc IncidentMessage
+	if err := json.NewDecoder(r.Body).Decode(&inc); err != nil {
+		http.Error(w, "Invalid request body", http.StatusBadRequest)
+		return
+	}
+
+	// Make sure Type is set correctly (for schema consistency)
+	if inc.Type == "" {
+		inc.Type = "incident_create"
+	}
+
+	// Publish via GossipSub
+	if err := s.pubSubManager.Broadcast(inc); err != nil {
+		log.Printf("Failed to broadcast message: %v\n", err)
+		http.Error(w, "Failed to broadcast", http.StatusInternalServerError)
+		return
+	}
+
+	log.Printf("HTTP /broadcast processed: sent incident %s to peers", inc.IncidentID)
+
+	w.WriteHeader(http.StatusAccepted)
+	json.NewEncoder(w).Encode(map[string]string{"status": "broadcasted"})
+}
+
+// handleEvents upgrades the GET request to a WebSocket connection
+func (s *APIServer) handleEvents(w http.ResponseWriter, r *http.Request) {
+	conn, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		log.Printf("Failed to upgrade WebSocket: %v", err)
+		return
+	}
+
+	// Register the new client
+	s.clientsMux.Lock()
+	s.clients[conn] = true
+	s.clientsMux.Unlock()
+
+	log.Printf("New WebSocket client connected: %s", conn.RemoteAddr().String())
+
+	// Read loop just to handle disconnects (we don't expect messages *from* the client on this WS yet)
+	go func() {
+		defer func() {
+			s.clientsMux.Lock()
+			delete(s.clients, conn)
+			s.clientsMux.Unlock()
+			conn.Close()
+			log.Printf("WebSocket client disconnected")
+		}()
+		for {
+			if _, _, err := conn.ReadMessage(); err != nil {
+				break
+			}
+		}
+	}()
+}
+
+// broadcastLoop reads from the p2p input channel and forwards to all connected WebSockets
+func (s *APIServer) broadcastLoop() {
+	for msg := range s.msgChan {
+		// Serialize
+		data, err := json.Marshal(msg)
+		if err != nil {
+			log.Printf("Failed to marshal WS broadcast message: %v", err)
+			continue
+		}
+
+		s.clientsMux.RLock()
+		for client := range s.clients {
+			// Write the serialized json message to the websocket client
+			if err := client.WriteMessage(websocket.TextMessage, data); err != nil {
+				log.Printf("Failed to send message to WS client: %v", err)
+				client.Close() // Typically best to close and let the read loop delete it
+			}
+		}
+		s.clientsMux.RUnlock()
+	}
+}

backend/p2p-node/discovery.go

@@ -0,0 +1,54 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/p2p/discovery/mdns"
+)
+
+const discoveryServiceTag = "openrescue.p2p"
+
+// discoveryNotifee gets notified when we find a new peer via mDNS discovery
+type discoveryNotifee struct {
+	h host.Host
+}
+
+// HandlePeerFound connects to peers discovered via mDNS
+func (n *discoveryNotifee) HandlePeerFound(pi peer.AddrInfo) {
+	fmt.Printf("mDNS Discovery: Found peer %s\n", pi.ID)
+	// Don't try to connect to ourselves
+	if pi.ID == n.h.ID() {
+		return
+	}
+
+	// Connect to the peer
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	defer cancel()
+	
+	err := n.h.Connect(ctx, pi)
+	if err != nil {
+		fmt.Printf("mDNS Discovery: Failed connecting to peer %s: %s\n", pi.ID.String(), err)
+	} else {
+		fmt.Printf("mDNS Discovery: Automatically connected to peer %s\n", pi.ID.String())
+	}
+}
+
+// setupDiscovery creates an mDNS discovery service
+func setupDiscovery(h host.Host) error {
+	s := mdns.NewMdnsService(h, discoveryServiceTag, &discoveryNotifee{h: h})
+	if s == nil {
+		return fmt.Errorf("failed creating mDNS service")
+	}
+	
+	if err := s.Start(); err != nil {
+		return fmt.Errorf("failed starting mDNS service: %v", err)
+	}
+	
+	log.Printf("mDNS Peer Discovery started (service tag: %s)\n", discoveryServiceTag)
+	return nil
+}

backend/p2p-node/go.mod

@@ -0,0 +1,98 @@
+module openrescue/p2p-node
+
+go 1.24.6
+
+require (
+	github.com/gorilla/websocket v1.5.3
+	github.com/libp2p/go-libp2p v0.47.0
+	github.com/libp2p/go-libp2p-pubsub v0.15.0
+)
+
+require (
+	github.com/benbjohnson/clock v1.3.5 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c // indirect
+	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect
+	github.com/dunglas/httpsfv v1.1.0 // indirect
+	github.com/flynn/noise v1.1.0 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
+	github.com/huin/goupnp v1.3.0 // indirect
+	github.com/ipfs/go-cid v0.5.0 // indirect
+	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
+	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
+	github.com/klauspost/cpuid/v2 v2.2.10 // indirect
+	github.com/koron/go-ssdp v0.0.6 // indirect
+	github.com/libp2p/go-buffer-pool v0.1.0 // indirect
+	github.com/libp2p/go-flow-metrics v0.2.0 // indirect
+	github.com/libp2p/go-libp2p-asn-util v0.4.1 // indirect
+	github.com/libp2p/go-msgio v0.3.0 // indirect
+	github.com/libp2p/go-netroute v0.3.0 // indirect
+	github.com/libp2p/go-reuseport v0.4.0 // indirect
+	github.com/libp2p/go-yamux/v5 v5.0.1 // indirect
+	github.com/libp2p/zeroconf/v2 v2.2.0 // indirect
+	github.com/marten-seemann/tcp v0.0.0-20210406111302-dfbc87cc63fd // indirect
+	github.com/miekg/dns v1.1.66 // indirect
+	github.com/mikioh/tcpinfo v0.0.0-20190314235526-30a79bb1804b // indirect
+	github.com/mikioh/tcpopt v0.0.0-20190314235656-172688c1accc // indirect
+	github.com/minio/sha256-simd v1.0.1 // indirect
+	github.com/mr-tron/base58 v1.2.0 // indirect
+	github.com/multiformats/go-base32 v0.1.0 // indirect
+	github.com/multiformats/go-base36 v0.2.0 // indirect
+	github.com/multiformats/go-multiaddr v0.16.1 // indirect
+	github.com/multiformats/go-multiaddr-dns v0.4.1 // indirect
+	github.com/multiformats/go-multiaddr-fmt v0.1.0 // indirect
+	github.com/multiformats/go-multibase v0.2.0 // indirect
+	github.com/multiformats/go-multicodec v0.9.1 // indirect
+	github.com/multiformats/go-multihash v0.2.3 // indirect
+	github.com/multiformats/go-multistream v0.6.1 // indirect
+	github.com/multiformats/go-varint v0.0.7 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58 // indirect
+	github.com/pion/datachannel v1.5.10 // indirect
+	github.com/pion/dtls/v2 v2.2.12 // indirect
+	github.com/pion/dtls/v3 v3.0.6 // indirect
+	github.com/pion/ice/v4 v4.0.10 // indirect
+	github.com/pion/interceptor v0.1.40 // indirect
+	github.com/pion/logging v0.2.3 // indirect
+	github.com/pion/mdns/v2 v2.0.7 // indirect
+	github.com/pion/randutil v0.1.0 // indirect
+	github.com/pion/rtcp v1.2.15 // indirect
+	github.com/pion/rtp v1.8.19 // indirect
+	github.com/pion/sctp v1.8.39 // indirect
+	github.com/pion/sdp/v3 v3.0.13 // indirect
+	github.com/pion/srtp/v3 v3.0.6 // indirect
+	github.com/pion/stun v0.6.1 // indirect
+	github.com/pion/stun/v3 v3.0.0 // indirect
+	github.com/pion/transport/v2 v2.2.10 // indirect
+	github.com/pion/transport/v3 v3.0.7 // indirect
+	github.com/pion/turn/v4 v4.0.2 // indirect
+	github.com/pion/webrtc/v4 v4.1.2 // indirect
+	github.com/prometheus/client_golang v1.22.0 // indirect
+	github.com/prometheus/client_model v0.6.2 // indirect
+	github.com/prometheus/common v0.64.0 // indirect
+	github.com/prometheus/procfs v0.16.1 // indirect
+	github.com/quic-go/qpack v0.6.0 // indirect
+	github.com/quic-go/quic-go v0.59.0 // indirect
+	github.com/quic-go/webtransport-go v0.10.0 // indirect
+	github.com/spaolacci/murmur3 v1.1.0 // indirect
+	github.com/wlynxg/anet v0.0.5 // indirect
+	go.uber.org/dig v1.19.0 // indirect
+	go.uber.org/fx v1.24.0 // indirect
+	go.uber.org/mock v0.5.2 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
+	golang.org/x/crypto v0.41.0 // indirect
+	golang.org/x/exp v0.0.0-20250606033433-dcc06ee1d476 // indirect
+	golang.org/x/mod v0.27.0 // indirect
+	golang.org/x/net v0.43.0 // indirect
+	golang.org/x/sync v0.16.0 // indirect
+	golang.org/x/sys v0.35.0 // indirect
+	golang.org/x/text v0.28.0 // indirect
+	golang.org/x/time v0.12.0 // indirect
+	golang.org/x/tools v0.36.0 // indirect
+	google.golang.org/protobuf v1.36.6 // indirect
+	lukechampine.com/blake3 v1.4.1 // indirect
+)