package brokeripc

import (
	"context"
	"errors"
	"io"
	"sync"

	"github.com/bnema/vev/internal/adapters/brokerwire"
	"github.com/bnema/vev/internal/adapters/sessionwire"
	"github.com/bnema/vev/internal/ports"
	"github.com/bnema/vev/internal/protocol/wire"
)

// Logical stream bridging.
//
// One brokerwire logical stream carries the session protocol between client
// and daemon. The broker relays complete opaque envelope payloads through
// bounded framed carriages; neither preamble nor session message is decoded.
//
// Identity and cancellation stay per stream: a stream the core loses, a stream
// whose local consumer stalled, and a stream the peer closed each settle alone,
// sending at most one typed StreamClosed and releasing only that stream's
// resources. The connection and every sibling stream stay up.

// serverStream is the broker-side half of one logical stream.
type serverStream struct {
	session  *serverSession
	id       ports.BrokerStreamID
	core     ports.BrokerEnvelopeStream
	pipe     *streamPipe
	carriage *carriage

	mu     sync.Mutex
	closed bool
	done   chan struct{}
}

// newServerStream builds one bridged stream over the admitted core connection.
func newServerStream(s *serverSession, id ports.BrokerStreamID, core ports.BrokerEnvelopeStream) (*serverStream, error) {
	if s == nil || core == nil || id == 0 {
		return nil, ErrConfig
	}
	st := &serverStream{session: s, id: id, core: core, done: make(chan struct{})}
	st.pipe = newStreamPipe(int(s.ceilings.StreamChunkLimit), st.sendFrame, st.grantCredit)
	st.carriage = newCarriage(st.pipe)
	return st, nil
}

// sendFrame writes one credited outbound frame to the client. A client that
// is not reading slows this relay, and through the daemon link the daemon,
// instead of losing its stream.
func (st *serverStream) sendFrame(data []byte) error {
	return st.session.send(brokerwire.ServerStreamData{
		Epoch: st.session.epoch, Connection: st.session.scope.Connection, Stream: st.id, Data: data,
	})
}

// grantCredit returns consumed inbound credit to the client.
func (st *serverStream) grantCredit(credit uint64) error {
	return st.session.send(brokerwire.StreamWindowUpdate{
		Epoch: st.session.epoch, Connection: st.session.scope.Connection, Stream: st.id, Credit: credit,
	})
}

// deliver accepts one inbound chunk from the connection reader.
func (st *serverStream) deliver(chunk []byte) error { return st.pipe.deliver(chunk) }

// start launches the three per-stream workers. Every one is joined by the
// session's WaitGroup before shutdown returns.
func (st *serverStream) start() {
	st.session.wg.Add(3)
	go st.relayClient()
	go st.relayServer()
	go st.watchCore()
}

// relayClient carries complete client envelopes without decoding.
func (st *serverStream) relayClient() {
	defer st.session.wg.Done()
	for {
		envelope, err := st.carriage.RecvBounded(wire.AbsoluteEnvelopeLimit)
		if err != nil {
			st.fail(cancelOrCause(st.session.ctx, streamFailure(err)))
			return
		}
		if err := st.core.SendEnvelope(envelope.Payload); err != nil {
			st.fail(coreFailure(st.core, err))
			return
		}
	}
}

// relayServer carries complete daemon envelopes without decoding.
func (st *serverStream) relayServer() {
	defer st.session.wg.Done()
	for {
		payload, err := st.core.RecvEnvelope()
		if err != nil {
			st.fail(coreFailure(st.core, err))
			return
		}
		if len(payload) == 0 || len(payload) > wire.AbsoluteEnvelopeLimit {
			st.fail(ErrProtocol)
			return
		}
		if err := st.carriage.Send(wire.Envelope{Payload: payload}); err != nil {
			st.fail(streamFailure(err))
			return
		}
	}
}

// watchCore publishes the core connection's terminal outcome even when no read
// is in flight. It settles the stream itself only for a failure: an orderly
// core end (Err()==nil) is left to relayServer, whose own core.RecvEnvelope()
// call observes the same terminal outcome right after forwarding any reply
// that was already in flight, and settles the stream from there. Settling here
// on every Done would race relayServer's in-flight carriage.Send and could
// close the carriage while a received reply is still being written out to the
// client, discarding it.
func (st *serverStream) watchCore() {
	defer st.session.wg.Done()
	select {
	case <-st.core.Done():
		if err := st.core.Err(); err != nil {
			st.fail(coreFailure(st.core, nil))
		}
	case <-st.done:
	}
}

// fail settles the stream once and reports it to the session, which retires its
// wire identity and sends at most one typed StreamClosed.
func (st *serverStream) fail(err error) {
	if st.close(err) {
		st.session.settleStream(st, err)
	}
}

// close releases the stream exactly once: the pipe unblocks a stalled reader,
// the carriage joins its writer, and the core connection closes.
func (st *serverStream) close(cause error) bool {
	if st == nil {
		return false
	}
	st.mu.Lock()
	if st.closed {
		st.mu.Unlock()
		return false
	}
	st.closed = true
	st.mu.Unlock()
	st.pipe.closeWith(cause)
	_ = st.carriage.Close()
	_ = st.core.Close()
	close(st.done)
	return true
}

// streamFailure classifies one wire-side stream error: an orderly carriage end
// is a peer close, not a transport failure.
func streamFailure(err error) error {
	if err == nil || errors.Is(err, io.EOF) {
		return ErrStreamGone
	}
	return err
}

// coreFailure classifies one core connection outcome. The core's typed terminal
// error wins; an orderly end is a peer close.
func coreFailure(core ports.BrokerEnvelopeStream, err error) error {
	if err == nil && core != nil {
		err = core.Err()
	}
	if err == nil {
		return ErrStreamGone
	}
	var failure ports.BrokerError
	if errors.As(err, &failure) {
		return err
	}
	if errors.Is(err, io.EOF) || errors.Is(err, context.Canceled) {
		return ErrStreamGone
	}
	return err
}

// orderlyStream reports whether one stream outcome closes the stream without a
// failure detail.
func orderlyStream(err error) bool {
	return err == nil || errors.Is(err, ErrStreamGone)
}

// startStream admits one inbound open on the wire tracker and starts its bridge
// once the broker core established the stream.
func (s *serverSession) startStream(m brokerwire.OpenStream) {
	if err := s.conn.OpenStream(m.Stream); err != nil {
		if sendErr := s.refuseStream(m.Stream, admissionError(err)); sendErr != nil {
			s.abort(sendErr)
		}
		return
	}
	request := ports.BrokerOpenStreamRequest{
		Epoch: m.Epoch, Purpose: m.Purpose, Admission: m.Admission, Name: m.Name,
		Local:      m.Local,
		Connection: m.Connection, Stream: m.Stream,
		Endpoint: m.Endpoint, Registration: m.Registration,
		Target: m.Target, Env: m.Env, Policy: m.Policy,
		StartMode: m.StartMode,
	}
	s.wg.Add(1)
	go func() {
		defer s.wg.Done()
		core, err := s.core.OpenEnvelopeStream(s.ctx, request)
		if err != nil {
			_, _ = s.conn.CloseStream(m.Stream)
			if sendErr := s.refuseStream(m.Stream, err); sendErr != nil {
				s.abort(sendErr)
			}
			return
		}
		if core == nil {
			_, _ = s.conn.CloseStream(m.Stream)
			if sendErr := s.refuseStream(m.Stream, errors.Join(ports.BrokerAdmissionInvalid, errors.New("brokeripc: core returned a nil logical connection"))); sendErr != nil {
				s.abort(sendErr)
			}
			return
		}
		stream, err := newServerStream(s, m.Stream, core)
		if err != nil {
			_ = core.Close()
			_, _ = s.conn.CloseStream(m.Stream)
			if sendErr := s.refuseStream(m.Stream, err); sendErr != nil {
				s.abort(sendErr)
			}
			return
		}
		if !s.registerStream(stream) {
			stream.close(ErrSessionClosed)
			return
		}
		if err := s.conn.StreamOpened(m.Stream); err != nil {
			s.settleStream(stream, errors.Join(ErrProtocol, err))
			return
		}
		if err := s.send(brokerwire.StreamOpened{Epoch: s.epoch, Connection: s.scope.Connection, Stream: m.Stream}); err != nil {
			s.settleStream(stream, transportFailure(err))
			return
		}
		stream.start()
	}()
}

// registerStream installs one bridge, or reports false when the session is
// closing.
func (s *serverSession) registerStream(st *serverStream) bool {
	s.streamsMu.Lock()
	defer s.streamsMu.Unlock()
	if s.ctx.Err() != nil {
		return false
	}
	s.streams[st.id] = st
	return true
}

// unregisterStream removes one bridge, if present.
func (s *serverSession) unregisterStream(id ports.BrokerStreamID) {
	s.streamsMu.Lock()
	delete(s.streams, id)
	s.streamsMu.Unlock()
}

// lookupStream returns the bridge for one stream identity.
func (s *serverSession) lookupStream(id ports.BrokerStreamID) *serverStream {
	s.streamsMu.Lock()
	defer s.streamsMu.Unlock()
	return s.streams[id]
}

// settleStream retires one stream's wire identity and reports its outcome to the
// peer. A session that is already shutting down sends nothing: the carriage is
// closing and the peer is gone.
func (s *serverSession) settleStream(st *serverStream, err error) {
	if st == nil {
		return
	}
	st.close(err)
	s.unregisterStream(st.id)
	_, _ = s.conn.CloseStream(st.id)
	// Retire the broker-side stream reservation as well, so a stream that
	// settled for any reason is released at the core and not only in its
	// carriage.
	_ = s.core.CloseStream(s.scope.Connection, st.id)
	if s.ctx.Err() != nil {
		return
	}
	detail := brokerwire.ErrorDetail{}
	if !orderlyStream(err) {
		detail = errorDetail(err)
	}
	if sendErr := s.send(brokerwire.StreamClosed{
		Epoch: s.epoch, Connection: s.scope.Connection, Stream: st.id,
		Error: detail, HasError: detail != (brokerwire.ErrorDetail{}),
	}); sendErr != nil {
		s.abort(sendErr)
	}
}

// closeStreams settles every bridged stream during shutdown, releasing each
// core connection and session carriage exactly once.
func (s *serverSession) closeStreams() {
	s.streamsMu.Lock()
	streams := make([]*serverStream, 0, len(s.streams))
	for _, st := range s.streams {
		streams = append(streams, st)
	}
	s.streams = make(map[ports.BrokerStreamID]*serverStream)
	s.streamsMu.Unlock()
	for _, st := range streams {
		st.close(ErrSessionClosed)
	}
}

// closeStreamByPeer retires one stream the peer closed. No StreamClosed is sent
// back: the peer already knows.
func (s *serverSession) closeStreamByPeer(id ports.BrokerStreamID) {
	st := s.lookupStream(id)
	if _, err := s.conn.CloseStream(id); err != nil && !errors.Is(err, brokerwire.ErrFutureStream) {
		// A close for a stream that was never allocated is a peer protocol
		// violation; a late close for a retired stream is simply discarded.
		if errors.Is(err, brokerwire.ErrConnectionState) || errors.Is(err, brokerwire.ErrScopeMismatch) {
			s.abort(errors.Join(ErrProtocol, err))
			return
		}
	}
	if st == nil {
		return
	}
	s.unregisterStream(id)
	st.close(ErrStreamGone)
	// Retire the broker-side stream reservation as well: the peer's close ends
	// the client-scoped stream, not merely its carriage.
	_ = s.core.CloseStream(s.scope.Connection, id)
}

// deliverStreamData routes one inbound client chunk to its bridge. Data beyond
// the granted credit settles only that stream; a chunk for an identity the peer
// never opened is a protocol violation.
func (s *serverSession) deliverStreamData(m brokerwire.ClientStreamData) error {
	st, err := s.streamFor(m.Stream)
	if st == nil || err != nil {
		return err
	}
	if err := st.deliver(m.Data); errors.Is(err, ErrStreamCredit) {
		s.settleStream(st, ErrStreamCredit)
	}
	// Any other error means the stream already reached its terminal outcome;
	// its data is discarded.
	return nil
}

// grantStreamCredit applies one client credit grant to its bridge.
func (s *serverSession) grantStreamCredit(m brokerwire.StreamWindowUpdate) error {
	st, err := s.streamFor(m.Stream)
	if st == nil || err != nil {
		return err
	}
	if err := st.pipe.granted(m.Credit); err != nil {
		s.settleStream(st, err)
	}
	return nil
}

// streamFor classifies one stream-scoped frame and returns its live bridge. A
// frame for a retired stream, or one settled between the tracker check and the
// lookup, yields no bridge and no error; a frame for a stream that was never
// allocated, or before it was confirmed open, is a peer protocol violation.
func (s *serverSession) streamFor(id ports.BrokerStreamID) (*serverStream, error) {
	disposition, err := s.conn.StreamData(id)
	if err != nil {
		return nil, errors.Join(ErrProtocol, err)
	}
	if disposition == brokerwire.StreamDiscarded {
		return nil, nil
	}
	return s.lookupStream(id), nil
}

// clientStream is the client-side half of one logical stream: the sessionwire
// client connection over the stream's credit-controlled pipe, plus the terminal outcome a
// ports.BrokerLogicalConnection exposes independently of reads.
type clientStream struct {
	ports.ClientConnection
	client   *client
	id       ports.BrokerStreamID
	pipe     *streamPipe
	carriage *carriage

	opened chan error
	mu     sync.Mutex
	closed bool
	err    error
	done   chan struct{}
}

// newClientStream builds one client-side stream over the broker connection.
func newClientStream(c *client, id ports.BrokerStreamID) (*clientStream, error) {
	if c == nil || id == 0 {
		return nil, ErrConfig
	}
	st := &clientStream{
		client: c,
		id:     id,
		opened: make(chan error, 1),
		done:   make(chan struct{}),
	}
	st.pipe = newStreamPipe(int(c.ceilings.StreamChunkLimit), st.sendFrame, st.grantCredit)
	st.carriage = newCarriage(st.pipe)
	st.ClientConnection = sessionwire.NewClientConnection(st.carriage)
	if st.ClientConnection == nil {
		return nil, ErrConfig
	}
	return st, nil
}

// sendFrame writes one credited outbound frame to the broker.
func (st *clientStream) sendFrame(data []byte) error {
	return st.client.send(brokerwire.ClientStreamData{
		Epoch: st.client.scope.Epoch, Connection: st.client.scope.Connection, Stream: st.id, Data: data,
	})
}

// grantCredit returns consumed inbound credit to the broker. It is written
// synchronously because a dropped grant would stall the stream forever; the
// write never waits long, since the broker's reader never blocks on a stream.
func (st *clientStream) grantCredit(credit uint64) error {
	return st.client.send(brokerwire.StreamWindowUpdate{
		Epoch: st.client.scope.Epoch, Connection: st.client.scope.Connection, Stream: st.id, Credit: credit,
	})
}

// deliver accepts one inbound server chunk.
func (st *clientStream) deliver(chunk []byte) error { return st.pipe.deliver(chunk) }

// Done closes when the stream reached its terminal outcome.
func (st *clientStream) Done() <-chan struct{} { return st.done }

// Err reports the terminal cause; nil is an orderly close.
func (st *clientStream) Err() error {
	st.mu.Lock()
	defer st.mu.Unlock()
	return st.err
}

// Close ends the stream: it sends one CloseStream frame best-effort, releases
// the carriage, and resolves any pending open. It is idempotent and
// concurrent-safe.
func (st *clientStream) Close() error {
	if st == nil {
		return nil
	}
	st.client.requestStreamClose(st.id)
	st.terminate(nil)
	return nil
}

// terminate settles the stream exactly once, publishes its terminal cause (nil
// for an orderly close), and resolves a pending open.
func (st *clientStream) terminate(cause error) bool {
	if st == nil {
		return false
	}
	st.mu.Lock()
	if st.closed {
		st.mu.Unlock()
		return false
	}
	st.closed = true
	st.err = cause
	st.mu.Unlock()
	st.pipe.closeWith(cause)
	_ = st.carriage.Close()
	st.client.unregisterStream(st.id)
	// Retire the identity on the connection's tracker as well, so a frame that
	// arrives after local retirement is classified as discarded.
	_, _ = st.client.conn.CloseStream(st.id)
	select {
	case st.opened <- cause:
	default:
	}
	close(st.done)
	return true
}

// fail settles the stream and tells the broker to retire it, so a stream-local
// failure never leaves the peer half-open.
func (st *clientStream) fail(err error) {
	if st.terminate(err) {
		st.client.requestStreamClose(st.id)
	}
}