package builtin

import (
	"context"
	"encoding/json"
	"strings"
	"testing"
	"time"
	"unicode/utf8"

	"github.com/denn-gubsky/loomcycle/internal/channels"
	"github.com/denn-gubsky/loomcycle/internal/providers"
	"github.com/denn-gubsky/loomcycle/internal/store"
	"github.com/denn-gubsky/loomcycle/internal/store/sqlite"
	"github.com/denn-gubsky/loomcycle/internal/tools"
)

// channelFixture builds a Channel tool over an in-memory SQLite store
// + a fresh Bus. The returned ctx has a sensible agent name + user_id
// + a default policy granting both publish and subscribe on
// "findings" (agent-scoped, queue, no TTL) and "alerts" (global,
// broadcast, 1h TTL). Tests override the policy when they want to
// exercise refusals.
func channelFixture(t *testing.T) (*Channel, context.Context, func()) {
	t.Helper()
	s, err := sqlite.Open(":memory:")
	if err != nil {
		t.Fatalf("sqlite.Open: %v", err)
	}
	tool := &Channel{
		Store:         s,
		Bus:           channels.NewBus(),
		MaxValueBytes: 65536,
		LongPollCapMS: 30000,
	}
	ctx := tools.WithAgentName(context.Background(), "researcher")
	ctx = tools.WithRunIdentity(ctx, tools.RunIdentityValue{UserID: "alice", AgentID: "a_test"})
	ctx = tools.WithChannelPolicy(ctx, tools.ChannelPolicyValue{
		Publish:   []string{"findings", "alerts"},
		Subscribe: []string{"findings", "alerts", "findings/*"},
		Channels: map[string]tools.ChannelDef{
			"findings":       {Name: "findings", Scope: "agent", MaxMessages: 1000, Semantic: "queue"},
			"alerts":         {Name: "alerts", Scope: "global", DefaultTTL: 3600, Semantic: "broadcast"},
			"findings/alpha": {Name: "findings/alpha", Scope: "agent", Semantic: "queue"},
		},
	})
	return tool, ctx, func() { _ = s.Close() }
}

func TestChannelTool_PublishSubscribeRoundTrip(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()

	for i := 0; i < 3; i++ {
		input := `{"op":"publish","channel":"findings","value":{"i":` + intToStr(i) + `}}`
		res, _ := tool.Execute(ctx, json.RawMessage(input))
		if res.IsError {
			t.Fatalf("publish %d: %s", i, res.Text)
		}
		time.Sleep(time.Microsecond)
	}
	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"subscribe","channel":"findings","max_messages":10}`))
	if res.IsError {
		t.Fatalf("subscribe: %s", res.Text)
	}
	// Check basic envelope structure — three messages + next_cursor.
	got := decodeResult(t, res.Text)
	msgs := got["messages"].([]any)
	if len(msgs) != 3 {
		t.Errorf("got %d msgs, want 3", len(msgs))
	}
	if next, _ := got["next_cursor"].(string); next == "" {
		t.Errorf("next_cursor empty in %s", res.Text)
	}
}

func TestChannelTool_PublishRefusedWithoutACL(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()
	// Replace policy: agent may subscribe but NOT publish to findings.
	ctx = tools.WithChannelPolicy(ctx, tools.ChannelPolicyValue{
		Publish:   []string{}, // no publish allowed
		Subscribe: []string{"findings"},
		Channels: map[string]tools.ChannelDef{
			"findings": {Name: "findings", Scope: "agent"},
		},
	})
	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"findings","value":{"x":1}}`))
	if !res.IsError {
		t.Fatalf("publish should be refused; got %s", res.Text)
	}
	if !strings.Contains(res.Text, "publish") {
		t.Errorf("refusal should mention publish; got %s", res.Text)
	}
}

func TestChannelTool_UnknownChannelRefused(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()
	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"nonexistent","value":{}}`))
	if !res.IsError {
		t.Fatal("publish to undeclared channel must be refused")
	}
	if !strings.Contains(res.Text, "not declared") {
		t.Errorf("refusal should mention 'not declared'; got %s", res.Text)
	}
}

func TestChannelTool_WildcardAllowlist(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()
	// findings/* is in subscribe allowlist, and findings/alpha is a
	// declared channel. Subscribe should succeed.
	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"subscribe","channel":"findings/alpha","max_messages":10}`))
	if res.IsError {
		t.Errorf("subscribe to findings/alpha should match findings/* wildcard; got %s", res.Text)
	}
}

func TestChannelTool_GlobalScopeChannel(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()
	// Publish to global-scope alerts. Even without a user_id on the
	// ctx, global publish should succeed (scope_id = "").
	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"alerts","value":{"level":"warn"}}`))
	if res.IsError {
		t.Fatalf("global publish: %s", res.Text)
	}
	// Subscribe should return that message.
	res, _ = tool.Execute(ctx, json.RawMessage(`{"op":"subscribe","channel":"alerts","max_messages":10}`))
	got := decodeResult(t, res.Text)
	msgs := got["messages"].([]any)
	if len(msgs) != 1 {
		t.Errorf("global subscribe got %d msgs, want 1", len(msgs))
	}
}

func TestChannelTool_AckMissingCursorErrors(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()
	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"ack","channel":"findings"}`))
	if !res.IsError {
		t.Error("ack without cursor should error")
	}
}

func TestChannelTool_PeekDoesNotAdvance(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()
	_, _ = tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"findings","value":{"x":1}}`))
	// Peek (no auto-ack).
	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"peek","channel":"findings","max_messages":10}`))
	if res.IsError {
		t.Fatal(res.Text)
	}
	// Subscribe should STILL see the message (peek didn't advance).
	res, _ = tool.Execute(ctx, json.RawMessage(`{"op":"subscribe","channel":"findings","max_messages":10}`))
	got := decodeResult(t, res.Text)
	msgs := got["messages"].([]any)
	if len(msgs) != 1 {
		t.Errorf("after peek, subscribe got %d msgs, want 1 (peek must not advance)", len(msgs))
	}
}

func TestChannelTool_ListChannelsReportsPolicy(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()
	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"list_channels"}`))
	if res.IsError {
		t.Fatal(res.Text)
	}
	if !strings.Contains(res.Text, `"publish"`) || !strings.Contains(res.Text, `"subscribe"`) {
		t.Errorf("list_channels should report publish + subscribe; got %s", res.Text)
	}
}

func TestChannelTool_InvalidJSONValueErrors(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()
	// value is a malformed JSON literal (unquoted identifier)
	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"findings","value":not_json}`))
	if !res.IsError {
		t.Error("invalid value json should error")
	}
}

func TestChannelTool_MaxValueBytesEnforced(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()
	tool.MaxValueBytes = 32
	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"findings","value":{"x":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}}`))
	if !res.IsError {
		t.Error("oversized payload should error")
	}
	if !strings.Contains(res.Text, "exceeds max") {
		t.Errorf("refusal should mention 'exceeds max'; got %s", res.Text)
	}
}

// ---- helpers ----

func decodeResult(t *testing.T, body string) map[string]any {
	t.Helper()
	var out map[string]any
	if err := json.Unmarshal([]byte(body), &out); err != nil {
		t.Fatalf("decode result %q: %v", body, err)
	}
	return out
}

func intToStr(i int) string {
	if i == 0 {
		return "0"
	}
	out := []byte{}
	for i > 0 {
		out = append([]byte{byte('0' + i%10)}, out...)
		i /= 10
	}
	return string(out)
}

// ---- v0.8.4 typed-audit-event tests ----

// captureEvents returns an EventEmitter that records every event
// into a slice so tests can assert on the structured payload.
func captureEvents() (tools.EventEmitterFunc, *[]providers.Event) {
	var got []providers.Event
	emit := func(ev providers.Event) { got = append(got, ev) }
	return emit, &got
}

// TestChannelTool_PublishEmitsTypedEvent pins that a successful
// Channel.publish surfaces EventChannelPublish with the full
// structured payload (channel id, scope, message_id, byte size,
// truncated preview).
func TestChannelTool_PublishEmitsTypedEvent(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()
	emit, got := captureEvents()
	ctx = tools.WithEventEmitter(ctx, emit)

	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"findings","value":{"i":42}}`))
	if res.IsError {
		t.Fatalf("publish: %s", res.Text)
	}

	var pubs []providers.Event
	for _, ev := range *got {
		if ev.Type == providers.EventChannelPublish {
			pubs = append(pubs, ev)
		}
	}
	if len(pubs) != 1 {
		t.Fatalf("got %d EventChannelPublish, want 1", len(pubs))
	}
	info := pubs[0].Channel
	if info == nil {
		t.Fatal("Event.Channel nil; want non-nil ChannelEventInfo")
	}
	if info.Channel != "findings" {
		t.Errorf("Channel = %q, want findings", info.Channel)
	}
	if info.MessageID == "" {
		t.Error("MessageID empty")
	}
	if info.Scope != "agent" {
		t.Errorf("Scope = %q, want agent", info.Scope)
	}
	if info.ScopeID != "researcher" {
		t.Errorf("ScopeID = %q, want researcher (the fixture's agent name)", info.ScopeID)
	}
	if info.PayloadBytes == 0 {
		t.Error("PayloadBytes = 0")
	}
	if !strings.Contains(info.PayloadPreview, `"i":42`) {
		t.Errorf("PayloadPreview = %q, want contains `\"i\":42`", info.PayloadPreview)
	}
	if info.DroppedOldest != 0 {
		t.Errorf("DroppedOldest = %d, want 0 on the first publish", info.DroppedOldest)
	}
}

// TestChannelTool_SubscribeEmitsOneDeliveryPerMessage pins that
// each message in a returned batch surfaces its own
// EventChannelDelivery, in order, with Cursor populated.
func TestChannelTool_SubscribeEmitsOneDeliveryPerMessage(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()
	emit, got := captureEvents()
	ctx = tools.WithEventEmitter(ctx, emit)

	for i := 0; i < 3; i++ {
		_, _ = tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"findings","value":{"i":`+intToStr(i)+`}}`))
		time.Sleep(time.Microsecond)
	}

	// Reset captured events so the subscribe phase is clean.
	*got = nil

	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"subscribe","channel":"findings","max_messages":10}`))
	if res.IsError {
		t.Fatalf("subscribe: %s", res.Text)
	}

	var deliveries []providers.Event
	for _, ev := range *got {
		if ev.Type == providers.EventChannelDelivery {
			deliveries = append(deliveries, ev)
		}
	}
	if len(deliveries) != 3 {
		t.Fatalf("got %d EventChannelDelivery, want 3 (one per message)", len(deliveries))
	}
	for i, ev := range deliveries {
		if ev.Channel == nil {
			t.Fatalf("delivery[%d].Channel nil", i)
		}
		if ev.Channel.MessageID == "" {
			t.Errorf("delivery[%d].MessageID empty", i)
		}
		if ev.Channel.Cursor != ev.Channel.MessageID {
			t.Errorf("delivery[%d].Cursor = %q, want MessageID %q (cursor = id of THIS msg)",
				i, ev.Channel.Cursor, ev.Channel.MessageID)
		}
		if ev.Channel.Channel != "findings" {
			t.Errorf("delivery[%d].Channel = %q, want findings", i, ev.Channel.Channel)
		}
		if i > 0 && deliveries[i].Channel.MessageID <= deliveries[i-1].Channel.MessageID {
			t.Errorf("delivery[%d].MessageID not strictly increasing: %q vs %q",
				i, deliveries[i-1].Channel.MessageID, deliveries[i].Channel.MessageID)
		}
	}
}

// TestChannelTool_NoEmitterIsSilent pins that the no-ctx-emitter
// path is a no-op — tools called without WithEventEmitter still
// work correctly, just without surfacing typed events.
func TestChannelTool_NoEmitterIsSilent(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()
	// No WithEventEmitter applied — fixture ctx has no emitter.

	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"findings","value":{}}`))
	if res.IsError {
		t.Errorf("publish with no emitter should still succeed: %s", res.Text)
	}
}

// TestChannelTool_PayloadPreviewIsValidUTF8AfterTruncation pins
// the UTF-8 safety contract: truncating a payload whose
// multi-byte rune straddles the byte cut MUST stop at the
// previous rune boundary, never produce invalid UTF-8.
//
// Without this guarantee, JSON consumers (the TS adapter's
// TextDecoder; strict json.Unmarshal in Go) would reject the
// preview as malformed.
func TestChannelTool_PayloadPreviewIsValidUTF8AfterTruncation(t *testing.T) {
	// 3-byte UTF-8 character (CJK). 100 of them = 300 bytes.
	// With max=200, naive byte-slicing would cut in the middle of
	// rune 67 (byte index 198..200 holds half of a 3-byte char).
	payload := strings.Repeat("漢", 100)
	got := truncateForEvent(payload, 200)
	if !utf8.ValidString(got) {
		t.Fatalf("truncated preview is not valid UTF-8: %q", got)
	}
	if !strings.HasSuffix(got, "…") {
		t.Errorf("expected ellipsis suffix on truncated preview; got %q", got[len(got)-5:])
	}
	// Should hold floor(200/3) = 66 runes + ellipsis = 67 runes,
	// 66*3 = 198 bytes + 3-byte ellipsis = 201 bytes total.
	if len(got) > 203 {
		t.Errorf("truncated length = %d bytes; want <= 203", len(got))
	}
}

// TestChannelTool_PayloadPreviewTruncatedAt200Chars pins that
// large payloads don't blow up the SSE wire — the preview is
// capped at 200 chars + an ellipsis suffix.
func TestChannelTool_PayloadPreviewTruncatedAt200Chars(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()
	emit, got := captureEvents()
	ctx = tools.WithEventEmitter(ctx, emit)

	// Construct a payload whose JSON-encoded form is well over 200 chars.
	big := strings.Repeat("a", 500)
	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"findings","value":"`+big+`"}`))
	if res.IsError {
		t.Fatalf("publish: %s", res.Text)
	}

	var info *providers.ChannelEventInfo
	for _, ev := range *got {
		if ev.Type == providers.EventChannelPublish {
			info = ev.Channel
			break
		}
	}
	if info == nil {
		t.Fatal("no EventChannelPublish captured")
	}
	if !strings.HasSuffix(info.PayloadPreview, "…") {
		t.Errorf("PayloadPreview should end with ellipsis when truncated; got %q", info.PayloadPreview[len(info.PayloadPreview)-10:])
	}
	// 200 chars + UTF-8 ellipsis (3 bytes). The rune count is 201; byte
	// length is 203. Either property pins the cap; we go with byte
	// length for simplicity since that's what len() returns.
	if len(info.PayloadPreview) > 203 {
		t.Errorf("PayloadPreview length = %d, want <= 203 (200 chars + 3-byte ellipsis)", len(info.PayloadPreview))
	}
	// PayloadBytes should reflect the FULL untruncated payload size.
	if info.PayloadBytes < 500 {
		t.Errorf("PayloadBytes = %d, want >= 500 (full payload size, not truncated)", info.PayloadBytes)
	}
}

// ---- v0.8.6 deferred publish ----

func TestChannelTool_DeferredPublishHiddenUntilVisible(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()
	tool.Scheduler = channels.NewScheduler(tool.Bus, 100)

	deferTo := time.Now().Add(150 * time.Millisecond).UTC().Format(time.RFC3339Nano)
	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"findings","value":{"k":"v"},"deliver_at":"`+deferTo+`"}`))
	if res.IsError {
		t.Fatalf("publish: %s", res.Text)
	}
	out := decodeResult(t, res.Text)
	if _, ok := out["visible_at"]; !ok {
		t.Errorf("deferred publish result missing visible_at: %v", out)
	}

	res, _ = tool.Execute(ctx, json.RawMessage(`{"op":"subscribe","channel":"findings","max_messages":10}`))
	out = decodeResult(t, res.Text)
	if msgs, _ := out["messages"].([]any); len(msgs) != 0 {
		t.Errorf("deferred message visible too early: got %d", len(msgs))
	}

	time.Sleep(200 * time.Millisecond)
	res, _ = tool.Execute(ctx, json.RawMessage(`{"op":"subscribe","channel":"findings","max_messages":10}`))
	out = decodeResult(t, res.Text)
	if msgs, _ := out["messages"].([]any); len(msgs) != 1 {
		t.Errorf("after visible_at: got %d messages, want 1", len(msgs))
	}
}

func TestChannelTool_DeferredAtRejectsBadFormat(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()

	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"findings","value":{},"deliver_at":"not-a-timestamp"}`))
	if !res.IsError {
		t.Fatalf("expected error on bad deliver_at; got %s", res.Text)
	}
	if !strings.Contains(res.Text, "deliver_at") {
		t.Errorf("error should mention deliver_at; got %q", res.Text)
	}
}

func TestChannelTool_DeferredAtInPastTreatedAsNow(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()

	past := time.Now().Add(-10 * time.Second).UTC().Format(time.RFC3339Nano)
	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"findings","value":{"k":"v"},"deliver_at":"`+past+`"}`))
	if res.IsError {
		t.Fatalf("publish: %s", res.Text)
	}
	out := decodeResult(t, res.Text)
	// Past deliver_at is treated as immediate — no visible_at field on result.
	if _, ok := out["visible_at"]; ok {
		t.Errorf("past deliver_at should not produce visible_at field on result: %v", out)
	}

	res, _ = tool.Execute(ctx, json.RawMessage(`{"op":"subscribe","channel":"findings","max_messages":10}`))
	out = decodeResult(t, res.Text)
	if msgs, _ := out["messages"].([]any); len(msgs) != 1 {
		t.Errorf("past deliver_at: got %d, want 1 (immediate)", len(msgs))
	}
}

func TestChannelTool_PublishedByUserIDFromCtx(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()

	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"findings","value":{"k":"v"}}`))
	if res.IsError {
		t.Fatalf("publish: %s", res.Text)
	}
	// Read back via subscribe + verify the store's row carries user_id.
	msgs, _, err := tool.Store.ChannelSubscribe(ctx, "", "findings", store.MemoryScopeAgent, "researcher", "", 10)
	if err != nil || len(msgs) != 1 {
		t.Fatalf("read back: msgs=%d err=%v", len(msgs), err)
	}
	if msgs[0].PublishedByUserID != "alice" {
		t.Errorf("PublishedByUserID = %q, want alice (from ctx)", msgs[0].PublishedByUserID)
	}
}

func TestChannelTool_AckRejectsLegacyCursorFormat(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()

	// Publish + try to ack with the v0.8.4-shaped cursor (raw msg_id).
	_, _ = tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"findings","value":{}}`))
	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"ack","channel":"findings","cursor":"msg_18aec3225c5a78b03dd59623"}`))
	if !res.IsError {
		t.Fatalf("legacy cursor should be rejected; got %s", res.Text)
	}
	if !strings.Contains(res.Text, "invalid channel cursor") {
		t.Errorf("error should explain invalid cursor; got %q", res.Text)
	}
}

// ---- v0.8.6 system-channels refusals ----

func TestChannelTool_AgentRefusedOnPublisherSystemChannel(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()

	// Override policy with a `publisher: system` channel.
	ctx = tools.WithChannelPolicy(ctx, tools.ChannelPolicyValue{
		Publish:   []string{"_system/heartbeat-1m"},
		Subscribe: []string{"_system/heartbeat-1m"},
		Channels: map[string]tools.ChannelDef{
			"_system/heartbeat-1m": {
				Name:      "_system/heartbeat-1m",
				Scope:     "global",
				Publisher: "system",
			},
		},
	})

	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"_system/heartbeat-1m","value":{}}`))
	if !res.IsError {
		t.Fatalf("expected refusal on publisher:system channel; got %s", res.Text)
	}
	if !strings.Contains(res.Text, "publisher: system") {
		t.Errorf("error should mention publisher: system; got %q", res.Text)
	}
}

func TestChannelTool_AgentRefusedOnSystemPrefix(t *testing.T) {
	tool, ctx, cleanup := channelFixture(t)
	defer cleanup()

	// Even if the operator forgot to set publisher: system but
	// declared a `_system/...` channel, the prefix itself blocks
	// agent publishes.
	ctx = tools.WithChannelPolicy(ctx, tools.ChannelPolicyValue{
		Publish: []string{"_system/alarms/info"},
		Channels: map[string]tools.ChannelDef{
			"_system/alarms/info": {
				Name:  "_system/alarms/info",
				Scope: "global",
				// Publisher empty — but the prefix should still refuse.
			},
		},
	})

	res, _ := tool.Execute(ctx, json.RawMessage(`{"op":"publish","channel":"_system/alarms/info","value":{}}`))
	if !res.IsError {
		t.Fatalf("expected refusal on _system/ prefix; got %s", res.Text)
	}
	if !strings.Contains(res.Text, "_system/") {
		t.Errorf("error should mention _system/ prefix; got %q", res.Text)
	}
}

// TestReadWithRetry covers the residual-race defensive retry that
// runs after Bus.Wait fires the waker. Three scenarios:
//
//  1. read succeeds on the first attempt — no retry cost.
//  2. read returns empty initially, then non-empty on attempt 2 —
//     retry saves the subscriber from a silent miss.
//  3. read returns empty on all 3 attempts — the function returns
//     empty (current behaviour); the caller decides.
//
// This is the safety net for the unexplained ~2% subscribe miss
// observed in x50 load testing: a publish committed, Bus.Notify
// fired, the waker woke, but an immediate re-read returned empty.
// The retry covers any commit/visibility window without making
// strong claims about the root mechanism.
func TestReadWithRetry(t *testing.T) {
	makeMsg := func(id string) store.ChannelMessage {
		return store.ChannelMessage{ID: id}
	}

	t.Run("succeeds_on_first_attempt", func(t *testing.T) {
		calls := 0
		read := func() ([]store.ChannelMessage, string, error) {
			calls++
			return []store.ChannelMessage{makeMsg("m1")}, "cur_x", nil
		}
		msgs, next, err := readWithRetry(read, "test", retryDiagnostics{})
		if err != nil {
			t.Fatal(err)
		}
		if len(msgs) != 1 || msgs[0].ID != "m1" {
			t.Errorf("msgs = %v, want [{m1}]", msgs)
		}
		if next != "cur_x" {
			t.Errorf("next = %q, want cur_x", next)
		}
		if calls != 1 {
			t.Errorf("calls = %d, want 1 (no retries needed)", calls)
		}
	})

	t.Run("retry_finds_message_on_second_attempt", func(t *testing.T) {
		calls := 0
		read := func() ([]store.ChannelMessage, string, error) {
			calls++
			if calls == 1 {
				return nil, "", nil // initial empty (the race)
			}
			return []store.ChannelMessage{makeMsg("m2")}, "cur_y", nil
		}
		msgs, _, err := readWithRetry(read, "test", retryDiagnostics{})
		if err != nil {
			t.Fatal(err)
		}
		if len(msgs) != 1 || msgs[0].ID != "m2" {
			t.Errorf("retry should have returned the message; got %v", msgs)
		}
		if calls != 2 {
			t.Errorf("calls = %d, want 2", calls)
		}
	})

	t.Run("all_three_attempts_empty_returns_empty", func(t *testing.T) {
		calls := 0
		read := func() ([]store.ChannelMessage, string, error) {
			calls++
			return nil, "", nil // always empty
		}
		msgs, next, err := readWithRetry(read, "test", retryDiagnostics{})
		if err != nil {
			t.Fatal(err)
		}
		if len(msgs) != 0 {
			t.Errorf("msgs should be empty, got %v", msgs)
		}
		if next != "" {
			t.Errorf("next should be empty, got %q", next)
		}
		if calls != 3 {
			t.Errorf("calls = %d, want 3 (full retry budget exhausted)", calls)
		}
	})

	t.Run("error_propagates_without_retry", func(t *testing.T) {
		calls := 0
		read := func() ([]store.ChannelMessage, string, error) {
			calls++
			return nil, "", errSentinel
		}
		_, _, err := readWithRetry(read, "test", retryDiagnostics{})
		if err != errSentinel {
			t.Errorf("err = %v, want %v", err, errSentinel)
		}
		if calls != 1 {
			t.Errorf("calls = %d, want 1 (errors don't retry)", calls)
		}
	})
}

// errSentinel — fixed sentinel error for the propagate-without-retry
// subtest above. Identity comparison only.
var errSentinel = sentinelErr{}

type sentinelErr struct{}

func (sentinelErr) Error() string { return "sentinel" }
