mesh

func TestMeshMultiNode(t *testing.T)

{
	requireMulticast(t)
	const nodeCount = 10
	portBase := 10000 + (int(time.Now().Unix()) % 5000)

	nodes := make([]*MeshBus, nodeCount)
	opts := make([]MeshOptions, nodeCount)

	ifi := findMulticastInterface()
	ifaceName := ""
	if ifi != nil {
		ifaceName = ifi.Name
	}

	// Create seed node (Node 0)
	seedAddr := fmt.Sprintf("127.0.0.1:%d", portBase)
	opts[0] = MeshOptions{
		Port:          portBase,
		Interface:     ifaceName,
		AdvertiseAddr: seedAddr,
		Heartbeat:     50 * time.Millisecond,
	}

	var err error
	nodes[0], err = NewMeshBus(opts[0])
	if err != nil {
		t.Fatalf("Failed to create seed node: %v", err)
	}
	defer nodes[0].Close()

	// Create other nodes, some with seed, some using multicast solely (simulated by same group)
	for i := 1; i < nodeCount; i++ {
		port := portBase + i
		addr := fmt.Sprintf("127.0.0.1:%d", port)

		o := MeshOptions{
			Port:          port,
			Interface:     ifaceName,
			AdvertiseAddr: addr,
			Heartbeat:     50 * time.Millisecond,
		}

		// Every node knows the seed to ensure discovery even if Multicast is restricted
		o.Peers = []string{seedAddr}

		opts[i] = o
		nodes[i], err = NewMeshBus(o)
		if err != nil {
			t.Fatalf("Failed to create node %d: %v", i, err)
		}
		defer nodes[i].Close()
	}

	// Wait for full mesh discovery
	t.Log("Waiting for nodes to discover each other...")
	// Increase timeout for discovery in potentially constrained environments
	discoveryCtx, discoveryCancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer discoveryCancel()

	for i, node := range nodes {
	discoveryLoop:
		for {
			peers := node.Peers()
			// We wait until most nodes are discovered.
			// In loopback/CI, we might not get 100% but we need enough for a valid test.
			if len(peers) >= nodeCount/2 {
				break discoveryLoop
			}
			select {
			case <-discoveryCtx.Done():
				t.Logf("Node %d only discovered %d/%d peers: %v", i, len(peers), nodeCount-1, peers)
				break discoveryLoop
			case <-time.After(200 * time.Millisecond):
			}
		}
	}

	// Verify propagation: Node 0 publishes, everyone else receives
	key := "multi-node-sync"
	verificationCtx, verificationCancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer verificationCancel()

	subs := make([]<-chan syncbus.Event, nodeCount)
	for i := 1; i < nodeCount; i++ {
		// Use verificationCtx so channels don't close prematurely if discovery took too long
		ch, _ := nodes[i].Subscribe(verificationCtx, key)
		subs[i] = ch
	}

	time.Sleep(500 * time.Millisecond) // Give time for subscriptions and gossip to settle

	if err := nodes[0].Publish(context.Background(), key); err != nil {
		t.Fatalf("Failed to publish from seed: %v", err)
	}

	var wg sync.WaitGroup
	for i := 1; i < nodeCount; i++ {
		wg.Add(1)
		go func(idx int, ch <-chan syncbus.Event) {
			defer wg.Done()
			select {
			case evt, ok := <-ch:
				if !ok {
					t.Errorf("Node %d: channel closed unexpectedly", idx)
					return
				}
				if evt.Key != key {
					t.Errorf("Node %d received wrong key: '%s' (expected '%s')", idx, evt.Key, key)
				}
			case <-verificationCtx.Done():
				// If we timed out, check if we missed it or if it's just slow
				t.Errorf("Node %d timed out waiting for event", idx)
			}
		}(i, subs[i])
	}
	wg.Wait()
}

func BenchmarkMeshPublish(b *testing.B)

{
	opts := MeshOptions{
		Port:      12000,
		Heartbeat: 1 * time.Hour, // Disable heartbeats for bench
	}
	node, _ := NewMeshBus(opts)
	defer node.Close()

	ctx := context.Background()
	key := "bench-key"

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		_ = node.Publish(ctx, key)
	}
}

func BenchmarkMeshPacketMarshal(b *testing.B)

{
	p := packet{
		Magic:   magicByte,
		Type:    typeInvalidate,
		NodeID:  [16]byte{1, 2, 3, 4},
		KeyHash: 12345678,
		KeyLen:  10,
		Key:     []byte("test-key-1"),
	}
	buf := make([]byte, 1500)

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		_, _ = p.marshal(buf)
	}
}

func requireMulticast(t *testing.T)

{
	t.Helper()
	addr, err := net.ResolveUDPAddr("udp4", "239.0.0.1:0")
	if err != nil {
		t.Skipf("requires multicast: cannot resolve 239.0.0.1: %v", err)
	}
	c, err := net.ListenPacket("udp4", "0.0.0.0:0")
	if err != nil {
		t.Skipf("requires multicast: cannot open UDP socket: %v", err)
	}
	defer c.Close()
	pc := ipv4.NewPacketConn(c)
	if err := pc.JoinGroup(nil, addr); err != nil {
		t.Skipf("requires a functional multicast environment: %v", err)
	}
}

func TestMeshIntegration(t *testing.T)

{
	requireMulticast(t)
	ifi := findMulticastInterface()
	ifaceName := ""
	if ifi != nil {
		ifaceName = ifi.Name
	}

	opts := MeshOptions{
		Port:      8000 + (int(time.Now().Unix()) % 1000), // Randomish port
		Group:     "239.0.0.1",
		Interface: ifaceName,
	}

	nodeA, err := NewMeshBus(opts)
	if err != nil {
		t.Fatalf("Failed to create nodeA: %v", err)
	}
	defer nodeA.Close()

	nodeB, err := NewMeshBus(opts)
	if err != nil {
		t.Fatalf("Failed to create nodeB: %v", err)
	}
	defer nodeB.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	key := "test-key"
	chB, err := nodeB.Subscribe(ctx, key)
	if err != nil {
		t.Fatalf("Failed to subscribe on nodeB: %v", err)
	}

	// Wait a bit for subscription to be ready
	time.Sleep(200 * time.Millisecond)

	if err := nodeA.Publish(ctx, key); err != nil {
		t.Fatalf("Failed to publish from nodeA: %v", err)
	}

	select {
	case evt := <-chB:
		if evt.Key != key {
			t.Errorf("Expected key %s, got %s", key, evt.Key)
		}
	case <-ctx.Done():
		t.Error("Timed out waiting for invalidation on nodeB")
	}
}

func TestMeshLoopback(t *testing.T)

{
	requireMulticast(t)
	ifi := findMulticastInterface()
	ifaceName := ""
	if ifi != nil {
		ifaceName = ifi.Name
	}

	opts := MeshOptions{
		Port:      8000 + (int(time.Now().Unix()) % 1000) + 1,
		Group:     "239.0.0.1",
		Interface: ifaceName,
	}

	node, err := NewMeshBus(opts)
	if err != nil {
		t.Fatalf("Failed to create node: %v", err)
	}
	defer node.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer cancel()

	key := "loop-key"
	ch, err := node.Subscribe(ctx, key)
	if err != nil {
		t.Fatalf("Failed to subscribe: %v", err)
	}

	if err := node.Publish(ctx, key); err != nil {
		t.Fatalf("Failed to publish: %v", err)
	}

	select {
	case <-ch:
		t.Error("Node received its own message (loopback should be filtered)")
	case <-ctx.Done():
		// Success
	}
}

func TestMeshGossip(t *testing.T)

{
	requireMulticast(t)
	portA := 9000 + (int(time.Now().Unix()) % 100)
	portB := portA + 1

	addrA := fmt.Sprintf("127.0.0.1:%d", portA)

	optsA := MeshOptions{
		Port:          portA,
		AdvertiseAddr: addrA,
		Heartbeat:     100 * time.Millisecond,
	}
	nodeA, err := NewMeshBus(optsA)
	if err != nil {
		t.Fatalf("Failed to create nodeA: %v", err)
	}
	defer nodeA.Close()

	optsB := MeshOptions{
		Port:          portB,
		AdvertiseAddr: fmt.Sprintf("127.0.0.1:%d", portB),
		Peers:         []string{addrA},
		Heartbeat:     100 * time.Millisecond,
	}
	nodeB, err := NewMeshBus(optsB)
	if err != nil {
		t.Fatalf("Failed to create nodeB: %v", err)
	}
	defer nodeB.Close()

	// Wait for gossip discovery
	found := false
	for i := 0; i < 20; i++ {
		peers := nodeA.Peers()
		for _, p := range peers {
			if p == optsB.AdvertiseAddr {
				found = true
				break
			}
		}
		if found {
			break
		}
		time.Sleep(100 * time.Millisecond)
	}

	if !found {
		t.Errorf("NodeA did not discover NodeB via gossip (Peers: %v)", nodeA.Peers())
	}
}

func NewMeshBus(opts MeshOptions) (*MeshBus, error)

{
	if opts.Port == 0 {
		opts.Port = 7946
	}
	if opts.Group == "" {
		opts.Group = "239.0.0.1"
	}

	addr, err := net.ResolveUDPAddr("udp4", fmt.Sprintf("%s:%d", opts.Group, opts.Port))
	if err != nil {
		return nil, fmt.Errorf("mesh: failed to resolve multicast address: %w", err)
	}

	// Configure socket to allow multiple listeners on the same port.
	lc := net.ListenConfig{
		Control: func(network, address string, c syscall.RawConn) error {
			var err error
			c.Control(func(fd uintptr) {
				_ = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_REUSEADDR, 1)
				_ = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, 15, 1)
			})
			return err
		},
	}

	c, err := lc.ListenPacket(context.Background(), "udp4", fmt.Sprintf("0.0.0.0:%d", opts.Port))
	if err != nil {
		return nil, fmt.Errorf("mesh: failed to listen on port %d: %w", opts.Port, err)
	}

	pconn := ipv4.NewPacketConn(c)

	var iface *net.Interface
	if opts.Interface != "" {
		iface, err = net.InterfaceByName(opts.Interface)
		if err != nil {
			_ = c.Close()
			return nil, fmt.Errorf("mesh: failed to find interface %s: %w", opts.Interface, err)
		}
	}

	if err := pconn.JoinGroup(iface, addr); err != nil {
		_ = c.Close()
		return nil, fmt.Errorf("mesh: failed to join group %s: %w", opts.Group, err)
	}

	if iface != nil {
		if err := pconn.SetMulticastInterface(iface); err != nil {
			_ = c.Close()
			return nil, fmt.Errorf("mesh: failed to set multicast interface: %w", err)
		}
	}

	// Enable loopback so multiple nodes on same host can hear each other.
	_ = pconn.SetMulticastLoopback(true)

	if opts.BatchInterval == 0 {
		opts.BatchInterval = 100 * time.Millisecond
	}
	if opts.BatchSize == 0 {
		opts.BatchSize = 20
	}

	ctx, cancel := context.WithCancel(context.Background())
	b := &MeshBus{
		opts:         opts,
		nodeID:       uuid.New(),
		conn:         c,
		pconn:        pconn,
		groupAddr:    addr,
		subs:         make(map[string][]chan syncbus.Event),
		knownPeers:   make(map[string]time.Time),
		resolvedAddr: make(map[string]*net.UDPAddr),
		publishCh:    make(chan string, 1000),
		pending:      make(map[string]struct{}),
		ctx:          ctx,
		cancel:       cancel,
	}

	go b.listen()
	go b.heartbeatLoop()
	go b.cleanupPeers()
	go b.runBatcher()

	return b, nil
}