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")
)

// 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)
	}
	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
	}
	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
	}
	switch {
	case ev.Seq <= d.lastSeq:
		// A duplicate cannot happen inside one pipe, so it is a defect rather than at-least-once
		// redelivery — and it is reported, not silently absorbed.
		return Envelope{}, fmt.Errorf("%w: seq %d after %d", ErrStreamGap, ev.Seq, d.lastSeq)
	case 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
}