package main

import (
	"strings"
	"testing"
	"time"

	"github.com/novox/mesh-controller/internal/broker"
)

// holding reports one message held by each named consumer (stream/name), nothing by the rest.
func holdingOne(names ...string) func(broker.Consumer) (uint64, error) {
	return func(c broker.Consumer) (uint64, error) {
		for _, n := range names {
			if n == c.Stream+"/"+c.Name {
				return 1, nil
			}
		}
		return 0, nil
	}
}

func deafObservations(now time.Time, held func(broker.Consumer) (uint64, error)) (string, int) {
	f := calm(now)
	loop, err := readLoop(now.Add(-loopDeafAfter-time.Minute), loopConsumers(), held)
	if err != nil {
		panic(err)
	}
	f.loop = loop
	obs := watchLoop(f)
	if len(obs) == 0 {
		return "", 0
	}
	return obs[0].Said, len(obs)
}

// 1063/AC1: a check in flight on a builder is not the controller's event loop being deaf.
func TestDeafLookIgnoresCheckWorkers(t *testing.T) {
	now := time.Now()
	for _, w := range broker.CheckWorkers() {
		if said, n := deafObservations(now, holdingOne(w.Stream+"/"+w.Name)); n != 0 {
			t.Fatalf("%s holding one message raised controller-deaf: %s", w.Name, said)
		}
	}
}

// 1063/AC2: a message held by one of the controller's own consumers is still deaf, naming its stream.
func TestDeafLookStillCountsOwnConsumers(t *testing.T) {
	now := time.Now()
	for _, c := range loopConsumers() {
		said, n := deafObservations(now, holdingOne(c.Stream+"/"+c.Name))
		if n != 1 || !strings.Contains(said, "1 on "+c.Stream) {
			t.Fatalf("%s/%s holding one message: %d observation(s), said %q", c.Stream, c.Name, n, said)
		}
	}
}

// 1063/AC3: the list is the controller's consumers and none of the check workers.
func TestDeafLookReadsExactlyTheControllersOwnConsumers(t *testing.T) {
	workers := map[string]bool{}
	for _, w := range broker.CheckWorkers() {
		workers[w.Stream+"/"+w.Name] = true
	}
	want := map[string]bool{}
	for _, c := range broker.MeshConsumers() {
		if !workers[c.Stream+"/"+c.Name] {
			want[c.Stream+"/"+c.Name] = true
		}
	}
	got := map[string]bool{}
	for _, c := range loopConsumers() {
		k := c.Stream + "/" + c.Name
		if workers[k] {
			t.Errorf("the deaf look reads the check worker %s", k)
		}
		got[k] = true
	}
	if len(got) != len(want) {
		t.Fatalf("deaf look reads %v, want %v", got, want)
	}
	for k := range want {
		if !got[k] {
			t.Errorf("the deaf look does not read %s", k)
		}
	}
}