create_aio

2025-09-27 09:33:40 +02:00
parent b5ad30cfbd
commit de22d7c0f2
13 changed files with 1387 additions and 907 deletions
--- a/internal/mesh/mesh.go
+++ b/internal/mesh/mesh.go
@@ -0,0 +1,451 @@
+package mesh
+
+import (
+	"bytes"
+	"context"
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"hash/crc32"
+	"io"
+	"log"
+	"net"
+	"net/http"
+	"os"
+	"slices"
+	"strings"
+	"sync"
+	"time"
+)
+
+/*** Types & Config ***/
+
+type Config struct {
+	BindAddr         string   // e.g. ":9090"
+	AdvertURL        string   // e.g. "http://10.0.0.5:9090"
+	Seeds            []string // other peers' mesh base URLs
+	ClusterSecret    string   // HMAC key
+	EnableDiscovery  bool
+	DiscoveryAddress string // "239.8.8.8:9898"
+}
+
+type Peer struct {
+	URL       string    `json:"url"`
+	LastSeen  time.Time `json:"lastSeen"`
+	Self      bool      `json:"self"`
+	OwnerHint int       `json:"ownerHint"` // optional
+}
+
+type Item struct {
+	ID        int64  `json:"id"`
+	Name      string `json:"name"`
+	UpdatedAt int64  `json:"updatedAt"`
+	Deleted   bool   `json:"deleted"` // <— NEU: Tombstone für Deletes
+}
+
+type Snapshot struct {
+	Items []Item `json:"items"`
+}
+
+// Callbacks that your app provides
+type Callbacks struct {
+	GetSnapshot func(ctx context.Context) (Snapshot, error)
+	ApplyRemote func(ctx context.Context, s Snapshot) error
+}
+
+/*** Node ***/
+
+type Node struct {
+	cfg    Config
+	cbs    Callbacks
+	self   Peer
+	mu     sync.RWMutex
+	peers  map[string]*Peer
+	client *http.Client
+	srv    *http.Server
+	stop   chan struct{}
+	wg     sync.WaitGroup
+}
+
+func New(cfg Config, cbs Callbacks) (*Node, error) {
+	if cfg.BindAddr == "" || cfg.AdvertURL == "" {
+		return nil, errors.New("mesh: BindAddr and AdvertURL required")
+	}
+	if cfg.ClusterSecret == "" {
+		return nil, errors.New("mesh: ClusterSecret required")
+	}
+	n := &Node{
+		cfg:   cfg,
+		cbs:   cbs,
+		self:  Peer{URL: cfg.AdvertURL, LastSeen: time.Now(), Self: true},
+		peers: make(map[string]*Peer),
+		client: &http.Client{
+			Timeout: 5 * time.Second,
+		},
+		stop: make(chan struct{}),
+	}
+	return n, nil
+}
+
+/*** HMAC helpers ***/
+
+func (n *Node) sign(b []byte) string {
+	m := hmac.New(sha256.New, []byte(n.cfg.ClusterSecret))
+	m.Write(b)
+	return hex.EncodeToString(m.Sum(nil))
+}
+
+func (n *Node) verify(b []byte, sig string) bool {
+	want := n.sign(b)
+	return hmac.Equal([]byte(want), []byte(sig))
+}
+
+/*** HTTP handlers (control plane) ***/
+
+func (n *Node) helloHandler(w http.ResponseWriter, r *http.Request) {
+	body, _ := io.ReadAll(r.Body)
+	if !n.verify(body, r.Header.Get("X-Mesh-Sig")) {
+		http.Error(w, "bad signature", http.StatusUnauthorized)
+		return
+	}
+	var p Peer
+	if err := json.Unmarshal(body, &p); err != nil {
+		http.Error(w, "bad json", http.StatusBadRequest)
+		return
+	}
+	p.LastSeen = time.Now()
+	n.mu.Lock()
+	if existing, ok := n.peers[p.URL]; ok {
+		existing.LastSeen = p.LastSeen
+	} else if p.URL != n.self.URL {
+		cp := p
+		n.peers[p.URL] = &cp
+	}
+	n.mu.Unlock()
+	w.WriteHeader(http.StatusNoContent)
+}
+
+func (n *Node) peersHandler(w http.ResponseWriter, r *http.Request) {
+	n.mu.RLock()
+	defer n.mu.RUnlock()
+	var list []Peer
+	list = append(list, n.self)
+	for _, p := range n.peers {
+		list = append(list, *p)
+	}
+	writeJSON(w, http.StatusOK, list)
+}
+
+func (n *Node) syncHandler(w http.ResponseWriter, r *http.Request) {
+	// verify signature
+	body, _ := io.ReadAll(r.Body)
+	if !n.verify(body, r.Header.Get("X-Mesh-Sig")) {
+		http.Error(w, "bad signature", http.StatusUnauthorized)
+		return
+	}
+	var s Snapshot
+	if err := json.Unmarshal(body, &s); err != nil {
+		http.Error(w, "bad json", http.StatusBadRequest)
+		return
+	}
+	// apply
+	if err := n.cbs.ApplyRemote(r.Context(), s); err != nil {
+		http.Error(w, "apply error: "+err.Error(), http.StatusInternalServerError)
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
+func writeJSON(w http.ResponseWriter, code int, v any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(code)
+	_ = json.NewEncoder(w).Encode(v)
+}
+
+/*** Serve ***/
+
+func (n *Node) Serve() error {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/mesh/peers", n.peersHandler)
+	mux.HandleFunc("/mesh/hello", n.helloHandler)
+	mux.HandleFunc("/mesh/sync", n.syncHandler)
+	n.srv = &http.Server{Addr: n.cfg.BindAddr, Handler: mux}
+
+	// background loops
+	n.wg.Add(1)
+	go func() {
+		defer n.wg.Done()
+		n.loopSeeder()
+	}()
+
+	if n.cfg.EnableDiscovery && n.cfg.DiscoveryAddress != "" {
+		n.wg.Add(2)
+		go func() {
+			defer n.wg.Done()
+			n.loopBeaconSend()
+		}()
+		go func() {
+			defer n.wg.Done()
+			n.loopBeaconRecv()
+		}()
+	}
+
+	n.wg.Add(1)
+	go func() {
+		defer n.wg.Done()
+		n.loopAntiEntropy()
+	}()
+
+	// http server
+	errc := make(chan error, 1)
+	go func() {
+		errc <- n.srv.ListenAndServe()
+	}()
+	select {
+	case err := <-errc:
+		return err
+	case <-n.stop:
+		return http.ErrServerClosed
+	}
+}
+
+func (n *Node) Close(ctx context.Context) error {
+	close(n.stop)
+	if n.srv != nil {
+		_ = n.srv.Shutdown(ctx)
+	}
+	n.wg.Wait()
+	return nil
+}
+
+/*** Loops ***/
+
+func (n *Node) loopSeeder() {
+	// attempt to hello known seeds every 5s at start, then every 30s
+	backoff := 5 * time.Second
+	for {
+		select {
+		case <-n.stop:
+			return
+		case <-time.After(backoff):
+		}
+		if len(n.cfg.Seeds) == 0 {
+			backoff = 30 * time.Second
+			continue
+		}
+		for _, s := range n.cfg.Seeds {
+			if s == "" || s == n.self.URL {
+				continue
+			}
+			_ = n.sendHello(s)
+		}
+		backoff = 30 * time.Second
+	}
+}
+
+func (n *Node) loopAntiEntropy() {
+	t := time.NewTicker(10 * time.Second)
+	defer t.Stop()
+	for {
+		select {
+		case <-n.stop:
+			return
+		case <-t.C:
+			n.mu.RLock()
+			targets := make([]string, 0, len(n.peers))
+			for url := range n.peers {
+				targets = append(targets, url)
+			}
+			n.mu.RUnlock()
+			if len(targets) == 0 {
+				continue
+			}
+			ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
+			snap, err := n.cbs.GetSnapshot(ctx)
+			cancel()
+			if err != nil {
+				continue
+			}
+			for _, url := range targets {
+				_ = n.sendSync(url, snap)
+			}
+		}
+	}
+}
+
+func (n *Node) loopBeaconSend() {
+	addr, err := net.ResolveUDPAddr("udp", n.cfg.DiscoveryAddress)
+	if err != nil {
+		log.Printf("mesh beacon send resolve: %v", err)
+		return
+	}
+	conn, err := net.DialUDP("udp", nil, addr)
+	if err != nil {
+		log.Printf("mesh beacon send dial: %v", err)
+		return
+	}
+	defer conn.Close()
+	type beacon struct {
+		URL string `json:"url"`
+	}
+	t := time.NewTicker(5 * time.Second)
+	defer t.Stop()
+	for {
+		select {
+		case <-n.stop:
+			return
+		case <-t.C:
+			b, _ := json.Marshal(beacon{URL: n.self.URL})
+			_, _ = conn.Write(b)
+		}
+	}
+}
+
+func (n *Node) loopBeaconRecv() {
+	addr, err := net.ResolveUDPAddr("udp", n.cfg.DiscoveryAddress)
+	if err != nil {
+		log.Printf("mesh beacon recv resolve: %v", err)
+		return
+	}
+	// enable multicast receive
+	l, err := net.ListenMulticastUDP("udp", nil, addr)
+	if err != nil {
+		log.Printf("mesh beacon recv listen: %v", err)
+		return
+	}
+	defer l.Close()
+	_ = l.SetReadBuffer(1 << 16)
+
+	buf := make([]byte, 2048)
+	for {
+		select {
+		case <-n.stop:
+			return
+		default:
+		}
+		_ = l.SetDeadline(time.Now().Add(6 * time.Second))
+		nr, _, err := l.ReadFromUDP(buf)
+		if err != nil {
+			if ne, ok := err.(net.Error); ok && ne.Timeout() {
+				continue
+			}
+			continue
+		}
+		var msg struct{ URL string }
+		if err := json.Unmarshal(buf[:nr], &msg); err == nil {
+			if msg.URL != "" && msg.URL != n.self.URL {
+				_ = n.sendHello(msg.URL)
+			}
+		}
+	}
+}
+
+/*** Outgoing ***/
+
+func (n *Node) sendHello(url string) error {
+	p := n.self
+	b, _ := json.Marshal(p)
+	req, _ := http.NewRequest(http.MethodPost, strings.TrimRight(url, "/")+"/mesh/hello", bytes.NewReader(b))
+	req.Header.Set("X-Mesh-Sig", n.sign(b))
+	resp, err := n.client.Do(req)
+	if err == nil {
+		io.Copy(io.Discard, resp.Body)
+		resp.Body.Close()
+	}
+	return err
+}
+
+func (n *Node) sendSync(url string, s Snapshot) error {
+	b, _ := json.Marshal(s)
+	req, _ := http.NewRequest(http.MethodPost, strings.TrimRight(url, "/")+"/mesh/sync", bytes.NewReader(b))
+	req.Header.Set("X-Mesh-Sig", n.sign(b))
+	resp, err := n.client.Do(req)
+	if err == nil {
+		io.Copy(io.Discard, resp.Body)
+		resp.Body.Close()
+	}
+	return err
+}
+
+// PeerList liefert eine Kopie der bekannten Peers inkl. Self.
+func (n *Node) PeerList() []Peer {
+	n.mu.RLock()
+	defer n.mu.RUnlock()
+	out := make([]Peer, 0, len(n.peers)+1)
+	out = append(out, n.self)
+	for _, p := range n.peers {
+		cp := *p
+		out = append(out, cp)
+	}
+	return out
+}
+
+// SyncNow verschickt sofort den aktuellen Snapshot an alle bekannten Peers.
+func (n *Node) SyncNow(ctx context.Context) error {
+	snap, err := n.cbs.GetSnapshot(ctx)
+	if err != nil {
+		return err
+	}
+	n.mu.RLock()
+	targets := make([]string, 0, len(n.peers))
+	for url := range n.peers {
+		targets = append(targets, url)
+	}
+	n.mu.RUnlock()
+	for _, u := range targets {
+		_ = n.sendSync(u, snap)
+	}
+	return nil
+}
+
+/*** Utilities ***/
+
+// OwnerHint is a simple, optional mapping to distribute responsibility.
+func OwnerHint(id int64, peers []string) int {
+	if len(peers) == 0 {
+		return 0
+	}
+	h := crc32.ChecksumIEEE([]byte(string(rune(id))))
+	return int(h % uint32(len(peers)))
+}
+
+// Helpers to load from ENV quickly
+func FromEnv() Config {
+	return Config{
+		BindAddr:         getenvDefault("MESH_BIND", ":9090"),
+		AdvertURL:        os.Getenv("MESH_ADVERT"),
+		Seeds:            splitCSV(os.Getenv("MESH_SEEDS")),
+		ClusterSecret:    os.Getenv("MESH_CLUSTER_SECRET"),
+		EnableDiscovery:  os.Getenv("MESH_ENABLE_DISCOVERY") == "true",
+		DiscoveryAddress: getenvDefault("MESH_DISCOVERY_ADDR", "239.8.8.8:9898"),
+	}
+}
+
+func splitCSV(s string) []string {
+	if strings.TrimSpace(s) == "" {
+		return nil
+	}
+	parts := strings.Split(s, ",")
+	for i := range parts {
+		parts[i] = strings.TrimSpace(parts[i])
+	}
+	// dedup
+	out := make([]string, 0, len(parts))
+	for _, p := range parts {
+		if p == "" || slices.Contains(out, p) {
+			continue
+		}
+		out = append(out, p)
+	}
+	return out
+}
+
+func getenvDefault(k, def string) string {
+	v := os.Getenv(k)
+	if v == "" {
+		return def
+	}
+	return v
+}