summercms/modules/lighthouse/centrifugo/client_test.go

package centrifugo

import (
	"context"
	"encoding/json"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	"git.golem15.com/golem15/summercms/modules/backpack"
	"git.golem15.com/golem15/summercms/modules/compass"
	"git.golem15.com/golem15/summercms/modules/lighthouse"
)

type apiCall struct {
	method, path, auth, contentType, body string
}

type fakeAPI struct {
	mu     sync.Mutex
	calls  []apiCall
	status int
	answer string
}

func newFakeAPI(t *testing.T) (*fakeAPI, *httptest.Server) {
	t.Helper()
	f := &fakeAPI{status: http.StatusOK, answer: `{}`}
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		body, _ := io.ReadAll(r.Body)
		f.mu.Lock()
		f.calls = append(f.calls, apiCall{r.Method, r.URL.Path, r.Header.Get("Authorization"), r.Header.Get("Content-Type"), string(body)})
		status, answer := f.status, f.answer
		f.mu.Unlock()
		w.WriteHeader(status)
		_, _ = io.WriteString(w, answer)
	}))
	t.Cleanup(srv.Close)
	return f, srv
}

func (f *fakeAPI) set(status int, answer string) {
	f.mu.Lock()
	f.status, f.answer = status, answer
	f.mu.Unlock()
}

func (f *fakeAPI) take() []apiCall {
	f.mu.Lock()
	defer f.mu.Unlock()
	out := f.calls
	f.calls = nil
	return out
}

func testClient(url, key string) *Client {
	c := NewClient(Config{APIURL: url + "/api", APIKey: key}, nil)
	c.now = func() time.Time { return fixedNow }
	return c
}

// TestClientRequests covers RT-01 and T-11-10: exact paths and bodies,
// the apikey header, the Carbon +00:00 timestamp, [] for an empty payload,
// any 2xx as success (including an error body on publish), errors that
// never carry the key, and no request at all without a key.
func TestClientRequests(t *testing.T) {
	f, srv := newFakeAPI(t)
	c := testClient(srv.URL, testAPIKey)
	ctx := context.Background()

	if err := c.Publish(ctx, "collection:5", "created.acme.widget", json.RawMessage(`{"z":1,"a":"x/y<z>"}`)); err != nil {
		t.Fatal(err)
	}
	if err := c.Publish(ctx, "collection:5", "pinged", nil); err != nil {
		t.Fatal(err)
	}
	if err := c.Broadcast(ctx, []string{"a:1", "b:2"}, "bulk", json.RawMessage(`{"count":2}`)); err != nil {
		t.Fatal(err)
	}
	if err := c.Broadcast(ctx, nil, "bulk", nil); err != nil {
		t.Fatal(err)
	}
	if err := c.Unsubscribe(ctx, 7, "collection:5"); err != nil {
		t.Fatal(err)
	}
	f.set(200, `{"result":{"presence":{"c1":{"user":"7"}}}}`)
	pres, err := c.Presence(ctx, "presence:room:1")
	if err != nil || pres["c1"] == nil {
		t.Fatalf("presence = %v, %v", pres, err)
	}
	f.set(200, `{"result":{"nodes":[{"name":"n1"}]}}`)
	info, err := c.Info(ctx)
	if err != nil || info["nodes"] == nil {
		t.Fatalf("info = %v, %v", info, err)
	}
	want := []apiCall{
		{"POST", "/api/publish", "apikey " + testAPIKey, "application/json", `{"channel":"collection:5","data":{"event":"created.acme.widget","payload":{"z":1,"a":"x/y<z>"},"timestamp":"2026-09-30T12:00:00+00:00"}}`},
		{"POST", "/api/publish", "apikey " + testAPIKey, "application/json", `{"channel":"collection:5","data":{"event":"pinged","payload":[],"timestamp":"2026-09-30T12:00:00+00:00"}}`},
		{"POST", "/api/broadcast", "apikey " + testAPIKey, "application/json", `{"channels":["a:1","b:2"],"data":{"event":"bulk","payload":{"count":2},"timestamp":"2026-09-30T12:00:00+00:00"}}`},
		{"POST", "/api/unsubscribe", "apikey " + testAPIKey, "application/json", `{"user":"7","channel":"collection:5"}`},
		{"POST", "/api/presence", "apikey " + testAPIKey, "application/json", `{"channel":"presence:room:1"}`},
		{"POST", "/api/info", "apikey " + testAPIKey, "application/json", `{}`},
	}
	got := f.take()
	if len(got) != len(want) {
		t.Fatalf("calls = %+v", got)
	}
	for i := range want {
		if got[i] != want[i] {
			t.Errorf("call %d = %+v\nwant %+v", i, got[i], want[i])
		}
	}

	t.Run("2xx_error_body_is_success_for_publish", func(t *testing.T) {
		f.set(200, `{"error":{"code":102,"message":"unknown channel"}}`)
		if err := c.Publish(ctx, "x:1", "e", nil); err != nil {
			t.Fatalf("publish with an error body = %v, want success like WinterCMS", err)
		}
		if _, err := c.Info(ctx); err == nil || !strings.Contains(err.Error(), "error 102: unknown channel") {
			t.Fatalf("info with an error body = %v", err)
		}
		f.set(200, `not json`)
		if _, err := c.Info(ctx); err == nil {
			t.Fatal("unreadable info answer accepted")
		}
		if p, err := c.Presence(ctx, "x"); err == nil || p == nil || len(p) != 0 {
			t.Fatalf("unreadable presence = %v, %v", p, err)
		}
		f.set(202, `{}`)
		if err := c.Unsubscribe(ctx, 1, "x"); err != nil {
			t.Fatalf("202 = %v", err)
		}
		if p, err := c.Presence(ctx, "x"); err != nil || p == nil {
			t.Fatalf("presence without result = %v, %v", p, err)
		}
		f.take()
	})

	t.Run("non_2xx_is_an_error_without_the_key", func(t *testing.T) {
		f.set(500, `{"secret":"`+testAPIKey+`"}`)
		for name, call := range map[string]func() error{
			"publish":     func() error { return c.Publish(ctx, "x:1", "e", nil) },
			"broadcast":   func() error { return c.Broadcast(ctx, []string{"x:1"}, "e", nil) },
			"unsubscribe": func() error { return c.Unsubscribe(ctx, 1, "x:1") },
			"presence":    func() error { _, err := c.Presence(ctx, "x:1"); return err },
			"info":        func() error { _, err := c.Info(ctx); return err },
		} {
			err := call()
			if err == nil || !strings.Contains(err.Error(), "HTTP 500") || strings.Contains(err.Error(), testAPIKey) {
				t.Errorf("%s: err = %v", name, err)
			}
		}
		f.take()
	})

	t.Run("unreachable_server", func(t *testing.T) {
		dead := httptest.NewServer(http.NotFoundHandler())
		url := dead.URL
		dead.Close()
		err := testClient(url, testAPIKey).Publish(ctx, "x:1", "e", nil)
		if err == nil || strings.Contains(err.Error(), testAPIKey) {
			t.Fatalf("err = %v", err)
		}
		if err := testClient("http://[::1", testAPIKey).Publish(ctx, "x:1", "e", nil); err == nil {
			t.Fatal("malformed URL accepted")
		}
		if err := testClient(srv.URL, testAPIKey).Publish(nil, "x:1", "e", json.RawMessage(`{"bad"`)); err == nil {
			t.Fatal("invalid payload JSON accepted")
		}
	})

	t.Run("empty_key_sends_nothing", func(t *testing.T) {
		off := testClient(srv.URL, "")
		f.take()
		if !errors.Is(off.Publish(ctx, "x", "e", nil), ErrNotConfigured) ||
			!errors.Is(off.Broadcast(ctx, []string{"x"}, "e", nil), ErrNotConfigured) ||
			!errors.Is(off.Unsubscribe(ctx, 1, "x"), ErrNotConfigured) {
			t.Fatal("publishing without a key did not return ErrNotConfigured")
		}
		if p, err := off.Presence(ctx, "x"); !errors.Is(err, ErrNotConfigured) || p == nil {
			t.Fatalf("presence = %v, %v", p, err)
		}
		if i, err := off.Info(ctx); !errors.Is(err, ErrNotConfigured) || i == nil {
			t.Fatalf("info = %v, %v", i, err)
		}
		if n := len(f.take()); n != 0 {
			t.Fatalf("%d requests sent without an API key", n)
		}
		if off.Enabled() || off.DebugInfo().APIKeySet || !c.DebugInfo().Enabled {
			t.Fatal("Enabled/DebugInfo")
		}
		var nilClient *Client
		if nilClient.Enabled() || nilClient.DebugInfo() != (DebugInfo{}) {
			t.Fatal("nil client")
		}
		raw, _ := json.Marshal(c.DebugInfo())
		if strings.Contains(string(raw), testAPIKey) {
			t.Fatal("DebugInfo carries the key")
		}
	})
}

// TestClientLoadConfig covers realtime.centrifugo.* parsing and the
// driver built from it.
func TestClientLoadConfig(t *testing.T) {
	def := LoadConfig(nil)
	if def.APIURL != DefaultAPIURL || def.TokenTTL != DefaultTokenTTL || def.WSURL != DefaultWSURL || def.TokenPath != DefaultTokenPath || def.SubscribePath != DefaultSubscribePath {
		t.Fatalf("defaults = %+v", def)
	}
	cfg, err := compass.Open(compass.Options{Dir: t.TempDir(), Env: "testing", Environ: []string{}})
	if err != nil {
		t.Fatal(err)
	}
	for k, v := range map[string]any{
		"realtime.driver":                    "centrifugo",
		"realtime.centrifugo.api_url":        "http://127.0.0.1:1/api/",
		"realtime.centrifugo.api_key":        " k ",
		"realtime.centrifugo.token_secret":   "s",
		"realtime.centrifugo.proxy_secret":   "p",
		"realtime.centrifugo.token_ttl":      "2m",
		"realtime.centrifugo.ws_url":         "wss://rt.example.test/ws",
		"realtime.centrifugo.token_path":     "/rt/token",
		"realtime.centrifugo.subscribe_path": "/rt/sub",
		"http.trusted_proxies":               []any{"10.0.0.0/8"},
	} {
		if err := cfg.Set(k, v); err != nil {
			t.Fatal(err)
		}
	}
	got := LoadConfig(cfg)
	if got.APIURL != "http://127.0.0.1:1/api" || got.APIKey != "k" || got.TokenSecret != "s" || got.ProxySecret != "p" ||
		got.TokenTTL != 2*time.Minute || got.WSURL != "wss://rt.example.test/ws" || got.TokenPath != "/rt/token" || got.SubscribePath != "/rt/sub" || len(got.TrustedProxies) != 1 {
		t.Fatalf("config = %+v", got)
	}
	svc, err := lighthouse.From(backpack.New(cfg))
	if err != nil {
		t.Fatal(err)
	}
	d, ok := svc.Driver().(*Driver)
	if !ok || d.Name() != DriverName || !d.Enabled() || d.Config().TokenPath != "/rt/token" || d.Client() == nil || !d.Issuer().Configured() {
		t.Fatalf("driver = %#v", svc.Driver())
	}
	routes := d.Routes()
	if len(routes) != 2 || routes[0].Path != "/rt/token" || routes[0].Surface != lighthouse.UserAuth || routes[1].Path != "/rt/sub" || routes[1].Surface != lighthouse.ServerToServer {
		t.Fatalf("routes = %+v", routes)
	}
	// Publish and Broadcast go through the client; an unreachable API is an
	// error, not a panic.
	if err := d.Publish(context.Background(), "a:1", "e", nil); err == nil {
		t.Fatal("publish to an unreachable API succeeded")
	}
	if err := d.Broadcast(context.Background(), []string{"a:1"}, "e", nil); err == nil {
		t.Fatal("broadcast to an unreachable API succeeded")
	}
}