homelab-frontend/internal/serviceadapter/informer.go

package serviceadapter

import (
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"sync"
	"time"
)

// Loader watches ServiceAdapter CRs via the Kubernetes REST API and populates the registry.
// No client-go dependency — uses the in-cluster service account token directly.
type Loader struct {
	registry  *Registry
	namespace string
	client    *http.Client
	token     string
	baseURL   string
	stopChan  chan struct{}
	once      sync.Once
}

// NewLoader creates a loader that reads ServiceAdapters from the k8s API.
// Returns an error if not running inside a cluster (missing SA token/ca).
func NewLoader(registry *Registry, namespace string) (*Loader, error) {
	if registry == nil {
		return nil, fmt.Errorf("registry cannot be nil")
	}
	if namespace == "" {
		namespace = "api"
	}

	tokenBytes, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/token")
	if err != nil {
		return nil, fmt.Errorf("not in cluster: %w", err)
	}

	caBytes, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/ca.crt")
	if err != nil {
		return nil, fmt.Errorf("missing CA cert: %w", err)
	}

	pool := x509.NewCertPool()
	pool.AppendCertsFromPEM(caBytes)

	// Use kube-api-proxy (nginx) in the same namespace to reach the API server.
	// This avoids needing direct egress to the API server ClusterIP which
	// standard NetworkPolicy can't allow through Cilium.
	proxyHost := os.Getenv("KUBE_API_PROXY_URL")
	if proxyHost == "" {
		proxyHost = "https://kube-api-proxy.api.svc.cluster.local:8443"
	}

	return &Loader{
		registry:  registry,
		namespace: namespace,
		token:     string(tokenBytes),
		baseURL:   proxyHost,
		client: &http.Client{
			Timeout: 10 * time.Second,
			Transport: &http.Transport{
				TLSClientConfig: &tls.Config{RootCAs: pool, InsecureSkipVerify: true},
			},
		},
		stopChan: make(chan struct{}),
	}, nil
}

// Start begins loading ServiceAdapters in the background. Non-blocking.
// Retries on failure so the gateway can start serving immediately.
func (l *Loader) Start(interval time.Duration) {
	go func() {
		// Initial load with retries (API server may be slow on startup)
		for attempt := 1; ; attempt++ {
			if err := l.load(); err != nil {
				backoff := time.Duration(min(attempt*5, 30)) * time.Second
				log.Printf("serviceadapter loader: attempt %d failed (%v), retry in %s", attempt, err, backoff)
				time.Sleep(backoff)
				continue
			}
			break
		}

		// Poll for changes
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			select {
			case <-ticker.C:
				if err := l.load(); err != nil {
					log.Printf("serviceadapter loader poll error: %v", err)
				}
			case <-l.stopChan:
				return
			}
		}
	}()
}

// Stop stops the background poll.
func (l *Loader) Stop() {
	l.once.Do(func() { close(l.stopChan) })
}

// load fetches all ServiceAdapters from the k8s API and syncs the registry.
func (l *Loader) load() error {
	url := fmt.Sprintf("%s/apis/gateway.riotpiao.com/v1/namespaces/%s/serviceadapters", l.baseURL, l.namespace)
	req, err := http.NewRequest("GET", url, nil)
	if err != nil {
		return err
	}
	req.Header.Set("Authorization", "Bearer "+l.token)
	req.Header.Set("Accept", "application/json")

	resp, err := l.client.Do(req)
	if err != nil {
		return fmt.Errorf("k8s API request: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode == 429 {
		return fmt.Errorf("k8s API: storage reinitializing (429)")
	}
	if resp.StatusCode != 200 {
		body, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("k8s API: %d %s", resp.StatusCode, string(body[:min(len(body), 200)]))
	}

	var list crList
	if err := json.NewDecoder(resp.Body).Decode(&list); err != nil {
		return fmt.Errorf("decoding response: %w", err)
	}

	// Sync: add/update all found, registry handles dedup
	for i := range list.Items {
		item := &list.Items[i]
		adapter := &ServiceAdapter{
			Name:        item.Metadata.Name,
			Namespace:   item.Metadata.Namespace,
			ServiceName: item.Spec.ServiceName,
			CreatedAt:   time.Now(),
		}
		adapter.Spec.Upstream.URL = item.Spec.Upstream.URL
		adapter.Spec.Upstream.TimeoutSeconds = item.Spec.Upstream.TimeoutSeconds
		adapter.Spec.Auth.Required = item.Spec.Auth.Required
		adapter.Spec.Auth.Capability = item.Spec.Auth.Capability

		if existing := l.registry.Get(item.Spec.ServiceName); existing != nil {
			l.registry.Update(adapter)
		} else {
			l.registry.Add(adapter)
			log.Printf("serviceadapter loaded: %s → %s", item.Spec.ServiceName, item.Spec.Upstream.URL)
		}
	}

	return nil
}

// Minimal JSON structs for the k8s list response — no client-go needed.
type crList struct {
	Items []crItem `json:"items"`
}

type crItem struct {
	Metadata struct {
		Name      string `json:"name"`
		Namespace string `json:"namespace"`
	} `json:"metadata"`
	Spec struct {
		ServiceName string `json:"serviceName"`
		Upstream    struct {
			URL            string `json:"url"`
			TimeoutSeconds int32  `json:"timeoutSeconds"`
		} `json:"upstream"`
		Auth struct {
			Required   bool   `json:"required"`
			Capability string `json:"capability"`
		} `json:"auth"`
	} `json:"spec"`
}