1package relay23import (4 "context"5 "encoding/json"6 "net/http"7 "net/http/httptest"8 "strings"9 "testing"10 "time"1112 "github.com/FelineStateMachine/tinyrelay/internal/event"13 "github.com/FelineStateMachine/tinyrelay/internal/syncprotocol"14 "github.com/coder/websocket"15)1617type testBackend struct{}1819func (testBackend) Publish(context.Context, event.Event, Session) (string, error) { return "", nil }20func (testBackend) Query(context.Context, []event.Filter, Session) ([]event.Event, error) {21 return nil, nil22}23func (testBackend) Count(context.Context, []event.Filter, Session) (any, error) { return 0, nil }2425type resultBackend struct{ reason string }2627func (b resultBackend) Publish(context.Context, event.Event, Session) (string, error) {28 return b.reason, nil29}30func (resultBackend) Query(context.Context, []event.Filter, Session) ([]event.Event, error) {31 return nil, nil32}33func (resultBackend) Count(context.Context, []event.Filter, Session) (any, error) { return 0, nil }3435type filterBackend struct{ resultBackend }3637func (filterBackend) CanReadFilter(event.Event, Session, *event.Filter) bool { return false }3839type gatedBackend struct {40 testBackend41 queryStarted chan struct{}42 allowQuery chan struct{}43 publish chan struct{}44 syncItems []syncprotocol.Item45}4647func (b *gatedBackend) Query(context.Context, []event.Filter, Session) ([]event.Event, error) {48 close(b.queryStarted)49 <-b.allowQuery50 return nil, nil51}52func (b *gatedBackend) Publish(context.Context, event.Event, Session) (string, error) {53 b.publish <- struct{}{}54 return "", nil55}56func (b *gatedBackend) Sync(context.Context, event.Filter, Session) ([]syncprotocol.Item, error) {57 return b.syncItems, nil58}5960func TestPendingQueueRejectsSlowConsumerBeforeUnboundedGrowth(t *testing.T) {61 r := New(testBackend{}, Config{MaxPendingBytes: 4})62 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))63 if err := c.enqueue("1234"); err != nil {64 t.Fatalf("first message: %v", err)65 }66 if err := c.enqueue("5"); err == nil {67 t.Fatal("expected slow-consumer rejection")68 }69 c.cancel()70}7172func TestPublishDoesNotRebroadcastDuplicate(t *testing.T) {73 r := New(resultBackend{reason: "duplicate: already have this event"}, Config{})74 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))75 c.subs["sub"] = subscription{filters: []event.Filter{{Kinds: []int{1}}}}76 r.clientsMu.Lock()77 r.clients[c] = struct{}{}78 r.clientsMu.Unlock()79 _, err := r.Publish(context.Background(), event.Event{ID: strings.Repeat("a", 64), Kind: 1}, Session{})80 if err != nil {81 t.Fatal(err)82 }83 c.mu.Lock()84 defer c.mu.Unlock()85 if len(c.queue) != 0 {86 t.Fatalf("duplicate publication was rebroadcast: %d queued messages", len(c.queue))87 }88}8990func TestFanoutUsesFilterAwarePrivacyGate(t *testing.T) {91 r := New(filterBackend{}, Config{})92 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))93 c.subs["sub"] = subscription{filters: []event.Filter{{Kinds: []int{1}}}}94 r.clientsMu.Lock()95 r.clients[c] = struct{}{}96 r.clientsMu.Unlock()97 r.BroadcastGenerated(event.Event{ID: strings.Repeat("a", 64), Kind: 1})98 c.mu.Lock()99 defer c.mu.Unlock()100 if len(c.queue) != 0 {101 t.Fatal("filter-aware reader gate was bypassed")102 }103}104105func TestPolicyChangeClosesSubscriptionsAndSyncSessions(t *testing.T) {106 r := New(testBackend{}, Config{})107 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))108 c.subs["sub"] = subscription{filters: []event.Filter{{Kinds: []int{1}}}}109 c.syncs["sync"] = &syncprotocol.Session{}110 r.clientsMu.Lock()111 r.clients[c] = struct{}{}112 r.clientsMu.Unlock()113 r.CloseSubscriptions("blocked: revoked")114 c.mu.Lock()115 defer c.mu.Unlock()116 if len(c.subs) != 0 || len(c.syncs) != 0 {117 t.Fatal("policy change retained a live subscription or sync session")118 }119 if len(c.queue) != 2 || !strings.Contains(string(c.queue[1].data), `"NEG-CLOSE","sync"`) {120 t.Fatalf("policy change did not close both protocol sessions: %+v", c.queue)121 }122}123124func TestApplyACLChangeInvalidatesSyncUnderPublicationFence(t *testing.T) {125 r := New(testBackend{}, Config{})126 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))127 c.syncs["sync"] = &syncprotocol.Session{}128 r.clientsMu.Lock()129 r.clients[c] = struct{}{}130 r.clientsMu.Unlock()131 called := false132 if _, err := r.ApplyACLChange(context.Background(), func(context.Context) (any, error) {133 called = true134 return "ok", nil135 }, "blocked: membership revoked"); err != nil {136 t.Fatal(err)137 }138 if !called {139 t.Fatal("ACL mutation callback did not run")140 }141 c.mu.Lock()142 defer c.mu.Unlock()143 if len(c.syncs) != 0 || len(c.queue) != 1 || !strings.Contains(string(c.queue[0].data), `"NEG-CLOSE","sync"`) {144 t.Fatalf("sync session survived ACL mutation: %+v", c.queue)145 }146}147148func TestAcceptedMembershipEventInvalidatesSyncBeforeFanout(t *testing.T) {149 r := New(testBackend{}, Config{ChangesAccess: func(e event.Event) bool { return e.Kind == event.KIND_LEAVE }})150 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))151 c.syncs["sync"] = &syncprotocol.Session{}152 c.subs["sub"] = subscription{filters: []event.Filter{{Kinds: []int{event.KIND_LEAVE}}}}153 r.clientsMu.Lock()154 r.clients[c] = struct{}{}155 r.clientsMu.Unlock()156 _, err := r.Publish(context.Background(), event.Event{ID: strings.Repeat("a", 64), Kind: event.KIND_LEAVE}, Session{})157 if err != nil {158 t.Fatal(err)159 }160 c.mu.Lock()161 defer c.mu.Unlock()162 if len(c.syncs) != 0 || len(c.subs) != 0 {163 t.Fatal("accepted membership event retained stale transport state")164 }165 if len(c.queue) < 2 || !queueContains(c.queue, `"NEG-CLOSE","sync"`) {166 t.Fatalf("membership revocation did not close sync before fanout: %+v", c.queue)167 }168}169170func queueContains(queue []outbound, value string) bool {171 for _, item := range queue {172 if strings.Contains(string(item.data), value) {173 return true174 }175 }176 return false177}178179func TestZeroMessageLimitIsUnlimited(t *testing.T) {180 r := New(testBackend{}, Config{MaxMessageBytes: 0})181 if r.cfg.MaxMessageBytes != 0 {182 t.Fatalf("zero message limit became %d", r.cfg.MaxMessageBytes)183 }184 if got := r.cfg.withDefaults().MaxMessageBytes; got != 0 {185 t.Fatalf("zero message limit became %d", got)186 }187}188189func TestIdlePingPongKeepsSessionAliveForLaterTraffic(t *testing.T) {190 r := New(testBackend{}, Config{ReadLimit: 20 * time.Millisecond, PingInterval: 5 * time.Millisecond})191 server := httptest.NewServer(http.HandlerFunc(r.HandleHTTP))192 defer server.Close()193 conn, _, err := websocket.Dial(context.Background(), "ws"+strings.TrimPrefix(server.URL, "http"), nil)194 if err != nil {195 t.Fatal(err)196 }197 defer conn.Close(websocket.StatusNormalClosure, "done")198199 readCtx, cancel := context.WithCancel(context.Background())200 defer cancel()201 texts := make(chan []byte, 8)202 readErr := make(chan error, 1)203 go func() {204 for {205 typ, data, err := conn.Read(readCtx)206 if err != nil {207 readErr <- err208 return209 }210 if typ == websocket.MessageText {211 texts <- data212 }213 }214 }()215 select {216 case <-texts:217 case err := <-readErr:218 t.Fatalf("read AUTH: %v", err)219 case <-time.After(time.Second):220 t.Fatal("timed out waiting for AUTH")221 }222 time.Sleep(50 * time.Millisecond)223 if err := conn.Write(context.Background(), websocket.MessageText, []byte(`["REQ","idle",{}]`)); err != nil {224 t.Fatalf("write REQ after idle: %v", err)225 }226 select {227 case data := <-texts:228 if !strings.Contains(string(data), `"EOSE","idle"`) {229 t.Fatalf("response after idle = %s", data)230 }231 case err := <-readErr:232 t.Fatalf("session closed during healthy idle: %v", err)233 case <-time.After(time.Second):234 t.Fatal("timed out waiting for EOSE after idle")235 }236 secret := strings.Repeat("1", 64)237 e := event.Event{Kind: 1, CreatedAt: time.Now().Unix(), Content: "after idle"}238 if err := event.Sign(&e, secret); err != nil {239 t.Fatal(err)240 }241 raw, _ := json.Marshal([]any{"EVENT", e})242 if err := conn.Write(context.Background(), websocket.MessageText, raw); err != nil {243 t.Fatalf("write EVENT after idle: %v", err)244 }245 select {246 case data := <-texts:247 if !strings.Contains(string(data), `"OK","`+e.ID+`",true`) {248 t.Fatalf("publish response after idle = %s", data)249 }250 case err := <-readErr:251 t.Fatalf("session closed before publish response: %v", err)252 case <-time.After(time.Second):253 t.Fatal("timed out waiting for publish response after idle")254 }255}256257func TestRemoteIPStripsPortWithoutTrustingForwardedHeaders(t *testing.T) {258 if got := remoteIP("192.0.2.10:7447"); got != "192.0.2.10" {259 t.Fatalf("remote IP = %q", got)260 }261 if got := remoteIP("[2001:db8::1]:7447"); got != "2001:db8::1" {262 t.Fatalf("IPv6 remote IP = %q", got)263 }264}265266func TestAuthUsesConnectionRelayURL(t *testing.T) {267 r := New(testBackend{}, Config{268 RelayURL: "wss://canonical.example/tenant",269 RequestRelayURL: func(*http.Request) string {270 return "wss://custom.example/tenant"271 },272 })273 c := newClient(r, nil, httptest.NewRequest("GET", "https://custom.example/tenant", nil))274 secret := strings.Repeat("1", 64)275 auth := event.Event{CreatedAt: time.Now().Unix(), Kind: event.KIND_AUTH, Tags: [][]string{{"relay", "wss://custom.example/tenant/"}, {"challenge", c.challenge}}, Content: ""}276 if err := event.Sign(&auth, secret); err != nil {277 t.Fatal(err)278 }279 c.handleAuth([]json.RawMessage{mustRaw(auth)})280 c.mu.Lock()281 if !strings.Contains(string(c.queue[len(c.queue)-1].data), `"OK","`+auth.ID+`",true`) {282 t.Fatalf("custom relay AUTH was rejected: %s", c.queue[len(c.queue)-1].data)283 }284 c.mu.Unlock()285286 bad := auth287 bad.Tags = [][]string{{"relay", "wss://other.example"}, {"challenge", c.challenge}}288 if err := event.Sign(&bad, secret); err != nil {289 t.Fatal(err)290 }291 c.handleAuth([]json.RawMessage{mustRaw(bad)})292 c.mu.Lock()293 defer c.mu.Unlock()294 if !strings.Contains(string(c.queue[len(c.queue)-1].data), `"OK","`+bad.ID+`",false`) {295 t.Fatalf("mismatched relay AUTH was accepted: %s", c.queue[len(c.queue)-1].data)296 }297}298299func mustRaw(v any) json.RawMessage {300 b, _ := json.Marshal(v)301 return b302}303304func TestNormalizeRelayIgnoresTrailingSlashOnly(t *testing.T) {305 if normalizeRelay("WSS://Relay.Example:8448/path///") != "relay.example" {306 t.Fatal("relay URL normalization should compare the relay host")307 }308}309310func TestREQQueryAndPublishShareTheHistoricalLiveFence(t *testing.T) {311 b := &gatedBackend{queryStarted: make(chan struct{}), allowQuery: make(chan struct{}), publish: make(chan struct{}, 1)}312 r := New(b, Config{})313 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))314 requestDone := make(chan struct{})315 go func() {316 c.handleRequest("REQ", []json.RawMessage{json.RawMessage(`"sub"`), json.RawMessage(`{}`)})317 close(requestDone)318 }()319 <-b.queryStarted320 publishDone := make(chan struct{})321 go func() {322 r.mu.Lock()323 _, _ = r.backend.Publish(context.Background(), event.Event{}, c.snapshot())324 r.mu.Unlock()325 close(publishDone)326 }()327 select {328 case <-b.publish:329 t.Fatal("publish crossed the historical query fence")330 case <-time.After(20 * time.Millisecond):331 }332 close(b.allowQuery)333 <-requestDone334 <-publishDone335 c.cancel()336}337338func TestCommitGeneratedSharesHistoricalLiveFence(t *testing.T) {339 b := &gatedBackend{queryStarted: make(chan struct{}), allowQuery: make(chan struct{}), publish: make(chan struct{}, 1)}340 r := New(b, Config{})341 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))342 queryDone := make(chan struct{})343 go func() {344 c.handleRequest("REQ", []json.RawMessage{json.RawMessage(`"sub"`), json.RawMessage(`{}`)})345 close(queryDone)346 }()347 <-b.queryStarted348 commitDone := make(chan struct{})349 go func() {350 _, _ = r.CommitGenerated(context.Background(), func(context.Context) (event.Event, error) {351 close(commitDone)352 return event.Event{}, nil353 })354 }()355 select {356 case <-commitDone:357 t.Fatal("generated commit crossed the historical query fence")358 case <-time.After(20 * time.Millisecond):359 }360 close(b.allowQuery)361 <-queryDone362 select {363 case <-commitDone:364 case <-time.After(time.Second):365 t.Fatal("generated commit did not proceed after query")366 }367 c.cancel()368}369370func TestNegOpenUsesAuthorizedSnapshotAndReportsProtocolResult(t *testing.T) {371 initiator, err := syncprotocol.NewSession([]syncprotocol.Item{{ID: strings.Repeat("a", 64), Timestamp: 1}}, true)372 if err != nil {373 t.Fatal(err)374 }375 message, err := initiator.Start()376 if err != nil {377 t.Fatal(err)378 }379 b := &gatedBackend{syncItems: []syncprotocol.Item{{ID: strings.Repeat("b", 64), Timestamp: 2}}}380 r := New(b, Config{})381 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))382 c.handleSync("NEG-OPEN", []json.RawMessage{json.RawMessage(`"sync"`), json.RawMessage(`{}`), json.RawMessage(`"` + message + `"`)})383 c.mu.Lock()384 defer c.mu.Unlock()385 if c.syncs["sync"] == nil {386 t.Fatal("NEG-OPEN did not retain a session after a non-empty response")387 }388 if len(c.queue) == 0 || !strings.Contains(string(c.queue[len(c.queue)-1].data), `"NEG-MSG"`) {389 t.Fatal("NEG-OPEN did not enqueue NEG-MSG")390 }391}