summercms/modules/tide/centrifugo_test.go

package tide

import (
	"context"
	"encoding/json"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"
)

func postJSON(t *testing.T, url, auth, body string) *http.Response {
	t.Helper()
	req, err := http.NewRequest(http.MethodPost, url, strings.NewReader(body))
	if err != nil {
		t.Fatal(err)
	}
	if auth != "" {
		req.Header.Set("Authorization", auth)
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = resp.Body.Close() })
	return resp
}

func TestCentrifugoRecorderRecordsPublishAndBroadcast(t *testing.T) {
	rec := NewCentrifugoRecorder(CentrifugoRecorderOptions{APIKey: "k1"})
	srv := httptest.NewServer(rec)
	t.Cleanup(srv.Close)

	resp := postJSON(t, srv.URL+"/api/publish", "apikey k1", `{"channel":"room:1","data":{}}`)
	body, _ := io.ReadAll(resp.Body)
	if resp.StatusCode != 200 || string(body) != `{"result":{}}` {
		t.Fatalf("publish answer %d %s", resp.StatusCode, body)
	}
	postJSON(t, srv.URL+"/api/broadcast", "apikey wrong", `{"channels":["a:1","b:2"]}`)
	resp = postJSON(t, srv.URL+"/api/presence", "apikey k1", `{"channel":"room:1"}`)
	body, _ = io.ReadAll(resp.Body)
	if string(body) != `{"result":{"presence":{}}}` {
		t.Fatalf("presence answer %s", body)
	}
	if resp := postJSON(t, srv.URL+"/api/other", "", `{}`); resp.StatusCode != 404 {
		t.Fatalf("unknown path %d", resp.StatusCode)
	}

	pubs := rec.Publications()
	if len(pubs) != 2 {
		t.Fatalf("publications %d, want 2 (presence is not recorded)", len(pubs))
	}
	if pubs[0].Path != "/api/publish" || !pubs[0].Authorization || pubs[0].Method != "POST" {
		t.Fatalf("first %+v", pubs[0])
	}
	if pubs[1].Path != "/api/broadcast" || pubs[1].Authorization {
		t.Fatalf("second %+v", pubs[1])
	}
	rec.Reset()
	if len(rec.Publications()) != 0 {
		t.Fatal("reset kept publications")
	}
}

func TestCentrifugoRecorderRefusesNonLoopback(t *testing.T) {
	rec := NewCentrifugoRecorder(CentrifugoRecorderOptions{})
	if err := rec.ListenAndServe(t.Context(), "0.0.0.0:0"); err == nil || !strings.Contains(err.Error(), "loopback") {
		t.Fatalf("want loopback refusal, got %v", err)
	}
}

func TestNormalizePublications(t *testing.T) {
	store := mustMemoryStore()
	store.Set("id:collection", "12")
	store.Set("id:album", "40")
	store.Set("jwt:alice", "not-an-id")
	pubs := []Publication{{
		Method: "POST",
		Path:   "/api/publish",
		Body: json.RawMessage(`{"channel":"collection:12","data":{"event":"deleted.acme.album",` +
			`"payload":{"id":40,"collection_id":12,"action":"deleted","actor":{"user_id":3,"name":null},` +
			`"timestamp":"2026-09-30T10:00:00+00:00","count":12},"timestamp":"2026-09-30T10:00:01+00:00"}}`),
	}}
	got, err := NormalizePublications(pubs, store)
	if err != nil {
		t.Fatal(err)
	}
	want := `{"channel":"collection:{{id:collection}}","data":{"event":"deleted.acme.album",` +
		`"payload":{"id":{{id:album}},"collection_id":{{id:collection}},"action":"deleted","actor":"{{actor}}",` +
		`"timestamp":"{{timestamp}}","count":12},"timestamp":"{{timestamp}}"}}`
	if string(got[0].Body) != want {
		t.Fatalf("normalised\n got %s\nwant %s", got[0].Body, want)
	}

	// A Z timestamp and an actor with extra keys are left alone.
	odd := []Publication{{Method: "POST", Path: "/p", Body: json.RawMessage(
		`{"data":{"payload":{"actor":{"user_id":1,"name":"x","role":"a"}},"timestamp":"2026-09-30T10:00:00Z"}}`)}}
	got, err = NormalizePublications(odd, store)
	if err != nil {
		t.Fatal(err)
	}
	if !strings.Contains(string(got[0].Body), `"2026-09-30T10:00:00Z"`) || !strings.Contains(string(got[0].Body), `"role":"a"`) {
		t.Fatalf("over-normalised %s", got[0].Body)
	}

	store.Set("id:other", "12")
	if _, err := NormalizePublications(pubs, store); err == nil || !strings.Contains(err.Error(), "ambiguous") {
		t.Fatalf("want ambiguity error, got %v", err)
	}
}

func TestDiffPublications(t *testing.T) {
	want := []Publication{{Method: "POST", Path: "/api/publish", Authorization: true,
		Body: json.RawMessage(`{"channel":"c:{{id:c}}","data":{"payload":{"id":{{id:a}},"n":1}}}`)}}
	same := []Publication{{Method: "POST", Path: "/api/publish", Authorization: true,
		Body: json.RawMessage(`{"data":{"payload":{"n":1,"id":{{id:a}}}},"channel":"c:{{id:c}}"}`)}}
	if d := DiffPublications(want, same); len(d) != 0 {
		t.Fatalf("key order must not matter: %+v", d)
	}
	stringID := []Publication{{Method: "POST", Path: "/api/broadcast", Authorization: false,
		Body: json.RawMessage(`{"channel":"c:{{id:c}}","data":{"payload":{"id":"{{id:a}}","n":1}}}`)}}
	d := DiffPublications(want, stringID)
	paths := map[string]bool{}
	for _, x := range d {
		paths[x.Path] = true
	}
	for _, p := range []string{"$[0].path", "$[0].authorization", "$[0].body.data.payload.id"} {
		if !paths[p] {
			t.Fatalf("missing diff at %s in %+v", p, d)
		}
	}
	if d := DiffPublications(want, nil); len(d) != 1 || d[0].Path != "$" {
		t.Fatalf("count diff %+v", d)
	}
}

func TestBroadcastGoldenRoundTrip(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "g.yaml")
	g := BroadcastGolden{Version: CurrentVersion, Name: "deleted", Flow: "flows/x#delete", Pending: "later",
		Publications: []Publication{{Method: "POST", Path: "/api/publish", Authorization: true,
			Body: json.RawMessage(`{"channel":"c:{{id:c}}","data":{"payload":{"id":{{id:a}}}}}`)}}}
	if err := WriteBroadcastGolden(path, g); err != nil {
		t.Fatal(err)
	}
	back, err := LoadBroadcastGolden(path)
	if err != nil {
		t.Fatal(err)
	}
	if back.Name != g.Name || back.Flow != g.Flow || back.Pending != g.Pending || len(back.Publications) != 1 {
		t.Fatalf("round trip %+v", back)
	}
	if d := DiffPublications(g.Publications, back.Publications); len(d) != 0 {
		t.Fatalf("round trip diff %+v", d)
	}

	leak := g
	leak.Publications = []Publication{{Method: "POST", Path: "/p",
		Body: json.RawMessage(`{"token":"eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIn0.c2lnbmF0dXJl"}`)}}
	if err := WriteBroadcastGolden(filepath.Join(dir, "leak.yaml"), leak); err == nil {
		t.Fatal("a JWT-shaped body must be refused")
	}
	if err := os.WriteFile(filepath.Join(dir, "bad.yaml"), []byte("version: 1\nname: x\nextra: 1\npublications: []\n"), 0o644); err != nil {
		t.Fatal(err)
	}
	if _, err := LoadBroadcastGolden(filepath.Join(dir, "bad.yaml")); err == nil {
		t.Fatal("unknown field must be rejected")
	}
}

func freeLoopbackAddr(t *testing.T) string {
	t.Helper()
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	addr := ln.Addr().String()
	_ = ln.Close()
	return addr
}

func TestRecordBroadcastsStep(t *testing.T) {
	listen := freeLoopbackAddr(t)
	publish := func(body string) {
		req, _ := http.NewRequest(http.MethodPost, "http://"+listen+"/api/publish", strings.NewReader(body))
		req.Header.Set("Authorization", "apikey test-key")
		if resp, err := http.DefaultClient.Do(req); err == nil {
			_ = resp.Body.Close()
		}
	}
	backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		switch r.Method {
		case http.MethodPost:
			publish(`{"channel":"room:5","data":{"event":"created","payload":{"id":77},"timestamp":"2026-09-30T10:00:00+00:00"}}`)
			_, _ = io.WriteString(w, `{"data":{"id":77}}`)
		case http.MethodDelete:
			publish(`{"channel":"room:5","data":{"event":"deleted","payload":{"id":77},"timestamp":"2026-09-30T10:00:02+00:00"}}`)
			_, _ = io.WriteString(w, `{"message":"deleted"}`)
		}
	}))
	t.Cleanup(backend.Close)

	store := mustMemoryStore()
	store.Set("id:room", "5")
	spec := Flow{Version: CurrentVersion, Name: "items", Steps: []Step{
		{ID: "create", Request: Request{Method: "POST", Path: "/items"},
			Capture: []CaptureRule{{From: "response.json", Path: "$.data.id", As: "id:item"}}},
		{ID: "delete", Request: Request{Method: "DELETE", Path: "/items/{{id:item}}"}},
	}}
	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
	defer cancel()
	g, err := RecordBroadcasts(ctx, spec, BroadcastConfig{
		Target: backend.URL, Listen: listen, APIKey: "test-key", Store: store,
		Step: "delete", IDs: []string{"id:room", "id:item"}, Settle: 50 * time.Millisecond,
	})
	if err != nil {
		t.Fatal(err)
	}
	if g.Name != "delete" || g.Flow != "items#delete" || len(g.Publications) != 1 {
		t.Fatalf("golden %+v", g)
	}
	want := `{"channel":"room:{{id:room}}","data":{"event":"deleted","payload":{"id":{{id:item}}},"timestamp":"{{timestamp}}"}}`
	if string(g.Publications[0].Body) != want || !g.Publications[0].Authorization {
		t.Fatalf("publication %+v body %s", g.Publications[0], g.Publications[0].Body)
	}

	if _, err := RecordBroadcasts(ctx, spec, BroadcastConfig{Target: "http://192.0.2.1:80", APIKey: "k"}); err == nil {
		t.Fatal("a non-loopback target must be refused")
	}
}

// TestCentrifugoRecorder covers the fake Centrifugo's edges (T-11-29):
// only publish and broadcast are recorded, the authorization flag is a
// comparison and the header value is never stored, other methods get 405,
// bodies over the cap get 413 and are not recorded, an empty configured
// key never counts as authorized, and ListenAndServe serves on loopback
// until its ctx ends.
func TestCentrifugoRecorder(t *testing.T) {
	rec := NewCentrifugoRecorder(CentrifugoRecorderOptions{APIKey: "rec-key"})
	srv := httptest.NewServer(rec)
	t.Cleanup(srv.Close)

	for _, path := range []string{"/api/unsubscribe", "/api/info", "/api/info/"} {
		resp := postJSON(t, srv.URL+path, "apikey rec-key", `{}`)
		body, _ := io.ReadAll(resp.Body)
		if resp.StatusCode != 200 || string(body) != `{"result":{}}` || resp.Header.Get("Content-Type") != "application/json" {
			t.Fatalf("%s answer %d %s", path, resp.StatusCode, body)
		}
	}
	get, err := http.Get(srv.URL + "/api/publish")
	if err != nil {
		t.Fatal(err)
	}
	_ = get.Body.Close()
	if get.StatusCode != http.StatusMethodNotAllowed || get.Header.Get("Allow") != http.MethodPost {
		t.Fatalf("GET publish = %d allow %q", get.StatusCode, get.Header.Get("Allow"))
	}
	big := `{"pad":"` + strings.Repeat("x", MaxPublicationBody) + `"}`
	if resp := postJSON(t, srv.URL+"/api/publish", "apikey rec-key", big); resp.StatusCode != http.StatusRequestEntityTooLarge {
		t.Fatalf("oversized body = %d", resp.StatusCode)
	}
	postJSON(t, srv.URL+"/api/publish/", "apikey rec-key", `{"channel":"a:1"}`)
	postJSON(t, srv.URL+"/api/publish", "Bearer rec-key", `{"channel":"a:2"}`)
	postJSON(t, srv.URL+"/api/publish", "", `{"channel":"a:3"}`)
	pubs := rec.Publications()
	if len(pubs) != 3 {
		t.Fatalf("publications = %d, want 3 (info, unsubscribe and the oversized body are not recorded)", len(pubs))
	}
	if !pubs[0].Authorization || pubs[1].Authorization || pubs[2].Authorization {
		t.Fatalf("authorization flags = %v %v %v", pubs[0].Authorization, pubs[1].Authorization, pubs[2].Authorization)
	}
	raw, _ := json.Marshal(pubs)
	if strings.Contains(string(raw), "rec-key") {
		t.Fatal("a recorded publication holds the Authorization value")
	}
	pubs[0].Body[0] = 'X'
	if rec.Publications()[0].Body[0] != '{' {
		t.Fatal("Publications returned the recorder's own body slice")
	}
	if NewCentrifugoRecorder(CentrifugoRecorderOptions{}).authorized("apikey ") {
		t.Fatal("an empty configured key authorized a request")
	}

	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	addr := ln.Addr().String()
	_ = ln.Close()
	ctx, cancel := context.WithCancel(t.Context())
	done := make(chan error, 1)
	go func() { done <- rec.ListenAndServe(ctx, addr) }()
	if err := waitListening(t.Context(), addr, done); err != nil {
		t.Fatal(err)
	}
	postJSON(t, "http://"+addr+"/api/broadcast", "apikey rec-key", `{"channels":["a:1"]}`)
	cancel()
	select {
	case err := <-done:
		if err != nil {
			t.Fatalf("ListenAndServe after cancel = %v", err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("recorder did not stop")
	}
	if err := rec.ListenAndServe(t.Context(), "127.0.0.1:bad"); err == nil {
		t.Fatal("an unusable address was accepted")
	}
	stopped := make(chan error, 1)
	stopped <- nil
	if err := waitListening(t.Context(), "127.0.0.1:1", stopped); err == nil || !strings.Contains(err.Error(), "stopped") {
		t.Fatalf("waitListening on a stopped recorder = %v", err)
	}
	cctx, ccancel := context.WithCancel(t.Context())
	ccancel()
	if err := sleepCtx(cctx, time.Second); err == nil {
		t.Fatal("sleepCtx ignored a cancelled ctx")
	}
}

// TestFlowIDNames covers the id variables a broadcast recording masks by
// default: placeholders in the path, query, body and headers, and id
// captures, sorted and without duplicates.
func TestFlowIDNames(t *testing.T) {
	spec := Flow{Steps: []Step{
		{Request: Request{Path: "/api/collections/{{id:collection}}/albums", Query: "artist={{id:artist}}", Headers: map[string]string{"X-Acme": "{{id:header}}"}},
			Capture: []CaptureRule{{As: "id:album"}, {As: "jwt:alice"}}},
		{Request: Request{Path: "/api/albums/{{id:album}}", Body: Body(`{"collection_id":{{id:collection}}}`)}},
	}}
	if got := strings.Join(flowIDNames(spec), ","); got != "id:album,id:artist,id:collection,id:header" {
		t.Fatalf("flowIDNames = %s", got)
	}
}