package guestcontrolplane import ( "context" "errors" "fmt" "log/slog" "net/http" "os" "strings" "sync" "time" "k8s.io/utils/ptr" "github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1" "github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/ingress" reporemote "github.com/johnandersen777/socialweb-computer-kcp/internal/atprotorepo/remote" "github.com/johnandersen777/socialweb-computer-kcp/internal/bus" "github.com/johnandersen777/socialweb-computer-kcp/internal/component" "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/guestcp" guestplane "github.com/johnandersen777/socialweb-computer-kcp/internal/guestcp/openbao" guestrelay "github.com/johnandersen777/socialweb-computer-kcp/internal/guestcp/relay" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/kcp" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/pinneddownloads" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/supervise" ) type Service struct { spec v1alpha1.GuestControlPlaneSpec deps Deps log *slog.Logger trust *component.Trust stateDir string grantSync time.Duration issuerRetry time.Duration downloads *pinneddownloads.Set mu sync.Mutex started bool registered bool readiness component.ReadySignal mux *http.ServeMux cancel context.CancelFunc wg sync.WaitGroup stopOnce sync.Once stopErr error cleanupMu sync.Mutex plane *guestplane.Plane relay *guestrelay.Relay queue guestcp.Queue unwatchQueue func() } func (s *Service) Start(ctx context.Context) error { runCtx, cancel := context.WithCancel(ctx) s.mu.Lock() if s.started { s.mu.Unlock() cancel() return errors.New("guestcontrolplane: the service is already started") } s.started = true s.cancel = cancel s.mu.Unlock() defer s.shutdown(ctx) stopTrust, err := s.trust.Follow(runCtx, s.deps.Instance) if err != nil { cancel() return err } defer stopTrust() if err := s.prepare(runCtx); err != nil { cancel() return err } s.mu.Lock() plane := s.plane s.mu.Unlock() if plane.Ready() != nil { s.readiness.Fail(errors.New("guestcontrolplane: the guest identity proxy is not configured")) } else { s.readiness.Signal() } <-runCtx.Done() return nil } func (s *Service) Ready() component.Readiness { return &s.readiness } func (s *Service) Endpoints() []component.Endpoint { s.mu.Lock() registered := s.registered plane := s.plane s.mu.Unlock() if !registered { return nil } out := []component.Endpoint{{Name: "http", Protocol: component.ProtocolHTTP, URL: s.logicalURL()}} if raw := s.relayURL(); raw != "" { protocol := component.ProtocolHTTPS if strings.HasPrefix(raw, "http://") { protocol = component.ProtocolHTTP } out = append(out, component.Endpoint{Name: "relay", Protocol: protocol, URL: raw}) } if plane != nil { if issuer := plane.OIDCIssuer(); issuer != "" { out = append(out, component.Endpoint{Name: "issuer", Protocol: component.ProtocolHTTPS, URL: issuer}) } } return out } func (s *Service) Stop(ctx context.Context) error { return s.shutdownWith(ctx) } func (s *Service) shutdown(ctx context.Context) error { ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), component.DefaultStopTimeout) defer cancel() return s.shutdownWith(ctx) } func (s *Service) shutdownWith(ctx context.Context) error { s.stopOnce.Do(func() { s.cleanupMu.Lock() defer s.cleanupMu.Unlock() s.mu.Lock() cancel := s.cancel relay := s.relay plane := s.plane unwatchQueue := s.unwatchQueue s.started = false s.registered = false s.mu.Unlock() if cancel != nil { cancel() } if unwatchQueue != nil { unwatchQueue() } if relay != nil { relay.Close() } if plane != nil { plane.Close() } if err := supervise.Join(ctx, s.wg.Wait); err != nil && s.stopErr == nil { s.stopErr = err } }) return s.stopErr } func (s *Service) prepare(ctx context.Context) error { if s.stateDir != "" { if err := os.MkdirAll(s.stateDir, 0o700); err != nil { return fmt.Errorf("guestcontrolplane: creating the state dir: %w", err) } } if s.deps.Bus == nil { return errors.New("guestcontrolplane: the handler registry is not available") } inputs, guests, grants, err := s.handles(ctx) if err != nil { return err } plane, err := guestplane.New(guestplane.Config{ Handle: s.handle(), DID: s.did(), Mount: s.spec.GuestLoginMount, Timeout: DefaultOpenBaoTimeout, GrantSync: s.grantSync, IssuerRetry: s.issuerRetry, HostKeyAttempts: int(ptr.Deref(s.spec.HostKeyAttempts, 0)), }, guestplane.Deps{ Inputs: inputs, Guests: guests, Grants: grants, RBAC: reporemote.NewNetwork(reporemote.NetworkConfig{Directory: ingress.NewDirectory(s.network().PLCDirectoryURL), HTTP: networkHTTPClient(s.deps.Hosts, s.network(), s.trust, DefaultOpenBaoTimeout)}), FallbackIssuer: s.relayURL, Publish: s.publishGuestEvent, Log: s.log, }) if err != nil { return err } if err := plane.Start(ctx); err != nil { return err } s.mu.Lock() s.plane = plane s.mu.Unlock() if err := s.openQueue(ctx, inputs); err != nil { s.log.Warn("the guest event queue is not open; guest events are refused", "err", err) } s.mux = s.routes() if err := s.register(); err != nil { return err } if err := s.startRelay(ctx, plane); err != nil { s.log.Warn("the guest relay did not start", "err", err) } s.log.Info("the guest control plane serves guests", "handler", s.logicalURL(), "handle", s.handle(), "issuer", plane.IssuerURL(), "mount", s.spec.GuestLoginMount, "downloads", len(s.downloads.List())) return nil } func (s *Service) handles(ctx context.Context) (guestcp.Source, guestcp.Guests, guestcp.Grants, error) { self := s.self() ctx = bus.WithCallerLabels(ctx, self, s.deps.Labels) inputs, err := bus.Resolve[guestcp.Source](ctx, s.deps.Bus, self, self, bus.Capability{Name: guestcp.CapabilityInputs}) if err != nil { return nil, nil, nil, fmt.Errorf("guestcontrolplane: resolving the guest inputs: %w", err) } guests, err := bus.Resolve[guestcp.Guests](ctx, s.deps.Bus, self, self, bus.Capability{Name: guestcp.CapabilityGuests}) if err != nil { return nil, nil, nil, fmt.Errorf("guestcontrolplane: resolving the guest report writer: %w", err) } grants, err := bus.Resolve[guestcp.Grants](ctx, s.deps.Bus, self, self, bus.Capability{Name: guestcp.CapabilityGrants}) if err != nil { return nil, nil, nil, fmt.Errorf("guestcontrolplane: resolving the grant writer: %w", err) } return inputs, guests, grants, nil } func (s *Service) startRelay(ctx context.Context, plane *guestplane.Plane) error { if !ptr.Deref(s.spec.RelayEnabled, false) { return nil } target, err := s.pdsRef() if err != nil { return err } keyFile := "" if s.stateDir != "" { keyFile = s.stateDir + "/guest-relay.jwk" } relay, err := guestrelay.New(guestrelay.Config{ Label: "guest-" + s.spec.Role, KeyFile: keyFile, IngressProxyHost: s.relayHost(), TLS: s.relayTLS(), }, guestrelay.Deps{ Handler: s.mux, Subscribe: s.subscribeDirect(ctx), ServiceAuth: busServiceAuth{bus: s.deps.Bus, self: s.self(), labels: s.deps.Labels, pds: target}, DID: plane.DID, Log: s.log, }) if err != nil { return err } s.mu.Lock() s.relay = relay s.mu.Unlock() s.wg.Go(func() { if err := relay.Run(ctx, plane.Recheck); err != nil && ctx.Err() == nil { s.log.Warn("the guest relay did not start", "err", err) } }) return nil } func (s *Service) relayURL() string { s.mu.Lock() relay := s.relay s.mu.Unlock() if relay == nil { return "" } return relay.URL() } func (s *Service) register() error { key := component.Key{Workspace: s.workspace(), Kind: Kind, Name: s.deps.Self.Name} if err := s.deps.Bus.RegisterHTTP(key, s.mux); err != nil { return err } s.mu.Lock() s.registered = true s.mu.Unlock() return nil } func (s *Service) logicalURL() string { return kcp.HandlerURL(s.workspace(), Kind, s.deps.Self.Name) } var _ component.Service = (*Service)(nil)