textmachine/platform/internal/ingest/decoder_test.go

package ingest

import (
	"context"
	"encoding/json"
	"errors"
	"io"
	"strings"
	"testing"
)

const helloLine = `{"seq":1,"type":"hello","time":"2026-08-04T10:00:00Z","data":{"stream_version":"1.0","engine_run_id":"tr_1","book_id":"gzr"}}`

func TestHelloIsRequiredFirst(t *testing.T) {
	d := NewDecoder(strings.NewReader(`{"seq":1,"type":"progress","data":{}}` + "\n"))
	if _, err := d.Hello(); !errors.Is(err, ErrNoHandshake) {
		t.Fatalf("want ErrNoHandshake, got %v", err)
	}
	if _, err := NewDecoder(strings.NewReader("")).Hello(); !errors.Is(err, ErrNoHandshake) {
		t.Fatalf("empty stream: want ErrNoHandshake, got %v", err)
	}
}

func TestMajorVersionRefused(t *testing.T) {
	line := strings.Replace(helloLine, `"stream_version":"1.0"`, `"stream_version":"2.0"`, 1)
	if _, err := NewDecoder(strings.NewReader(line)).Hello(); !errors.Is(err, ErrUnsupportedVersion) {
		t.Fatalf("want ErrUnsupportedVersion, got %v", err)
	}
}

func TestMinorVersionAndUnknownFieldsTolerated(t *testing.T) {
	line := strings.Replace(helloLine, `"stream_version":"1.0"`, `"stream_version":"1.7","future":42`, 1)
	d := NewDecoder(strings.NewReader(line + "\n" + `{"seq":2,"type":"weather","data":{"sky":"grey"}}`))
	if _, err := d.Hello(); err != nil {
		t.Fatalf("minor bump must be accepted: %v", err)
	}
	ev, err := d.Next()
	if err != nil {
		t.Fatalf("unknown event type must reach the sink: %v", err)
	}
	if ev.Type != "weather" {
		t.Fatalf("type = %q", ev.Type)
	}
}

func TestSequenceGapAndReplayAreReported(t *testing.T) {
	for name, second := range map[string]string{
		"gap":     `{"seq":5,"type":"progress","data":{}}`,
		"replay":  `{"seq":1,"type":"progress","data":{}}`,
		"reverse": `{"seq":0,"type":"progress","data":{}}`,
	} {
		t.Run(name, func(t *testing.T) {
			d := NewDecoder(strings.NewReader(helloLine + "\n" + second))
			if _, err := d.Hello(); err != nil {
				t.Fatal(err)
			}
			if _, err := d.Next(); !errors.Is(err, ErrStreamGap) {
				t.Fatalf("want ErrStreamGap, got %v", err)
			}
		})
	}
}

func TestIngestFeedsSinkInOrder(t *testing.T) {
	stream := helloLine + "\n" +
		`{"seq":2,"type":"progress","data":{"draft":{"done":1,"total":10},"edit":{"done":0,"total":10}}}` + "\n" +
		"\n" + // a blank line is not an event
		`{"seq":3,"type":"ceiling","data":{"halted":true}}` + "\n"
	s := &recordingSink{}
	if err := Ingest(context.Background(), strings.NewReader(stream), s); err != nil {
		t.Fatalf("ingest: %v", err)
	}
	if s.hello.EngineRunID != "tr_1" {
		t.Fatalf("hello not bound: %+v", s.hello)
	}
	if got := len(s.applied); got != 2 {
		t.Fatalf("applied %d events, want 2", got)
	}
	if s.applied[0].Type != TypeProgress || s.applied[1].Type != TypeCeiling {
		t.Fatalf("order: %v", s.applied)
	}
}

func TestIngestStopsAtMalformedLine(t *testing.T) {
	stream := helloLine + "\n" + "{not json\n"
	s := &recordingSink{}
	err := Ingest(context.Background(), strings.NewReader(stream), s)
	if err == nil || errors.Is(err, io.EOF) {
		t.Fatalf("want a decode error, got %v", err)
	}
	if len(s.applied) != 0 {
		t.Fatalf("nothing may be applied from a broken stream, got %d", len(s.applied))
	}
}

type recordingSink struct {
	hello   Hello
	applied []Envelope
}

func (s *recordingSink) Begin(_ context.Context, h Hello) error { s.hello = h; return nil }

func (s *recordingSink) Apply(_ context.Context, ev Envelope, _ Cursor) error {
	s.applied = append(s.applied, ev)
	return nil
}

// TestANewFieldInsideAKnownEventIsIgnored is the seam's minor rule at the level the next engine
// release will actually exercise it: not a new event TYPE and not a new top-level key, but a new
// MEMBER inside the payload of an event this build already reads.
//
// ⚠ It is pinned separately from TestMinorVersionAndUnknownFieldsTolerated because that one proves a
// different thing — an unknown key in `hello` and an unknown event name — and neither would fail if
// somebody made a payload decode strict. The engine pack landing beside this one adds "how much was
// missing" to `ceiling` and moves the stream's minor; this reader has to survive both, and the cost
// of finding out that it does not is a run that halts on its ceiling and is materialized as nothing.
//
// ⚠ WHY THE PIN LIVES HERE AND NOT BESIDE THE CONSUMER. The code that actually reads these payloads
// is `internal/pgstore`'s sink, whose decode is a small wrapper around the same call this test makes
// — `json.Unmarshal(ev.Data, into)` with the event name folded into the error (internal/pgstore/sink.go,
// `func decode`). So there is ONE decode, not two, and pinning it here needs no database. What a
// reader must not conclude is that the sink is therefore free to become strict: if it ever grows a
// `DisallowUnknownFields`, this test keeps passing and the seam breaks anyway.
//
// ⚠ AND `StreamVersion` IS DELIBERATELY NOT BUMPED FOR ANY OF THIS. The engine moved to 1.3; this
// decoder still says 1.1, and that is correct rather than stale — the constant states what THIS build
// UNDERSTANDS, not what the other side writes (see its own doc comment), and this build reads none of
// the new members. Bumping it to look current would be exactly the false claim about our own
// vocabulary that the constant exists to prevent.
func TestANewFieldInsideAKnownEventIsIgnored(t *testing.T) {
	// The field names are the REAL ones the engine pack landing beside this one adds to `ceiling`
	// (`shortfall_micro_usd`), plus a nested one nobody has proposed — so the pin covers both "the
	// member we know is coming" and "any member at all".
	line := strings.Replace(helloLine, `"stream_version":"1.0"`, `"stream_version":"1.3"`, 1)
	d := NewDecoder(strings.NewReader(line + "\n" +
		`{"seq":2,"type":"ceiling","data":{"halted":true,"scope":"book","shortfall_micro_usd":1234,"whatever":{"nested":[1,2]}}}`))
	if _, err := d.Hello(); err != nil {
		t.Fatalf("a later minor must be accepted: %v", err)
	}
	ev, err := d.Next()
	if err != nil {
		t.Fatalf("an event carrying a field this build does not know must still decode: %v", err)
	}
	var c Ceiling
	if err := json.Unmarshal(ev.Data, &c); err != nil {
		t.Fatalf("the payload of a known event must decode past an unknown member: %v", err)
	}
	if !c.Halted || c.Scope != ScopeBook {
		t.Errorf("the members this build DOES know were lost: %+v", c)
	}

	// ⚠ THE SAME EVENT WITHOUT THE NEW MEMBER, because that is the shape this reader will meet FIRST.
	// The engine writes `shortfall_micro_usd` with `omitempty`: it is absent for a `day`-scope halt
	// and for a run whose ledger read failed at the moment of refusal. A pin whose every frame carries
	// the field would not cover the common case at all.
	d2 := NewDecoder(strings.NewReader(line + "\n" +
		`{"seq":2,"type":"ceiling","data":{"halted":true,"scope":"day"}}`))
	if _, err := d2.Hello(); err != nil {
		t.Fatal(err)
	}
	ev2, err := d2.Next()
	if err != nil {
		t.Fatalf("a ceiling without the new member did not decode: %v", err)
	}
	var c2 Ceiling
	if err := json.Unmarshal(ev2.Data, &c2); err != nil || !c2.Halted || c2.Scope != ScopeDay {
		t.Errorf("a day-scope halt with no shortfall member: %+v (%v)", c2, err)
	}

	// ⚠ AND A NESTED unknown OBJECT, which is a different path through a decoder than an unknown
	// SCALAR: a strict struct refuses them at different points, so a pin that only ever adds a number
	// can stay green while an object breaks. 1.3 adds `money: {units_resolved, units_deferred}` to
	// `finished` — this build reads only `outcome` there and must go on doing so.
	d3 := NewDecoder(strings.NewReader(line + "\n" +
		`{"seq":2,"type":"finished","data":{"outcome":"ceiling","money":{"units_resolved":7,"units_deferred":2}}}`))
	if _, err := d3.Hello(); err != nil {
		t.Fatal(err)
	}
	ev3, err := d3.Next()
	if err != nil {
		t.Fatalf("a finished carrying a nested unknown object did not decode: %v", err)
	}
	var f Finished
	if err := json.Unmarshal(ev3.Data, &f); err != nil {
		t.Fatalf("the payload of `finished` must decode past a nested unknown object: %v", err)
	}
	if f.Outcome != OutcomeCeiling {
		t.Errorf("the outcome this build DOES read was lost: %+v", f)
	}
}