textmachine/platform/internal/ingest/tail_test.go

package ingest

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"testing"
)

// tailSink records what the tailer decided to materialize, and nothing else: what these tests are
// about is the READER's rule for at-least-once delivery, not what a sink does with an event.
type tailSink struct {
	hellos  []Hello
	applied []Envelope
	cursors []Cursor
	// failFrom makes the sink refuse every event from this seq on, which is how a database outage
	// mid-stream looks to the reader.
	failFrom int64
}

func (s *tailSink) Begin(_ context.Context, h Hello) error {
	s.hellos = append(s.hellos, h)
	return nil
}

func (s *tailSink) Apply(_ context.Context, ev Envelope, c Cursor) error {
	if s.failFrom > 0 && ev.Seq >= s.failFrom {
		return errors.New("database is down")
	}
	s.applied = append(s.applied, ev)
	s.cursors = append(s.cursors, c)
	return nil
}

func (s *tailSink) seqs() []int64 {
	out := make([]int64, 0, len(s.applied))
	for _, e := range s.applied {
		out = append(out, e.Seq)
	}
	return out
}

func line(t *testing.T, seq int64, typ Type, data any) string {
	t.Helper()
	raw, err := json.Marshal(data)
	if err != nil {
		t.Fatal(err)
	}
	b, err := json.Marshal(Envelope{Seq: seq, Type: typ, Data: raw})
	if err != nil {
		t.Fatal(err)
	}
	return string(b) + "\n"
}

func hello(t *testing.T, seq int64, runID string) string {
	return line(t, seq, TypeHello, Hello{StreamVersion: StreamVersion, EngineRunID: runID, BookID: "bk"})
}

func progress(t *testing.T, seq int64, done int) string {
	return line(t, seq, TypeProgress, Progress{Draft: Counter{Done: done, Total: 10}})
}

func journal(t *testing.T, content string) string {
	t.Helper()
	path := filepath.Join(t.TempDir(), JournalFile)
	if err := os.WriteFile(path, []byte(content), 0o600); err != nil {
		t.Fatal(err)
	}
	return path
}

// The tailer is started when the unit is; the engine writes its first line whenever it gets there.
// Waiting is the correct behaviour and it must not be an error, or every run would log a failure
// during its first seconds — and today, with no emitter at all, forever.
func TestAnAbsentJournalIsNotAFailure(t *testing.T) {
	_, _, err := Tail(t.Context(), filepath.Join(t.TempDir(), JournalFile), "", Position{}, &tailSink{})
	if !errors.Is(err, ErrNoJournal) {
		t.Fatalf("missing journal gave %v, want ErrNoJournal", err)
	}
}

// A line without its newline is a line still being written. Reading it as an event turns an ordinary
// interleaving with the writer into a decode error on a paid run.
func TestAHalfWrittenLineIsLeftForNextTime(t *testing.T) {
	partial := hello(t, 1, "eng-1") + progress(t, 2, 1) + `{"seq":3,"type":"progress","dat`
	path := journal(t, partial)
	sink := &tailSink{}
	pos, id, err := Tail(t.Context(), path, "", Position{}, sink)
	if err != nil {
		t.Fatal(err)
	}
	if id != "eng-1" {
		t.Fatalf("engine run id %q", id)
	}
	// The handshake moves the cursor like any other line: its effect is nothing, its seq is the
	// cursor, and without it a resume after the hello reads the next line as a gap.
	if got := sink.seqs(); len(got) != 2 || got[0] != 1 || got[1] != 2 {
		t.Fatalf("applied %v, want 1,2", got)
	}
	// The cursor stands before the partial line, so the completed form of it is read next time.
	if want := int64(len(hello(t, 1, "eng-1")) + len(progress(t, 2, 1))); pos.Offset != want {
		t.Fatalf("offset %d, want %d (before the partial line)", pos.Offset, want)
	}
	// Now the writer finishes the line and appends another.
	rest := `a":{"draft":{"done":2,"total":10}}}` + "\n" + progress(t, 4, 3)
	f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o600)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := f.WriteString(rest); err != nil {
		t.Fatal(err)
	}
	f.Close()
	if _, _, err := Tail(t.Context(), path, id, pos, sink); err != nil {
		t.Fatal(err)
	}
	if got := sink.seqs(); len(got) != 4 || got[2] != 3 || got[3] != 4 {
		t.Fatalf("applied %v, want 1,2,3,4", got)
	}
}

// PD-105, ratified by D39.119: delivery is at-least-once, so re-reading the line the cursor stands
// on is NORMAL. It must be idempotent and it must not be fatal — the previous decoder called it
// ErrStreamGap, and after the PD-12 fix that killed a paid run over one duplicated line.
func TestARedeliveredLineIsNormalAndChangesNothing(t *testing.T) {
	body := hello(t, 1, "eng-1") + progress(t, 2, 1) + progress(t, 3, 2)
	path := journal(t, body)
	sink := &tailSink{}
	pos, id, err := Tail(t.Context(), path, "", Position{}, sink)
	if err != nil {
		t.Fatal(err)
	}
	// Re-read the WHOLE journal from the beginning with the cursor already at seq 3: every line is a
	// duplicate, and not one of them may be applied again or raise an error.
	again := &tailSink{}
	pos2, _, err := Tail(t.Context(), path, id, Position{LastSeq: pos.LastSeq, LastHash: pos.LastHash}, again)
	if err != nil {
		t.Fatalf("re-reading an already applied journal: %v", err)
	}
	if len(again.applied) != 0 || len(again.hellos) != 0 {
		t.Fatalf("a re-read applied %d events and %d handshakes; both must be zero", len(again.applied), len(again.hellos))
	}
	if pos2.Offset != pos.Offset {
		t.Errorf("the byte hint did not catch up on a re-read: %d, want %d", pos2.Offset, pos.Offset)
	}
}

// The one case a high-water mark cannot absorb: the same seq saying something different. Quietly
// skipping it materializes whichever copy was read first (research/25 §эмиттер).
func TestTheSameSeqWithADifferentPayloadIsRefused(t *testing.T) {
	first := journal(t, hello(t, 1, "eng-1")+progress(t, 2, 1))
	sink := &tailSink{}
	pos, id, err := Tail(t.Context(), first, "", Position{}, sink)
	if err != nil {
		t.Fatal(err)
	}
	// The same seq, a different payload — a rewritten journal, or two engine instances on one run.
	rewritten := journal(t, hello(t, 1, "eng-1")+progress(t, 2, 7))
	if _, _, err := Tail(t.Context(), rewritten, id, Position{LastSeq: pos.LastSeq, LastHash: pos.LastHash}, sink); !errors.Is(err, ErrPayloadConflict) {
		t.Fatalf("a changed payload gave %v, want ErrPayloadConflict", err)
	}
}

// A gap means lines were lost, and reading further cannot recover them: the ratified answer is the
// resync channel, so the reader has to say so rather than materialize a projection with a hole.
func TestALostLineIsReportedRatherThanSkipped(t *testing.T) {
	path := journal(t, hello(t, 1, "eng-1")+progress(t, 2, 1)+progress(t, 5, 4))
	sink := &tailSink{}
	if _, _, err := Tail(t.Context(), path, "", Position{}, sink); !errors.Is(err, ErrStreamGap) {
		t.Fatalf("a skipped seq gave %v, want ErrStreamGap", err)
	}
	if got := sink.seqs(); len(got) != 2 || got[1] != 2 {
		t.Fatalf("applied %v: everything before the gap must still land", got)
	}
}

// The journal is per BOOK and append-only, so a resumed run appends a SECOND handshake with its own
// run id and a seq that starts over. A reader that treated it as corruption would stop exactly when
// the run resumed; one that adopted it would mix two attempts into one attempt's counters.
func TestAnotherAttemptsStreamInTheSameJournalIsSkipped(t *testing.T) {
	body := hello(t, 1, "eng-1") + progress(t, 2, 1) +
		hello(t, 1, "eng-2") + progress(t, 2, 5) + progress(t, 3, 6)
	path := journal(t, body)

	// Attempt one owns the first stream and never sees the second.
	first := &tailSink{}
	pos, id, err := Tail(t.Context(), path, "", Position{}, first)
	if err != nil {
		t.Fatal(err)
	}
	if id != "eng-1" {
		t.Fatalf("adopted %q, want eng-1", id)
	}
	if got := first.seqs(); len(got) != 2 || got[1] != 2 {
		t.Fatalf("attempt one applied %v, want 1,2 of eng-1 and nothing of eng-2", got)
	}
	if pos.Offset != int64(len(body)) {
		t.Errorf("the reader did not walk past the other attempt's lines: offset %d of %d", pos.Offset, len(body))
	}

	// Attempt two starts where its own lines begin — the journal's size when it was admitted.
	start := int64(len(hello(t, 1, "eng-1") + progress(t, 2, 1)))
	second := &tailSink{}
	if _, id, err = Tail(t.Context(), path, "", Position{Offset: start}, second); err != nil {
		t.Fatal(err)
	}
	if id != "eng-2" {
		t.Fatalf("attempt two adopted %q, want eng-2", id)
	}
	if got := second.seqs(); len(got) != 3 || got[1] != 2 || got[2] != 3 {
		t.Fatalf("attempt two applied %v, want 1,2,3 of eng-2", got)
	}
}

// A journal that already holds SOMEBODY ELSE's stream at this attempt's starting offset — an
// operator running tmctl by hand in the book's directory is the realistic source.
//
// ⚠ Reproduced before the fix, and it was three defects in one line: the reader adopted the foreign
// handshake, materialized its events onto this attempt (a `ceiling` among them, which pauses a run
// nothing had paused), and then refused the attempt's OWN handshake for carrying a seq that was not
// 1 — quarantining the projection of a paying run for good. The cure is that the platform NAMES the
// stream before the unit starts (runs.engineStreamID) and hands the name over, so `want` is never
// empty and there is nothing left to adopt.
//
// Mutation caught: going back to `mine := want != "" && pos.Offset > 0`, and dropping the id from
// the spawn record so the reader is handed an empty one.
func TestAForeignStreamAtOurOffsetIsNeverAdopted(t *testing.T) {
	foreign := hello(t, 1, "SOMEONE-ELSE") + line(t, 2, TypeCeiling, Ceiling{Halted: true, Scope: ScopeBook})
	body := foreign + hello(t, 1, "ours") + progress(t, 2, 7)
	path := journal(t, body)

	// The attempt was admitted when the journal was empty, so its byte hint is 0 and it reads the
	// foreign lines on the way to its own.
	sink := &tailSink{}
	pos, id, err := Tail(t.Context(), path, "ours", Position{}, sink)
	if err != nil {
		t.Fatalf("a foreign stream in the journal broke the read: %v", err)
	}
	if id != "ours" {
		t.Fatalf("the reader ended up on stream %q", id)
	}
	if len(sink.hellos) != 0 {
		t.Errorf("Begin was called for a stream this attempt did not own: %+v", sink.hellos)
	}
	if got := sink.seqs(); len(got) != 2 || got[0] != 1 || got[1] != 2 {
		t.Fatalf("applied %v, want only our own 1,2 — a foreign ceiling event pauses a live run", got)
	}
	if pos.Offset != int64(len(body)) || pos.LastSeq != 2 {
		t.Errorf("cursor %+v after the whole journal", pos)
	}

	// And the case the byte HINT makes dangerous. An attempt's hint is the journal's SIZE when it was
	// admitted, so it is non-zero before the attempt has read a line — and if a foreign process is
	// writing at that moment, the hint lands INSIDE that process's stream, past its handshake. There
	// is then no hello above those lines to say whose they are.
	//
	// ⚠ This is why the reader asks about the CURSOR (`LastSeq > 0`, and the handshake is seq 1) and
	// not about the hint. Under "my offset is non-zero ⇒ these lines are mine" the foreign
	// continuation below is JUDGED against our cursor, its seq reads as a gap, and the projection of a
	// perfectly healthy attempt is quarantined for good.
	//
	// Mutation caught: `mine := want != "" && pos.Offset > 0`.
	before := hello(t, 1, "SOMEONE-ELSE") + progress(t, 2, 1)
	body = before + progress(t, 3, 2) + hello(t, 1, "ours") + progress(t, 2, 9)
	path = journal(t, body)
	restarted := &tailSink{}
	pos, _, err = Tail(t.Context(), path, "ours", Position{Offset: int64(len(before))}, restarted)
	if err != nil {
		t.Fatalf("a hint landing inside somebody else's stream broke the read: %v", err)
	}
	if got := restarted.seqs(); len(got) != 2 || got[0] != 1 || got[1] != 2 {
		t.Errorf("applied %v, want only our own 1,2", got)
	}
	if pos.Offset != int64(len(body)) {
		t.Errorf("the reader did not walk past the foreign lines: offset %d of %d", pos.Offset, len(body))
	}
}

// The same journal read from the START by an attempt that already knows its own id. This is what a
// lost byte hint produces, and it is where the reader used to quarantine a perfectly healthy run:
// the previous attempt's seq 2 landed on our cursor's seq 2 with a different payload, which is
// exactly the corruption signal. Nothing but the last handshake says whose a line is.
func TestARereadFromTheStartDoesNotMistakeAnotherAttemptForCorruption(t *testing.T) {
	body := hello(t, 1, "eng-1") + progress(t, 2, 1) +
		hello(t, 1, "eng-2") + progress(t, 2, 5) + progress(t, 3, 6)
	path := journal(t, body)
	// Attempt two, mid-run: it knows its id and its cursor, and its byte hint is gone.
	own := &tailSink{}
	first, _, err := Tail(t.Context(), path, "eng-2", Position{}, own) // no byte hint at all
	if err != nil {
		t.Fatalf("re-reading from the start: %v", err)
	}
	if got := own.seqs(); len(got) != 3 || got[1] != 2 || got[2] != 3 {
		t.Fatalf("applied %v, want 1,2,3 of eng-2 and nothing of eng-1", got)
	}
	if first.Offset != int64(len(body)) {
		t.Errorf("offset %d, want %d", first.Offset, len(body))
	}
}

// Nothing is materialized before a handshake we own, even when the reader already knows its own
// engine run id. A journal that opens with events instead of a hello is the documented rot path
// (ad-hoc unversioned JSON), and judging those lines against OUR cursor would either apply another
// process's counters or report our own healthy stream as corrupt.
func TestEventsBeforeAnyHandshakeAreNotJudgedAgainstOurCursor(t *testing.T) {
	headless := progress(t, 2, 5) + progress(t, 3, 6) + hello(t, 1, "eng-1") + progress(t, 2, 1)
	path := journal(t, headless)
	sink := &tailSink{}
	pos, id, err := Tail(t.Context(), path, "eng-1", Position{}, sink)
	if err != nil {
		t.Fatalf("a journal that opens without a handshake: %v", err)
	}
	if id != "eng-1" {
		t.Fatalf("engine run id %q", id)
	}
	if got := sink.seqs(); len(got) != 2 || got[0] != 1 || got[1] != 2 {
		t.Fatalf("applied %v, want only the two lines that follow our own handshake", got)
	}
	if pos.LastSeq != 2 {
		t.Errorf("cursor at seq %d", pos.LastSeq)
	}
}

// The offset is a HINT and the append-only guarantee is what makes it usable. A journal that got
// SHORTER broke that guarantee, and reading on at a stale offset would decode the middle of a line.
func TestAJournalThatShrankIsRefusedRatherThanMisread(t *testing.T) {
	path := journal(t, hello(t, 1, "eng-1"))
	_, _, err := Tail(t.Context(), path, "eng-1", Position{Offset: 100_000, LastSeq: 5}, &tailSink{})
	if err == nil || !strings.Contains(err.Error(), "append-only") {
		t.Fatalf("a shrunken journal gave %v, want a refusal naming the broken guarantee", err)
	}
}

// The cursor the sink is handed has to be the position AFTER the line, or a crash between the effect
// and the next read re-reads a line the sink has already applied — harmless by luck, not by design.
func TestTheCursorHandedToTheSinkIsPastTheLine(t *testing.T) {
	h, p2 := hello(t, 1, "eng-1"), progress(t, 2, 1)
	path := journal(t, h+p2)
	sink := &tailSink{}
	if _, _, err := Tail(t.Context(), path, "", Position{}, sink); err != nil {
		t.Fatal(err)
	}
	if len(sink.cursors) != 2 {
		t.Fatalf("%d cursors, want one per applied line", len(sink.cursors))
	}
	if want := int64(len(h)); sink.cursors[0].Offset != want {
		t.Errorf("handshake cursor offset %d, want %d", sink.cursors[0].Offset, want)
	}
	if want := int64(len(h) + len(p2)); sink.cursors[1].Offset != want {
		t.Errorf("cursor offset %d, want %d", sink.cursors[1].Offset, want)
	}
	if len(sink.cursors[1].SHA256) != 32 {
		t.Errorf("cursor carries no line hash: %v", sink.cursors[1].SHA256)
	}
}

// A stream whose major version changed means something different by the same words; materializing it
// is worse than not materializing it.
func TestAMajorVersionBumpIsRefused(t *testing.T) {
	body, err := json.Marshal(Envelope{Seq: 1, Type: TypeHello,
		Data: json.RawMessage(`{"stream_version":"9.9","engine_run_id":"eng-1"}`)})
	if err != nil {
		t.Fatal(err)
	}
	path := journal(t, string(body)+"\n")
	if _, _, err := Tail(t.Context(), path, "", Position{}, &tailSink{}); !errors.Is(err, ErrUnsupportedVersion) {
		t.Fatalf("a major bump gave %v, want ErrUnsupportedVersion", err)
	}
}

// Half of the ratified idempotency key lives in the handshake; without it every run's events would
// collapse into one namespace instead of failing.
func TestAHandshakeWithoutAnEngineRunIdIsRefused(t *testing.T) {
	body := fmt.Sprintf(`{"seq":1,"type":"hello","data":{"stream_version":%q,"engine_run_id":""}}`+"\n", StreamVersion)
	path := journal(t, body)
	if _, _, err := Tail(t.Context(), path, "", Position{}, &tailSink{}); !errors.Is(err, ErrBadHandshake) {
		t.Fatalf("an anonymous handshake gave %v, want ErrBadHandshake", err)
	}
}

// A sink that fails stops the read where it is: the cursor must not advance past an effect that was
// never applied.
func TestAFailingSinkDoesNotMoveTheCursor(t *testing.T) {
	path := journal(t, hello(t, 1, "eng-1")+progress(t, 2, 1)+progress(t, 3, 2))
	sink := &tailSink{failFrom: 2}
	pos, _, err := Tail(t.Context(), path, "", Position{}, sink)
	if err == nil {
		t.Fatal("a failing sink was not reported")
	}
	if pos.LastSeq != 1 {
		t.Errorf("cursor moved to seq %d past an event the sink refused", pos.LastSeq)
	}
}

// A line past the buffer is REFUSED, not swallowed. Returning io.EOF for it would make the tailer
// report "caught up" at that byte forever: nothing after it would ever be materialized, with no
// error and no quarantine.
func TestALinePastTheBufferIsRefusedRatherThanRead(t *testing.T) {
	huge := `{"seq":2,"type":"progress","data":{"note":"` + strings.Repeat("x", maxLine) + `"}}`
	path := journal(t, hello(t, 1, "eng-1")+huge+"\n")
	_, _, err := Tail(t.Context(), path, "", Position{}, &tailSink{})
	if err == nil || !strings.Contains(err.Error(), "exceeds") {
		t.Fatalf("an over-long line gave %v, want a refusal naming the limit", err)
	}
}

// A blank line is not an event, but it IS bytes: an offset that did not advance past it is short by
// its length from then on, permanently.
func TestABlankLineStillMovesTheOffset(t *testing.T) {
	body := hello(t, 1, "eng-1") + "\n" + progress(t, 2, 1)
	path := journal(t, body)
	pos, _, err := Tail(t.Context(), path, "", Position{}, &tailSink{})
	if err != nil {
		t.Fatal(err)
	}
	if pos.Offset != int64(len(body)) {
		t.Errorf("offset %d after a journal of %d bytes containing a blank line", pos.Offset, len(body))
	}
}

// Neither half of the cursor may advance past an effect the sink refused — the byte hint as much as
// the seq, because the hint is what decides where the next sweep starts reading.
func TestNeitherHalfOfTheCursorAdvancesPastARefusedEffect(t *testing.T) {
	h, p2 := hello(t, 1, "eng-1"), progress(t, 2, 1)
	path := journal(t, h+p2+progress(t, 3, 2))
	sink := &tailSink{failFrom: 3}
	pos, _, err := Tail(t.Context(), path, "", Position{}, sink)
	if err == nil {
		t.Fatal("a failing sink was not reported")
	}
	if pos.LastSeq != 2 {
		t.Errorf("cursor seq %d past a refused event", pos.LastSeq)
	}
	if want := int64(len(h) + len(p2)); pos.Offset != want {
		t.Errorf("cursor offset %d, want %d — the byte hint moved past a refused event", pos.Offset, want)
	}
}

// A handshake that is NOT OURS is skipped whatever it says. The journal is per BOOK, so a foreign
// process in it is ordinary — an operator's own tmctl, an older build with another major, a broken
// build writing an empty id or a seq that is not 1. A reader that ran this build's handshake rules
// over every hello BEFORE asking whose it was would turn a foreign line's defects into errors of
// OUR read, `quarantines()` calls those terminal, and the projection of the paying attempt that
// merely shares the file would stop for good (PD-214, PD-426).
//
// Mutation caught: moving the ownership test back below any one of the three checks.
func TestAForeignHandshakeThisBuildCannotReadIsSkippedRatherThanQuarantiningOurs(t *testing.T) {
	for name, foreign := range map[string]string{
		"another major":          line(t, 1, TypeHello, Hello{StreamVersion: "9.9", EngineRunID: "SOMEONE-ELSE"}),
		"no engine run id":       line(t, 1, TypeHello, Hello{StreamVersion: StreamVersion, EngineRunID: ""}),
		"a seq that is not 1":    line(t, 7, TypeHello, Hello{StreamVersion: StreamVersion, EngineRunID: "SOMEONE-ELSE"}),
		"all three at once":      line(t, 16, TypeHello, Hello{StreamVersion: "0.1", EngineRunID: ""}),
		"a legal one, for scale": hello(t, 1, "SOMEONE-ELSE"),
	} {
		t.Run(name, func(t *testing.T) {
			// The foreign stream, its own events, then ours. Our lines are the only ones that may land.
			body := foreign + line(t, 2, TypeCeiling, Ceiling{Halted: true, Scope: ScopeBook}) +
				hello(t, 1, "ours") + progress(t, 2, 4)
			path := journal(t, body)
			sink := &tailSink{}
			pos, id, err := Tail(t.Context(), path, "ours", Position{}, sink)
			if err != nil {
				t.Fatalf("a foreign handshake broke the read of our own stream: %v", err)
			}
			if id != "ours" {
				t.Fatalf("the reader ended up on stream %q", id)
			}
			if len(sink.hellos) != 0 {
				t.Errorf("Begin was called for a stream this attempt did not own: %+v", sink.hellos)
			}
			if got := sink.seqs(); len(got) != 2 || got[0] != 1 || got[1] != 2 {
				t.Fatalf("applied %v, want only our own 1,2 — the foreign ceiling event pauses a live run", got)
			}
			if pos.Offset != int64(len(body)) || pos.LastSeq != 2 {
				t.Errorf("cursor %+v after the whole journal of %d bytes", pos, len(body))
			}
		})
	}
}

// The same three checks stay in force for OUR OWN handshake: a stream this build cannot read is still
// a stream it cannot read, and materializing it on a guess is worse than stopping. Ownership decides
// first only for lines that are NOT ours.
func TestOurOwnHandshakeThisBuildCannotReadStillStopsTheProjection(t *testing.T) {
	for name, tc := range map[string]struct {
		ours string
		want error
	}{
		"another major":       {line(t, 1, TypeHello, Hello{StreamVersion: "9.9", EngineRunID: "ours"}), ErrUnsupportedVersion},
		"a seq that is not 1": {line(t, 16, TypeHello, Hello{StreamVersion: StreamVersion, EngineRunID: "ours"}), ErrBadHandshake},
	} {
		t.Run(name, func(t *testing.T) {
			path := journal(t, tc.ours+progress(t, 2, 1))
			sink := &tailSink{}
			if _, _, err := Tail(t.Context(), path, "ours", Position{}, sink); !errors.Is(err, tc.want) {
				t.Fatalf("our own unreadable handshake gave %v, want %v", err, tc.want)
			}
			if got := sink.seqs(); len(got) != 0 {
				t.Errorf("applied %v after a handshake this build refused", got)
			}
		})
	}
}

// With NO name yet the reader ADOPTS the first handshake it meets, and adoption validates in full —
// lowering the checks there would take a stream of a foreign major as our own, the same defect from
// the other side. Legacy shape (attempts predate the platform naming their stream), pinned because
// the ordering change above must not reach it.
//
// Mutation caught: applying the ownership shortcut when `want` is empty.
func TestAnAdoptedHandshakeIsStillValidatedInFull(t *testing.T) {
	for name, tc := range map[string]struct {
		first string
		want  error
	}{
		"another major":       {line(t, 1, TypeHello, Hello{StreamVersion: "9.9", EngineRunID: "eng-1"}), ErrUnsupportedVersion},
		"no engine run id":    {line(t, 1, TypeHello, Hello{StreamVersion: StreamVersion, EngineRunID: ""}), ErrBadHandshake},
		"a seq that is not 1": {line(t, 16, TypeHello, Hello{StreamVersion: StreamVersion, EngineRunID: "eng-1"}), ErrBadHandshake},
	} {
		t.Run(name, func(t *testing.T) {
			path := journal(t, tc.first+progress(t, 2, 1))
			sink := &tailSink{}
			if _, _, err := Tail(t.Context(), path, "", Position{}, sink); !errors.Is(err, tc.want) {
				t.Fatalf("adopting %s gave %v, want %v", name, err, tc.want)
			}
			if len(sink.hellos) != 0 {
				t.Errorf("Begin was called for a handshake this build refused: %+v", sink.hellos)
			}
		})
	}
}

// A handshake that does not DECODE has no owner: the engine run id lives inside the payload, so the
// reader cannot skip it as somebody else's — and it must not, because a payload it cannot read on
// our own stream would then pass as foreign and our stream would go quiet with nothing saying why.
// The decode stays above the ownership question whether or not the reader knows its name.
//
// Mutation caught: moving the decode below the ownership test (which cannot compile as written,
// so the landing is a `default` that skips on decode failure when `want` is known).
func TestAHandshakeThatDoesNotDecodeIsRefusedEvenWhenTheReaderKnowsItsName(t *testing.T) {
	broken := `{"seq":1,"type":"hello","data":"not an object"}` + "\n"
	path := journal(t, broken+hello(t, 1, "ours")+progress(t, 2, 1))
	for _, want := range []string{"", "ours"} {
		sink := &tailSink{}
		_, _, err := Tail(t.Context(), path, want, Position{}, sink)
		if err == nil || !strings.Contains(err.Error(), "hello payload") {
			t.Fatalf("want=%q: an undecodable handshake gave %v, want a refusal naming the payload", want, err)
		}
		if got := sink.seqs(); len(got) != 0 {
			t.Errorf("want=%q: applied %v past a handshake that does not decode", want, got)
		}
	}
}

// THE PASS BOUNDARY, which is the shape drainJournal actually runs in: the sweep reads what the
// journal has gained, persists the cursor, and comes back a minute later with nothing but that
// cursor. Ownership is not part of it — `mine` is re-derived from `LastSeq > 0` — so a reader that
// walked PAST a foreign handshake would hand the next pass a cursor claiming a region that belongs
// to somebody else: the stranger's next event lands on this attempt (a `ceiling` pauses a paying
// run) and its seq meeting ours quarantines a healthy projection. The reader stops at the foreign
// handshake instead and leaves the byte hint on it.
//
// Mutation caught: returning `held` (the advanced offset) from the foreign-handshake branch;
// dropping the `pos.LastSeq > 0` or the `named` half of the stop.
func TestAForeignStreamDoesNotOwnOurAttemptOnTheNextPass(t *testing.T) {
	for name, foreign := range map[string]string{
		"a legal foreign handshake": hello(t, 1, "SOMEONE-ELSE"),
		"a foreign major":           line(t, 1, TypeHello, Hello{StreamVersion: "9.9", EngineRunID: "SOMEONE-ELSE"}),
		"one with no id at all":     line(t, 1, TypeHello, Hello{StreamVersion: StreamVersion, EngineRunID: ""}),
	} {
		t.Run(name, func(t *testing.T) {
			runOne(t, hello(t, 1, "ours")+progress(t, 2, 1), foreign)
			// The BOUNDARY of the same property, and it is the shape П1 describes: our engine wrote
			// its handshake and died, so exactly ONE line of ours is applied when the stranger's
			// handshake arrives. A guard that starts one line later reads that stranger's region as
			// ours and lets its halting `ceiling` land on a paying attempt.
			runOne(t, hello(t, 1, "ours"), foreign)
		})
	}
}

// runOne walks the two passes of the property for one journal prefix of ours.
func runOne(t *testing.T, ours, foreign string) {
	t.Helper()
	// Our own prefix is one line per seq, numbered from 1, so its line count IS the cursor the park
	// must leave behind — which is the point of running this with a one-line prefix as well as a
	// two-line one: the guard's boundary is at the first applied line, not at the second.
	mine := int64(strings.Count(ours, "\n"))
	{
		sink := &tailSink{}
		pos, _, err := Tail(t.Context(), journal(t, ours+foreign), "ours", Position{}, sink)
		// The park is an ANSWER, not a failure: our own lines are applied and the stranger's are
		// not, and the caller is told which of the two silences this is.
		if !errors.Is(err, ErrForeignStreamAhead) {
			t.Fatalf("a foreign handshake answered %v, want ErrForeignStreamAhead", err)
		}
		if got := sink.seqs(); int64(len(got)) != mine {
			t.Fatalf("pass one applied %v, want our own %d line(s)", got, mine)
		}
		if pos.Offset != int64(len(ours)) || pos.LastSeq != mine {
			t.Fatalf("pass one left the cursor at %+v, want it on the foreign handshake at offset %d with our seq %d",
				pos, len(ours), mine)
		}
		// Pass two: the stranger writes on. Its ceiling would pause OUR run; its seq 2 is a
		// different payload at the number our cursor stands on.
		// The stranger's own continuation: its seq 2 both as an event that would HALT our run and as a
		// payload at a number our cursor may stand on, plus one past it.
		for _, next := range []string{
			line(t, 2, TypeCeiling, Ceiling{Halted: true, Scope: ScopeBook}),
			progress(t, 2, 9),
			progress(t, 3, 9),
		} {
			sink := &tailSink{}
			pos2, _, err := Tail(t.Context(), journal(t, ours+foreign+next), "ours", pos, sink)
			if !errors.Is(err, ErrForeignStreamAhead) {
				t.Fatalf("the stranger's line answered %v, want the same park (never a quarantine)", err)
			}
			if got := sink.seqs(); len(got) != 0 {
				t.Fatalf("the stranger's line was applied to our attempt: %v", got)
			}
			if pos2.Offset != pos.Offset || pos2.LastSeq != pos.LastSeq || !bytes.Equal(pos2.LastHash, pos.LastHash) {
				t.Fatalf("the cursor moved over a region that is not ours: %+v, want %+v", pos2, pos)
			}
		}
	}
}

// The stop is for the attempt whose stream the platform NAMED, and only once our own lines are in:
// before that, our handshake may still be further down the file (a foreign process wrote at the byte
// this attempt was admitted on), and stopping would strand a run whose stream is right there. The
// nameless legacy path walks past instead, and TestAnotherAttemptsStreamInTheSameJournalIsSkipped is
// where that half lives.
func TestAForeignStreamBeforeOursIsWalkedPastRatherThanStoppedOn(t *testing.T) {
	foreign := hello(t, 1, "SOMEONE-ELSE") + line(t, 2, TypeCeiling, Ceiling{Halted: true, Scope: ScopeBook})
	body := foreign + hello(t, 1, "ours") + progress(t, 2, 7)
	sink := &tailSink{}
	pos, id, err := Tail(t.Context(), journal(t, body), "ours", Position{}, sink)
	if err != nil {
		t.Fatalf("a foreign stream before ours: %v", err)
	}
	if id != "ours" || pos.Offset != int64(len(body)) || pos.LastSeq != 2 {
		t.Fatalf("the reader ended on %q at %+v, want it past the stranger and on our seq 2", id, pos)
	}
	if got := sink.seqs(); len(got) != 2 || got[0] != 1 || got[1] != 2 {
		t.Fatalf("applied %v, want only our own 1,2", got)
	}
}

// The park is REPORTED, not swallowed: the caller has to tell "our region ended here" from "caught
// up", because the two leave the identical cursor and only one of them means the projection has
// stopped speaking. A reader that returned nil here left a paying run's screen frozen with no error,
// no quarantine and nothing in any listing.
//
// Mutation caught: returning nil instead of ErrForeignStreamAhead from the park.
func TestTheParkIsReportedSoTheCallerCanTellItFromBeingCaughtUp(t *testing.T) {
	ours := hello(t, 1, "ours") + progress(t, 2, 1)
	path := journal(t, ours+hello(t, 1, "SOMEONE-ELSE")+progress(t, 2, 9))
	sink := &tailSink{}
	pos, _, err := Tail(t.Context(), path, "ours", Position{}, sink)
	if !errors.Is(err, ErrForeignStreamAhead) {
		t.Fatalf("the park answered %v, want ErrForeignStreamAhead", err)
	}
	if pos.Offset != int64(len(ours)) || pos.LastSeq != 2 {
		t.Fatalf("the park moved the cursor: %+v, want it on the foreign handshake at offset %d", pos, len(ours))
	}
	if got := sink.seqs(); len(got) != 2 {
		t.Fatalf("applied %v, want our own two lines and nothing of the stranger's", got)
	}
	// Being caught up is the other answer, and it is not this one.
	if _, _, err := Tail(t.Context(), journal(t, ours), "ours", Position{}, &tailSink{}); err != nil {
		t.Fatalf("a journal with nothing but our own lines answered %v, want nil", err)
	}
}