Refactor: Centralize App Startup and Config Loading (#33)

Reviewed-on: #33

internal/bootstrap/bootstrap.go

@@ -0,0 +1,97 @@
+package bootstrap
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"git.fossy.my.id/bagas/tunnel-please-controller/internal/config"
+	"git.fossy.my.id/bagas/tunnel-please-controller/internal/db/sqlc/repository"
+	"git.fossy.my.id/bagas/tunnel-please-controller/internal/server"
+	"github.com/jackc/pgx/v5"
+	"github.com/lestrrat-go/httprc/v3"
+	"github.com/lestrrat-go/jwx/v3/jwk"
+)
+
+type bootstrap struct {
+	config     config.Config
+	errChan    chan error
+	signalChan chan os.Signal
+}
+
+type Bootstrap interface {
+	Run() error
+}
+
+func New(config config.Config) Bootstrap {
+	errChan := make(chan error, 5)
+	signalChan := make(chan os.Signal, 1)
+	return &bootstrap{
+		config:     config,
+		errChan:    errChan,
+		signalChan: signalChan,
+	}
+}
+
+func (b *bootstrap) Run() error {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	signal.Notify(b.signalChan, os.Interrupt, syscall.SIGTERM)
+
+	dbURL := b.config.DatabaseURL()
+	repo, err := startDatabase(ctx, dbURL)
+	if err != nil {
+		return err
+	}
+
+	client := httprc.NewClient()
+	jwkCache, err := jwk.NewCache(ctx, client)
+	if err != nil {
+		return err
+	}
+	s := server.New(repo, jwkCache, b.config)
+
+	b.errChan = make(chan error, 2)
+
+	go func() {
+		if err = s.StartAPI(ctx, b.config.ApiAddr()); err != nil && !errors.Is(err, context.Canceled) {
+			b.errChan <- err
+		}
+	}()
+
+	go func() {
+		if err = s.StartController(ctx, b.config.ControllerAddr()); err != nil && !errors.Is(err, context.Canceled) {
+			b.errChan <- err
+		}
+	}()
+
+	select {
+	case err = <-b.errChan:
+		return fmt.Errorf("service error: %w", err)
+	case sig := <-b.signalChan:
+		log.Printf("Received signal %s, initiating graceful shutdown", sig)
+		cancel()
+		return nil
+	}
+}
+
+func startDatabase(ctx context.Context, dbURL string) (*repository.Queries, error) {
+	connect, err := pgx.Connect(ctx, dbURL)
+	if err != nil {
+		return nil, err
+	}
+
+	defer func(connect *pgx.Conn, ctx context.Context) {
+		err = connect.Close(ctx)
+		if err != nil {
+			panic(err)
+		}
+	}(connect, ctx)
+
+	return repository.New(connect), nil
+}

internal/config/config.go

@@ -1,8 +1,81 @@
 package config
 
-import "os"
+import (
+	"crypto/rand"
+	"encoding/base64"
+	"errors"
+	"log"
+	"os"
 
-func Getenv(key, defaultValue string) string {
+	"github.com/joho/godotenv"
+)
+
+type config struct {
+	databaseUrl    string
+	controllerAddr string
+	apiAddr        string
+	authToken      string
+	jwksUrl        string
+}
+
+type Config interface {
+	DatabaseURL() string
+	ControllerAddr() string
+	ApiAddr() string
+	AuthToken() string
+	JwksURL() string
+}
+
+func (c *config) DatabaseURL() string    { return c.databaseUrl }
+func (c *config) ControllerAddr() string { return c.controllerAddr }
+func (c *config) ApiAddr() string        { return c.apiAddr }
+func (c *config) AuthToken() string      { return c.authToken }
+func (c *config) JwksURL() string        { return c.jwksUrl }
+
+func MustLoad() (Config, error) {
+	if err := loadEnvFile(); err != nil {
+		return nil, err
+	}
+
+	cfg, err := parse()
+	if err != nil {
+		return nil, err
+	}
+
+	return cfg, nil
+}
+
+func loadEnvFile() error {
+	if _, err := os.Stat(".env"); err == nil {
+		return godotenv.Load(".env")
+	}
+	return nil
+}
+
+func parse() (*config, error) {
+	databaseUrl := getenv("DATABASE_URL", "")
+	if databaseUrl == "" {
+		return nil, errors.New("DATABASE_URL environment variable not set")
+	}
+	controllerAddr := getenv("CONTROLLER_ADDR", ":8080")
+	apiAddr := getenv("API_ADDR", ":8081")
+	authToken := getenv("AUTH_TOKEN", "")
+	if authToken == "" {
+		authToken = generateAuthToken()
+		log.Printf("No AUTH_TOKEN provided. Generated token: %s", authToken)
+	}
+	jwksUrl := getenv("JWKS_URL", "")
+
+	return &config{
+		databaseUrl:    databaseUrl,
+		controllerAddr: controllerAddr,
+		apiAddr:        apiAddr,
+		authToken:      authToken,
+		jwksUrl:        jwksUrl,
+	}, nil
+}
+
+func getenv(key, defaultValue string) string {
 	val := os.Getenv(key)
 	if val == "" {
 		val = defaultValue
@@ -10,3 +83,11 @@ func Getenv(key, defaultValue string) string {
 
 	return val
 }
+
+func generateAuthToken() string {
+	buf := make([]byte, 32)
+	if _, err := rand.Read(buf); err != nil {
+		panic(err)
+	}
+	return base64.StdEncoding.EncodeToString(buf)
+}

internal/db/sqlc/migrations/000001_init_schema.down.sql

@@ -0,0 +1,10 @@
+DROP TABLE IF EXISTS public.verification;
+DROP TABLE IF EXISTS public.jwks;
+DROP TABLE IF EXISTS public.session;
+DROP TABLE IF EXISTS public.account;
+DROP TABLE IF EXISTS public."user";
+
+DROP TABLE IF EXISTS drizzle.__drizzle_migrations;
+DROP SCHEMA IF EXISTS drizzle;
+
+DROP EXTENSION IF EXISTS pgcrypto;

internal/db/sqlc/migrations/000001_init_schema.up.sql

@@ -0,0 +1,88 @@
+CREATE EXTENSION IF NOT EXISTS pgcrypto;
+
+CREATE TABLE public."user" (
+    id TEXT PRIMARY KEY,
+    ssh_identifier TEXT NOT NULL DEFAULT substr(encode(gen_random_bytes(16), 'hex'), 1, 32),
+    name TEXT NOT NULL,
+    email TEXT NOT NULL,
+    email_verified BOOLEAN NOT NULL DEFAULT false,
+    image TEXT,
+    created_at TIMESTAMP WITHOUT TIME ZONE NOT NULL DEFAULT now(),
+    updated_at TIMESTAMP WITHOUT TIME ZONE NOT NULL DEFAULT now()
+);
+
+CREATE UNIQUE INDEX user_email_unique
+    ON public."user"(email);
+
+CREATE UNIQUE INDEX user_ssh_identifier_unique
+    ON public."user"(ssh_identifier);
+
+CREATE TABLE public.account (
+    id TEXT PRIMARY KEY,
+    account_id TEXT NOT NULL,
+    provider_id TEXT NOT NULL,
+    user_id TEXT NOT NULL,
+    access_token TEXT,
+    refresh_token TEXT,
+    id_token TEXT,
+    access_token_expires_at TIMESTAMP WITHOUT TIME ZONE,
+    refresh_token_expires_at TIMESTAMP WITHOUT TIME ZONE,
+    scope TEXT,
+    password TEXT,
+    created_at TIMESTAMP WITHOUT TIME ZONE NOT NULL DEFAULT now(),
+    updated_at TIMESTAMP WITHOUT TIME ZONE NOT NULL
+);
+
+CREATE INDEX account_userId_idx
+    ON public.account(user_id);
+
+ALTER TABLE public.account
+    ADD CONSTRAINT account_user_id_user_id_fk
+        FOREIGN KEY (user_id)
+            REFERENCES public."user"(id)
+            ON DELETE CASCADE;
+
+
+CREATE TABLE public.session (
+    id TEXT PRIMARY KEY,
+    expires_at TIMESTAMP WITHOUT TIME ZONE NOT NULL,
+    token TEXT NOT NULL,
+    created_at TIMESTAMP WITHOUT TIME ZONE NOT NULL DEFAULT now(),
+    updated_at TIMESTAMP WITHOUT TIME ZONE NOT NULL,
+    ip_address TEXT,
+    user_agent TEXT,
+    user_id TEXT NOT NULL
+);
+
+CREATE UNIQUE INDEX session_token_unique
+    ON public.session(token);
+
+CREATE INDEX session_userId_idx
+    ON public.session(user_id);
+
+ALTER TABLE public.session
+    ADD CONSTRAINT session_user_id_user_id_fk
+        FOREIGN KEY (user_id)
+            REFERENCES public."user"(id)
+            ON DELETE CASCADE;
+
+
+CREATE TABLE public.jwks (
+    id TEXT PRIMARY KEY,
+    public_key TEXT NOT NULL,
+    private_key TEXT NOT NULL,
+    created_at TIMESTAMP WITHOUT TIME ZONE NOT NULL,
+    expires_at TIMESTAMP WITHOUT TIME ZONE
+);
+
+CREATE TABLE public.verification (
+    id TEXT PRIMARY KEY,
+    identifier TEXT NOT NULL,
+    value TEXT NOT NULL,
+    expires_at TIMESTAMP WITHOUT TIME ZONE NOT NULL,
+    created_at TIMESTAMP WITHOUT TIME ZONE NOT NULL DEFAULT now(),
+    updated_at TIMESTAMP WITHOUT TIME ZONE NOT NULL DEFAULT now()
+);
+
+CREATE INDEX verification_identifier_idx
+    ON public.verification(identifier);

internal/db/sqlc/queries/query.sql

@@ -0,0 +1,5 @@
+-- name: GetVerifiedEmailBySSHIdentifier :one
+SELECT email
+FROM public."user"
+WHERE ssh_identifier = $1
+  AND email_verified = true;

internal/db/sqlc/repository/db.go

@@ -0,0 +1,32 @@
+// Code generated by sqlc. DO NOT EDIT.
+// versions:
+//   sqlc v1.27.0
+
+package repository
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+type DBTX interface {
+	Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error)
+	Query(context.Context, string, ...interface{}) (pgx.Rows, error)
+	QueryRow(context.Context, string, ...interface{}) pgx.Row
+}
+
+func New(db DBTX) *Queries {
+	return &Queries{db: db}
+}
+
+type Queries struct {
+	db DBTX
+}
+
+func (q *Queries) WithTx(tx pgx.Tx) *Queries {
+	return &Queries{
+		db: tx,
+	}
+}

internal/db/sqlc/repository/models.go

@@ -0,0 +1,56 @@
+// Code generated by sqlc. DO NOT EDIT.
+// versions:
+//   sqlc v1.27.0
+
+package repository
+
+import (
+	"github.com/jackc/pgx/v5/pgtype"
+)
+
+type Account struct {
+	ID                    string
+	AccountID             string
+	ProviderID            string
+	UserID                string
+	AccessToken           pgtype.Text
+	RefreshToken          pgtype.Text
+	IDToken               pgtype.Text
+	AccessTokenExpiresAt  pgtype.Timestamp
+	RefreshTokenExpiresAt pgtype.Timestamp
+	Scope                 pgtype.Text
+	Password              pgtype.Text
+	CreatedAt             pgtype.Timestamp
+	UpdatedAt             pgtype.Timestamp
+}
+
+type Session struct {
+	ID        string
+	ExpiresAt pgtype.Timestamp
+	Token     string
+	CreatedAt pgtype.Timestamp
+	UpdatedAt pgtype.Timestamp
+	IpAddress pgtype.Text
+	UserAgent pgtype.Text
+	UserID    string
+}
+
+type User struct {
+	ID            string
+	SshIdentifier string
+	Name          string
+	Email         string
+	EmailVerified bool
+	Image         pgtype.Text
+	CreatedAt     pgtype.Timestamp
+	UpdatedAt     pgtype.Timestamp
+}
+
+type Verification struct {
+	ID         string
+	Identifier string
+	Value      string
+	ExpiresAt  pgtype.Timestamp
+	CreatedAt  pgtype.Timestamp
+	UpdatedAt  pgtype.Timestamp
+}

internal/db/sqlc/repository/query.sql.go

@@ -0,0 +1,24 @@
+// Code generated by sqlc. DO NOT EDIT.
+// versions:
+//   sqlc v1.27.0
+// source: query.sql
+
+package repository
+
+import (
+	"context"
+)
+
+const getVerifiedEmailBySSHIdentifier = `-- name: GetVerifiedEmailBySSHIdentifier :one
+SELECT email
+FROM public."user"
+WHERE ssh_identifier = $1
+  AND email_verified = true
+`
+
+func (q *Queries) GetVerifiedEmailBySSHIdentifier(ctx context.Context, sshIdentifier string) (string, error) {
+	row := q.db.QueryRow(ctx, getVerifiedEmailBySSHIdentifier, sshIdentifier)
+	var email string
+	err := row.Scan(&email)
+	return email, err
+}

internal/db/sqlc/sqlc.yaml

@@ -0,0 +1,26 @@
+version: "2"
+sql:
+  - engine: "postgresql"
+    queries: "queries"
+    schema: "migrations"
+    gen:
+      go:
+        package: "repository"
+        out: "repository"
+        sql_package: "pgx/v5"
+        overrides:
+          - db_type: "uuid"
+            go_type:
+              import: "github.com/google/uuid"
+              type: "UUID"
+          - db_type: "uuid"
+            go_type:
+              import: "github.com/google/uuid"
+              type: "UUID"
+              pointer: true
+            nullable: true
+          - db_type: "timestamptz"
+            go_type:
+              type: "*time.Time"
+              pointer: true
+            nullable: true

internal/server/server.go

@@ -0,0 +1,788 @@
+package server
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log"
+	"net"
+	"net/http"
+	"reflect"
+	"sync"
+	"time"
+
+	"git.fossy.my.id/bagas/tunnel-please-controller/internal/config"
+	"git.fossy.my.id/bagas/tunnel-please-controller/internal/db/sqlc/repository"
+	proto "git.fossy.my.id/bagas/tunnel-please-grpc/gen"
+	"github.com/lestrrat-go/jwx/v3/jwk"
+	"github.com/lestrrat-go/jwx/v3/jwt"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/health"
+	"google.golang.org/grpc/health/grpc_health_v1"
+	"google.golang.org/grpc/keepalive"
+	"google.golang.org/grpc/reflection"
+	"google.golang.org/grpc/status"
+)
+
+const (
+	defaultSubscriberResponseWait = 5 * time.Second
+	jwkRegisterTimeout            = 5 * time.Second
+)
+
+type Subscriber struct {
+	node      chan *proto.Node
+	events    chan *proto.Events
+	done      chan struct{}
+	closeOnce sync.Once
+	mu        sync.Mutex
+}
+type Server struct {
+	Database    *repository.Queries
+	Subscribers map[string]*Subscriber
+	mu          *sync.RWMutex
+	jwkCache    *jwk.Cache
+	config      config.Config
+	proto.UnimplementedEventServiceServer
+	proto.UnimplementedUserServiceServer
+}
+
+func New(database *repository.Queries, jwkCache *jwk.Cache, config config.Config) *Server {
+	return &Server{
+		Database:    database,
+		Subscribers: make(map[string]*Subscriber),
+		mu:          new(sync.RWMutex),
+		jwkCache:    jwkCache,
+		config:      config,
+	}
+}
+
+func (s *Server) Check(ctx context.Context, request *proto.CheckRequest) (*proto.CheckResponse, error) {
+	user, err := s.Database.GetVerifiedEmailBySSHIdentifier(ctx, request.GetAuthToken())
+	if err != nil || user == "" {
+		return &proto.CheckResponse{
+			Response: proto.AuthorizationResponse_MESSAGE_TYPE_UNAUTHORIZED,
+			User:     "",
+		}, err
+	}
+
+	return &proto.CheckResponse{
+		Response: proto.AuthorizationResponse_MESSAGE_TYPE_AUTHORIZED,
+		User:     user,
+	}, nil
+}
+
+func (s *Server) Subscribe(event grpc.BidiStreamingServer[proto.Node, proto.Events]) error {
+	ctx := event.Context()
+	recv, err := event.Recv()
+	if err != nil {
+		return err
+	}
+	if recv == nil {
+		return status.Error(codes.InvalidArgument, "missing authentication event")
+	}
+	if recv.GetType() != proto.EventType_AUTHENTICATION {
+		return status.Errorf(codes.InvalidArgument, "invalid event type: %s", recv.GetType())
+	}
+	payload, ok := recv.GetPayload().(*proto.Node_AuthEvent)
+	if !ok || payload == nil || payload.AuthEvent == nil {
+		return status.Error(codes.InvalidArgument, "missing auth payload")
+	}
+	identity := payload.AuthEvent.Identity
+	if identity == "" {
+		return status.Error(codes.InvalidArgument, "missing identity")
+	}
+	token := payload.AuthEvent.AuthToken
+	if token != s.config.AuthToken() {
+		return status.Error(codes.Unauthenticated, "invalid auth token")
+	}
+
+	log.Printf("Client %s authenticated successfully", identity)
+
+	requestChan := &Subscriber{
+		node:   make(chan *proto.Node, 10),
+		events: make(chan *proto.Events, 10),
+		done:   make(chan struct{}),
+	}
+
+	if err = s.AddEventSubscriber(identity, requestChan); err != nil {
+		return status.Error(codes.AlreadyExists, err.Error())
+	}
+	defer func() {
+		s.RemoveEventSubscriber(identity)
+		log.Printf("Client %s disconnected and unsubscribed", identity)
+	}()
+
+	return processEventStream(ctx, requestChan, event)
+}
+
+func processEventStream(ctx context.Context, requestChan *Subscriber, event grpc.BidiStreamingServer[proto.Node, proto.Events]) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-requestChan.done:
+			return nil
+		case request, ok := <-requestChan.events:
+			if !ok {
+				return nil
+			}
+			if request == nil {
+				continue
+			}
+			log.Printf("Received event request: %v", request)
+			switch request.GetType() {
+			case proto.EventType_SLUG_CHANGE:
+				payload, ok := request.GetPayload().(*proto.Events_SlugEvent)
+				if !ok || payload == nil || payload.SlugEvent == nil {
+					log.Printf("invalid slug change payload")
+					continue
+				}
+				slugEvent := payload.SlugEvent
+				log.Printf("Processing slug change event: old=%s, new=%s", slugEvent.Old, slugEvent.New)
+				if err := event.Send(request); err != nil {
+					return err
+				}
+				recv, err := recvClientWithTimeout(ctx, requestChan.done, event, defaultSubscriberResponseWait)
+				if err != nil {
+					return err
+				}
+				select {
+				case <-ctx.Done():
+					return ctx.Err()
+				case requestChan.node <- recv:
+				}
+				log.Printf("Received slug change event: %v", recv)
+			case proto.EventType_GET_SESSIONS:
+				log.Printf("Processing session event")
+				if err := event.Send(request); err != nil {
+					return err
+				}
+				recv, err := recvClientWithTimeout(ctx, requestChan.done, event, defaultSubscriberResponseWait)
+				if err != nil {
+					return err
+				}
+				select {
+				case <-ctx.Done():
+					return ctx.Err()
+				case requestChan.node <- recv:
+				}
+				log.Printf("Received SESSIONS  event: %v", recv)
+			case proto.EventType_TERMINATE_SESSION:
+				log.Printf("Processing terminate event")
+				if err := event.Send(request); err != nil {
+					return err
+				}
+				recv, err := recvClientWithTimeout(ctx, requestChan.done, event, defaultSubscriberResponseWait)
+				if err != nil {
+					return err
+				}
+				select {
+				case <-ctx.Done():
+					return ctx.Err()
+				case requestChan.node <- recv:
+				}
+				log.Printf("Received terminate event: %v", recv)
+			default:
+				log.Printf("Unknown event type: %v", request.GetType())
+			}
+		}
+	}
+}
+
+func recvClientWithTimeout(ctx context.Context, done <-chan struct{}, event grpc.BidiStreamingServer[proto.Node, proto.Events], timeout time.Duration) (*proto.Node, error) {
+	respCh := make(chan *proto.Node, 1)
+	errCh := make(chan error, 1)
+
+	go func() {
+		recv, err := event.Recv()
+		if err != nil {
+			errCh <- err
+			return
+		}
+		respCh <- recv
+	}()
+
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case <-done:
+		return nil, status.Error(codes.Canceled, "subscriber removed")
+	case err := <-errCh:
+		return nil, err
+	case <-timer.C:
+		return nil, status.Error(codes.DeadlineExceeded, "client response timeout")
+	case recv := <-respCh:
+		return recv, nil
+	}
+}
+
+func (s *Server) AddEventSubscriber(identity string, req *Subscriber) error {
+	if identity == "" || req == nil {
+		return fmt.Errorf("invalid subscriber")
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if _, exist := s.Subscribers[identity]; exist {
+		return fmt.Errorf("identity %s already subscribed", identity)
+	}
+	s.Subscribers[identity] = req
+	return nil
+}
+
+func (s *Server) RemoveEventSubscriber(identity string) {
+	s.mu.Lock()
+	sub := s.Subscribers[identity]
+	delete(s.Subscribers, identity)
+	s.mu.Unlock()
+
+	if sub != nil {
+		sub.closeOnce.Do(func() { close(sub.done) })
+	}
+}
+
+func (s *Server) GetEventSubscriber(identity string) (*Subscriber, error) {
+	if identity == "" {
+		return nil, status.Error(codes.InvalidArgument, "missing identity")
+	}
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	req, ok := s.Subscribers[identity]
+	if !ok {
+		return nil, status.Errorf(codes.NotFound, "identity %s not subscribed", identity)
+	}
+	return req, nil
+}
+
+type SubscriberResult struct {
+	Identity string
+	Response *proto.Node
+	Err      error
+}
+
+func (s *Server) notifyAllSubscriber(ctx context.Context, recvChan <-chan *proto.Events, resultChan chan<- []SubscriberResult) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case controllerReq, ok := <-recvChan:
+			if !ok {
+				return
+			}
+			if controllerReq == nil {
+				continue
+			}
+
+			s.mu.RLock()
+			subs := make(map[string]*Subscriber, len(s.Subscribers))
+			for id, sub := range s.Subscribers {
+				subs[id] = sub
+			}
+			s.mu.RUnlock()
+			results := make([]SubscriberResult, 0, len(subs))
+			for id, sub := range subs {
+				select {
+				case <-ctx.Done():
+					return
+				case <-sub.done:
+					results = append(results, SubscriberResult{Identity: id, Err: status.Error(codes.Canceled, "subscriber removed")})
+					continue
+				case sub.events <- controllerReq:
+				default:
+					results = append(results, SubscriberResult{Identity: id, Err: status.Error(codes.Unavailable, "controller channel blocked")})
+					continue
+				}
+				select {
+				case <-ctx.Done():
+					return
+				case <-sub.done:
+					results = append(results, SubscriberResult{Identity: id, Err: status.Error(codes.Canceled, "subscriber removed")})
+				case resp, ok := <-sub.node:
+					if !ok {
+						results = append(results, SubscriberResult{Identity: id, Err: status.Error(codes.FailedPrecondition, "client channel closed")})
+						continue
+					}
+					results = append(results, SubscriberResult{Identity: id, Response: resp})
+				}
+			}
+			select {
+			case <-ctx.Done():
+				return
+			case resultChan <- results:
+			}
+		}
+	}
+}
+
+type Slug struct {
+	Old string `json:"old"`
+	New string `json:"new"`
+}
+
+func (s *Server) StartAPI(ctx context.Context, Addr string) error {
+	handler := http.NewServeMux()
+	httpServer := http.Server{
+		Addr:         Addr,
+		Handler:      handler,
+		ReadTimeout:  15 * time.Second,
+		WriteTimeout: 15 * time.Second,
+		IdleTimeout:  60 * time.Second,
+	}
+	jwkURL := s.config.JwksURL()
+	if jwkURL != "" {
+		registerCtx, cancel := context.WithTimeout(ctx, jwkRegisterTimeout)
+		defer cancel()
+
+		if err := s.jwkCache.Register(registerCtx, jwkURL); err != nil {
+			return fmt.Errorf("failed to register jwk cache: %w", err)
+		}
+	}
+	handler.HandleFunc("PATCH /api/session/{node}", func(writer http.ResponseWriter, request *http.Request) {
+		writeError := func(status int, msg string) {
+			writer.Header().Set("Content-Type", "application/json")
+			writer.WriteHeader(status)
+			_ = json.NewEncoder(writer).Encode(map[string]string{"error": msg})
+		}
+
+		var token jwt.Token
+		var err error
+		var keyset jwk.Set
+		if jwkURL != "" {
+			keyset, err = s.jwkCache.Lookup(request.Context(), jwkURL)
+			if err != nil {
+				log.Printf("jwks lookup failed: %v", err)
+				writeError(http.StatusBadGateway, "unable to fetch jwks")
+				return
+			}
+
+			token, err = jwt.ParseRequest(request, jwt.WithKeySet(keyset))
+			if err != nil {
+				log.Printf("jwt parse failed: %v", err)
+				writeError(http.StatusUnauthorized, "invalid or expired token")
+				return
+			}
+		} else {
+			token, err = jwt.ParseRequest(request, jwt.WithVerify(false))
+			if err != nil {
+				log.Printf("jwt parse failed (no verification): %v", err)
+				writeError(http.StatusBadRequest, "invalid token")
+				return
+			}
+		}
+
+		var email string
+		err = token.Get("email", &email)
+		if err != nil {
+			log.Printf("email claim not found: %v", err)
+			writeError(http.StatusBadRequest, "missing email claim in token")
+			return
+		}
+		if email == "" {
+			writeError(http.StatusBadRequest, "empty email claim in token")
+			return
+		}
+		node := request.PathValue("node")
+		if node == "" {
+			writeError(http.StatusBadRequest, "no node specified")
+			return
+		}
+		var slug *Slug
+		if err := json.NewDecoder(request.Body).Decode(&slug); err != nil {
+			writeError(http.StatusBadRequest, "invalid request body")
+			return
+		}
+		subscriber, err := s.GetEventSubscriber(node)
+		if err != nil {
+			writeError(http.StatusBadRequest, "no node found")
+			return
+		}
+
+		subscriber.events <- &proto.Events{
+			Type: proto.EventType_SLUG_CHANGE,
+			Payload: &proto.Events_SlugEvent{
+				SlugEvent: &proto.SlugChangeEvent{
+					User: email,
+					Old:  slug.Old,
+					New:  slug.New,
+				},
+			},
+		}
+
+		select {
+		case response := <-subscriber.node:
+			resp, ok := response.Payload.(*proto.Node_SlugEventResponse)
+			if !ok {
+				writeError(http.StatusInternalServerError, "received an unexpected response from the node")
+				return
+			}
+			if !resp.SlugEventResponse.Success {
+				writeError(http.StatusBadRequest, resp.SlugEventResponse.Message)
+				return
+			}
+			log.Printf("Received slug change response: %v", response)
+			writer.WriteHeader(http.StatusNoContent)
+		case <-request.Context().Done():
+		}
+	})
+
+	handler.HandleFunc("DELETE /api/session/{node}/{type}/{session}", func(writer http.ResponseWriter, request *http.Request) {
+		writeError := func(status int, msg string) {
+			writer.Header().Set("Content-Type", "application/json")
+			writer.WriteHeader(status)
+			_ = json.NewEncoder(writer).Encode(map[string]string{"error": msg})
+		}
+
+		var token jwt.Token
+		var err error
+		var keyset jwk.Set
+		if jwkURL != "" {
+			keyset, err = s.jwkCache.Lookup(request.Context(), jwkURL)
+			if err != nil {
+				log.Printf("jwks lookup failed: %v", err)
+				writeError(http.StatusBadGateway, "unable to fetch jwks")
+				return
+			}
+
+			token, err = jwt.ParseRequest(request, jwt.WithKeySet(keyset))
+			if err != nil {
+				log.Printf("jwt parse failed: %v", err)
+				writeError(http.StatusUnauthorized, "invalid or expired token")
+				return
+			}
+		} else {
+			token, err = jwt.ParseRequest(request, jwt.WithVerify(false))
+			if err != nil {
+				log.Printf("jwt parse failed (no verification): %v", err)
+				writeError(http.StatusBadRequest, "invalid token")
+				return
+			}
+		}
+
+		var email string
+		err = token.Get("email", &email)
+		if err != nil {
+			log.Printf("email claim not found: %v", err)
+			writeError(http.StatusBadRequest, "missing email claim in token")
+			return
+		}
+		if email == "" {
+			writeError(http.StatusBadRequest, "empty email claim in token")
+			return
+		}
+		node := request.PathValue("node")
+		if node == "" {
+			writeError(http.StatusBadRequest, "no node specified")
+			return
+		}
+
+		sessionTypeRaw := request.PathValue("type")
+		if node == "" {
+			writeError(http.StatusBadRequest, "no type specified")
+			return
+		}
+
+		var tunnelType proto.TunnelType
+		if sessionTypeRaw == "http" {
+			tunnelType = proto.TunnelType_HTTP
+		} else if sessionTypeRaw == "tcp" {
+			tunnelType = proto.TunnelType_TCP
+		} else {
+			writeError(http.StatusBadRequest, "invalid session type specified")
+			return
+		}
+
+		session := request.PathValue("session")
+		if node == "" {
+			writeError(http.StatusBadRequest, "no node specified")
+			return
+		}
+
+		subscriber, err := s.GetEventSubscriber(node)
+		if err != nil {
+			writeError(http.StatusBadRequest, "no node found")
+			return
+		}
+		subscriber.events <- &proto.Events{
+			Type: proto.EventType_TERMINATE_SESSION,
+			Payload: &proto.Events_TerminateSessionEvent{
+				TerminateSessionEvent: &proto.TerminateSessionEvent{
+					User:       email,
+					TunnelType: tunnelType,
+					Slug:       session,
+				},
+			},
+		}
+		select {
+		case response := <-subscriber.node:
+			resp, ok := response.Payload.(*proto.Node_TerminateSessionEventResponse)
+			if !ok {
+				writeError(http.StatusInternalServerError, "received an unexpected response from the node")
+				return
+			}
+			if !resp.TerminateSessionEventResponse.Success {
+				writeError(http.StatusBadRequest, resp.TerminateSessionEventResponse.Message)
+				return
+			}
+			log.Printf("Received terminate session response: %v", response)
+			writer.WriteHeader(http.StatusNoContent)
+		case <-request.Context().Done():
+		}
+	})
+
+	handler.HandleFunc("/api/sessions", func(writer http.ResponseWriter, request *http.Request) {
+		writeError := func(status int, msg string) {
+			writer.Header().Set("Content-Type", "application/json")
+			writer.WriteHeader(status)
+			_ = json.NewEncoder(writer).Encode(map[string]string{"error": msg})
+		}
+
+		var token jwt.Token
+		var err error
+		if jwkURL != "" {
+			keyset, err := s.jwkCache.Lookup(request.Context(), jwkURL)
+			if err != nil {
+				log.Printf("jwks lookup failed: %v", err)
+				writeError(http.StatusBadGateway, "unable to fetch jwks")
+				return
+			}
+
+			token, err = jwt.ParseRequest(request, jwt.WithKeySet(keyset))
+			if err != nil {
+				log.Printf("jwt parse failed: %v", err)
+				writeError(http.StatusUnauthorized, "invalid or expired token")
+				return
+			}
+		} else {
+			token, err = jwt.ParseRequest(request, jwt.WithVerify(false))
+			if err != nil {
+				log.Printf("jwt parse failed (no verification): %v", err)
+				writeError(http.StatusBadRequest, "invalid token")
+				return
+			}
+		}
+
+		var email string
+		err = token.Get("email", &email)
+		if err != nil {
+			log.Printf("email claim not found: %v", err)
+			writeError(http.StatusBadRequest, "missing email claim in token")
+			return
+		}
+		if email == "" {
+			writeError(http.StatusBadRequest, "empty email claim in token")
+			return
+		}
+
+		results := s.broadcastAndCollect(request.Context(), func(ctx context.Context, subscriber *Subscriber) (interface{}, bool) {
+			receive, err := s.sendAndReceive(ctx, subscriber, &proto.Events{
+				Type: proto.EventType_GET_SESSIONS,
+				Payload: &proto.Events_GetSessionsEvent{
+					GetSessionsEvent: &proto.GetSessionsEvent{
+						Identity: email,
+					},
+				},
+			}, defaultSubscriberResponseWait)
+			if err != nil {
+				log.Printf("get sessions request failed: %v", err)
+				return nil, false
+			}
+			if receive == nil {
+				log.Printf("get sessions request returned nil response")
+				return nil, false
+			}
+			payload, ok := receive.Payload.(*proto.Node_GetSessionsEvent)
+			if !ok || payload == nil || payload.GetSessionsEvent == nil {
+				log.Printf("unexpected get sessions payload type: %T", receive.Payload)
+				return nil, false
+			}
+			return payload.GetSessionsEvent.Details, true
+		})
+
+		flatten := flattenInterfaces(results)
+
+		writer.Header().Set("Content-Type", "application/json")
+
+		if len(flatten) == 0 {
+			_, err := writer.Write([]byte("[]"))
+			if err != nil {
+				log.Printf("write empty sessions response failed: %v", err)
+			}
+			return
+		}
+
+		marshal, err := json.Marshal(flatten)
+		if err != nil {
+			log.Printf("marshal sessions failed: %v", err)
+			writeError(http.StatusInternalServerError, "failed to marshal sessions")
+			return
+		}
+		_, err = writer.Write(marshal)
+		if err != nil {
+			log.Printf("write sessions response failed: %v", err)
+			return
+		}
+	})
+
+	errCh := make(chan error, 1)
+	go func() {
+		log.Printf("Listening api on %s", httpServer.Addr)
+		errCh <- httpServer.ListenAndServe()
+	}()
+
+	select {
+	case <-ctx.Done():
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		_ = httpServer.Shutdown(shutdownCtx)
+		return ctx.Err()
+	case err := <-errCh:
+		if errors.Is(err, http.ErrServerClosed) {
+			return nil
+		}
+		return err
+	}
+}
+
+func flattenInterfaces(values []interface{}) []interface{} {
+	var flat []interface{}
+	for _, v := range values {
+		if v == nil {
+			continue
+		}
+		rv := reflect.ValueOf(v)
+		if rv.Kind() == reflect.Slice {
+			for i := 0; i < rv.Len(); i++ {
+				flat = append(flat, rv.Index(i).Interface())
+			}
+			continue
+		}
+		flat = append(flat, v)
+	}
+	return flat
+}
+
+func (s *Server) StartController(ctx context.Context, Addr string) error {
+	listener, err := net.Listen("tcp", Addr)
+	if err != nil {
+		return err
+	}
+
+	kaParams := keepalive.ServerParameters{
+		MaxConnectionIdle:     0,
+		MaxConnectionAge:      0,
+		MaxConnectionAgeGrace: 0,
+		Time:                  30 * time.Second,
+		Timeout:               10 * time.Second,
+	}
+	kaPolicy := keepalive.EnforcementPolicy{
+		MinTime:             5 * time.Second,
+		PermitWithoutStream: true,
+	}
+
+	grpcServer := grpc.NewServer(
+		grpc.KeepaliveParams(kaParams),
+		grpc.KeepaliveEnforcementPolicy(kaPolicy),
+	)
+	reflection.Register(grpcServer)
+
+	proto.RegisterEventServiceServer(grpcServer, s)
+	proto.RegisterUserServiceServer(grpcServer, s)
+
+	healthServer := health.NewServer()
+	grpc_health_v1.RegisterHealthServer(grpcServer, healthServer)
+	healthServer.SetServingStatus("", grpc_health_v1.HealthCheckResponse_SERVING)
+
+	serveErr := make(chan error, 1)
+	go func() {
+		log.Printf("Listening controller on %s", s.config.ControllerAddr())
+		serveErr <- grpcServer.Serve(listener)
+	}()
+
+	select {
+	case <-ctx.Done():
+		grpcServer.GracefulStop()
+		err = <-serveErr
+		if err != nil {
+			return err
+		}
+		return ctx.Err()
+	case err = <-serveErr:
+		return err
+	}
+}
+
+func (s *Server) sendAndReceive(ctx context.Context, sub *Subscriber, req *proto.Events, wait time.Duration) (*proto.Node, error) {
+	if sub == nil || req == nil {
+		return nil, status.Error(codes.InvalidArgument, "missing subscriber or request")
+	}
+
+	sub.mu.Lock()
+	defer sub.mu.Unlock()
+
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case <-sub.done:
+		return nil, status.Error(codes.Canceled, "subscriber removed")
+	case sub.events <- req:
+	}
+
+	timer := time.NewTimer(wait)
+	defer timer.Stop()
+
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case <-sub.done:
+		return nil, status.Error(codes.Canceled, "subscriber removed")
+	case <-timer.C:
+		return nil, status.Error(codes.DeadlineExceeded, "subscriber response timeout")
+	case resp, ok := <-sub.node:
+		if !ok {
+			return nil, status.Error(codes.FailedPrecondition, "client channel closed")
+		}
+		return resp, nil
+	}
+}
+
+func (s *Server) snapshotSubscribers() []*Subscriber {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	subs := make([]*Subscriber, 0, len(s.Subscribers))
+	for _, sub := range s.Subscribers {
+		subs = append(subs, sub)
+	}
+	return subs
+}
+
+func (s *Server) broadcastAndCollect(ctx context.Context, worker func(context.Context, *Subscriber) (interface{}, bool)) []interface{} {
+	subs := s.snapshotSubscribers()
+	if len(subs) == 0 {
+		return nil
+	}
+
+	resultsCh := make(chan interface{}, len(subs))
+	var wg sync.WaitGroup
+	for _, sub := range subs {
+		wg.Add(1)
+		go func(subscriber *Subscriber) {
+			defer wg.Done()
+			if val, ok := worker(ctx, subscriber); ok {
+				resultsCh <- val
+			}
+		}(sub)
+	}
+
+	wg.Wait()
+	close(resultsCh)
+
+	var results []interface{}
+	for val := range resultsCh {
+		results = append(results, val)
+	}
+	return results
+}