johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package xrpcsubscription

import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"log/slog"
	"net"
	"net/http"
	"sync"
	"time"

	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/ingress"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/component"
	contract "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/ingress"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/hostport"
)

const shutdownTimeout = 5 * time.Second

type Config struct {
	Host string

	Port int

	Hostname string

	Hostnames []string

	AdditionalHosts []string

	AllowedDIDs []string

	AllowedHosts []string

	PLCURL string

	TLS *tls.Config

	RelayURL string

	RelayTimeout time.Duration

	KeepaliveInterval time.Duration

	KeepaliveTimeout time.Duration

	ReconnectGrace time.Duration

	NonceTTL time.Duration

	Log *slog.Logger
}

var _ contract.Provider = (*Provider)(nil)

type Provider struct {
	cfg Config

	mu sync.Mutex

	log *slog.Logger

	server *ingress.Server

	http *http.Server

	listener net.Listener

	stopCh chan struct{}

	stopOnce sync.Once

	readiness component.ReadySignal
}

func New(cfg Config) *Provider {
	return &Provider{cfg: cfg}
}

func (p *Provider) Start(ctx context.Context) error {
	cfg := p.cfg
	log := cfg.Log
	if log == nil {
		log = slog.New(slog.DiscardHandler)
	}
	hostname := hostport.Primary(cfg.Hostname, cfg.Hostnames)
	server, err := ingress.NewServer(ingress.Config{
		Hostname:          hostname,
		RelayTimeout:      cfg.RelayTimeout,
		KeepaliveInterval: cfg.KeepaliveInterval,
		KeepaliveTimeout:  cfg.KeepaliveTimeout,
		ReconnectGrace:    cfg.ReconnectGrace,
		NonceTTL:          cfg.NonceTTL,
		AdditionalHosts:   append([]string(nil), cfg.AdditionalHosts...),
		AllowedDIDs:       append([]string(nil), cfg.AllowedDIDs...),
		AllowedHosts:      append([]string(nil), cfg.AllowedHosts...),
		PLCURL:            cfg.PLCURL,
		Logger:            log,
	})
	if err != nil {
		return fmt.Errorf("ingressproxy: building the server: %w", err)
	}
	rawListener, err := net.Listen("tcp", hostport.Listen("", cfg.Host, cfg.Port))
	if err != nil {
		server.Close()
		return fmt.Errorf("ingressproxy: listening: %w", err)
	}
	var listener net.Listener = rawListener
	if cfg.TLS != nil {
		listener = tls.NewListener(rawListener, cfg.TLS)
	}
	httpServer := &http.Server{
		Handler:           server.Handler(),
		ReadHeaderTimeout: 10 * time.Second,
	}
	stopCh := make(chan struct{})

	p.mu.Lock()
	if p.http != nil {
		p.mu.Unlock()
		_ = listener.Close()
		server.Close()
		return errors.New("ingressproxy: the service already started")
	}
	p.log = log
	p.server = server
	p.http = httpServer
	p.listener = listener
	p.stopCh = stopCh
	p.stopOnce = sync.Once{}
	p.mu.Unlock()

	log.Info("ingressproxy_listening", "provider", "xrpc-subscription", "address", listener.Addr().String(),
		"hostname", hostname, "relay", cfg.RelayURL)

	errCh := make(chan error, 1)
	go func() { errCh <- httpServer.Serve(listener) }()
	p.readiness.Signal()

	select {
	case <-ctx.Done():
	case <-stopCh:
	case serveErr := <-errCh:
		server.Close()
		p.finish()
		if serveErr != nil && !errors.Is(serveErr, http.ErrServerClosed) {
			return fmt.Errorf("ingressproxy: serving: %w", serveErr)
		}
		return nil
	}

	p.shutdown(ctx)
	server.Close()
	serveErr := <-errCh
	p.finish()
	if serveErr != nil && !errors.Is(serveErr, http.ErrServerClosed) {
		return fmt.Errorf("ingressproxy: serving: %w", serveErr)
	}
	return nil
}

func (p *Provider) Ready() component.Readiness {
	return &p.readiness
}

func (p *Provider) Endpoints() []component.Endpoint {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.listener == nil {
		return nil
	}
	address := p.listener.Addr().String()
	hostname := hostport.Primary(p.cfg.Hostname, p.cfg.Hostnames)
	if p.cfg.TLS != nil {
		return []component.Endpoint{
			{Name: "https", Protocol: component.ProtocolHTTPS, URL: "https://" + address, Address: address},
			{Name: "control", Protocol: component.ProtocolHTTPS, URL: "https://" + hostport.HostnameOnly(hostname), Address: address},
		}
	}
	return []component.Endpoint{
		{Name: "http", Protocol: component.ProtocolHTTP, URL: "http://" + address, Address: address},
		{Name: "control", Protocol: component.ProtocolHTTP, URL: "http://" + hostport.HostnameOnly(hostname), Address: address},
	}
}

func (p *Provider) Stop(ctx context.Context) error {
	p.mu.Lock()
	stopCh := p.stopCh
	p.mu.Unlock()
	if stopCh == nil {
		return nil
	}
	p.stopOnce.Do(func() { close(stopCh) })
	p.shutdown(ctx)
	return nil
}

func (p *Provider) shutdown(ctx context.Context) {
	p.mu.Lock()
	httpServer := p.http
	p.mu.Unlock()
	if httpServer == nil {
		return
	}
	ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), shutdownTimeout)
	defer cancel()
	if err := httpServer.Shutdown(ctx); err != nil {
		_ = httpServer.Close()
	}
}

func (p *Provider) finish() {
	p.mu.Lock()
	p.listener = nil
	p.http = nil
	p.mu.Unlock()
}