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"
)

const defaultPort = "8931"

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

// 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
	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)
	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
	}
	writeJSON(w, pingResponse{
		NodeID: n.Kdb.NodeID(),
		Name:   n.Name,
		Vector: vector,
	})
}

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
}

func (n *Node) handleBatch(w http.ResponseWriter, r *http.Request) {
	body, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, err.Error(), 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
	}
	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,
	})
}

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()
	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)
}

// Run executes one anti-entropy sweep against the given peer addresses.
func Run(kdb *db.KnoxDB, peers []string) {
	myID := kdb.NodeID()
	for _, addr := range peers {
		addr = strings.TrimSpace(addr)
		if addr == "" {
			continue
		}
		if strings.HasPrefix(addr, kdb.NodeID()+":") {
			continue
		}
		c := &Client{Addr: addr, Timeout: 10 * time.Second}
		p, err := c.Ping()
		if err != nil {
			log.Printf("[gossip] ping %s: %v", addr, err)
			continue
		}
		if p.NodeID == myID {
			continue // never talk to ourselves (or an aliased address)
		}

		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)
					continue
				}
				pulled += n
			}
		}

		pushed, _ := c.Push(kdb, p.Vector)

		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)
		}
	}
}

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
}

// 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
}