fenrir/internal/notifications/shared_test.go

// Copyright (c) 2026 Storm Dragon
// SPDX-License-Identifier: GPL-3.0-or-later

package notifications

import (
	"context"
	"encoding/json"
	"fmt"
	"net"
	"os"
	"os/exec"
	"reflect"
	"strings"
	"sync"
	"testing"
	"time"
)

func test_client(t *testing.T, namespace string) *Client {
	t.Helper()
	c := New(context.Background(), os.Geteuid(), namespace)
	t.Cleanup(c.Close)
	return c
}

func wait_history(t *testing.T, c *Client, want int) Snapshot {
	t.Helper()
	deadline := time.NewTimer(5 * time.Second)
	defer deadline.Stop()
	for {
		if state := c.Snapshot(); len(state.Entries) == want {
			return state
		}
		select {
		case <-c.Updates():
		case <-deadline.C:
			t.Fatalf("history has %d entries, want %d", len(c.Snapshot().Entries), want)
		}
	}
}

func TestSharedHistoryConcurrentPublishingBoundAndSingleCue(t *testing.T) {
	namespace := fmt.Sprintf(".%x", random_id())
	var clients []*Client
	for range 4 {
		clients = append(clients, test_client(t, namespace))
	}
	var workers sync.WaitGroup
	for index, client := range clients {
		workers.Go(func() {
			for count := range 30 {
				client.Publish(fmt.Sprintf("Reader %d message %d", index, count), time.Now())
			}
		})
	}
	workers.Wait()
	for _, client := range clients {
		wait_history(t, client, Limit)
	}
	// Reconcile after reaching the limit: all replicas must agree on the same
	// newest 100 IDs rather than just having the same length.
	for range 3 {
		for _, client := range clients {
			if _, err := client.exchange(context.Background(), false); err != nil {
				t.Fatal(err)
			}
		}
	}
	want := clients[0].Snapshot().Entries
	for _, client := range clients[1:] {
		if !reflect.DeepEqual(client.Snapshot().Entries, want) {
			t.Fatal("readers disagree about bounded history")
		}
	}
	accepted := make(chan bool, len(clients))
	for _, client := range clients {
		workers.Go(func() { accepted <- client.Claim_sound(context.Background()) })
	}
	workers.Wait()
	close(accepted)
	count := 0
	for allowed := range accepted {
		if allowed {
			count++
		}
	}
	if count != 1 {
		t.Fatal("multiple readers admitted the notification cue", count)
	}
	for _, client := range clients {
		if client.Claim_sound(context.Background()) {
			t.Fatal("a repeated cue escaped the shared cooldown")
		}
	}
}

func TestCoordinatorTakeoverAndHistoryLifetime(t *testing.T) {
	namespace := fmt.Sprintf(".%x", random_id())
	owner := test_client(t, namespace)
	owner.Publish("Before takeover", time.Now())
	if !owner.Claim_sound(context.Background()) {
		t.Fatal("first cue was rejected")
	}
	survivor := test_client(t, namespace)
	wait_history(t, survivor, 1)
	// Synchronize the cue timestamp before terminating the coordinator.
	if _, err := survivor.exchange(context.Background(), false); err != nil {
		t.Fatal(err)
	}
	owner.Close()
	if survivor.Claim_sound(context.Background()) {
		t.Fatal("coordinator exit reset the cooldown")
	}
	survivor.Publish("After takeover", time.Now())
	if _, err := survivor.exchange(context.Background(), false); err != nil {
		t.Fatal(err)
	}
	joining := test_client(t, namespace)
	wait_history(t, joining, 2)
	survivor.Close()
	joining.Close()
	fresh := test_client(t, namespace)
	if _, err := fresh.exchange(context.Background(), false); err != nil {
		t.Fatal(err)
	}
	if len(fresh.Snapshot().Entries) != 0 || !fresh.Claim_sound(context.Background()) {
		t.Fatal("history or cooldown survived the last reader")
	}
}

func TestCoordinatorExitFlushesNotificationToSurvivingReader(t *testing.T) {
	namespace := fmt.Sprintf(".%x", random_id())
	owner := test_client(t, namespace)
	if _, err := owner.exchange(context.Background(), false); err != nil {
		t.Fatal(err)
	}
	survivor := test_client(t, namespace)
	if _, err := survivor.exchange(context.Background(), false); err != nil {
		t.Fatal(err)
	}
	owner.Publish("Immediately before exit", time.Now())
	owner.Close()
	wait_history(t, survivor, 1)
	if _, err := survivor.exchange(context.Background(), false); err != nil {
		t.Fatal(err)
	}
	if got := survivor.Snapshot().Entries[0].Text; got != "Immediately before exit" {
		t.Fatal(got)
	}
}

func TestMutedSoundRequestsUseSharedCooldown(t *testing.T) {
	namespace := fmt.Sprintf(".%x", random_id())
	first, second := test_client(t, namespace), test_client(t, namespace)
	played := make(chan struct{}, 4)
	play := func(context.Context) { played <- struct{}{} }
	first.Sound(play)
	second.Sound(play)
	select {
	case <-played:
	case <-time.After(3 * time.Second):
		t.Fatal("muted notification cue was not delivered")
	}
	first.Close()
	second.Close()
	if len(played) != 0 {
		t.Fatal("muted readers played multiple cues")
	}
}

func TestCanceledClaimDoesNotConsumeCooldownAndNamespacesStaySeparate(t *testing.T) {
	namespace := fmt.Sprintf(".%x", random_id())
	first := test_client(t, namespace)
	second := test_client(t, namespace+".other")
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	if first.Claim_sound(ctx) {
		t.Fatal("canceled cue was accepted")
	}
	first.Publish("Private history", time.Now())
	if !first.Claim_sound(context.Background()) || !second.Claim_sound(context.Background()) {
		t.Fatal("unrelated histories shared a cooldown")
	}
	if len(second.Snapshot().Entries) != 0 {
		t.Fatal("history leaked to an unrelated group")
	}
}

func TestBrokerRejectsInvalidFramesAndAdmitsCueAfterCooldown(t *testing.T) {
	namespace := fmt.Sprintf(".%x", random_id())
	c := test_client(t, namespace)
	if _, err := c.exchange(context.Background(), false); err != nil {
		t.Fatal(err)
	}
	for _, frame := range []string{
		`{"Version":2}`,
		`{"Version":1,"Seed":{"Entries":[{"ID":"bad","Text":"bad"}]},"Claim":true}`,
		`{"Version":1,"Seed":{"Entries":[],"LastSound":"2999-01-01T00:00:00Z"},"Claim":true}`,
	} {
		connection, err := net.DialTimeout("unix", c.address, time.Second)
		if err != nil {
			t.Fatal(err)
		}
		connection.SetDeadline(time.Now().Add(time.Second))
		fmt.Fprintln(connection, frame)
		var result response
		err = json.NewDecoder(connection).Decode(&result)
		connection.Close()
		if err == nil {
			t.Fatal("invalid frame was accepted", frame)
		}
	}
	c.brokerMu.Lock()
	b := c.broker
	c.brokerMu.Unlock()
	b.mu.Lock()
	b.state.LastSound = time.Now().Add(-SoundInterval)
	b.mu.Unlock()
	if !c.Claim_sound(context.Background()) {
		t.Fatal("expired cooldown did not admit a cue")
	}
}

type processRequest struct {
	Publish string
	Claim   bool
	Stop    bool
}

type processResult struct {
	State   Snapshot
	Allowed bool
}

func TestNotificationProcess(t *testing.T) {
	namespace := os.Getenv("FENRIR_NOTIFICATION_PROCESS")
	if namespace == "" {
		return
	}
	c := New(context.Background(), os.Geteuid(), namespace)
	encoder := json.NewEncoder(os.Stdout)
	decoder := json.NewDecoder(os.Stdin)
	_ = encoder.Encode(processResult{})
	for {
		var message processRequest
		if decoder.Decode(&message) != nil || message.Stop {
			c.Close()
			os.Exit(0)
		}
		if message.Publish != "" {
			c.Publish(message.Publish, time.Now())
		}
		result, err := c.exchange(context.Background(), message.Claim)
		if err != nil {
			t.Fatal(err)
		}
		_ = encoder.Encode(processResult{State: result.State, Allowed: result.Allowed})
	}
}

type notificationProcess struct {
	command *exec.Cmd
	input   *json.Encoder
	output  *json.Decoder
	stopped bool
}

func start_process(t *testing.T, namespace string) *notificationProcess {
	t.Helper()
	command := exec.Command(os.Args[0], "-test.run=^TestNotificationProcess$")
	command.Env = append(os.Environ(), "FENRIR_NOTIFICATION_PROCESS="+namespace)
	command.Stderr = os.Stderr
	input, err := command.StdinPipe()
	if err != nil {
		t.Fatal(err)
	}
	output, err := command.StdoutPipe()
	if err != nil {
		t.Fatal(err)
	}
	if err := command.Start(); err != nil {
		t.Fatal(err)
	}
	p := &notificationProcess{command: command, input: json.NewEncoder(input), output: json.NewDecoder(output)}
	t.Cleanup(func() { p.stop(t) })
	var ready processResult
	if err := p.output.Decode(&ready); err != nil {
		t.Fatal(err)
	}
	return p
}

func (p *notificationProcess) send(t *testing.T, message processRequest) processResult {
	t.Helper()
	if err := p.input.Encode(message); err != nil {
		t.Fatal(err)
	}
	var result processResult
	if err := p.output.Decode(&result); err != nil {
		t.Fatal(err)
	}
	return result
}

func (p *notificationProcess) stop(t *testing.T) {
	t.Helper()
	if p.stopped {
		return
	}
	p.stopped = true
	_ = p.input.Encode(processRequest{Stop: true})
	done := make(chan error, 1)
	go func() { done <- p.command.Wait() }()
	select {
	case err := <-done:
		if err != nil {
			t.Error(err)
		}
	case <-time.After(5 * time.Second):
		_ = p.command.Process.Kill()
		<-done
		t.Error("notification fixture did not exit")
	}
}

func TestFourProcessesShareHistoryAndCooldownAcrossCoordinatorExit(t *testing.T) {
	for _, crash := range []bool{false, true} {
		t.Run(fmt.Sprintf("crash=%v", crash), func(t *testing.T) {
			check_four_processes(t, crash)
		})
	}
}

func check_four_processes(t *testing.T, crash bool) {
	t.Helper()
	namespace := fmt.Sprintf(".%x", random_id())
	processes := []*notificationProcess{start_process(t, namespace)}
	want := processes[0].send(t, processRequest{Publish: "One targeted notification", Claim: true})
	if len(want.State.Entries) != 1 || !want.Allowed {
		t.Fatal(want)
	}
	for range 3 {
		processes = append(processes, start_process(t, namespace))
	}
	for _, process := range processes[1:] {
		got := process.send(t, processRequest{Claim: true})
		if !reflect.DeepEqual(got.State.Entries, want.State.Entries) || got.Allowed {
			t.Fatal("a replica lost history or claimed another cue", got)
		}
	}
	if crash {
		processes[0].stopped = true
		if err := processes[0].command.Process.Kill(); err != nil {
			t.Fatal(err)
		}
		_ = processes[0].command.Wait()
	} else {
		processes[0].stop(t)
	}
	for _, process := range processes[1:] {
		got := process.send(t, processRequest{Claim: true})
		if !reflect.DeepEqual(got.State.Entries, want.State.Entries) || got.Allowed {
			t.Fatal("coordinator exit lost history or cooldown", got)
		}
	}
	processes[1].send(t, processRequest{Publish: "Another targeted notification"})
	for _, process := range processes[1:] {
		got := process.send(t, processRequest{})
		if len(got.State.Entries) != 2 || !strings.Contains(got.State.Entries[1].Text, "Another") {
			t.Fatal(got)
		}
	}
}