package agent

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"sync"
)

// A translated line is consumed by Process.readLineWithLimit, whose limit
// includes the terminating newline. Keep the producer's limit identical and
// account for the exact encoded wire bytes rather than the source event size.
const (
	codexTranslatedOutputQueueEntries = 128
	codexTranslatedOutputMaxLineBytes = maxProcessLineBytes
	codexTranslatedOutputByteBudget   = codexTranslatedOutputMaxLineBytes
)

var errCodexTranslatedOutputSuperseded = errors.New("codex translated output generation was superseded")

// codexTranslatedReader acknowledges EOF only after Process has requested the
// read following the final translated line. Process handles each complete line
// before making that next read, so this is the driver-side terminal-consumption
// fence used by Wait.
type codexTranslatedReader struct {
	reader   *io.PipeReader
	consumed chan struct{}
	once     sync.Once
}

func newCodexTranslatedReader(reader *io.PipeReader) *codexTranslatedReader {
	return &codexTranslatedReader{reader: reader, consumed: make(chan struct{})}
}

func (reader *codexTranslatedReader) Read(p []byte) (int, error) {
	if reader == nil || reader.reader == nil {
		return 0, io.ErrClosedPipe
	}
	n, err := reader.reader.Read(p)
	if err != nil {
		reader.finish()
	}
	return n, err
}

func (reader *codexTranslatedReader) Close() error {
	if reader == nil || reader.reader == nil {
		return nil
	}
	err := reader.reader.Close()
	reader.finish()
	return err
}

func (reader *codexTranslatedReader) finish() {
	if reader == nil {
		return
	}
	reader.once.Do(func() { close(reader.consumed) })
}

// codexTranslatedOutput is a generation-scoped, lossless output queue.
//
// Producers back-pressure when either the entry queue or byte budget is full.
// Replacement first aborts the old queue, which releases every producer before
// it waits for the transport generation write fence. A single line that cannot
// fit in Process's parser limit (or in the configured queue budget) is a
// transport-fatal error; it is never silently dropped.
type codexTranslatedOutput struct {
	mu sync.Mutex

	lines          chan string
	done           chan struct{}
	spaceAvailable chan struct{}
	idle           chan struct{}

	maxLineBytes    int
	maxQueuedBytes  int
	queuedBytes     int
	activeEnqueues  int
	aborted         bool
	abortOnce       sync.Once
	closeOnce       sync.Once
	flushOnce       sync.Once
	writerCloseOnce sync.Once
	writerCloser    io.Closer
	writerCanceled  bool
	flushed         chan struct{}

	fatalErr        error
	fatalHandler    func(error)
	fatalDispatched bool
}

func (output *codexTranslatedOutput) attachWriter(writer io.Writer) {
	if output == nil {
		return
	}
	closer, ok := writer.(io.Closer)
	if !ok {
		return
	}
	output.mu.Lock()
	output.writerCloser = closer
	canceled := output.writerCanceled
	output.mu.Unlock()
	if canceled {
		output.closeWriter()
	}
}

func (output *codexTranslatedOutput) closeWriter() {
	if output == nil {
		return
	}
	output.mu.Lock()
	closer := output.writerCloser
	output.mu.Unlock()
	if closer == nil {
		return
	}
	output.writerCloseOnce.Do(func() {
		logIgnoredCleanupError("failed to close codex translated output writer", closer.Close())
	})
}

func (output *codexTranslatedOutput) isAborted() bool {
	if output == nil {
		return true
	}
	output.mu.Lock()
	defer output.mu.Unlock()
	return output.aborted
}

func (output *codexTranslatedOutput) isCanceled() bool {
	if output == nil {
		return true
	}
	output.mu.Lock()
	defer output.mu.Unlock()
	return output.writerCanceled
}

func newCodexTranslatedOutput(entryCapacity, maxLineBytes, maxQueuedBytes int) *codexTranslatedOutput {
	if entryCapacity <= 0 {
		entryCapacity = 1
	}
	if maxLineBytes <= 0 {
		maxLineBytes = codexTranslatedOutputMaxLineBytes
	}
	if maxQueuedBytes <= 0 {
		maxQueuedBytes = maxLineBytes
	}
	idle := make(chan struct{})
	close(idle)
	return &codexTranslatedOutput{
		lines:          make(chan string, entryCapacity),
		done:           make(chan struct{}),
		spaceAvailable: make(chan struct{}),
		idle:           idle,
		flushed:        make(chan struct{}),
		maxLineBytes:   maxLineBytes,
		maxQueuedBytes: maxQueuedBytes,
	}
}

func (output *codexTranslatedOutput) finishFlush() {
	if output == nil {
		return
	}
	output.flushOnce.Do(func() {
		close(output.flushed)
	})
}

func (output *codexTranslatedOutput) waitFlushed() {
	if output == nil {
		return
	}
	<-output.flushed
}

func (output *codexTranslatedOutput) setFatalHandler(handler func(error)) {
	if output == nil {
		return
	}
	var fatalErr error
	output.mu.Lock()
	output.fatalHandler = handler
	if output.fatalErr != nil && !output.fatalDispatched && handler != nil {
		output.fatalDispatched = true
		fatalErr = output.fatalErr
	}
	output.mu.Unlock()
	if fatalErr != nil {
		handler(fatalErr)
	}
}

func (output *codexTranslatedOutput) fatalError() error {
	if output == nil {
		return nil
	}
	output.mu.Lock()
	defer output.mu.Unlock()
	return output.fatalErr
}

func (output *codexTranslatedOutput) fail(err error) error {
	if output == nil || err == nil {
		return err
	}
	var handler func(error)
	var fatalErr error
	output.mu.Lock()
	if output.fatalErr == nil {
		output.fatalErr = err
	}
	fatalErr = output.fatalErr
	if !output.fatalDispatched && output.fatalHandler != nil {
		output.fatalDispatched = true
		handler = output.fatalHandler
	}
	output.mu.Unlock()
	output.cancel()
	if handler != nil {
		handler(fatalErr)
	}
	return fatalErr
}

// abort is idempotent and deliberately closes neither lines nor the translated
// writer. It only releases producers. A transport EOF seals the queue after
// abort so already-translated terminal/transcript events can still drain.
func (output *codexTranslatedOutput) abort() {
	if output == nil {
		return
	}
	output.abortOnce.Do(func() {
		output.mu.Lock()
		output.aborted = true
		close(output.done)
		output.signalSpaceLocked()
		output.mu.Unlock()
	})
}

// cancel is used for replacement, explicit cleanup, and output failure. Unlike
// a source-side EOF, those paths must close the io.PipeWriter immediately to
// release a flusher blocked behind a missing Process consumer.
func (output *codexTranslatedOutput) cancel() {
	if output == nil {
		return
	}
	output.abort()
	output.mu.Lock()
	output.writerCanceled = true
	output.mu.Unlock()
	output.closeWriter()
}

// seal stops producers and closes the item channel without canceling the
// writer. The flusher drains every admitted line in order, then closes writer.
func (output *codexTranslatedOutput) seal() {
	if output == nil {
		return
	}
	output.abort()
	output.closeOnce.Do(func() {
		output.mu.Lock()
		idle := output.idle
		output.mu.Unlock()
		<-idle
		close(output.lines)
	})
}

func (output *codexTranslatedOutput) close() {
	if output == nil {
		return
	}
	output.cancel()
	output.seal()
}

func (output *codexTranslatedOutput) beginEnqueue() error {
	output.mu.Lock()
	defer output.mu.Unlock()
	if output.aborted {
		if output.fatalErr != nil {
			return output.fatalErr
		}
		return errCodexTranslatedOutputSuperseded
	}
	if output.activeEnqueues == 0 {
		output.idle = make(chan struct{})
	}
	output.activeEnqueues++
	return nil
}

func (output *codexTranslatedOutput) finishEnqueue() {
	output.mu.Lock()
	output.activeEnqueues--
	if output.activeEnqueues == 0 {
		close(output.idle)
	}
	output.mu.Unlock()
}

func (output *codexTranslatedOutput) enqueue(line string, wireBytes int) error {
	if output == nil {
		return errCodexTranslatedOutputSuperseded
	}
	if wireBytes > output.maxLineBytes {
		return output.fail(fmt.Errorf(
			"codex translated output line is %d bytes, exceeding the %d-byte process parser limit",
			wireBytes,
			output.maxLineBytes,
		))
	}
	if wireBytes > output.maxQueuedBytes {
		return output.fail(fmt.Errorf(
			"codex translated output line is %d bytes, exceeding the %d-byte output queue budget",
			wireBytes,
			output.maxQueuedBytes,
		))
	}
	if err := output.beginEnqueue(); err != nil {
		return err
	}
	defer output.finishEnqueue()

	reserved := false
	for !reserved {
		output.mu.Lock()
		if output.aborted {
			err := output.fatalErr
			output.mu.Unlock()
			if err != nil {
				return err
			}
			return errCodexTranslatedOutputSuperseded
		}
		if output.queuedBytes <= output.maxQueuedBytes-wireBytes {
			output.queuedBytes += wireBytes
			reserved = true
			output.mu.Unlock()
			break
		}
		spaceAvailable := output.spaceAvailable
		output.mu.Unlock()

		select {
		case <-spaceAvailable:
		case <-output.done:
		}
	}

	select {
	case output.lines <- line:
		return nil
	case <-output.done:
		output.release(wireBytes)
		if err := output.fatalError(); err != nil {
			return err
		}
		return errCodexTranslatedOutputSuperseded
	}
}

func (output *codexTranslatedOutput) release(wireBytes int) {
	if output == nil || wireBytes <= 0 {
		return
	}
	output.mu.Lock()
	output.queuedBytes -= wireBytes
	if output.queuedBytes < 0 {
		output.queuedBytes = 0
	}
	output.signalSpaceLocked()
	output.mu.Unlock()
}

func (output *codexTranslatedOutput) signalSpaceLocked() {
	close(output.spaceAvailable)
	output.spaceAvailable = make(chan struct{})
}

func encodeCodexTranslatedEvent(event translatedEvent) (line string, wireBytes int, err error) {
	var encoded bytes.Buffer
	encoder := json.NewEncoder(&encoded)
	// json.Marshal expands <, >, and & to six-byte escape sequences. That can
	// turn a valid app-server payload into a line larger than Process accepts.
	// Disable HTML escaping and validate the final encoded bytes below instead.
	encoder.SetEscapeHTML(false)
	if err := encoder.Encode(event); err != nil {
		return "", 0, err
	}
	data := encoded.Bytes()
	if len(data) == 0 || data[len(data)-1] != '\n' {
		return "", 0, errors.New("translated event encoder omitted line terminator")
	}
	return string(data[:len(data)-1]), len(data), nil
}