coder/coderd/tailnet.go

package coderd

import (
	"bufio"
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"net"
	"net/http"
	"net/http/httptrace"
	"net/http/httputil"
	"net/netip"
	"net/url"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/google/uuid"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tailscale/wireguard-go/device"
	"go.opentelemetry.io/otel/trace"
	"golang.org/x/xerrors"
	"tailscale.com/derp"
	"tailscale.com/tailcfg"

	"cdr.dev/slog/v3"
	"github.com/coder/coder/v2/coderd/httpmw/loggermw"
	"github.com/coder/coder/v2/coderd/tracing"
	"github.com/coder/coder/v2/coderd/workspaceapps"
	"github.com/coder/coder/v2/coderd/workspaceapps/appurl"
	"github.com/coder/coder/v2/codersdk"
	"github.com/coder/coder/v2/codersdk/workspacesdk"
	"github.com/coder/coder/v2/site"
	"github.com/coder/coder/v2/tailnet"
	"github.com/coder/coder/v2/tailnet/proto"
	"github.com/coder/quartz"
)

var tailnetTransport *http.Transport

func init() {
	tp, valid := http.DefaultTransport.(*http.Transport)
	if !valid {
		panic("dev error: default transport is the wrong type")
	}
	tailnetTransport = tp.Clone()
	// We do not want to respect the proxy settings from the environment, since
	// all network traffic happens over wireguard.
	tailnetTransport.Proxy = nil
}

var _ workspaceapps.AgentProvider = (*ServerTailnet)(nil)

// NewServerTailnet creates a new tailnet intended for use by coderd.
func NewServerTailnet(
	ctx context.Context,
	logger slog.Logger,
	derpServer *derp.Server,
	dialer tailnet.ControlProtocolDialer,
	derpForceWebSockets bool,
	blockEndpoints bool,
	traceProvider trace.TracerProvider,
) (*ServerTailnet, error) {
	logger = logger.Named("servertailnet")
	conn, err := tailnet.NewConn(&tailnet.Options{
		Addresses:           []netip.Prefix{tailnet.TailscaleServicePrefix.RandomPrefix()},
		DERPForceWebSockets: derpForceWebSockets,
		Logger:              logger,
		BlockEndpoints:      blockEndpoints,
	})
	if err != nil {
		return nil, xerrors.Errorf("create tailnet conn: %w", err)
	}
	serverCtx, cancel := context.WithCancel(ctx)

	derpConnects := prometheus.NewCounter(prometheus.CounterOpts{
		Namespace: "coder",
		Subsystem: "servertailnet",
		Name:      "derp_connects_total",
		Help:      "Number of times the server tailnet connected to the embedded DERP relay. Stays 0 when the embedded relay is disabled.",
	})

	// This is set to allow local DERP traffic to be proxied through memory
	// instead of needing to hit the external access URL. Don't use the ctx
	// given in this callback, it's only valid while connecting.
	if derpServer != nil {
		conn.SetDERPRegionDialer(embeddedDERPRegionDialer(ctx, logger, derpServer, derpConnects))
	}

	tracer := traceProvider.Tracer(tracing.TracerName)

	controller := tailnet.NewController(logger, dialer)
	// it's important to set the DERPRegionDialer above _before_ we set the DERP map so that if
	// there is an embedded relay, we use the local in-memory dialer.
	controller.DERPCtrl = tailnet.NewBasicDERPController(logger, nil, conn)
	coordCtrl := NewMultiAgentController(serverCtx, logger, tracer, conn)
	controller.CoordCtrl = coordCtrl
	// TODO: support controller.TelemetryCtrl

	tn := &ServerTailnet{
		ctx:                 serverCtx,
		cancel:              cancel,
		logger:              logger,
		tracer:              tracer,
		conn:                conn,
		coordinatee:         conn,
		controller:          controller,
		coordCtrl:           coordCtrl,
		transport:           tailnetTransport.Clone(),
		clock:               quartz.NewReal(),
		readPeerDiagnostics: conn.GetPeerDiagnostics,
		peerDiagnosticsSlot: make(chan struct{}, 1),
		connsPerAgent: prometheus.NewGaugeVec(prometheus.GaugeOpts{
			Namespace: "coder",
			Subsystem: "servertailnet",
			Name:      "open_connections",
			Help:      "Total number of TCP connections currently open to workspace agents.",
		}, []string{"network"}),
		totalConns: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "coder",
			Subsystem: "servertailnet",
			Name:      "connections_total",
			Help:      "Total number of TCP connections made to workspace agents.",
		}, []string{"network"}),
		derpConnects: derpConnects,
		agentUnreachable: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "coder",
			Subsystem: "servertailnet",
			Name:      "agent_unreachable_total",
			Help:      "Number of connection attempts where the workspace agent did not answer in time, by reason: no_node (this server has no node for the agent), peer_lost (the agent's or this server's coordinator connection ended), no_handshake (no WireGuard handshake yet), handshake_stale (last handshake over 180s ago), handshake_ok (last handshake under 180s ago), diagnostics_timeout (peer state not readable within 1s), client_canceled (client left within 5s).",
		}, []string{"reason"}),
		awaitReachable: prometheus.NewHistogram(prometheus.HistogramOpts{
			Namespace: "coder",
			Subsystem: "servertailnet",
			Name:      "await_reachable_seconds",
			Help:      "Time until a workspace agent answered the first ping when a new connection was opened.",
			Buckets:   []float64{0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60},
		}),
	}
	tn.transport.DialContext = tn.dialContext
	// These options are mostly just picked at random, and they can likely be
	// fine-tuned further. Generally, users are running applications in dev mode
	// which can generate hundreds of requests per page load, so we increased
	// MaxIdleConnsPerHost from 2 to 6 and removed the limit of total idle
	// conns.
	tn.transport.MaxIdleConnsPerHost = 6
	tn.transport.MaxIdleConns = 0
	tn.transport.IdleConnTimeout = 10 * time.Minute
	// We intentionally don't verify the certificate chain here.
	// The connection to the workspace is already established and most
	// apps are already going to be accessed over plain HTTP, this config
	// simply allows apps being run over HTTPS to be accessed without error --
	// many of which may be using self-signed certs.
	tn.transport.TLSClientConfig = &tls.Config{
		MinVersion: tls.VersionTLS12,
		//nolint:gosec
		InsecureSkipVerify: true,
	}

	tn.controller.Run(tn.ctx)
	return tn, nil
}

// Conn is used to access the underlying tailnet conn of the ServerTailnet. It
// should only be used for read-only purposes.
func (s *ServerTailnet) Conn() *tailnet.Conn {
	return s.conn
}

func (s *ServerTailnet) Describe(descs chan<- *prometheus.Desc) {
	s.connsPerAgent.Describe(descs)
	s.totalConns.Describe(descs)
	s.derpConnects.Describe(descs)
	s.agentUnreachable.Describe(descs)
	s.awaitReachable.Describe(descs)
}

func (s *ServerTailnet) Collect(metrics chan<- prometheus.Metric) {
	s.connsPerAgent.Collect(metrics)
	s.totalConns.Collect(metrics)
	s.derpConnects.Collect(metrics)
	s.agentUnreachable.Collect(metrics)
	s.awaitReachable.Collect(metrics)
}

type ServerTailnet struct {
	ctx    context.Context
	cancel func()

	logger slog.Logger
	tracer trace.Tracer

	// in prod, these are the same, but coordinatee is a subset of Conn's
	// methods which makes some tests easier.
	conn        *tailnet.Conn
	coordinatee tailnet.Coordinatee

	controller *tailnet.Controller
	coordCtrl  *MultiAgentController

	transport *http.Transport
	clock     quartz.Clock

	connsPerAgent    *prometheus.GaugeVec
	totalConns       *prometheus.CounterVec
	derpConnects     prometheus.Counter
	agentUnreachable *prometheus.CounterVec
	awaitReachable   prometheus.Histogram

	// peerDiagnosticsSlot holds one token while a peer diagnostics read is
	// in flight.
	peerDiagnosticsSlot chan struct{}
	readPeerDiagnostics func(agentID uuid.UUID) tailnet.PeerDiagnostics
}

func (s *ServerTailnet) ReverseProxy(targetURL, dashboardURL *url.URL, agentID uuid.UUID, app appurl.ApplicationURL, wildcardHostname string) *httputil.ReverseProxy {
	// Rewrite the targetURL's Host to point to the agent's IP. This is
	// necessary because due to TCP connection caching, each agent needs to be
	// addressed invidivually. Otherwise, all connections get dialed as
	// "localhost:port", causing connections to be shared across agents.
	tgt := *targetURL
	_, port, _ := net.SplitHostPort(tgt.Host)
	tgt.Host = net.JoinHostPort(tailnet.TailscaleServicePrefix.AddrFromUUID(agentID).String(), port)

	proxy := httputil.NewSingleHostReverseProxy(&tgt)
	proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, theErr error) {
		theErr = s.reportProxyDialFailure(r.Context(), agentID, theErr)
		var (
			desc           = "Failed to proxy request to application: " + theErr.Error()
			additionalInfo = ""
			actions        = []site.Action{}
		)

		var tlsError tls.RecordHeaderError
		if (errors.As(theErr, &tlsError) && tlsError.Msg == "first record does not look like a TLS handshake") ||
			errors.Is(theErr, http.ErrSchemeMismatch) {
			// If the error is due to an HTTP/HTTPS mismatch, we can provide a
			// more helpful error message with redirect buttons.
			switchURL := url.URL{
				Scheme: dashboardURL.Scheme,
			}
			_, protocol, isPort := app.PortInfo()
			if isPort {
				targetProtocol := "https"
				if protocol == "https" {
					targetProtocol = "http"
				}
				app = app.ChangePortProtocol(targetProtocol)

				switchURL.Host = fmt.Sprintf("%s%s", app.String(), strings.TrimPrefix(wildcardHostname, "*"))
				actions = append(actions, site.Action{
					URL:  switchURL.String(),
					Text: fmt.Sprintf("Switch to %s", strings.ToUpper(targetProtocol)),
				})
				additionalInfo += fmt.Sprintf("This error seems to be due to an app protocol mismatch, try switching to %s.", strings.ToUpper(targetProtocol))
			}
		}

		site.RenderStaticErrorPage(w, r, site.ErrorPageData{
			Status:      http.StatusBadGateway,
			Title:       "Bad Gateway",
			Description: desc,
			Actions: append(actions, []site.Action{
				{
					Text: "Retry",
				},
				{
					URL:  dashboardURL.String(),
					Text: "Back to site",
				},
			}...),
			AdditionalInfo: additionalInfo,
		})
	}
	proxy.Director = s.director(agentID, proxy.Director)
	proxy.Transport = s.transport

	return proxy
}

type agentIDKey struct{}

// director makes sure agentIDKey and a dialState are set on the context in the
// reverse proxy. The transport uses the agent ID to pick which agent to dial,
// and the dial and the error handler share the dialState.
func (*ServerTailnet) director(agentID uuid.UUID, prev func(req *http.Request)) func(req *http.Request) {
	return func(req *http.Request) {
		ds := &dialState{}
		ctx := context.WithValue(req.Context(), agentIDKey{}, agentID)
		ctx = context.WithValue(ctx, dialStateKey{}, ds)
		// A pooled connection skips dialContext, so the dial phases cannot show
		// that the request has a connection. GotConn fires either way. GetConn
		// runs on the request goroutine before each attempt, so a retry clears
		// the connection the previous attempt used.
		ctx = httptrace.WithClientTrace(ctx, &httptrace.ClientTrace{
			GetConn: func(string) { ds.setGotConn(false) },
			GotConn: func(httptrace.GotConnInfo) { ds.setGotConn(true) },
		})
		*req = *req.WithContext(ctx)
		prev(req)
	}
}

// proxyDialTimeout caps a reverse proxy dial. The HTTP transport keeps dialing
// after the request ends, so without a cap an unreachable agent is pinged
// until the ServerTailnet closes.
const proxyDialTimeout = time.Minute

// dialPhase is the step of a reverse proxy dial that was in progress when the
// request ended.
type dialPhase int

const (
	dialPhaseNone dialPhase = iota
	dialPhaseAwaitReachable
	dialPhaseTCPDial
	dialPhaseDone
)

func (p dialPhase) String() string {
	switch p {
	case dialPhaseAwaitReachable:
		return "await_reachable"
	case dialPhaseTCPDial:
		return "tcp_dial"
	case dialPhaseDone:
		return "done"
	default:
		return "none"
	}
}

// dialState is shared by the request goroutine and the transport's dial
// goroutine through the request context. The error handler reads it to tell
// a dial that never finished from a failure on an established connection.
type dialState struct {
	mu      sync.Mutex
	phase   dialPhase
	since   time.Time
	gotConn bool
}

type dialStateKey struct{}

func dialStateFromContext(ctx context.Context) *dialState {
	ds, _ := ctx.Value(dialStateKey{}).(*dialState)
	return ds
}

func (d *dialState) set(phase dialPhase) {
	if d == nil {
		return
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	d.phase = phase
	d.since = time.Now()
}

// get returns the current phase and how long it has been in progress.
func (d *dialState) get() (dialPhase, time.Duration) {
	if d == nil {
		return dialPhaseNone, 0
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.phase, time.Since(d.since)
}

func (d *dialState) setGotConn(gotConn bool) {
	if d == nil {
		return
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	d.gotConn = gotConn
}

// hadConn reports whether the transport gave the request a connection.
func (d *dialState) hadConn() bool {
	if d == nil {
		return false
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.gotConn
}

func (s *ServerTailnet) dialContext(ctx context.Context, network, addr string) (net.Conn, error) {
	agentID, ok := ctx.Value(agentIDKey{}).(uuid.UUID)
	if !ok {
		return nil, xerrors.Errorf("no agent id attached")
	}
	// Only the connect steps below use the context. The returned conn is not
	// tied to it, so canceling after the dial is safe.
	ctx, cancel := context.WithTimeout(ctx, proxyDialTimeout)
	defer cancel()

	// The reverse proxy's error handler records unreachable agents on the
	// request goroutine when the request ends, which can be before this dial
	// returns, so this path uses acquireAgent, which neither logs nor counts.
	ds := dialStateFromContext(ctx)
	ds.set(dialPhaseAwaitReachable)
	conn, release, err := s.acquireAgent(ctx, agentID)
	if err != nil {
		return nil, xerrors.Errorf("acquire agent conn: %w", err)
	}

	ds.set(dialPhaseTCPDial)
	nc, err := conn.DialContext(ctx, network, addr)
	if err != nil {
		release()
		return nil, xerrors.Errorf("dial context: %w", err)
	}
	ds.set(dialPhaseDone)

	s.connsPerAgent.WithLabelValues("tcp").Inc()
	s.totalConns.WithLabelValues("tcp").Inc()
	return &instrumentedConn{
		Conn:          &netConnCloser{Conn: nc, close: release},
		agentID:       agentID,
		connsPerAgent: s.connsPerAgent,
	}, nil
}

// reportProxyDialFailure records a reverse proxy request that ended while its
// dial was still in progress. It runs on the request goroutine before the 502
// is written, which is the last point where the request log line can still
// take fields: the transport detaches the dial from the request deadline, so
// the dial goroutine learns of the failure later, or not at all. Errors on an
// established connection are returned unchanged.
func (s *ServerTailnet) reportProxyDialFailure(ctx context.Context, agentID uuid.UUID, err error) error {
	ds := dialStateFromContext(ctx)
	// The transport may give the request a pooled connection while the dial it
	// started is still waiting on the agent. An error after that is from the
	// connection the request used, so the dial phase does not describe it.
	if ds.hadConn() {
		return err
	}
	phase, elapsed := ds.get()
	switch phase {
	case dialPhaseAwaitReachable:
		// A request that is still waiting can also see an error from before
		// the first ping, such as a failed coordinator send. AgentConn does
		// not count those, so neither does this path.
		if !errors.Is(err, errAgentUnreachable) && ctx.Err() == nil {
			return err
		}
		unreachable := s.recordAgentUnreachable(ctx, agentID, elapsed, slog.F("dial_phase", phase.String()))
		if errors.Is(err, errAgentUnreachable) {
			return err
		}
		return xerrors.Errorf("%s: %w", unreachable.Error(), err)
	case dialPhaseTCPDial:
		// The agent answered a ping, so the tunnel is up and the failure is
		// above it: the netstack dial or an app that is not accepting. Name
		// the phase on the log line, but do not count it as unreachable.
		if rl := loggermw.RequestLoggerFromContext(ctx); rl != nil {
			rl.WithFields(
				slog.F("agent_id", agentID),
				slog.F("dial_phase", phase.String()),
				slog.F("tcp_dial_after", elapsed),
			)
		}
		return err
	default:
		return err
	}
}

// errAgentUnreachable is returned by acquireAgent when the agent did not
// answer a ping before the context ended.
var errAgentUnreachable = xerrors.New("agent is unreachable")

func (s *ServerTailnet) AgentConn(ctx context.Context, agentID uuid.UUID) (workspacesdk.AgentConn, func(), error) {
	start := s.clock.Now()
	conn, release, err := s.acquireAgent(ctx, agentID)
	cause := context.Cause(ctx)
	if errors.Is(err, errAgentUnreachable) &&
		!errors.Is(cause, workspacesdk.ErrDialAbandoned) &&
		!errors.Is(cause, workspacesdk.ErrReadinessProbeTimeout) {
		return nil, nil, s.recordAgentUnreachable(ctx, agentID, s.clock.Since(start))
	}
	return conn, release, err
}

// acquireAgent adds the agent to the coordinator and waits for it to answer a
// ping. On timeout it returns errAgentUnreachable and leaves logging and
// metrics to the caller, which knows whether a request is still waiting.
func (s *ServerTailnet) acquireAgent(ctx context.Context, agentID uuid.UUID) (workspacesdk.AgentConn, func(), error) {
	var (
		conn workspacesdk.AgentConn
		ret  func()
	)

	s.logger.Debug(s.ctx, "acquiring agent", slog.F("agent_id", agentID))
	err := s.coordCtrl.ensureAgent(agentID)
	if err != nil {
		return nil, nil, xerrors.Errorf("ensure agent: %w", err)
	}
	ret = s.coordCtrl.acquireTicket(agentID)

	conn = workspacesdk.NewAgentConn(s.conn, workspacesdk.AgentConnOptions{
		AgentID:   agentID,
		CloseFunc: func() error { return workspacesdk.ErrSkipClose },
		Logger:    s.logger,
	})

	// Since we now have an open conn, be careful to close it if we error
	// without returning it to the user.

	start := s.clock.Now()
	reachable := conn.AwaitReachable(ctx)
	if !reachable {
		ret()
		return nil, nil, errAgentUnreachable
	}
	s.awaitReachable.Observe(s.clock.Since(start).Seconds())

	return conn, ret, nil
}

// peerDiagnosticsTimeout caps how long a failed connection waits for peer
// diagnostics, which need WireGuard engine locks that may be busy during an
// incident.
const peerDiagnosticsTimeout = time.Second

// clientCanceledWait is the longest wait that still counts as the client
// leaving when the context was canceled. A connected agent answers its first
// ping well within this time.
const clientCanceledWait = 5 * time.Second

// recordAgentUnreachable counts a connection attempt that the agent did not
// answer and returns an error carrying the peer state. The fields go on the
// request log line, or on a warn line when the context has no request logger.
// A client that left quickly gets a plain error and no warn line.
func (s *ServerTailnet) recordAgentUnreachable(ctx context.Context, agentID uuid.UUID, after time.Duration, extra ...slog.Field) error {
	fields := append([]slog.Field{
		slog.F("agent_id", agentID),
		slog.F("unreachable_after", after),
	}, extra...)
	reason := "diagnostics_timeout"

	// A quick cancellation is the client leaving, not the agent failing.
	// A long wait counts as unreachable however the context ended, since a
	// load balancer giving up on a hung request also cancels it.
	clientCanceled := errors.Is(ctx.Err(), context.Canceled) && after < clientCanceledWait
	if clientCanceled {
		reason = "client_canceled"
	} else if d, ok := s.peerDiagnostics(agentID); ok {
		fields = append(fields, unreachableFields(d)...)
		reason = unreachableReason(d, s.clock.Now())
	}
	fields = append(fields, slog.F("reason", reason))
	s.agentUnreachable.WithLabelValues(reason).Inc()

	if rl := loggermw.RequestLoggerFromContext(ctx); rl != nil {
		rl.WithFields(fields...)
	} else if !clientCanceled {
		// Callers such as chatd have no request log line, so without this the
		// counter rises with no record of which agent failed.
		s.logger.Warn(ctx, "agent is unreachable", fields...)
	}
	if clientCanceled {
		return xerrors.New("agent is unreachable: client canceled")
	}
	return &workspaceapps.AgentUnreachableError{Fields: fields}
}

// peerDiagnostics reads the peer state within peerDiagnosticsTimeout. The
// read takes WireGuard engine locks that may be held during an incident, so
// one read runs at a time and other callers wait for the slot within the
// same timeout instead of each parking a goroutine on the lock.
func (s *ServerTailnet) peerDiagnostics(agentID uuid.UUID) (tailnet.PeerDiagnostics, bool) {
	timeout := s.clock.NewTimer(peerDiagnosticsTimeout, "peerDiagnostics")
	defer timeout.Stop()
	select {
	case s.peerDiagnosticsSlot <- struct{}{}:
	case <-timeout.C:
		return tailnet.PeerDiagnostics{}, false
	}
	diagCh := make(chan tailnet.PeerDiagnostics, 1)
	go func() {
		defer func() { <-s.peerDiagnosticsSlot }()
		diagCh <- s.readPeerDiagnostics(agentID)
	}()
	select {
	case d := <-diagCh:
		return d, true
	case <-timeout.C:
		return tailnet.PeerDiagnostics{}, false
	}
}

// unreachableFields describes the peer state for the request log line.
// server_preferred_derp is this server's home relay. The peer_* fields
// describe the agent.
func unreachableFields(d tailnet.PeerDiagnostics) []slog.Field {
	fields := []slog.Field{
		slog.F("peer_node_received", d.ReceivedNode != nil),
		slog.F("server_preferred_derp", d.PreferredDERP),
	}
	if !d.LastWireguardHandshake.IsZero() {
		fields = append(fields, slog.F("peer_last_handshake", d.LastWireguardHandshake))
	}
	if d.ReceivedNode != nil {
		// The node's DERP is "127.3.3.40:N" where N is the region ID.
		region := d.ReceivedNode.DERP
		if _, after, found := strings.Cut(region, ":"); found {
			region = after
		}
		fields = append(fields,
			slog.F("peer_lost", d.Lost),
			slog.F("peer_preferred_derp", region),
			slog.F("peer_tx_bytes", d.TxBytes),
			slog.F("peer_rx_bytes", d.RxBytes),
		)
	}
	return fields
}

// unreachableReason is the reason label for agent_unreachable_total. A lost
// peer comes before the handshake checks because the tunnel can look healthy
// while the control plane is down. A handshake older than WireGuard's session
// limit means our handshakes went unanswered while we waited.
func unreachableReason(d tailnet.PeerDiagnostics, now time.Time) string {
	switch {
	// The node was never received, or was removed after a disconnect or a
	// lost timeout.
	case d.ReceivedNode == nil:
		return "no_node"
	// The agent's coordinator connection dropped, or this server's connection
	// to the coordinator ended.
	case d.Lost:
		return "peer_lost"
	case d.LastWireguardHandshake.IsZero():
		return "no_handshake"
	case now.Sub(d.LastWireguardHandshake) > device.RejectAfterTime:
		return "handshake_stale"
	default:
		return "handshake_ok"
	}
}

func (s *ServerTailnet) ServeHTTPDebug(w http.ResponseWriter, r *http.Request) {
	s.conn.MagicsockServeHTTPDebug(w, r)
}

// embeddedDERPRegionDialer serves the embedded DERP region over an in-memory
// pipe. A nil conn makes the DERP client dial the access URL, so after ctx is
// done it returns a closed conn: the handshake fails and magicsock backs off.
func embeddedDERPRegionDialer(ctx context.Context, logger slog.Logger, derpServer *derp.Server, derpConnects prometheus.Counter) func(context.Context, *tailcfg.DERPRegion) net.Conn {
	return func(_ context.Context, region *tailcfg.DERPRegion) net.Conn {
		if !region.EmbeddedRelay {
			return nil
		}
		left, right := net.Pipe()
		if ctx.Err() != nil {
			_ = right.Close()
			return left
		}
		derpConnects.Inc()
		logger.Debug(ctx, "connecting to embedded DERP via in-memory pipe")
		go func() {
			defer left.Close()
			defer right.Close()
			brw := bufio.NewReadWriter(bufio.NewReader(right), bufio.NewWriter(right))
			derpServer.Accept(ctx, right, brw, "internal")
		}()
		return left
	}
}

type netConnCloser struct {
	net.Conn
	close func()
}

func (c *netConnCloser) Close() error {
	c.close()
	return c.Conn.Close()
}

func (s *ServerTailnet) Close() error {
	s.logger.Info(s.ctx, "closing server tailnet")
	defer s.logger.Debug(s.ctx, "server tailnet close complete")
	s.cancel()
	_ = s.conn.Close()
	s.transport.CloseIdleConnections()
	s.coordCtrl.Close()
	<-s.controller.Closed()
	return nil
}

type instrumentedConn struct {
	net.Conn

	agentID       uuid.UUID
	closeOnce     sync.Once
	connsPerAgent *prometheus.GaugeVec
}

func (c *instrumentedConn) Close() error {
	c.closeOnce.Do(func() {
		c.connsPerAgent.WithLabelValues("tcp").Dec()
	})
	return c.Conn.Close()
}

// MultiAgentController is a tailnet.CoordinationController for connecting to multiple workspace
// agents.  It keeps track of connection times to the agents, and removes them on a timer if they
// have no active connections and haven't been used in a while.
type MultiAgentController struct {
	*tailnet.BasicCoordinationController

	logger slog.Logger
	tracer trace.Tracer

	mu sync.Mutex
	// connectionTimes is a map of agents the server wants to keep a connection to. It
	// contains the last time the agent was connected to.
	connectionTimes map[uuid.UUID]time.Time
	// tickets is a map of destinations to a set of connection tickets, representing open
	// connections to the destination
	tickets      map[uuid.UUID]map[uuid.UUID]struct{}
	coordination *tailnet.BasicCoordination

	cancel              context.CancelFunc
	expireOldAgentsDone chan struct{}
}

func (m *MultiAgentController) New(client tailnet.CoordinatorClient) tailnet.CloserWaiter {
	b := m.NewCoordination(client)
	// resync all destinations
	m.mu.Lock()
	defer m.mu.Unlock()
	m.coordination = b
	for agentID := range m.connectionTimes {
		err := b.SendRequest(&proto.CoordinateRequest{
			AddTunnel: &proto.CoordinateRequest_Tunnel{Id: agentID[:]},
		})
		if err != nil {
			m.logger.Error(context.Background(), "failed to re-add tunnel", slog.F("agent_id", agentID),
				slog.Error(err))
			b.SendErr(err)
			_ = client.Close()
			m.coordination = nil
			break
		}
	}
	return b
}

func (m *MultiAgentController) ensureAgent(agentID uuid.UUID) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	_, ok := m.connectionTimes[agentID]
	// If we don't have the agent, subscribe.
	if !ok {
		m.logger.Debug(context.Background(),
			"subscribing to agent", slog.F("agent_id", agentID))
		if m.coordination != nil {
			err := m.coordination.SendRequest(&proto.CoordinateRequest{
				AddTunnel: &proto.CoordinateRequest_Tunnel{Id: agentID[:]},
			})
			if err != nil {
				err = xerrors.Errorf("subscribe agent: %w", err)
				m.coordination.SendErr(err)
				_ = m.coordination.CloseClient()
				m.coordination = nil
				return err
			}
		}
		m.tickets[agentID] = map[uuid.UUID]struct{}{}
	}
	m.connectionTimes[agentID] = time.Now()
	return nil
}

func (m *MultiAgentController) acquireTicket(agentID uuid.UUID) (release func()) {
	id := uuid.New()
	m.mu.Lock()
	defer m.mu.Unlock()
	m.tickets[agentID][id] = struct{}{}

	return func() {
		m.mu.Lock()
		defer m.mu.Unlock()
		delete(m.tickets[agentID], id)
	}
}

func (m *MultiAgentController) expireOldAgents(ctx context.Context) {
	defer close(m.expireOldAgentsDone)
	defer m.logger.Debug(context.Background(), "stopped expiring old agents")
	const (
		tick   = 5 * time.Minute
		cutoff = 30 * time.Minute
	)

	ticker := time.NewTicker(tick)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}

		m.doExpireOldAgents(ctx, cutoff)
	}
}

func (m *MultiAgentController) doExpireOldAgents(ctx context.Context, cutoff time.Duration) {
	// TODO: add some attrs to this.
	ctx, span := m.tracer.Start(ctx, tracing.FuncName())
	defer span.End()

	start := time.Now()
	deletedCount := 0

	m.mu.Lock()
	defer m.mu.Unlock()
	m.logger.Debug(ctx, "pruning inactive agents", slog.F("agent_count", len(m.connectionTimes)))
	for agentID, lastConnection := range m.connectionTimes {
		// If no one has connected since the cutoff and there are no active
		// connections, remove the agent.
		if time.Since(lastConnection) > cutoff && len(m.tickets[agentID]) == 0 {
			if m.coordination != nil {
				err := m.coordination.SendRequest(&proto.CoordinateRequest{
					RemoveTunnel: &proto.CoordinateRequest_Tunnel{Id: agentID[:]},
				})
				if err != nil {
					m.logger.Debug(ctx, "unsubscribe expired agent", slog.Error(err), slog.F("agent_id", agentID))
					m.coordination.SendErr(xerrors.Errorf("unsubscribe expired agent: %w", err))
					// close the client because we do not want to do a graceful disconnect by
					// closing the coordination.
					_ = m.coordination.CloseClient()
					m.coordination = nil
					// Here we continue deleting any inactive agents: there is no point in
					// re-establishing tunnels to expired agents when we eventually reconnect.
				}
			}
			deletedCount++
			delete(m.connectionTimes, agentID)
		}
	}
	m.logger.Debug(ctx, "pruned inactive agents",
		slog.F("deleted", deletedCount),
		slog.F("took", time.Since(start)),
	)
}

func (m *MultiAgentController) Close() {
	m.cancel()
	<-m.expireOldAgentsDone
}

func NewMultiAgentController(ctx context.Context, logger slog.Logger, tracer trace.Tracer, coordinatee tailnet.Coordinatee) *MultiAgentController {
	m := &MultiAgentController{
		BasicCoordinationController: &tailnet.BasicCoordinationController{
			Logger:      logger,
			Coordinatee: coordinatee,
			SendAcks:    false, // we are a client, connecting to multiple agents
			Initiator:   codersdk.DisconnectInitiatorServer,
			Direction:   codersdk.ConnectionDirectionServerToAgent,
		},
		logger:              logger,
		tracer:              tracer,
		connectionTimes:     make(map[uuid.UUID]time.Time),
		tickets:             make(map[uuid.UUID]map[uuid.UUID]struct{}),
		expireOldAgentsDone: make(chan struct{}),
	}
	ctx, m.cancel = context.WithCancel(ctx)
	go m.expireOldAgents(ctx)
	return m
}

type Pinger interface {
	Ping(context.Context) (time.Duration, error)
}

// InmemTailnetDialer is a tailnet.ControlProtocolDialer that connects to a Coordinator and DERPMap
// service running in the same memory space.
type InmemTailnetDialer struct {
	CoordPtr *atomic.Pointer[tailnet.Coordinator]
	DERPFn   func() *tailcfg.DERPMap
	Logger   slog.Logger
	ClientID uuid.UUID
	// DatabaseHealthCheck is used to validate that the store is reachable.
	DatabaseHealthCheck Pinger
	// Clock is optional and defaults to the real clock.
	Clock quartz.Clock
}

func (a *InmemTailnetDialer) Dial(ctx context.Context, _ tailnet.ResumeTokenController) (tailnet.ControlProtocolClients, error) {
	if a.DatabaseHealthCheck != nil {
		if _, err := a.DatabaseHealthCheck.Ping(ctx); err != nil {
			return tailnet.ControlProtocolClients{}, xerrors.Errorf("%w: %v", codersdk.ErrDatabaseNotReachable, err)
		}
	}

	coord := a.CoordPtr.Load()
	if coord == nil {
		return tailnet.ControlProtocolClients{}, xerrors.Errorf("tailnet coordinator not initialized")
	}
	coordClient := tailnet.NewInMemoryCoordinatorClient(
		a.Logger, a.ClientID, tailnet.SingleTailnetCoordinateeAuth{}, *coord)
	clock := a.Clock
	if clock == nil {
		clock = quartz.NewReal()
	}
	derpClient := newPollingDERPClient(a.DERPFn, a.Logger, clock)
	return tailnet.ControlProtocolClients{
		Closer:      closeAll{coord: coordClient, derp: derpClient},
		Coordinator: coordClient,
		DERP:        derpClient,
	}, nil
}

func newPollingDERPClient(derpFn func() *tailcfg.DERPMap, logger slog.Logger, clock quartz.Clock) tailnet.DERPClient {
	ctx, cancel := context.WithCancel(context.Background())
	a := &pollingDERPClient{
		fn:       derpFn,
		ctx:      ctx,
		cancel:   cancel,
		logger:   logger,
		clock:    clock,
		ch:       make(chan *tailcfg.DERPMap),
		loopDone: make(chan struct{}),
	}
	go a.pollDERP()
	return a
}

// pollingDERPClient is a DERP client that just calls a function on a polling
// interval
type pollingDERPClient struct {
	fn          func() *tailcfg.DERPMap
	logger      slog.Logger
	clock       quartz.Clock
	ctx         context.Context
	cancel      context.CancelFunc
	loopDone    chan struct{}
	lastDERPMap *tailcfg.DERPMap
	ch          chan *tailcfg.DERPMap
}

// Close the DERP client
func (a *pollingDERPClient) Close() error {
	a.cancel()
	<-a.loopDone
	return nil
}

func (a *pollingDERPClient) Recv() (*tailcfg.DERPMap, error) {
	select {
	case <-a.ctx.Done():
		return nil, a.ctx.Err()
	case dm := <-a.ch:
		return dm, nil
	}
}

func (a *pollingDERPClient) pollDERP() {
	defer close(a.loopDone)
	defer a.logger.Debug(a.ctx, "polling DERPMap exited")

	// Fetch immediately so the tailnet can publish its node without waiting
	// for the first tick.
	ticker := a.clock.NewTicker(5*time.Second, "pollDERP")
	defer ticker.Stop()

	for {
		newDerpMap := a.fn()
		if !tailnet.CompareDERPMaps(a.lastDERPMap, newDerpMap) {
			select {
			case <-a.ctx.Done():
				return
			case a.ch <- newDerpMap:
			}
		}

		select {
		case <-a.ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

type closeAll struct {
	coord tailnet.CoordinatorClient
	derp  tailnet.DERPClient
}

func (c closeAll) Close() error {
	cErr := c.coord.Close()
	dErr := c.derp.Close()
	if cErr != nil {
		return cErr
	}
	return dErr
}