knox/internal/watch/gossip.go

// Package watch implements the knox peer-to-peer observation sync.
//
// A node advertises a small HTTP/JSON API (ping, log pull, batch push). Peers
// periodically exchange knowledge vectors, pull what they lack, and push their
// own observations. Observations are immutable and merged idempotently on the
// (node_id, hcl) locator; materialized state is rebuilt by reconcile after a
// pull.
package watch

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"strings"
	"time"

	"github.com/david/knox/internal/db"
	"github.com/david/knox/internal/metrics"
)

const defaultPort = "8931"

type Node struct {
	Kdb     *db.KnoxDB
	Name    string
	Addr    string // advertised base URL, e.g. http://192.168.1.20:8931
	Metrics *metrics.Metrics
}

// pingResponse is the anti-entropy summary returned by /v1/ping.
type pingResponse struct {
	NodeID string           `json:"node_id"`
	Name   string           `json:"name"`
	Vector map[string]int64 `json:"vector"` // node_id → max hcl
	Peers  []db.PeerInfo    `json:"peers"`  // swarm membership this node knows
	MaxHCL *int64           `json:"max_hcl,omitempty"`
}

func (n *Node) Handler() http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("GET /v1/ping", n.handlePing)
	mux.HandleFunc("GET /v1/log", n.handleLog)
	mux.HandleFunc("POST /v1/obs/batch", n.handleBatch)
	mux.HandleFunc("GET /v1/diff", n.handleDiff)
	return mux
}

func (n *Node) handlePing(w http.ResponseWriter, r *http.Request) {
	vector, err := n.Kdb.KnowledgeVector()
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	peers, err := n.Kdb.ShareablePeers()
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	writeJSON(w, pingResponse{
		NodeID: n.Kdb.NodeID(),
		Name:   n.Name,
		Vector: vector,
		Peers:  peers,
	})
}

func (n *Node) handleLog(w http.ResponseWriter, r *http.Request) {
	q := r.URL.Query()
	nodeID := q.Get("node")
	var after int64
	fmt.Sscanf(q.Get("after"), "%d", &after)

	if nodeID == "" {
		http.Error(w, "node param required", http.StatusBadRequest)
		return
	}
	rows, err := n.Kdb.ObservationsAfter(nodeID, after, 500)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	writeJSON(w, map[string]any{
		"node":   nodeID,
		"after":  after,
		"cursor": nextCursor(rows),
		"rows":   rows,
	})
}

func nextCursor(rows []db.GossipObservation) int64 {
	if len(rows) == 0 {
		return 0
	}
	return rows[len(rows)-1].HCL
}

// maxBatchRows bounds the number of observations a peer may push in one POST.
// Pull already pages at 500 rows, so any larger batch is at best redundant and
// at worst a flood; capping keeps memory and insert work bounded.
const maxBatchRows = 1000

func (n *Node) handleBatch(w http.ResponseWriter, r *http.Request) {
	r.Body = http.MaxBytesReader(w, r.Body, 4<<20) // 4 MiB
	body, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, "batch too large or unreadable", http.StatusBadRequest)
		return
	}
	var rows []db.GossipObservation
	if err := json.Unmarshal(body, &rows); err != nil {
		http.Error(w, "bad json: "+err.Error(), http.StatusBadRequest)
		return
	}
	if len(rows) > maxBatchRows {
		http.Error(w, fmt.Sprintf("batch too large: %d rows (max %d)", len(rows), maxBatchRows), http.StatusBadRequest)
		return
	}
	inserted, err := n.Kdb.PushObservations(rows)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	writeJSON(w, map[string]any{
		"accepted": inserted,
		"conflict": len(rows) - inserted,
	})
}

// DiffSummary is the divergence snapshot served at /v1/diff: everything this
// node observes, plus the status of auto-threaded threads (for tombstones).
type DiffSummary struct {
	NodeID       string            `json:"node_id"`
	Name         string            `json:"name"`
	Fingerprints []string          `json:"fingerprints"`
	ThreadStatus map[string]string `json:"thread_status"`
	ObsCount     int               `json:"obs_count"`
}

func (n *Node) handleDiff(w http.ResponseWriter, r *http.Request) {
	fps, err := n.Kdb.DistinctFingerprints()
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	threads, err := n.Kdb.ThreadStatusByCluster()
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	list := make([]string, 0, len(fps))
	for f := range fps {
		list = append(list, f)
	}
	writeJSON(w, DiffSummary{
		NodeID:       n.Kdb.NodeID(),
		Name:         n.Name,
		Fingerprints: list,
		ThreadStatus: threads,
		ObsCount:     len(fps),
	})
}

func writeJSON(w http.ResponseWriter, v any) {
	w.Header().Set("Content-Type", "application/json")
	enc := json.NewEncoder(w)
	enc.SetIndent("", "  ")
	_ = enc.Encode(v)
}

// Client is the pull/push half used by the exchange loop.
type Client struct {
	NodeID  string
	Addr    string
	Timeout time.Duration
}

func (c *Client) Ping() (*pingResponse, error) {
	var p pingResponse
	if err := c.get(c.Addr+"/v1/ping", &p); err != nil {
		return nil, err
	}
	return &p, nil
}

// Pull fetches a peer node's observations newer than the local cursor, looping
// through cursor pages until caught up. Returns newly inserted count.
func (c *Client) Pull(remoteNodeID string, after int64, kdb *db.KnoxDB) (int, error) {
	total := 0
	for {
		var resp struct {
			Rows []db.GossipObservation `json:"rows"`
		}
		url := fmt.Sprintf("%s/v1/log?node=%s&after=%d", c.Addr, remoteNodeID, after)
		if err := c.get(url, &resp); err != nil {
			return total, err
		}
		if len(resp.Rows) == 0 {
			break
		}
		n, err := kdb.PushObservations(resp.Rows)
		if err != nil {
			return total, err
		}
		total += n
		last := resp.Rows[len(resp.Rows)-1].HCL
		if last <= after {
			break // no progress — bail to avoid an infinite loop
		}
		after = last
		if len(resp.Rows) < 500 {
			break
		}
	}
	return total, nil
}

// Push sends this node's own observations newer than what the peer reported.
// Ownership by node_id prevents echo: we never resend observations we merely
// received from another node.
func (c *Client) Push(kdb *db.KnoxDB, peerVector map[string]int64) (int, error) {
	local := kdb.NodeID()
	after := peerVector[local] // what the peer already has from us
	rows, err := kdb.ObservationsAfter(local, after, 500)
	if err != nil {
		return 0, err
	}
	if len(rows) == 0 {
		return 0, nil
	}
	body, _ := json.Marshal(rows)
	req, err := http.NewRequest(http.MethodPost, c.Addr+"/v1/obs/batch", bytes.NewReader(body))
	if err != nil {
		return 0, err
	}
	req.Header.Set("Content-Type", "application/json")
	resp, err := (&http.Client{Timeout: c.Timeout}).Do(req)
	if err != nil {
		return 0, err
	}
	defer resp.Body.Close()
	if resp.StatusCode/100 != 2 {
		return 0, fmt.Errorf("push to %s: %s", c.Addr, resp.Status)
	}
	var out struct {
		Accepted int `json:"accepted"`
	}
	_ = json.NewDecoder(resp.Body).Decode(&out)
	return out.Accepted, nil
}

func (c *Client) get(url string, v any) error {
	req, err := http.NewRequest(http.MethodGet, url, nil)
	if err != nil {
		return err
	}
	resp, err := (&http.Client{Timeout: c.Timeout}).Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("%s %s: %s", resp.Request.Method, resp.Request.URL.String(), resp.Status)
	}
	return json.NewDecoder(resp.Body).Decode(v)
}

// Diff fetches the peer's divergence summary.
func (c *Client) Diff() (*DiffSummary, error) {
	var d DiffSummary
	if err := c.get(c.Addr+"/v1/diff", &d); err != nil {
		return nil, err
	}
	return &d, nil
}

// Run executes one anti-entropy + membership sweep.
//
// Peers are a fusion of the statically configured list (KNOX_PEERS) and peers
// previously discovered and persisted in the peers table (swarm join): syncing
// with any one node reveals who else is in the swarm, and those nodes are then
// swept too. There is no relay of observations — only membership is shared; each
// node pulls/pushes directly with every other node it learns about.
//
// m, when non-nil, receives gossip event counters (nil for one-shot CLI runs).
func Run(kdb *db.KnoxDB, m *metrics.Metrics, static []string) int {
	myID := kdb.NodeID()

	// Seed the work queue with static config plus persisted discoveries.
	persisted, _ := kdb.SwarmPeerAddrs()
	work := make([]string, 0, len(static)+len(persisted))
	work = append(work, static...)
	work = append(work, persisted...)

	seen := make(map[string]bool) // addr → handled (also suppresses self)
	queue := 0
	pulledTotal := 0
	for queue < len(work) {
		addr := strings.TrimSpace(work[queue])
		queue++
		if addr == "" || seen[addr] {
			continue
		}
		seen[addr] = true

		c := &Client{Addr: addr, Timeout: 10 * time.Second}
		p, err := c.Ping()
		if err != nil {
			log.Printf("[gossip] ping %s: %v", addr, err)
			if m != nil {
				m.IncrementErrors()
			}
			continue
		}
		if p.NodeID == myID {
			continue // never talk to ourselves (or an aliased address)
		}

		// Membership discovery: learn who else is in the swarm and enqueue
		// their addresses for direct sweeps.
		for _, pi := range p.Peers {
			if pi.PeerID == "" || pi.PeerID == myID || pi.Addr == "" {
				continue
			}
			if err := kdb.MergePeer(pi.PeerID, pi.Addr, pi.Name); err != nil {
				log.Printf("[gossip] merge peer %s: %v", pi.PeerID, err)
				continue
			}
			if !seen[pi.Addr] {
				work = append(work, pi.Addr)
			}
		}

		pulled := 0
		for remoteNode, remoteHCL := range p.Vector {
			localHCL := kdbVectorGet(kdb, remoteNode)
			if remoteNode == myID {
				continue // the peer cannot tell us about ourselves
			}
			if remoteHCL > localHCL {
				n, err := c.Pull(remoteNode, localHCL, kdb)
				if err != nil {
					log.Printf("[gossip] pull %s@%s: %v", remoteNode, addr, err)
					if m != nil {
						m.IncrementErrors()
					}
					continue
				}
				pulled += n
			}
		}
		pulledTotal += pulled
		if m != nil {
			m.IncrementPull(pulled)
		}

		pushed, err := c.Push(kdb, p.Vector)
		if err != nil {
			log.Printf("[gossip] push %s: %v", addr, err)
			if m != nil {
				m.IncrementErrors()
			}
		}
		if m != nil {
			m.IncrementPush(pushed)
		}

		if err := kdb.UpsertPeer(p.NodeID, addr, p.Name, vectorMax(p.Vector)); err != nil {
			log.Printf("[gossip] peer upsert: %v", err)
		}
		if pulled > 0 {
			log.Printf("[gossip] synced with %s (%s): pulled %d obs, pushed %d", p.NodeID, addr, pulled, pushed)
		} else {
			log.Printf("[gossip] synced with %s (%s): in sync (pushed %d)", p.NodeID, addr, pushed)
		}
	}
	return pulledTotal
}

func kdbVectorGet(kdb *db.KnoxDB, nodeID string) int64 {
	kv, err := kdb.KnowledgeVector()
	if err != nil {
		return 0
	}
	return kv[nodeID]
}

func vectorMax(v map[string]int64) int64 {
	var m int64
	for _, h := range v {
		if h > m {
			m = h
		}
	}
	return m
}

// ListenAddr returns the advertised HTTP address (KNOX_PEER_ADDR or default).
func ListenAddr() (addr string) {
	if addr = os.Getenv("KNOX_PEER_ADDR"); addr != "" {
		return addr
	}
	return "localhost:" + defaultPort
}

const defaultMetricsPort = "8932"

// MetricsAddr returns the Prometheus scrape address (KNOX_METRICS_ADDR or
// default). It is a separate port from gossip so scraping never contends with
// the peer protocol.
func MetricsAddr() (addr string) {
	if addr = os.Getenv("KNOX_METRICS_ADDR"); addr != "" {
		return addr
	}
	return "localhost:" + defaultMetricsPort
}

// PeerAddrs returns the configured peer list (KNOX_PEERS, comma-separated).
func PeerAddrs() []string {
	raw := os.Getenv("KNOX_PEERS")
	if raw == "" {
		return nil
	}
	var out []string
	for _, p := range strings.Split(raw, ",") {
		p = strings.TrimSpace(p)
		if p != "" {
			out = append(out, p)
		}
	}
	return out
}