johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package relay

import (
	"cmp"
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"log/slog"
	"net"
	"net/http"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"k8s.io/utils/ptr"

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	atrelay "github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/relay"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/bus"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/component"
	contract "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/relay"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/hostport"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/kcp"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/networkdial"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/servetls"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/strs"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/relay/local"
)

const (
	Kind = "Relay"

	PdsKind = "Pds"

	AnyName = "*"
)

type Deps struct {
	component.Instance
	Registry *component.Registry
	TLS      *tls.Config
	Egress   component.Egress
}

type Service struct {
	spec v1alpha1.RelaySpec
	deps Deps

	serveHost     string
	hostnames     []string
	plcURL        string
	insecureHTTP  bool
	allowedHosts  []string
	crawlInterval time.Duration
	tlsConfig     *tls.Config

	mu       sync.Mutex
	relay    *local.Relay
	key      component.Key
	baseURL  string
	stopCh   chan struct{}
	stopOnce sync.Once
	ready    bool

	readiness component.ReadySignal
}

func New(spec v1alpha1.RelaySpec, deps Deps) (*Service, error) {
	s := &Service{
		spec:          spec,
		deps:          deps,
		hostnames:     append([]string(nil), spec.Hostnames...),
		crawlInterval: time.Duration(ptr.Deref(spec.CrawlIntervalSeconds, 0)) * time.Second,
	}
	port := 0
	if spec.Serve != nil {
		s.serveHost = spec.Serve.Host
		port = int(spec.Serve.Port)
	}
	if spec.Network != nil {
		s.plcURL = strings.TrimRight(strings.TrimSpace(spec.Network.PLCDirectoryURL), "/")
		s.insecureHTTP = strings.HasPrefix(strings.ToLower(s.plcURL), "http://")
	}
	if spec.TLS != nil {
		s.hostnames = strs.AppendUniqueFold(s.hostnames, spec.TLS.Hostnames...)
	}
	if spec.Network != nil {
		s.hostnames = strs.AppendUniqueFold(s.hostnames, hostport.NetworkHostnames(spec.Network, port)...)
	}
	if len(spec.PdsRefs) > 0 {
		s.insecureHTTP = true
	}
	if !s.anyPDS() {
		for _, ref := range spec.PdsRefs {
			s.allowedHosts = strs.AppendUniqueFold(s.allowedHosts, pdsHost(refKind(ref.Kind), ref.Name))
		}
	}
	if spec.TLS != nil && spec.TLS.Enabled {
		certFile, _ := servetls.CertificateFiles(spec.TLS.CertRef, spec.Storage, s.hostnames)
		tlsConfig, err := servetls.Config(certFile, deps.TLS, deps.Log)
		if err != nil {
			return nil, fmt.Errorf("relay: %w", err)
		}
		s.tlsConfig = tlsConfig
	}
	return s, nil
}

var _ component.Service = (*Service)(nil)

func (s *Service) stateDir() string {
	if s.spec.Storage != nil && s.spec.Storage.DataDir != "" {
		return s.spec.Storage.DataDir
	}
	return s.deps.StateDir
}

func refKind(kind string) string {
	return cmp.Or(strings.TrimSpace(kind), PdsKind)
}

func pdsHost(kind, name string) string {
	kind, name = hostport.Normalize(kind), hostport.Normalize(name)
	if kind == "" || name == "" || name == AnyName {
		return ""
	}
	return hostport.Normalize(kind + "." + name)
}

func (s *Service) anyPDS() bool {
	for _, ref := range s.spec.PdsRefs {
		if strings.TrimSpace(ref.Name) == AnyName {
			return true
		}
	}
	return false
}

func (s *Service) upstream(key component.Key) (string, bool) {
	for _, ref := range s.spec.PdsRefs {
		if !strings.EqualFold(refKind(ref.Kind), key.Kind) {
			continue
		}
		name := strings.TrimSpace(ref.Name)
		if name != AnyName && !strings.EqualFold(name, key.Name) {
			continue
		}
		if host := pdsHost(key.Kind, key.Name); host != "" {
			return host, true
		}
	}
	return "", false
}

func (s *Service) observe(crawler contract.Crawler, key component.Key) {
	host, ok := s.upstream(key)
	if !ok {
		return
	}
	if _, ok := s.deps.Bus.HandlerForHost(host); ok {
		crawler.Crawl(host)
		return
	}
	crawler.Forget(host)
}

func (s *Service) follow(crawler contract.Crawler) func() {
	if len(s.spec.PdsRefs) == 0 {
		return func() {}
	}
	cancel := s.deps.Registry.Subscribe(func(key component.Key) { s.observe(crawler, key) })
	for _, ref := range s.spec.PdsRefs {
		if strings.TrimSpace(ref.Name) != AnyName {
			s.observe(crawler, component.Key{Workspace: s.deps.Self.Workspace, Kind: refKind(ref.Kind), Name: strings.TrimSpace(ref.Name)})
		}
	}
	if s.anyPDS() {
		for _, entry := range s.deps.Registry.List("") {
			s.observe(crawler, entry.Key)
		}
	}
	return cancel
}

func (s *Service) Start(ctx context.Context) error {
	ctx = s.deps.Caller(ctx)
	handlers := s.deps.Bus
	if handlers == nil {
		return errors.New("relay: no handler registry is available")
	}
	log := s.deps.Log
	if log == nil {
		log = slog.New(slog.DiscardHandler)
	}
	stateFile := ""
	if dir := s.stateDir(); dir != "" {
		stateFile = filepath.Join(dir, "relay-state.json")
	}
	tlsConfig := s.deps.TLS
	if tlsConfig == nil {
		tlsConfig = s.tlsConfig
	}
	hostname := hostport.Primary(s.serveHost, s.hostnames)
	relay, err := local.New(ctx, local.Config{
		Hostname:      hostname,
		PLCURL:        s.plcURL,
		InsecureHTTP:  s.insecureHTTP,
		AllowedHosts:  s.allowedHosts,
		StateFile:     stateFile,
		CrawlInterval: s.crawlInterval,
		TLS:           tlsConfig,
	}, local.Deps{
		Log:         log,
		HTTPClient:  &http.Client{Transport: handlers.Transport(s.deps.Egress.Client(0).Transport)},
		DialContext: handlerDialer(handlers, externalDialContext(s.deps.Egress)),
	})
	if err != nil {
		return fmt.Errorf("relay: opening the crawl state: %w", err)
	}
	key := s.deps.Key()
	key.Kind = cmp.Or(key.Kind, Kind)
	key.Name = cmp.Or(key.Name, "default")
	stopCh := make(chan struct{})
	s.mu.Lock()
	if s.stopCh != nil {
		s.mu.Unlock()
		relay.Core().Close()
		return errors.New("relay: the service already started")
	}
	s.stopCh = stopCh
	s.mu.Unlock()
	host := kcp.KindServiceHost(key.Kind, key.Workspace)
	if err := handlers.RegisterHTTP(key, relay.Handler()); err != nil {
		s.mu.Lock()
		s.stopCh = nil
		s.mu.Unlock()
		relay.Core().Close()
		return fmt.Errorf("relay: registering the handler: %w", err)
	}

	s.mu.Lock()
	s.relay = relay
	s.key = key
	s.baseURL = "http://" + host
	s.stopOnce = sync.Once{}
	s.ready = true
	s.mu.Unlock()

	log.Info("relay_registered", "host", host, "hostname", hostname)
	s.readiness.Signal()

	runCtx, cancel := context.WithCancel(ctx)
	var stopper sync.WaitGroup
	defer stopper.Wait()
	defer cancel()
	stopper.Go(func() {
		select {
		case <-stopCh:
			cancel()
		case <-runCtx.Done():
		}
	})
	unfollow := s.follow(relay)
	_ = relay.Run(runCtx)
	unfollow()
	s.finish()
	return nil
}

func (s *Service) Ready() component.Readiness {
	return &s.readiness
}

func (s *Service) Endpoints() []component.Endpoint {
	s.mu.Lock()
	defer s.mu.Unlock()
	if !s.ready || s.baseURL == "" {
		return nil
	}
	return []component.Endpoint{
		{Name: "http", Protocol: component.ProtocolHTTP, URL: s.baseURL},
		{Name: "subscribeRepos", Protocol: component.ProtocolWS, URL: "ws" + strings.TrimPrefix(s.baseURL, "http") + "/xrpc/com.atproto.sync.subscribeRepos"},
	}
}

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

func (s *Service) Relay() *atrelay.Relay {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.relay == nil {
		return nil
	}
	return s.relay.Core()
}

func (s *Service) Store() atrelay.Store {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.relay == nil {
		return nil
	}
	return s.relay.Store()
}

func (s *Service) Address() string {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.baseURL == "" {
		return ""
	}
	return strings.TrimPrefix(s.baseURL, "http://")
}

func (s *Service) AddSink(sink atrelay.EventSink) {
	if relay := s.Relay(); relay != nil {
		relay.AddSink(sink)
	}
}

func (s *Service) finish() {
	s.mu.Lock()
	s.ready = false
	s.stopCh = nil
	s.mu.Unlock()
}

func externalDialContext(egress component.Egress) func(context.Context, string, string) (net.Conn, error) {
	dialer, err := networkdial.NewDialerWithOptions(networkdial.DialOptions{
		Hosts:          egress.Hosts,
		TLSConfig:      egress.TLS,
		AllowSystemDNS: true,
	})
	if err != nil {
		return (&net.Dialer{Timeout: 10 * time.Second}).DialContext
	}
	return dialer.DialContext
}

func handlerDialer(handlers *bus.Bus, external func(context.Context, string, string) (net.Conn, error)) func(context.Context, string, string) (net.Conn, error) {
	return func(ctx context.Context, network, address string) (net.Conn, error) {
		if _, ok := handlers.HandlerForHost(address); ok {
			return handlers.DialHTTP(ctx, address)
		}
		return external(ctx, network, address)
	}
}