Maven/internal/event/event_test.go

package event

import (
	"strings"
	"sync"
	"testing"
	"time"
)

var testNow = time.Date(2026, 8, 1, 9, 30, 0, 0, time.UTC)

func TestNormalizeFillsDefaults(t *testing.T) {
	got := Event{Source: "poll:zenmoney", Title: "  spent   today  "}.Normalize(testNow)
	if got.Title != "spent today" {
		t.Errorf("title = %q, want collapsed whitespace", got.Title)
	}
	if got.Priority != PriorityNormal {
		t.Errorf("priority = %q, want %q", got.Priority, PriorityNormal)
	}
	if got.Kind != KindFact {
		t.Errorf("kind = %q, want %q", got.Kind, KindFact)
	}
	if !got.OccurredAt.Equal(testNow) {
		t.Errorf("occurred_at = %v, want %v", got.OccurredAt, testNow)
	}
}

func TestNormalizeKeepsRealOccurredAt(t *testing.T) {
	// A wg handshake carries the handshake instant, not "now". Flattening that
	// would make every intake look like it happened at notice time.
	real := testNow.Add(-3 * time.Hour)
	got := Event{Source: "infer:wg", Title: "wg_handshake", OccurredAt: real}.Normalize(testNow)
	if !got.OccurredAt.Equal(real) {
		t.Errorf("occurred_at = %v, want the supplied %v", got.OccurredAt, real)
	}
}

func TestNormalizeTruncatesOnRuneBoundary(t *testing.T) {
	long := strings.Repeat("я", TitleMaxRunes+50)
	got := Event{Source: "rss:x", Title: long}.Normalize(testNow)
	r := []rune(got.Title)
	if len(r) != TitleMaxRunes+1 { // +1 for the ellipsis marker
		t.Fatalf("title runes = %d, want %d", len(r), TitleMaxRunes+1)
	}
	if r[len(r)-1] != '…' {
		t.Errorf("truncated title does not mark the cut: %q", string(r[len(r)-3:]))
	}
	for _, c := range r[:TitleMaxRunes] {
		if c != 'я' {
			t.Fatalf("truncation broke a rune: got %q", c)
		}
	}
}

func TestNormalizeRejectsUnknownPriority(t *testing.T) {
	got := Event{Source: "s", Title: "t", Priority: "URGENT!!"}.Normalize(testNow)
	if got.Priority != PriorityNormal {
		t.Errorf("priority = %q, want %q", got.Priority, PriorityNormal)
	}
}

func TestValid(t *testing.T) {
	base := Event{Source: "rss:tech", Kind: KindNote, Title: "заголовок", OccurredAt: testNow}
	if !base.Valid() {
		t.Fatal("well-formed event reported invalid")
	}
	for name, mut := range map[string]func(Event) Event{
		"no source": func(e Event) Event { e.Source = ""; return e },
		"no title":  func(e Event) Event { e.Title = ""; return e },
		"no time":   func(e Event) Event { e.OccurredAt = time.Time{}; return e },
		"bad kind":  func(e Event) Event { e.Kind = "whatever"; return e },
	} {
		if mut(base).Valid() {
			t.Errorf("%s: reported valid", name)
		}
	}
}

func TestSourceKind(t *testing.T) {
	cases := map[string]string{
		"rss:tech":     KindNote,
		"crawl:kernel": KindNote,
		// No email rule: a WriteFact under an email source is a fact, not a
		// captured task. The mail path builds its own task envelope.
		"email:inbox":   KindFact,
		"probe:netdata": KindHealth,
		"ambient:notif": KindFact,
		"tap:voice":     KindFact,
	}
	for src, want := range cases {
		if got := SourceKind(src, KindFact); got != want {
			t.Errorf("SourceKind(%q) = %q, want %q", src, got, want)
		}
	}
}

func TestBusNilIsANoOp(t *testing.T) {
	// The whole adoption story depends on this: an intake path calls Publish
	// unconditionally, and a daemon with no bus behaves as it did before.
	var b *Bus
	b.Publish(Event{Source: "s", Kind: KindFact, Title: "t"}, testNow)
	b.Subscribe(func(Event) { t.Error("nil bus delivered to a subscriber") })
	if got := b.Recent(10); got != nil {
		t.Errorf("Recent on nil bus = %v, want nil", got)
	}
	if got := b.Len(); got != 0 {
		t.Errorf("Len on nil bus = %d, want 0", got)
	}
}

func TestBusRecentIsNewestFirst(t *testing.T) {
	b := NewBus(8)
	for _, title := range []string{"one", "two", "three"} {
		b.Publish(Event{Source: "rss:t", Kind: KindNote, Title: title}, testNow)
	}
	got := b.Recent(0)
	if len(got) != 3 {
		t.Fatalf("len = %d, want 3", len(got))
	}
	want := []string{"three", "two", "one"}
	for i, w := range want {
		if got[i].Title != w {
			t.Errorf("Recent()[%d] = %q, want %q", i, got[i].Title, w)
		}
	}
	if lim := b.Recent(2); len(lim) != 2 || lim[0].Title != "three" {
		t.Errorf("Recent(2) = %v, want the two newest", lim)
	}
}

func TestBusRingEvicts(t *testing.T) {
	b := NewBus(3)
	for _, title := range []string{"a", "b", "c", "d", "e"} {
		b.Publish(Event{Source: "s", Kind: KindFact, Title: title}, testNow)
	}
	if b.Len() != 3 {
		t.Fatalf("Len = %d, want the capacity 3", b.Len())
	}
	got := b.Recent(0)
	want := []string{"e", "d", "c"}
	for i, w := range want {
		if got[i].Title != w {
			t.Errorf("Recent()[%d] = %q, want %q", i, got[i].Title, w)
		}
	}
}

func TestBusDropsInvalid(t *testing.T) {
	b := NewBus(4)
	b.Publish(Event{Kind: KindFact, Title: "no source"}, testNow)
	b.Publish(Event{Source: "s", Kind: KindFact}, testNow)
	if b.Len() != 0 {
		t.Errorf("Len = %d, want 0 — an envelope with no provenance must not be kept", b.Len())
	}
}

func TestBusSubscriberPanicDoesNotBreakIntake(t *testing.T) {
	b := NewBus(4)
	var seen int
	b.Subscribe(func(Event) { panic("observer is broken") })
	b.Subscribe(func(Event) { seen++ })
	b.Publish(Event{Source: "s", Kind: KindFact, Title: "t"}, testNow)
	if seen != 1 {
		t.Errorf("healthy subscriber called %d times, want 1", seen)
	}
	if b.Len() != 1 {
		t.Errorf("event not recorded despite a panicking subscriber")
	}
}

func TestBusConcurrentPublish(t *testing.T) {
	b := NewBus(256)
	var wg sync.WaitGroup
	for i := 0; i < 16; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := 0; j < 10; j++ {
				b.Publish(Event{Source: "s", Kind: KindFact, Title: "t"}, testNow)
			}
		}()
	}
	wg.Wait()
	if b.Len() != 160 {
		t.Errorf("Len = %d, want 160", b.Len())
	}
}

// The ring is insertion-ordered and the page calls itself newest first, so the
// two only agree if the column is notice time. A cold feed read publishes a
// week of items in feed order, which used to make the OccurredAt column walk
// forwards and backwards on the same page.
func TestRecentIsOrderedByNoticeTimeNotByWhenThingsHappened(t *testing.T) {
	b := NewBus(8)
	// Published in feed order, six days old first, and a mark stamped now.
	for i, e := range []Event{
		{Source: "rss:t", Kind: KindNote, Title: "six days ago", OccurredAt: testNow.Add(-6 * 24 * time.Hour)},
		{Source: "rss:t", Kind: KindNote, Title: "two days ago", OccurredAt: testNow.Add(-2 * 24 * time.Hour)},
		{Source: "ambient:notif", Kind: KindFact, Title: "a meeting at six", OccurredAt: testNow.Add(9 * time.Hour)},
	} {
		b.Publish(e, testNow.Add(time.Duration(i)*time.Second))
	}
	got := b.Recent(0)
	want := []string{"a meeting at six", "two days ago", "six days ago"}
	for i, w := range want {
		if got[i].Title != w {
			t.Errorf("Recent()[%d] = %q, want %q (notice order)", i, got[i].Title, w)
		}
	}
	// Notice time is monotone down the page even where occurrence time is not.
	for i := 1; i < len(got); i++ {
		if got[i].NoticedAt.After(got[i-1].NoticedAt) {
			t.Errorf("NoticedAt is not descending at %d", i)
		}
	}
	if got[0].OccurredAt.Before(got[1].OccurredAt) {
		t.Fatal("this fixture is supposed to have occurrence time out of order")
	}
	// A caller does not get to claim when Maven noticed something.
	forged := Event{Source: "s", Kind: KindFact, Title: "t", NoticedAt: testNow.Add(100 * time.Hour)}
	b.Publish(forged, testNow)
	if n := b.Recent(1)[0].NoticedAt; !n.Equal(testNow) {
		t.Errorf("NoticedAt = %v, want the publish instant %v", n, testNow)
	}
}