knox/internal/cmd/gossip.go

package cmd

import (
	"fmt"
	"time"

	"github.com/david/knox/internal/db"
	"github.com/david/knox/internal/watch"
	"github.com/spf13/cobra"
)

// NewGossipCmd exposes peer status/diff for the gossip protocol.
func NewGossipCmd(kdb *db.KnoxDB) *cobra.Command {
	cmd := &cobra.Command{
		Use:   "gossip",
		Short: "Inspect peer-to-peer observation sync",
	}
	cmd.AddCommand(
		&cobra.Command{
			Use:   "status",
			Short: "Show known peers and knowledge vectors",
			RunE: func(c *cobra.Command, args []string) error {
				peers, err := kdb.ListPeers()
				if err != nil {
					return err
				}
				vector, err := kdb.KnowledgeVector()
				if err != nil {
					return err
				}
				fmt.Printf("node_id: %s\n", kdb.NodeID())
				fmt.Printf("peers (static: %d configured):\n", len(watch.PeerAddrs()))
				for _, p := range peers {
					fmt.Printf("  %-24s %-28s last=%s  vector=%d\n", p.PeerID, p.Addr, p.LastHandshake, p.Cursor)
				}
				if len(peers) == 0 {
					fmt.Println("  (none — set KNOX_PEERS to sync with other nodes)")
				}
				fmt.Println("knowledge vector (node_id -> max hcl):")
				for nid, hcl := range vector {
					if nid == kdb.NodeID() {
						continue
					}
					fmt.Printf("  %-24s -> %d\n", nid, hcl)
				}
				return nil
			},
		},
		&cobra.Command{
			Use:   "diff <peer-url>",
			Short: "Show observation/thread divergence with a peer",
			Args:  cobra.ExactArgs(1),
			RunE: func(c *cobra.Command, args []string) error {
				return gossipDiff(kdb, args[0])
			},
		},
		&cobra.Command{
			Use:   "sync",
			Short: "Run one anti-entropy sweep against configured peers",
			RunE: func(c *cobra.Command, args []string) error {
				peers := watch.PeerAddrs()
				if len(peers) == 0 {
					return fmt.Errorf("no peers configured (set KNOX_PEERS)")
				}
				watch.Run(kdb, nil, peers)
				created, linked, err := watch.Reconcile(kdb)
				if err != nil {
					return err
				}
				fmt.Printf("sync done: %d threads created, %d observations linked\n", created, linked)
				return nil
			},
		},
	)
	return cmd
}

// gossipDiff compares this node's observation log + auto-threads with a peer's
// via /v1/diff. Never writes; it is the preview for what a sync would adopt.
func gossipDiff(kdb *db.KnoxDB, peerAddr string) error {
	c := &watch.Client{Addr: peerAddr, Timeout: 10 * time.Second}
	remote, err := c.Diff()
	if err != nil {
		return err
	}
	local, err := kdb.DistinctFingerprints()
	if err != nil {
		return err
	}

	remoteFP := make(map[string]bool, len(remote.Fingerprints))
	for _, f := range remote.Fingerprints {
		remoteFP[f] = true
	}

	var peerOnly, localOnly []string
	for f := range remoteFP {
		if !local[f] {
			peerOnly = append(peerOnly, f)
		}
	}
	for f := range local {
		if !remoteFP[f] {
			localOnly = append(localOnly, f)
		}
	}

	fmt.Printf("peer:       %s (%s)\n", remote.Name, remote.NodeID)
	fmt.Printf("this node:  %s\n", kdb.NodeID())
	fmt.Printf("fingerprints: peer=%d this=%d\n", len(remoteFP), len(local))
	fmt.Printf("  peer-only (would pull): %d\n", len(peerOnly))
	for _, f := range peerOnly[:min(10, len(peerOnly))] {
		fmt.Printf("    %s\n", f)
	}
	fmt.Printf("  local-only (would lose): %d\n", len(localOnly))
	for _, f := range localOnly[:min(10, len(localOnly))] {
		fmt.Printf("    %s\n", f)
	}

	// Tombstone divergence on auto-threaded threads.
	localThreads, err := kdb.ThreadStatusByCluster()
	if err != nil {
		return err
	}
	remoteThreads := remote.ThreadStatus
	var resolvedPeer, revokedLocal []string
	for k, st := range remoteThreads {
		if st == "resolved" {
			if ls, ok := localThreads[k]; !ok || ls != "resolved" {
				resolvedPeer = append(resolvedPeer, k)
			}
		}
	}
	for k, st := range localThreads {
		if st == "resolved" {
			if rs, ok := remoteThreads[k]; !ok || rs != "resolved" {
				revokedLocal = append(revokedLocal, k)
			}
		}
	}
	if len(resolvedPeer) > 0 {
		fmt.Printf("threads resolved on peer but not here (tombstone to adopt): %d\n", len(resolvedPeer))
	}
	if len(revokedLocal) > 0 {
		fmt.Printf("threads resolved here but not on peer (would resurrect on sync): %d\n", len(revokedLocal))
	}
	if len(peerOnly) == 0 && len(localOnly) == 0 && len(resolvedPeer) == 0 && len(revokedLocal) == 0 {
		fmt.Println("in sync.")
	}
	return nil
}

func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}