package node

import (
	"context"
	"fmt"
	"time"

	"go.uber.org/zap"

	"iop/apps/node/internal/transport"
	runtime "iop/packages/go/execution"
	iop "iop/proto/gen/iop"
)

// OnProviderTunnelRequest handles an incoming ProviderTunnelRequest from a transport Session.
func (n *Node) OnProviderTunnelRequest(ctx context.Context, sess *transport.Session, req *iop.ProviderTunnelRequest) error {
	n.logger.Info("provider tunnel request received",
		zap.String("run_id", req.GetRunId()),
		zap.String("tunnel_id", req.GetTunnelId()),
		zap.String("adapter", req.GetAdapter()),
		zap.String("target", req.GetTarget()),
	)

	tr, err := providerTunnelRequestFromProto(req)
	if err != nil {
		n.sendTunnelError(sess, tr, fmt.Errorf("node: %w", err))
		return fmt.Errorf("node: %w", err)
	}

	n.configSetMu.RLock()
	configLocked := true
	defer func() {
		if configLocked {
			n.configSetMu.RUnlock()
		}
	}()

	adapter, err := n.router.LookupAdapter(tr.Adapter)
	if err != nil {
		n.sendTunnelError(sess, tr, fmt.Errorf("node: lookup adapter: %w", err))
		return fmt.Errorf("node: lookup adapter: %w", err)
	}

	tunnelAdapter, ok := adapter.(runtime.ProviderTunnelAdapter)
	if !ok {
		err := fmt.Errorf("node: adapter %q does not support tunneling", tr.Adapter)
		n.sendTunnelError(sess, tr, err)
		return err
	}

	var caps runtime.Capabilities
	if c, capsErr := adapter.Capabilities(ctx); capsErr == nil {
		caps = c
	}
	admission := n.admissionFor(tr.Adapter, caps)
	ticket, err := admission.acquire()
	if err != nil {
		n.logger.Warn("provider tunnel admission rejected",
			zap.String("run_id", tr.RunID),
			zap.String("tunnel_id", tr.TunnelID),
			zap.String("adapter", tr.Adapter),
			zap.String("reason", string(admissionRejectReason(err))),
		)
		n.sendTunnelError(sess, tr, err)
		return fmt.Errorf("node: provider tunnel %s: %w", tr.TunnelID, err)
	}
	preProviderOwned := true
	defer func() {
		if preProviderOwned {
			ticket.release()
		}
	}()

	// Consume only after adapter capacity admission, immediately before handing
	// the request to the adapter. A rejected or queued-out request never owns
	// plaintext provider credential bytes.
	material, err := n.consumeTunnelCredential(ctx, req, &tr)
	if err != nil {
		n.sendTunnelError(sess, tr, err)
		return err
	}

	var sender protoSender = noopSender{}
	var seq healthObservationSequencer
	nodeID := n.nodeID
	nodeAlias := ""
	if sess != nil {
		if sess.IsAlive() {
			sender = sess
			seq = sess
		}
		nodeID = sess.NodeID()
		nodeAlias = sess.Alias()
	}

	observer := newAttemptObserver(n.watchdogClock, time.Duration(tr.ResponseStallTimeoutMS)*time.Millisecond)
	sink := &tunnelSink{
		sess:      sender,
		nodeID:    nodeID,
		nodeAlias: nodeAlias,
		observer:  observer,
	}

	execCtx, cancel := context.WithCancel(ctx)
	if tr.TimeoutSec > 0 {
		execCtx, cancel = context.WithTimeout(ctx, time.Duration(tr.TimeoutSec)*time.Second)
	}

	h := &runHandle{
		runID:     tr.RunID,
		adapter:   tr.Adapter,
		target:    tr.Target,
		sessionID: normalizeSessionID(tr.SessionID),
		cancel:    cancel,
		done:      make(chan struct{}),
	}
	n.runs.register(h)

	probe := healthProbeFor(adapter, caps.AdapterName, caps.InstanceKey, tr.Target)

	configLocked = false
	n.configSetMu.RUnlock()
	preProviderOwned = false
	return n.executeTunnelAttempt(execCtx, cancel, tunnelAdapter, tr, sink, ticket, h, material, probe, seq)
}

func (n *Node) sendTunnelError(sess *transport.Session, tr runtime.ProviderTunnelRequest, err error) {
	if sess == nil || !sess.IsAlive() {
		return
	}
	tf := &iop.ProviderTunnelFrame{
		RunId:     tr.RunID,
		TunnelId:  tr.TunnelID,
		Sequence:  0,
		Kind:      iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR,
		Error:     err.Error(),
		Timestamp: time.Now().UnixNano(),
		NodeId:    sess.NodeID(),
		NodeAlias: sess.Alias(),
	}
	_ = sess.Send(tf)
}
