package ingest

import (
	"errors"
	"io"
	"strings"
	"testing"
)

// The decoder is the only place where bytes produced by another process become platform state, so
// it is fuzzed rather than merely exampled. The oracles are the invariants the rest of the ingest
// path is built on; each is stated as "if the decoder said yes, then …", because a refusal is
// always an acceptable answer and only an ACCEPTANCE can be wrong.
//
//  1. it never panics, whatever arrives;
//  2. an accepted handshake identifies the stream (engine_run_id non-empty — half of the
//     idempotency key) and is seq 1;
//  3. accepted events increase seq by exactly one, so a gap can never be silently absorbed;
//  4. no event is ever returned before a successful handshake;
//  5. hello never appears again after the handshake, at any version.
func FuzzDecoder(f *testing.F) {
	f.Add(helloLine)
	f.Add(helloLine + "\n" + `{"seq":2,"type":"progress","data":{"draft":{"done":1,"total":2}}}`)
	f.Add(helloLine + "\n" + helloLine)
	f.Add(`{"seq":9,"type":"hello","data":{"stream_version":"1.0","engine_run_id":"x"}}`)
	f.Add(`{"seq":1,"type":"hello","data":{"stream_version":"1.0","engine_run_id":""}}`)
	f.Add(`{"seq":1,"type":"hello","data":{"stream_version":"9.9","engine_run_id":"x"}}`)
	f.Add("\n\n\n")
	f.Add("{not json")

	f.Fuzz(func(t *testing.T, stream string) {
		d := NewDecoder(strings.NewReader(stream))

		// Oracle 4: nothing may come out before the handshake.
		if _, err := d.Next(); !errors.Is(err, ErrNoHandshake) {
			t.Fatalf("Next before Hello returned %v, want ErrNoHandshake", err)
		}

		h, err := d.Hello()
		if err != nil {
			return // a refusal is always allowed
		}
		// Oracle 2.
		if h.EngineRunID == "" {
			t.Fatal("accepted a handshake with no engine_run_id: half the idempotency key")
		}
		if maj, err := major(h.StreamVersion); err != nil || maj != 1 {
			t.Fatalf("accepted stream version %q", h.StreamVersion)
		}

		last := int64(1) // oracle 2: the handshake is seq 1 or it is refused
		for {
			ev, err := d.Next()
			if err != nil {
				if errors.Is(err, io.EOF) {
					return
				}
				return // any refusal is allowed; only acceptances are constrained
			}
			// Oracle 3.
			if ev.Seq != last+1 {
				t.Fatalf("accepted seq %d after %d", ev.Seq, last)
			}
			// Oracle 5.
			if ev.Type == TypeHello {
				t.Fatal("accepted a second handshake mid-stream")
			}
			last = ev.Seq
		}
	})
}

func TestHandshakeMustIdentifyTheStream(t *testing.T) {
	t.Run("no engine_run_id", func(t *testing.T) {
		line := strings.Replace(helloLine, `"engine_run_id":"tr_1"`, `"engine_run_id":""`, 1)
		if _, err := NewDecoder(strings.NewReader(line)).Hello(); !errors.Is(err, ErrBadHandshake) {
			t.Fatalf("want ErrBadHandshake, got %v", err)
		}
	})
	t.Run("handshake is seq 1", func(t *testing.T) {
		line := strings.Replace(helloLine, `"seq":1`, `"seq":4`, 1)
		if _, err := NewDecoder(strings.NewReader(line)).Hello(); !errors.Is(err, ErrStreamGap) {
			t.Fatalf("want ErrStreamGap, got %v", err)
		}
	})
	t.Run("no second handshake", func(t *testing.T) {
		second := strings.Replace(helloLine, `"seq":1`, `"seq":2`, 1)
		// A major version the gate would have refused on line 1, arriving on line 2.
		second = strings.Replace(second, `"stream_version":"1.0"`, `"stream_version":"9.9"`, 1)
		d := NewDecoder(strings.NewReader(helloLine + "\n" + second))
		if _, err := d.Hello(); err != nil {
			t.Fatal(err)
		}
		if _, err := d.Next(); !errors.Is(err, ErrRepeatedHello) {
			t.Fatalf("want ErrRepeatedHello, got %v", err)
		}
	})
}