package ingest

import (
	"bufio"
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"os"
)

// JournalFile is the engine's event journal, in the BOOK's directory (D39.106 §2).
const JournalFile = "events.jsonl"

var (
	// ErrPayloadConflict is the same seq carrying a different payload. Re-reading a line is normal;
	// re-reading it and finding it CHANGED is not, and it cannot be reconciled by reading on —
	// either the journal was rewritten or two engine instances overlapped on one run id
	// (research/25 §эмиттер). The attempt is quarantined rather than materialized on a guess.
	ErrPayloadConflict = errors.New("ingest: a seq was re-read with a different payload")
	// ErrNoJournal is a journal that does not exist yet. Not a failure: the tailer is started when
	// the unit is, and the engine writes its first line whenever it gets there.
	ErrNoJournal = errors.New("ingest: journal not present yet")
	// ErrForeignStreamAhead is a handshake of ANOTHER stream met after this attempt has already
	// applied lines of its own. It is not corruption and the caller must not quarantine on it: the
	// read simply stops one line short, at the byte that handshake starts on, and the cursor stays
	// there so ownership can be decided again on the next pass (see tailFrom).
	//
	// It is REPORTED rather than swallowed because the state it leaves is otherwise invisible: the
	// projection stops with no error, no quarantine and no moving cursor, and a caller that cannot
	// tell it from "caught up" leaves a paying run's screen frozen with nothing saying why.
	ErrForeignStreamAhead = errors.New("ingest: another stream begins here and this attempt's own region has ended")
)

// Position is where the reader stands in one journal.
//
// LastSeq is half of the RATIFIED cursor (engine_run_id, seq); Offset is a HINT, and the difference
// is not pedantry — a hint may be wrong and is then discarded, whereas the cursor decides whether an
// event has already been applied. Both are committed in the SAME transaction as the effect, which
// is what makes at-least-once delivery safe to re-read.
type Position struct {
	Offset   int64
	LastSeq  int64
	LastHash []byte
}

// Cursor travels with an event so a sink can persist the two together.
type Cursor struct {
	// Offset is the byte position just past the line this event came from.
	Offset int64
	// SHA256 of the raw line. Stored for the highest applied seq only: it is what turns "I have seen
	// this seq" into "I have seen this seq and it said the same thing" (PD-105).
	SHA256 []byte
}

// Tail applies every COMPLETE line the journal has gained since `from`, and returns where it now
// stands. It returns ErrNoJournal while the file is absent, which the caller treats as "not yet".
//
// A partial trailing line is deliberately left alone: the writer appends, and a line without its
// newline is a line still being written, not a malformed one. Reading it would turn an ordinary
// interleaving into a decode error on a paid run.
//
// want is the engine run id this attempt owns, which the platform CHOOSES before the unit is created
// (runs.engineStreamID). Lines belonging to any other engine run id are skipped — handshakes this
// build could not even validate included: the journal is per BOOK and append-only, so a resumed run
// appends a second hello to the same file, and a reader that treated it as a corruption would stop
// exactly when the run resumed. What a foreign handshake SAYS is never judged; only whose it is.
//
// ⚠ Empty means the attempt was started before the platform named its stream, and then — and only
// then — the first hello at or after the starting offset is ADOPTED. That fallback is the shape of a
// measured defect and is kept only for attempts that predate the naming: a journal is per book, so
// "the first hello at my offset" is whatever was written there, and a foreign one had its events
// materialized onto this attempt and then quarantined the projection when the real handshake
// arrived at a seq that was not 1.
func Tail(ctx context.Context, path, want string, from Position, sink Sink) (Position, string, error) {
	f, err := os.Open(path)
	if errors.Is(err, fs.ErrNotExist) {
		return from, want, ErrNoJournal
	}
	if err != nil {
		return from, want, fmt.Errorf("ingest: open journal: %w", err)
	}
	defer f.Close()

	// A journal shorter than our hint is a journal that was replaced, which the ratified form says
	// cannot happen (append-only). Rather than read garbage at a stale offset, start over and say so.
	if st, err := f.Stat(); err == nil && st.Size() < from.Offset {
		return Position{}, want, fmt.Errorf("ingest: journal %s shrank from %d to %d bytes: it is not append-only", path, from.Offset, st.Size())
	}
	if from.Offset > 0 {
		if _, err := f.Seek(from.Offset, io.SeekStart); err != nil {
			return from, want, fmt.Errorf("ingest: seek journal: %w", err)
		}
	}
	return tailFrom(ctx, f, want, from, sink)
}

func tailFrom(ctx context.Context, r io.Reader, want string, pos Position, sink Sink) (Position, string, error) {
	br := bufio.NewReaderSize(r, maxLine)
	// Which stream the current lines belong to. A line carries a seq and NO stream id, so the only
	// thing that says whose it is, is the last handshake above it. We are inside our own region when
	// we have already APPLIED lines of it — which is what a non-zero last_seq means, the handshake
	// itself being seq 1.
	//
	// ⚠ It used to ask about the byte OFFSET instead, and that was wrong at exactly one moment: a new
	// attempt starts at the journal's current SIZE, so its offset is non-zero before it has read a
	// thing, and every line written there was assumed to be ours. Anything a foreign process appended
	// in that window — an operator's own tmctl in the book's directory — was then materialized onto
	// this attempt, `ceiling` events included.
	//
	// Getting this wrong is not a subtle loss in either direction: a previous attempt's seq 2 read
	// against our cursor's seq 2 is a different payload at the same number, i.e. exactly the
	// corruption signal, so a resumed run quarantined itself the first time the reader walked the
	// journal from the start.
	mine := want != "" && pos.LastSeq > 0
	// Whether the PLATFORM named this attempt's stream before the unit was created
	// (runs.EngineStreamID). A named attempt knows its own id for the life of the attempt, so the
	// question "is this line ours" has an answer at every byte; an attempt that predates the naming
	// adopts the first handshake it meets and can only answer from that point on.
	named := want != ""
	for {
		if err := ctx.Err(); err != nil {
			return pos, want, err
		}
		line, err := readLine(br)
		if errors.Is(err, io.EOF) {
			return pos, want, nil // caught up; the rest of the line, if any, is still being written
		}
		if err != nil {
			return pos, want, err
		}
		// Where the cursor stands ONCE this line has been applied. It is what the sink persists, in
		// the same transaction as the effect, so a crash between the two cannot skip the line.
		at := pos.Offset + int64(len(line))
		body := bytes.TrimRight(line, "\n")
		if len(bytes.TrimSpace(body)) == 0 {
			pos.Offset = at // a blank line is not an event
			continue
		}
		var ev Envelope
		if err := json.Unmarshal(body, &ev); err != nil {
			return pos, want, fmt.Errorf("ingest: malformed line at offset %d: %w", pos.Offset, err)
		}
		next, id, own, err := apply(ctx, ev, body, want, mine, named,
			Position{Offset: at, LastSeq: pos.LastSeq, LastHash: pos.LastHash}, sink)
		if errors.Is(err, ErrForeignStreamAhead) {
			// STOP HERE, and leave the cursor where it stands — one byte before the foreign
			// handshake. That the byte hint does not move is the whole point rather than a cost:
			// ownership is decided by the last handshake above a line, it is NOT persisted with the
			// cursor, and the next pass re-derives it from `pos.LastSeq > 0` alone. Walking past a
			// foreign handshake would therefore hand the NEXT pass a cursor that says "these lines
			// are ours" over a region that belongs to somebody else — and the foreign process's next
			// event lands on this attempt: a `ceiling` pauses a paying run, a `unit_done` credits
			// chapters nobody bought, and a foreign seq that meets ours quarantines a healthy
			// projection. Parking here costs one re-read of this line per sweep, and it buys an
			// answer that survives the pass.
			//
			// ⚠ WHOSE that stream is, is not knowable from here, and the tempting answer is wrong:
			// a second process of THIS SAME attempt announces itself under a DIFFERENT id. The
			// platform hands the attempt's id to every spawn of it (runs.engineStreamID takes run and
			// attempt, not the try), and an engine that finds that id has already written events for
			// this book mints a fresh one and carries on (backend pipeline.openEmitter). So a
			// handshake that is not ours may be an operator's own tmctl — or our own run, respawned.
			// Either way its lines are not ours to apply, and either way the read stops here; what
			// changes is that the caller cannot treat the state as short-lived.
			return pos, want, ErrForeignStreamAhead
		}
		if err != nil {
			return pos, want, err
		}
		pos, want, mine = next, id, own
	}
}

// apply decides what one decoded line means for the cursor and hands it to the sink. pos.Offset is
// already the position PAST this line, so whatever it returns is where the reader now stands. mine
// says whether the lines at this point belong to the attempt being tailed, and named whether the
// platform gave this attempt its stream id before the unit started.
func apply(ctx context.Context, ev Envelope, body []byte, want string, mine, named bool, pos Position, sink Sink) (Position, string, bool, error) {
	sum := sha256.Sum256(body)
	held := Position{Offset: pos.Offset, LastSeq: pos.LastSeq, LastHash: pos.LastHash}
	if ev.Type == TypeHello {
		var h Hello
		if err := json.Unmarshal(ev.Data, &h); err != nil {
			// Refused BEFORE the question of whose it is, and it cannot be otherwise: the engine run
			// id lives inside this payload, so a handshake that does not decode has no owner — it can
			// no more be skipped as somebody else's than read as ours.
			return pos, want, mine, fmt.Errorf("ingest: hello payload: %w", err)
		}
		if named && pos.LastSeq > 0 && h.EngineRunID != want {
			// Our own region of this journal is over and a stranger's begins. The caller stops here
			// rather than reading on, because ownership does not travel with the cursor.
			return pos, want, mine, ErrForeignStreamAhead
		}
		if want != "" && h.EngineRunID != want {
			// Another attempt's stream begins here — or something that only looks like one: an
			// operator's own tmctl in the book's directory, an older build speaking another major, a
			// broken build writing an empty id or a seq that is not 1. The journal is per BOOK and
			// append-only, so all of those are ordinary, and none of them is ours to judge: run the
			// rules below over a foreign handshake and its defects become errors of OUR read, which
			// `quarantines()` calls terminal — the projection of the paying attempt that merely
			// shares the file stops for good (PD-214, PD-426). Whose a line is decides before what it
			// says.
			return held, want, false, nil
		}
		// From here the handshake is ours by name or, with no name yet, the one this reader ADOPTS —
		// and adoption validates in full. Lowering these checks below the ownership question on the
		// nameless path would take a stream of a foreign major as our own: the same defect from the
		// other side.
		if err := checkVersion(h.StreamVersion); err != nil {
			return pos, want, mine, err
		}
		if h.EngineRunID == "" {
			return pos, want, mine, fmt.Errorf("%w: no engine_run_id", ErrBadHandshake)
		}
		// The handshake is seq 1 by definition, and here it also MOVES the cursor — so a hello at seq 0
		// would leave the cursor at zero and the first real event would then be dropped as a duplicate
		// of it. The pipe decoder has always required this (decoder.go); the file reader must too.
		if ev.Seq != 1 {
			return pos, want, mine, fmt.Errorf("%w: hello carries seq %d, want 1", ErrBadHandshake, ev.Seq)
		}
		if want == "" {
			// The handshake of the attempt this tailer was started for.
			if err := sink.Begin(ctx, h); err != nil {
				return pos, want, mine, err
			}
			want = h.EngineRunID
		}
		mine = true // our own handshake: adopted, or read again
		// and then it falls through to the ordinary rules below. The handshake is seq 1 of the stream,
		// so it MOVES THE CURSOR like any other line — and it has to, or the byte hint outlives a
		// last_seq that stayed at zero and the very next line reads as a gap. Its effect on the read
		// model is nothing; its effect on the cursor is the point.
	}
	if want == "" || !mine {
		// Either no handshake of ours has been seen yet, or these lines belong to another attempt of
		// the same book — the journal is per BOOK and append-only, so both are ordinary. Nothing is
		// materialized and, crucially, nothing is JUDGED: a foreign seq compared against our cursor
		// reads as a payload conflict at the same number.
		return held, want, mine, nil
	}
	switch {
	case ev.Seq == pos.LastSeq:
		// At-least-once delivery: re-reading the line the cursor stands on is NORMAL (PD-105, ratified
		// by D39.119) and must be idempotent. The same seq with a DIFFERENT payload is the one case a
		// high-water mark cannot absorb.
		if len(pos.LastHash) > 0 && !bytes.Equal(pos.LastHash, sum[:]) {
			return pos, want, mine, fmt.Errorf("%w: seq %d", ErrPayloadConflict, ev.Seq)
		}
		return held, want, mine, nil
	case ev.Seq < pos.LastSeq:
		// Already applied and already surpassed. Only the highest seq keeps a hash, so this one rests
		// on the journal's append-only guarantee rather than being verified — stated because an
		// unstated limit is the kind that gets read as a check.
		return held, want, mine, nil
	case ev.Seq > pos.LastSeq+1:
		return pos, want, mine, fmt.Errorf("%w: seq %d after %d", ErrStreamGap, ev.Seq, pos.LastSeq)
	}
	if err := sink.Apply(ctx, ev, Cursor{Offset: pos.Offset, SHA256: sum[:]}); err != nil {
		return pos, want, mine, err
	}
	return Position{Offset: pos.Offset, LastSeq: ev.Seq, LastHash: sum[:]}, want, mine, nil
}

// readLine returns one line INCLUDING its newline, or io.EOF when no complete line is available. A
// line longer than the buffer is refused rather than grown: events carry counters and ids, never
// text, so a megabyte line means the journal is not what we think it is.
//
// ⚠ Only a real end of file means "caught up". Mapping EVERY error to io.EOF — which the first
// version did — turns a failing disk into "nothing new": materialization would stop and no sweep
// would ever say why.
func readLine(br *bufio.Reader) ([]byte, error) {
	line, err := br.ReadSlice('\n')
	switch {
	case err == nil:
		return line, nil
	case errors.Is(err, bufio.ErrBufferFull):
		return nil, fmt.Errorf("ingest: journal line exceeds %d bytes", maxLine)
	case errors.Is(err, io.EOF):
		// Includes a trailing partial line: it is not finished being written.
		return nil, io.EOF
	default:
		return nil, fmt.Errorf("ingest: read journal: %w", err)
	}
}