package gatewaymain import ( "context" "errors" "fmt" "log/slog" "net" "net/http" "net/url" "os" "os/signal" "strings" "sync" "syscall" "time" "github.com/bluesky-social/indigo/atproto/identity" "k8s.io/client-go/tools/clientcmd" "github.com/johnandersen777/socialweb-computer-kcp/internal/clientcert" "github.com/johnandersen777/socialweb-computer-kcp/internal/foldmain" "github.com/johnandersen777/socialweb-computer-kcp/internal/kcpclient" "github.com/johnandersen777/socialweb-computer-kcp/internal/kcpserver" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/opt" "github.com/johnandersen777/socialweb-computer-kcp/internal/plcresolver" "github.com/johnandersen777/socialweb-computer-kcp/internal/plcsweep" "github.com/johnandersen777/socialweb-computer-kcp/internal/sweepkcp" "github.com/johnandersen777/socialweb-computer-kcp/internal/tenantapply" "github.com/johnandersen777/socialweb-computer-kcp/internal/vault/openbao" "github.com/johnandersen777/socialweb-computer-kcp/internal/xrpcserver" ) const directoryCacheCapacity = 10_000 func Main(ctx context.Context, inv foldmain.Invocation) int { if err := run(ctx, inv); err != nil { fmt.Fprintln(inv.Stderr, "socialweb-gateway:", err) return 1 } return 0 } func run(ctx context.Context, inv foldmain.Invocation) error { cfg, err := load(inv.Env) if err != nil { return err } log := slog.New(slog.NewJSONHandler(inv.Stderr, &slog.HandlerOptions{Level: cfg.LogLevel})) slog.SetDefault(log) deps, err := buildDeps(cfg) if err != nil { return err } g, err := Build(ctx, cfg, deps, log) if err != nil { return err } log.Info("serving", "addr", cfg.Addr, "kcp", cfg.KCPHost, "kcpKubeconfig", cfg.KCPKubeconfig, "audiences", cfg.Audiences, "xrpc", xrpcserver.XRPCPrefix, ) return serve(ctx, cfg, g, log, inv) } func buildDeps(cfg Config) (Deps, error) { dir, err := directory(cfg) if err != nil { return Deps{}, err } kcpRestConfig, err := clientcmd.BuildConfigFromFlags("", cfg.KCPKubeconfig) if err != nil { return Deps{}, fmt.Errorf("reading %s: %w", envKCPKubeconfig, err) } kcpserver.FollowTokenFile(kcpRestConfig) owners := ownerCredentials(cfg.ClusterStateDir) tenantAPI, err := tenantapply.New(owners) if err != nil { return Deps{}, err } certs := &clientcert.Issuer{StateDir: cfg.ClusterStateDir, CAFile: cfg.KCPPublicCAFile, Logins: openbao.NewLogins(nil), Log: slog.Default()} client, err := kcpclient.New(kcpclient.Options{ Host: cfg.KCPHost, PublicHost: cfg.KCPPublicHost, RestConfig: kcpRestConfig, Kubeconfigs: certs, Tenant: kcpclient.TenantConfig(owners), NodeClaimAgent: cfg.NodeClaimAgent, NodeClaimQuota: cfg.NodeClaimQuota, HostProcessWorkspace: cfg.HostProcessWorkspace, }) if err != nil { return Deps{}, err } certs.Recorder = client worklist, err := sweepkcp.NewWorklist(sweepkcp.Options{Host: cfg.KCPHost, RestConfig: kcpRestConfig}) if err != nil { return Deps{}, fmt.Errorf("the sweep's worklist: %w", err) } resolver, err := plcresolver.New(plcresolver.Options{ URL: cfg.PLCURL, Client: &http.Client{Timeout: cfg.PLCTimeout}, }) if err != nil { return Deps{}, fmt.Errorf("the sweep's resolver: %w", err) } revoker := &sweepkcp.Revoker{Deny: client, FanOut: client, Erase: client} return Deps{ Store: client, Deny: client, Feed: client, Directory: dir, Revoker: client, Certs: certs, Tenant: tenantAPI, Sweep: &plcsweep.Sweeper{ Worklist: worklist, Resolver: resolver, Revoker: revoker, }, }, nil } const ListenerName = "xrpc" func loopbackURL(raw string) bool { u, err := url.Parse(raw) if err != nil { return false } if u.Hostname() == "localhost" { return true } ip := net.ParseIP(u.Hostname()) return ip != nil && ip.IsLoopback() } func directory(cfg Config) (identity.Directory, error) { hardened := identity.DefaultDirectory() cached, ok := hardened.(*identity.CacheDirectory) if !ok { return nil, fmt.Errorf( "indigo's identity.DefaultDirectory() is a %T, not a *identity.CacheDirectory, so its "+ "revalidation window cannot be set", hardened) } base, ok := cached.Inner.(*identity.BaseDirectory) if !ok { return nil, fmt.Errorf( "indigo's identity.DefaultDirectory() wraps a %T, not a *identity.BaseDirectory, so the "+ "PLC, did:web and handle HTTP clients cannot be bounded", cached.Inner) } timeout := cfg.PLCTimeout timeout = opt.Positive(timeout, DefaultPLCTimeout) if cfg.PLCURL != "" { base.PLCURL = strings.TrimSuffix(cfg.PLCURL, "/") } base.PLCClient = &http.Client{Timeout: timeout} if loopbackURL(base.PLCURL) { base.HTTPClient = http.Client{Timeout: timeout} } else { base.HTTPClient.Timeout = timeout } return identity.NewCacheDirectory( base, directoryCacheCapacity, cfg.DirectoryCacheTTL, cfg.DirectoryCacheTTL, cfg.DirectoryCacheTTL, ), nil } func serve(ctx context.Context, cfg Config, g *Gateway, log *slog.Logger, inv foldmain.Invocation) error { ctx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM) defer stop() server := &http.Server{ Handler: g.Handler(), ReadHeaderTimeout: 5 * time.Second, IdleTimeout: 2 * time.Minute, } ln, err := net.Listen("tcp", cfg.Addr) if err != nil { return err } log.Info("listening", "addr", ln.Addr().String()) inv.Listened(ListenerName, ln.Addr()) var background sync.WaitGroup background.Go(func() { if err := g.Run(ctx); err != nil { log.Error("the background work stopped", "err", err) } }) served := make(chan error, 1) go func() { served <- server.Serve(ln) }() var serveErr error select { case serveErr = <-served: stop() case <-ctx.Done(): log.Info("signal received, draining") } g.Close() drainCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), cfg.ShutdownTimeout) defer cancel() drainErr := server.Shutdown(drainCtx) if serveErr == nil { serveErr = <-served } if drainErr != nil { return fmt.Errorf("draining after %s: %w", cfg.ShutdownTimeout, drainErr) } background.Wait() if serveErr != nil && !errors.Is(serveErr, http.ErrServerClosed) { return serveErr } return nil }