package ingest

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"io"
)

var (
	// ErrUnsupportedVersion is a major-version refusal: materializing a stream whose meaning
	// changed is worse than not materializing it.
	ErrUnsupportedVersion = errors.New("ingest: unsupported stream version")
	// ErrNoHandshake is a stream whose first line is not the handshake. Ad-hoc unversioned JSON is
	// the documented rot path (docker jsonmessage) and is refused at the door.
	ErrNoHandshake = errors.New("ingest: stream does not open with hello")
	// ErrStreamGap is a seq that skipped or went backwards: lines were lost. Recovery is not
	// guesswork — the caller reconciles from `tmctl status --json`, the ratified resync channel.
	ErrStreamGap = errors.New("ingest: sequence gap")
	// ErrBadHandshake is a handshake that parses but does not identify the stream. Half of the
	// ratified idempotency key lives in it, so an empty engine_run_id would collapse every run's
	// events into one namespace rather than fail (PD-10a).
	ErrBadHandshake = errors.New("ingest: incomplete handshake")
	// ErrRepeatedHello is a second handshake mid-stream: the process on the other end restarted, or
	// two streams were spliced. Either way the identity behind the seq numbers changed, and the
	// version gate only ever inspected line 1 (PD-10c).
	ErrRepeatedHello = errors.New("ingest: hello after the handshake")
)

// maxLine caps one event. Events carry counters and ids, never text — the translated text travels
// as artifacts — so a megabyte line means the stream is not what we think it is.
const maxLine = 1 << 20

// Decoder reads the NDJSON event stream.
type Decoder struct {
	sc      *bufio.Scanner
	hello   Hello
	greeted bool
	lastSeq int64
}

func NewDecoder(r io.Reader) *Decoder {
	sc := bufio.NewScanner(r)
	sc.Buffer(make([]byte, 0, 64<<10), maxLine)
	return &Decoder{sc: sc}
}

// Hello reads and validates the handshake. It must be called before Next.
func (d *Decoder) Hello() (Hello, error) {
	ev, err := d.line()
	if err != nil {
		if errors.Is(err, io.EOF) {
			return Hello{}, ErrNoHandshake
		}
		return Hello{}, err
	}
	if ev.Type != TypeHello {
		return Hello{}, fmt.Errorf("%w: first line is %q", ErrNoHandshake, ev.Type)
	}
	// The handshake is seq 1 by definition. Without this check a stream whose first lines were lost
	// still opens, and the loss is undetectable: the gap check below only compares neighbours.
	if ev.Seq != 1 {
		return Hello{}, fmt.Errorf("%w: hello carries seq %d, want 1", ErrStreamGap, ev.Seq)
	}
	var h Hello
	if err := json.Unmarshal(ev.Data, &h); err != nil {
		return Hello{}, fmt.Errorf("ingest: hello payload: %w", err)
	}
	if err := checkVersion(h.StreamVersion); err != nil {
		return Hello{}, err
	}
	if h.EngineRunID == "" {
		return Hello{}, fmt.Errorf("%w: no engine_run_id", ErrBadHandshake)
	}
	d.hello, d.greeted = h, true
	d.lastSeq = ev.Seq
	return h, nil
}

// Next returns the next event, or io.EOF at the end of the stream. An unknown event type is
// returned as-is: tolerating it is the minor-version rule, and dropping it is the sink's decision.
func (d *Decoder) Next() (Envelope, error) {
	if !d.greeted {
		return Envelope{}, ErrNoHandshake
	}
	ev, err := d.line()
	if err != nil {
		return Envelope{}, err
	}
	if ev.Type == TypeHello {
		return Envelope{}, fmt.Errorf("%w: seq %d", ErrRepeatedHello, ev.Seq)
	}
	// Exactly one step, in one direction. A repeat is as wrong as a gap: inside one pipe there is
	// no at-least-once redelivery to absorb, so both are reported.
	if ev.Seq != d.lastSeq+1 {
		return Envelope{}, fmt.Errorf("%w: seq %d after %d", ErrStreamGap, ev.Seq, d.lastSeq)
	}
	d.lastSeq = ev.Seq
	return ev, nil
}

func (d *Decoder) line() (Envelope, error) {
	for d.sc.Scan() {
		raw := d.sc.Bytes()
		if len(raw) == 0 {
			continue // a blank line is not an event
		}
		var ev Envelope
		if err := json.Unmarshal(raw, &ev); err != nil {
			return Envelope{}, fmt.Errorf("ingest: malformed line: %w", err)
		}
		return ev, nil
	}
	if err := d.sc.Err(); err != nil {
		return Envelope{}, fmt.Errorf("ingest: read stream: %w", err)
	}
	return Envelope{}, io.EOF
}