KamadoPool/api/internal/logmon/tailer_test.go

package logmon

import (
	"context"
	"io"
	"log/slog"
	"os"
	"sync"
	"testing"
	"time"
)

func discardLog() *slog.Logger {
	return slog.New(slog.NewTextHandler(io.Discard, nil))
}

// ---- handleLine unit tests (no filesystem, no goroutines) ----

func TestHandleLine_Solved(t *testing.T) {
	tl := New("/unused", discardLog())
	tl.handleLine("Solved and confirmed block 840123\n")
	select {
	case ev := <-tl.Events:
		if ev.Height != 840123 {
			t.Errorf("Height = %d, want 840123", ev.Height)
		}
	default:
		t.Fatal("no BlockEvent emitted for solved line")
	}
}

func TestHandleLine_AttemptCarriedToBlock(t *testing.T) {
	tl := New("/unused", discardLog())
	tl.handleLine("Possible block solve diff 32768 !\n")
	tl.handleLine("Solved and confirmed block 900000\n")

	select {
	case <-tl.Attempts:
	default:
		t.Fatal("no AttemptEvent emitted for attempt line")
	}
	select {
	case ev := <-tl.Events:
		if ev.ShareDiff != 32768 {
			t.Errorf("ShareDiff = %v, want 32768", ev.ShareDiff)
		}
		if ev.Height != 900000 {
			t.Errorf("Height = %d, want 900000", ev.Height)
		}
	default:
		t.Fatal("no BlockEvent emitted")
	}
}

func TestHandleLine_SubmittingVariant(t *testing.T) {
	tl := New("/unused", discardLog())
	tl.handleLine("Submitting possible block solve share diff 65536 !\n")
	select {
	case ev := <-tl.Attempts:
		if ev.ShareDiff != 65536 {
			t.Errorf("ShareDiff = %v, want 65536", ev.ShareDiff)
		}
	default:
		t.Fatal("no AttemptEvent for submitting variant")
	}
	// Must NOT also emit a BlockEvent.
	select {
	case ev := <-tl.Events:
		t.Errorf("unexpected BlockEvent from attempt line: %+v", ev)
	default:
	}
}

func TestHandleLine_DiffResetAfterBlock(t *testing.T) {
	tl := New("/unused", discardLog())
	tl.handleLine("Possible block solve diff 16384 !\n")
	tl.handleLine("Solved and confirmed block 100\n")
	// Second block — no preceding attempt line; diff must NOT carry over.
	tl.handleLine("Solved and confirmed block 101\n")

	drain := func() BlockEvent {
		select {
		case ev := <-tl.Events:
			return ev
		default:
			t.Fatal("no BlockEvent")
			return BlockEvent{}
		}
	}
	ev100 := drain()
	ev101 := drain()
	if ev100.ShareDiff != 16384 {
		t.Errorf("block 100 ShareDiff = %v, want 16384", ev100.ShareDiff)
	}
	if ev101.ShareDiff != 0 {
		t.Errorf("block 101 ShareDiff = %v, want 0 (no preceding attempt)", ev101.ShareDiff)
	}
}

func TestHandleLine_Unrelated(t *testing.T) {
	tl := New("/unused", discardLog())
	for _, line := range []string{
		"Stratifier started\n",
		"New block hash 000000abc\n",
		"Worker authenticated\n",
	} {
		tl.handleLine(line)
	}
	select {
	case ev := <-tl.Events:
		t.Errorf("unexpected BlockEvent from unrelated line: %+v", ev)
	default:
	}
	select {
	case ev := <-tl.Attempts:
		t.Errorf("unexpected AttemptEvent from unrelated line: %+v", ev)
	default:
	}
}

// ---- Tailer.Run integration tests (filesystem + goroutines) ----

func tempLog(t *testing.T) string {
	t.Helper()
	f, err := os.CreateTemp("", "ckpool*.log")
	if err != nil {
		t.Fatal(err)
	}
	f.Close()
	t.Cleanup(func() { _ = os.Remove(f.Name()) })
	return f.Name()
}

func appendLines(t *testing.T, path string, lines ...string) {
	t.Helper()
	f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0)
	if err != nil {
		t.Fatal(err)
	}
	defer f.Close()
	for _, l := range lines {
		if _, err := f.WriteString(l); err != nil {
			t.Fatal(err)
		}
	}
}

// TestTailerRun_BasicRead starts the tailer on an empty file, appends log
// lines, and verifies the expected events are emitted.
func TestTailerRun_BasicRead(t *testing.T) {
	path := tempLog(t)
	tl := New(path, discardLog())
	tl.PollWait = 20 * time.Millisecond

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	go tl.Run(ctx)

	// Give the tailer time to open the empty file and park at EOF.
	time.Sleep(40 * time.Millisecond)

	appendLines(t, path,
		"Possible block solve diff 16384 !\n",
		"Solved and confirmed block 100\n",
	)

	select {
	case <-tl.Attempts:
	case <-ctx.Done():
		t.Fatal("timed out waiting for AttemptEvent")
	}
	select {
	case ev := <-tl.Events:
		if ev.Height != 100 {
			t.Errorf("Height = %d, want 100", ev.Height)
		}
		if ev.ShareDiff != 16384 {
			t.Errorf("ShareDiff = %v, want 16384", ev.ShareDiff)
		}
	case <-ctx.Done():
		t.Fatal("timed out waiting for BlockEvent")
	}
}

// TestTailerRun_CursorResume verifies that a new tailer honours a saved
// cursor and does not re-emit lines that were already processed.
func TestTailerRun_CursorResume(t *testing.T) {
	path := tempLog(t)

	var mu sync.Mutex
	var savedIno uint64
	var savedOff int64

	tl := New(path, discardLog())
	tl.PollWait = 20 * time.Millisecond
	tl.SaveCursor = func(ino uint64, off int64) {
		mu.Lock()
		savedIno, savedOff = ino, off
		mu.Unlock()
	}

	ctx1, cancel1 := context.WithTimeout(context.Background(), 3*time.Second)
	go tl.Run(ctx1)
	time.Sleep(40 * time.Millisecond)

	// Phase 1: write 2 blocks and drain both events.
	appendLines(t, path,
		"Solved and confirmed block 200\n",
		"Solved and confirmed block 201\n",
	)
	for i := 0; i < 2; i++ {
		select {
		case <-tl.Events:
		case <-ctx1.Done():
			t.Fatalf("phase 1: only got %d events, want 2", i)
		}
	}
	// Allow the tailer to hit EOF and save the cursor.
	time.Sleep(100 * time.Millisecond)
	cancel1()
	// Give the goroutine time to exit and force-save.
	time.Sleep(60 * time.Millisecond)

	mu.Lock()
	ino, off := savedIno, savedOff
	mu.Unlock()
	if ino == 0 {
		t.Fatal("cursor was never saved after phase 1")
	}

	// Phase 2: append a third block, start a new tailer from the cursor.
	// It must see ONLY block 202, not replays of 200 or 201.
	appendLines(t, path, "Solved and confirmed block 202\n")

	tl2 := New(path, discardLog())
	tl2.PollWait = 20 * time.Millisecond
	tl2.LoadCursor = func() (uint64, int64, bool) { return ino, off, true }

	ctx2, cancel2 := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel2()
	go tl2.Run(ctx2)

	select {
	case ev := <-tl2.Events:
		if ev.Height != 202 {
			t.Errorf("resumed tailer emitted height %d, want 202 (cursor resume may be broken)", ev.Height)
		}
	case <-ctx2.Done():
		t.Fatal("timed out waiting for block 202 after cursor resume")
	}
}

// TestTailerRun_Rotation verifies that when the log file is replaced
// (inode changes), the tailer re-opens it from the start and reads the
// new content.
func TestTailerRun_Rotation(t *testing.T) {
	path := tempLog(t)
	tl := New(path, discardLog())
	tl.PollWait = 30 * time.Millisecond

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	go tl.Run(ctx)
	time.Sleep(50 * time.Millisecond)

	// Write a block to the original file and collect it.
	appendLines(t, path, "Solved and confirmed block 300\n")
	select {
	case ev := <-tl.Events:
		if ev.Height != 300 {
			t.Errorf("before rotation: height = %d, want 300", ev.Height)
		}
	case <-ctx.Done():
		t.Fatal("timed out waiting for pre-rotation block")
	}

	// Simulate log rotation: replace the file at the same path.
	newF, err := os.CreateTemp("", "ckpool_new*.log")
	if err != nil {
		t.Fatal(err)
	}
	newF.WriteString("Solved and confirmed block 301\n")
	newF.Close()
	if err := os.Rename(newF.Name(), path); err != nil {
		t.Fatal(err)
	}

	select {
	case ev := <-tl.Events:
		if ev.Height != 301 {
			t.Errorf("after rotation: height = %d, want 301", ev.Height)
		}
	case <-ctx.Done():
		t.Fatal("timed out waiting for post-rotation block")
	}
}