perler-beads-server/internal/server/websocket_requests_test.go

package server

import (
	"context"
	"encoding/json"
	"github.com/google/uuid"
	"github.com/gorilla/websocket"
	"github.com/jackc/pgx/v5/pgxpool"
	"net/http"
	"os"
	"perler-beads-server/internal/domain"
	"strings"
	"testing"
	"time"
)

func wsEvents(c *websocket.Conn) <-chan map[string]any {
	events := make(chan map[string]any, 8)
	go func() {
		defer close(events)
		for {
			var v map[string]any
			if e := c.ReadJSON(&v); e != nil {
				return
			}
			events <- v
		}
	}()
	return events
}
func expectChanged(t *testing.T, events <-chan map[string]any) {
	t.Helper()
	select {
	case event, open := <-events:
		if !open || event["type"] != "tasks.changed" || len(event) != 1 {
			t.Fatalf("expected private-free invalidation, got %+v", event)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("committed task did not push notification")
	}
}
func expectQuiet(t *testing.T, events <-chan map[string]any) {
	t.Helper()
	select {
	case event := <-events:
		t.Fatalf("unexpected notification: %+v", event)
	case <-time.After(200 * time.Millisecond):
	}
}
func TestWebSocketCommittedTasksStoreIsolationAndReplay(t *testing.T) {
	dsn := os.Getenv("TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("TEST_DATABASE_URL absent; actual committed task push test skipped")
	}
	cfg, e := pgxpool.ParseConfig(dsn)
	if e != nil {
		t.Fatal(e)
	}
	if !strings.Contains(cfg.ConnConfig.Database, "test") {
		t.Fatal("requires isolated test database")
	}
	cfg.MaxConns = 1
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	db, e := pgxpool.NewWithConfig(ctx, cfg)
	if e != nil {
		t.Fatal(e)
	}
	defer db.Close()
	if e = Migrate(ctx, db); e != nil {
		t.Fatal(e)
	}
	s, h := wsTestServer(t)
	s.DB = db
	store := "ws-" + uuid.NewString()
	other := store + "-other"
	guest := wsToken(t, s, "guest", store)
	admin := wsToken(t, s, "admin", store)
	worker := wsConnect(t, h, h.URL)
	wsAuth(t, worker, admin)
	remote := wsConnect(t, h, h.URL)
	wsAuth(t, remote, wsToken(t, s, "staff", other))
	events, remoteEvents := wsEvents(worker), wsEvents(remote)
	if e = s.put(ctx, "tables", store, store, "", []domain.Table{{Code: "T1", Label: "Test", Enabled: true}}); e != nil {
		t.Fatal(e)
	}
	defer func() {
		cleanup, stop := context.WithTimeout(context.Background(), 5*time.Second)
		defer stop()
		var ids []string
		rows, e := db.Query(cleanup, `SELECT DISTINCT owner_id FROM documents WHERE kind='request' AND store_id=$1`, store)
		if e == nil {
			for rows.Next() {
				var id string
				rows.Scan(&id)
				ids = append(ids, id)
			}
			rows.Close()
		}
		for _, id := range ids {
			db.Exec(cleanup, `DELETE FROM request_keys WHERE user_id=$1`, id)
			db.Exec(cleanup, `DELETE FROM visits WHERE user_id=$1`, id)
		}
		db.Exec(cleanup, `DELETE FROM documents WHERE store_id=$1`, store)
	}()
	call := func(method, path, token, body string) (int, json.RawMessage) {
		t.Helper()
		request, e := http.NewRequestWithContext(ctx, method, h.URL+path, strings.NewReader(body))
		if e != nil {
			t.Fatal(e)
		}
		request.Header.Set("Content-Type", "application/json")
		request.Header.Set("Authorization", "Bearer "+token)
		response, e := http.DefaultClient.Do(request)
		if e != nil {
			t.Fatal(e)
		}
		defer response.Body.Close()
		var result struct {
			Code int
			Data json.RawMessage
		}
		if e = json.NewDecoder(response.Body).Decode(&result); e != nil {
			t.Fatal(e)
		}
		return result.Code, result.Data
	}
	if code, _ := call("POST", "/api/visits", guest, `{"storeId":"`+store+`","tableCode":"T1"}`); code != 200 {
		t.Fatalf("visit failed%d", code)
	}
	requestID := uuid.NewString()
	body := `{"kind":"help","note":"private note","requestId":"` + requestID + `"}`
	code, data := call("POST", "/api/service-requests", guest, body)
	if code != 200 {
		t.Fatalf("create failed%d %s", code, data)
	}
	var task domain.Request
	json.Unmarshal(data, &task)
	expectChanged(t, events)
	expectQuiet(t, remoteEvents)
	var stored domain.Request
	if e = s.get(ctx, "request", task.ID, &stored); e != nil || stored.ID != task.ID {
		t.Fatal("push before durable commit")
	}
	if code, replay := call("POST", "/api/service-requests", guest, body); code != 200 || string(replay) != string(data) { // JSON object key order can change in PostgreSQL; compare ID instead.
		var repeated domain.Request
		json.Unmarshal(replay, &repeated)
		if code != 200 || repeated.ID != task.ID {
			t.Fatalf("bad replay%d %s", code, replay)
		}
	}
	expectQuiet(t, events)
	conflict := strings.Replace(body, "private note", "different note", 1)
	if code, _ := call("POST", "/api/service-requests", guest, conflict); code != 409 {
		t.Fatalf("conflict%d", code)
	}
	expectQuiet(t, events)
	if code, _ := call("PATCH", "/api/service-requests/"+task.ID, admin, `{"status":"accepted"}`); code != 200 {
		t.Fatalf("patch%d", code)
	}
	expectChanged(t, events)
	expectQuiet(t, remoteEvents)
	if code, _ := call("PATCH", "/api/service-requests/"+task.ID, guest, `{"status":"cancelled"}`); code != 401 {
		t.Fatalf("unauthorized%d", code)
	}
	expectQuiet(t, events)
}
func TestTaskHubCoalescesWithoutBlocking(t *testing.T) {
	h := newTaskHub()
	c := &taskSocket{user: domain.User{StoreID: "one"}, changed: make(chan struct{}, 1), stop: make(chan wsClose, 1)}
	if !h.add(c) {
		t.Fatal("add")
	}
	done := make(chan struct{})
	go func() {
		for i := 0; i < 10000; i++ {
			h.notify("one")
			h.notify("other")
		}
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("slow browser blocked broadcaster")
	}
	if len(c.changed) != 1 {
		t.Fatal("notifications must merge")
	}
	h.remove(c)
	h.notify("one")
	h.shutdown()
	if h.add(c) {
		t.Fatal("shutdown hub accepted client")
	}
}