tinyrelayfile
tinyrelay/internal/replication/replication_test.go/
internal/replication/replication_test.go raw
1package replication
2
3import (
4 "bytes"
5 "context"
6 "encoding/json"
7 "fmt"
8 "net"
9 "net/http"
10 "net/http/httptest"
11 "net/url"
12 "os"
13 "strings"
14 "testing"
15 "time"
16
17 "github.com/FelineStateMachine/tinyrelay/internal/event"
18 "github.com/FelineStateMachine/tinyrelay/internal/storage"
19 "github.com/FelineStateMachine/tinyrelay/internal/work"
20 "github.com/coder/websocket"
21)
22
23type routeBook struct {
24 writes map[string][]string
25 reads map[string][]string
26}
27
28type countingDiscovery struct{ calls int }
29
30func (d *countingDiscovery) DiscoverRelays(context.Context, string) ([]string, error) {
31 d.calls++
32 return []string{"wss://remote.example"}, nil
33}
34
35type fakeTransport struct {
36 result DeliveryResult
37 event event.Event
38}
39
40type recordingTransport struct {
41 events []event.Event
42}
43
44func (r *recordingTransport) Send(_ context.Context, _ string, e event.Event) (DeliveryResult, error) {
45 r.events = append(r.events, e)
46 return DeliveryResult{Accepted: true}, nil
47}
48
49type fakePush struct{ sent map[string][]string }
50
51type countingRoundTripper struct{ calls int }
52
53type fixedResolver []net.IPAddr
54
55func (r fixedResolver) LookupIPAddr(context.Context, string) ([]net.IPAddr, error) {
56 return []net.IPAddr(r), nil
57}
58
59func TestWebsocketDialerBlocksPrivateByDefault(t *testing.T) {
60 _, _, err := (WebsocketDialer{Resolver: fixedResolver{{IP: net.ParseIP("127.0.0.1")}}}).Dial(context.Background(), "ws://relay.test/socket")
61 if err == nil || !strings.Contains(err.Error(), "private or local") {
62 t.Fatalf("expected private relay rejection, got %v", err)
63 }
64}
65
66func TestWebsocketDialerAllowsPrivateWhenExplicitlyEnabled(t *testing.T) {
67 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
68 conn, err := websocket.Accept(w, req, nil)
69 if err != nil {
70 return
71 }
72 defer conn.Close(websocket.StatusNormalClosure, "done")
73 _, _, _ = conn.Read(req.Context())
74 }))
75 t.Cleanup(server.Close)
76 parsed, err := url.Parse(server.URL)
77 if err != nil {
78 t.Fatal(err)
79 }
80 target := "ws://" + parsed.Host
81 socket, _, err := (WebsocketDialer{AllowPrivate: true}).Dial(context.Background(), target)
82 if err != nil {
83 t.Fatalf("allow private dial: %v", err)
84 }
85 _ = socket.Close(nil)
86}
87
88func TestWebsocketDialerRejectsDNSPinningToPrivateAddress(t *testing.T) {
89 resolver := fixedResolver{{IP: net.ParseIP("192.168.1.20")}}
90 _, _, err := (WebsocketDialer{Resolver: resolver}).Dial(context.Background(), "ws://relay.test/socket")
91 if err == nil || !strings.Contains(err.Error(), "private or local") {
92 t.Fatalf("expected DNS-pinned private rejection, got %v", err)
93 }
94}
95
96func TestWebsocketDialerBlocksCGNATAndReservedAddresses(t *testing.T) {
97 for _, address := range []string{"100.87.226.30", "198.18.0.1", "240.0.0.1", "224.0.0.1"} {
98 resolver := fixedResolver{{IP: net.ParseIP(address)}}
99 _, _, err := (WebsocketDialer{Resolver: resolver}).Dial(context.Background(), "ws://relay.test/socket")
100 if err == nil || !strings.Contains(err.Error(), "private or local") {
101 t.Errorf("address %s was not blocked: %v", address, err)
102 }
103 }
104}
105
106func (r *countingRoundTripper) RoundTrip(*http.Request) (*http.Response, error) {
107 r.calls++
108 return &http.Response{StatusCode: 200, Body: http.NoBody, Header: make(http.Header)}, nil
109}
110
111type fakeBackupProvider struct{ restored BackupState }
112
113func (f *fakeBackupProvider) Snapshot(context.Context) (BackupState, error) {
114 return BackupState{Config: json.RawMessage(`{"owner":"owner"` + `}`), Blobs: []BackupObject{{Name: "blob", SHA256: "hash", Content: []byte("data")}}}, nil
115}
116func (f *fakeBackupProvider) Restore(_ context.Context, state BackupState) error {
117 f.restored = state
118 return nil
119}
120
121func (f *fakePush) Send(_ context.Context, target string, e event.Event) (DeliveryResult, error) {
122 if f.sent == nil {
123 f.sent = map[string][]string{}
124 }
125 f.sent[target] = append(f.sent[target], e.ID)
126 return DeliveryResult{Accepted: true}, nil
127}
128
129func (f *fakeTransport) Send(_ context.Context, _ string, e event.Event) (DeliveryResult, error) {
130 f.event = e
131 return f.result, nil
132}
133
134func openReplicationStore(t *testing.T) *storage.Store {
135 t.Helper()
136 store, err := storage.Open(context.Background(), t.TempDir()+"/relay.db")
137 if err != nil {
138 t.Fatal(err)
139 }
140 t.Cleanup(func() { _ = store.Close() })
141 return store
142}
143
144func signedEvent(t *testing.T, createdAt int64) event.Event {
145 t.Helper()
146 secret := "1111111111111111111111111111111111111111111111111111111111111111"
147 e := event.Event{CreatedAt: createdAt, Kind: 1, Tags: [][]string{}, Content: "hello"}
148 if err := event.Sign(&e, secret); err != nil {
149 t.Fatal(err)
150 }
151 return e
152}
153
154func (r routeBook) WriteRelays(string) []string { return r.writes["author"] }
155func (r routeBook) ReadRelays(string) []string { return r.reads["recipient"] }
156
157func TestPrepareIntentsPreservesNIP65RoutingAndExclusions(t *testing.T) {
158 e := event.Event{ID: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", PubKey: "author", Kind: 1}
159 // The planner only needs wire identity here; storage validates complete events.
160 book := routeBook{writes: map[string][]string{"author": {"wss://write.example"}}, reads: map[string][]string{"recipient": {"wss://read.example"}}}
161 policy := Policy{Enabled: true, SelfPubKey: "self", ReadMembersOnly: false}
162 got := PrepareIntents(context.Background(), e, OriginClient, policy, book)
163 if len(got) != 1 || got[0].Kind != "delivery" || got[0].Target != "wss://write.example" {
164 t.Fatalf("intents = %#v", got)
165 }
166 if got[0].EventID != e.ID {
167 t.Fatalf("event id = %q", got[0].EventID)
168 }
169 for _, excluded := range []event.Event{
170 {ID: e.ID, PubKey: e.PubKey, Kind: event.KIND_DM},
171 {ID: e.ID, PubKey: e.PubKey, Kind: 1, Tags: [][]string{{"-", "protected"}}},
172 } {
173 if intents := PrepareIntents(context.Background(), excluded, OriginClient, policy, book); len(intents) != 0 {
174 t.Fatalf("excluded event planned: %#v", intents)
175 }
176 }
177}
178
179func TestPrepareIntentsAreDeterministicAndDeduplicated(t *testing.T) {
180 e := event.Event{ID: "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", PubKey: "author", Kind: 1, Tags: [][]string{{"p", "recipient"}, {"p", "recipient"}}}
181 book := routeBook{writes: map[string][]string{"author": {"wss://same.example", "wss://same.example"}}, reads: map[string][]string{"recipient": {"wss://same.example"}}}
182 got := PrepareIntents(context.Background(), e, OriginClient, Policy{Enabled: true}, book)
183 if len(got) != 1 {
184 t.Fatalf("deduped intents = %#v", got)
185 }
186 var payload map[string]any
187 if err := json.Unmarshal([]byte(got[0].Payload), &payload); err != nil {
188 t.Fatal(err)
189 }
190 if payload["origin"] != string(OriginClient) {
191 t.Fatalf("payload = %#v", payload)
192 }
193}
194
195func TestPrepareIntentsStopsForMembersOnlyPolicy(t *testing.T) {
196 e := event.Event{ID: "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee", PubKey: "author", Kind: 1}
197 book := routeBook{writes: map[string][]string{"author": {"wss://write.example"}}}
198 if got := PrepareIntents(context.Background(), e, OriginClient, Policy{Enabled: true, ReadMembersOnly: true}, book); len(got) != 0 {
199 t.Fatalf("members-only intents = %#v", got)
200 }
201}
202
203func TestRelayDiscoveryPublicationUsesConfiguredOutbox(t *testing.T) {
204 e := event.Event{ID: "abababababababababababababababababababababababababababababababab", PubKey: "author", Kind: event.KIND_RELAY_DISCOVERY}
205 book := routeBook{writes: map[string][]string{"author": {"wss://operator.example"}}}
206 got := PrepareIntents(context.Background(), e, OriginServer, Policy{Enabled: true, SelfPubKey: "author"}, book)
207 if len(got) != 1 || got[0].Target != "wss://operator.example" {
208 t.Fatalf("peer publication intents = %#v", got)
209 }
210}
211
212func TestServiceRetainsUnresolvedOutboxRoutingAsDurableWork(t *testing.T) {
213 store := openReplicationStore(t)
214 service, err := NewService(Config{Store: store, Policy: Policy{Enabled: true}})
215 if err != nil {
216 t.Fatal(err)
217 }
218 e := event.Event{ID: "ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff", PubKey: "author", Kind: 1}
219 intents := service.Prepare(e, OriginClient)
220 if len(intents) != 1 || intents[0].Kind != "delivery-discovery" {
221 t.Fatalf("intents = %#v", intents)
222 }
223 if _, err := service.Queue().Enqueue(context.Background(), work.Intent{Kind: intents[0].Kind, EventID: intents[0].EventID, Target: intents[0].Target, Payload: intents[0].Payload}); err != nil {
224 t.Fatal(err)
225 }
226}
227
228func TestPrepareDoesNotPerformRemoteDiscoveryInPublishPath(t *testing.T) {
229 store := openReplicationStore(t)
230 discovery := &countingDiscovery{}
231 service, err := NewService(Config{Store: store, Policy: Policy{Enabled: true}, Discovery: discovery})
232 if err != nil {
233 t.Fatal(err)
234 }
235 e := event.Event{ID: "1212121212121212121212121212121212121212121212121212121212121212", PubKey: "author", Kind: 1}
236 intents := service.Prepare(e, OriginClient)
237 if len(intents) != 1 || intents[0].Kind != "delivery-discovery" {
238 t.Fatalf("intents = %#v", intents)
239 }
240 if discovery.calls != 0 {
241 t.Fatalf("discovery calls during prepare = %d", discovery.calls)
242 }
243}
244
245func TestServiceWorkersRunIndependentHandlersConcurrently(t *testing.T) {
246 store := openReplicationStore(t)
247 slowStarted := make(chan struct{})
248 fastDone := make(chan struct{})
249 service, err := NewService(Config{Store: store, Workers: 2})
250 if err != nil {
251 t.Fatal(err)
252 }
253 handlers := service.Handlers()
254 handlers["slow-test"] = func(ctx context.Context, _ work.Intent) error { close(slowStarted); <-ctx.Done(); return ctx.Err() }
255 handlers["fast-test"] = func(context.Context, work.Intent) error { close(fastDone); return nil }
256 if _, err := service.Queue().Enqueue(context.Background(), work.Intent{Kind: "slow-test", EventID: "slow", Target: "one"}); err != nil {
257 t.Fatal(err)
258 }
259 if _, err := service.Queue().Enqueue(context.Background(), work.Intent{Kind: "fast-test", EventID: "fast", Target: "two"}); err != nil {
260 t.Fatal(err)
261 }
262 ctx, cancel := context.WithCancel(context.Background())
263 defer cancel()
264 done := make(chan error, 1)
265 go func() { done <- work.RunPool(ctx, service.Queue(), handlers, work.PoolOptions{Workers: 2}) }()
266 select {
267 case <-slowStarted:
268 case <-time.After(time.Second):
269 t.Fatal("slow handler did not start")
270 }
271 select {
272 case <-fastDone:
273 case <-time.After(time.Second):
274 t.Fatal("fast handler was blocked by slow handler")
275 }
276 cancel()
277 <-done
278}
279
280func TestCallbackHandlerSkipsRevokedRegistration(t *testing.T) {
281 store := openReplicationStore(t)
282 registration := signedEvent(t, 100)
283 registration.Kind = event.KIND_PUSH_REGISTRATION
284 registration.Tags = [][]string{{"d", "main"}, {"relay", "wss://relay.example"}, {"callback", "https://push.example"}, {"filter", `{"kinds":[1]}`}}
285 if err := event.Sign(&registration, "1111111111111111111111111111111111111111111111111111111111111111"); err != nil {
286 t.Fatal(err)
287 }
288 if _, err := store.Save(context.Background(), registration, storage.SaveOptions{Now: 101}); err != nil {
289 t.Fatal(err)
290 }
291 e := signedEvent(t, 102)
292 if _, err := store.Save(context.Background(), e, storage.SaveOptions{Now: 103}); err != nil {
293 t.Fatal(err)
294 }
295 var old PushRegistration
296 old, err := ParsePushRegistration(registration, CallbackPolicy{HostOrigins: []string{"https://push.example"}, OwnerOrigins: []string{"https://push.example"}}, "wss://relay.example")
297 if err != nil {
298 t.Fatal(err)
299 }
300 payload, _ := json.Marshal(struct {
301 Registration PushRegistration `json:"registration"`
302 }{old})
303 transport := &countingRoundTripper{}
304 handler := NewCallbackHandlerWithPolicy(store, CallbackClient{HTTP: &http.Client{Transport: transport}}, func() CallbackPolicy {
305 return CallbackPolicy{HostOrigins: []string{"https://other.example"}, OwnerOrigins: []string{"https://other.example"}}
306 }, func() bool { return true }, nil)
307 err = handler(context.Background(), work.Intent{Kind: "callback", EventID: e.ID, Target: old.ID, Payload: string(payload)})
308 if err != nil {
309 t.Fatal(err)
310 }
311 if transport.calls != 0 {
312 t.Fatalf("revoked callback was sent: %d", transport.calls)
313 }
314}
315
316func TestCallbackHandlerRechecksDeliveredEventVisibility(t *testing.T) {
317 store := openReplicationStore(t)
318 registration := signedEvent(t, 200)
319 registration.Kind = event.KIND_PUSH_REGISTRATION
320 registration.Tags = [][]string{{"d", "main"}, {"relay", "wss://relay.example"}, {"callback", "https://push.example"}, {"filter", `{"kinds":[4]}`}}
321 if err := event.Sign(&registration, "1111111111111111111111111111111111111111111111111111111111111111"); err != nil {
322 t.Fatal(err)
323 }
324 if _, err := store.Save(context.Background(), registration, storage.SaveOptions{Now: 201}); err != nil {
325 t.Fatal(err)
326 }
327 delivered := signedEvent(t, 202)
328 delivered.Kind = 4
329 if err := event.Sign(&delivered, "1111111111111111111111111111111111111111111111111111111111111111"); err != nil {
330 t.Fatal(err)
331 }
332 if _, err := store.Save(context.Background(), delivered, storage.SaveOptions{Now: 203}); err != nil {
333 t.Fatal(err)
334 }
335 parsed, err := ParsePushRegistration(registration, CallbackPolicy{HostOrigins: []string{"https://push.example"}, OwnerOrigins: []string{"https://push.example"}}, "wss://relay.example")
336 if err != nil {
337 t.Fatal(err)
338 }
339 payload, _ := json.Marshal(struct {
340 Registration PushRegistration `json:"registration"`
341 }{parsed})
342 transport := &countingRoundTripper{}
343 handler := NewCallbackHandlerWithEventVisibility(store, CallbackClient{HTTP: &http.Client{Transport: transport}}, func() CallbackPolicy {
344 return CallbackPolicy{HostOrigins: []string{"https://push.example"}, OwnerOrigins: []string{"https://push.example"}}
345 }, func() bool { return true }, nil, func(context.Context, string, event.Event) bool { return false })
346 if err := handler(context.Background(), work.Intent{Kind: "callback", EventID: delivered.ID, Target: parsed.ID, Payload: string(payload)}); err != nil {
347 t.Fatal(err)
348 }
349 if transport.calls != 0 {
350 t.Fatalf("invisible callback was sent: %d", transport.calls)
351 }
352}
353
354func TestSafeArtifactPathRejectsTraversal(t *testing.T) {
355 if _, err := safeArtifactPath("/tmp/tenant", "../other.json"); err == nil {
356 t.Fatal("path traversal accepted")
357 }
358}
359
360func TestManagementMethodsHaveConcreteDispatch(t *testing.T) {
361 store := openReplicationStore(t)
362 dataDir := t.TempDir()
363 service, err := NewService(Config{Store: store, DataDir: dataDir})
364 if err != nil {
365 t.Fatal(err)
366 }
367 if err := os.WriteFile(dataDir+"/old.jsonl", []byte(""), 0600); err != nil {
368 t.Fatal(err)
369 }
370 for _, method := range service.Methods() {
371 var params []json.RawMessage
372 switch method {
373 case "addjob":
374 params = []json.RawMessage{json.RawMessage(`{"id":"m1","kind":"dump"}`)}
375 case "removejob", "runjob", "deletedump", "deletebackup":
376 params = []json.RawMessage{json.RawMessage(`"missing"`)}
377 case "pullfrom":
378 params = []json.RawMessage{json.RawMessage(`"wss://relay.example"`)}
379 case "backfill":
380 params = []json.RawMessage{json.RawMessage(`["wss://relay.example"]`)}
381 case "dumpnow":
382 params = []json.RawMessage{json.RawMessage(`"new.jsonl"`)}
383 case "backupnow":
384 params = []json.RawMessage{json.RawMessage(`"new.backup.json"`)}
385 }
386 _, dispatchErr := service.ExecuteRaw(context.Background(), method, params)
387 if dispatchErr != nil && strings.Contains(dispatchErr.Error(), "unsupported management method") {
388 t.Fatalf("%s has no implementation: %v", method, dispatchErr)
389 }
390 }
391}
392
393func TestBackfillDiscoversOwnerReadRelaysAndFiltersOwner(t *testing.T) {
394 store := openReplicationStore(t)
395 service, err := NewService(Config{
396 Store: store,
397 DataDir: t.TempDir(),
398 Owner: func() string { return strings.Repeat("a", 64) },
399 Directory: testRelayDirectory{reads: []string{"https://history.example"}},
400 })
401 if err != nil {
402 t.Fatal(err)
403 }
404 result, err := service.ExecuteRaw(context.Background(), "backfill", nil)
405 if err != nil {
406 t.Fatal(err)
407 }
408 spec := result.(JobSpec)
409 if len(spec.Relays) != 1 || spec.Relays[0] != "wss://history.example" {
410 t.Fatalf("unexpected discovered relays: %#v", spec.Relays)
411 }
412 var filter event.Filter
413 if err := json.Unmarshal([]byte(spec.Filter), &filter); err != nil {
414 t.Fatal(err)
415 }
416 if len(filter.Authors) != 1 || filter.Authors[0] != strings.Repeat("a", 64) {
417 t.Fatalf("unexpected owner filter: %#v", filter)
418 }
419 if _, err := service.ExecuteRaw(context.Background(), "backfill", []json.RawMessage{json.RawMessage(`{}`)}); err != nil {
420 t.Fatalf("empty guided plan should discover relays: %v", err)
421 }
422}
423
424type testRelayDirectory struct {
425 reads []string
426 all []string
427}
428
429func (d testRelayDirectory) WriteRelays(string) []string { return nil }
430func (d testRelayDirectory) ReadRelays(string) []string { return append([]string(nil), d.reads...) }
431func (d testRelayDirectory) RelayList(string) []string {
432 if d.all != nil {
433 return append([]string(nil), d.all...)
434 }
435 return append([]string(nil), d.reads...)
436}
437
438func TestBackupProviderStateIsIntegritySealedAndRestored(t *testing.T) {
439 store := openReplicationStore(t)
440 provider := &fakeBackupProvider{}
441 archive, err := CreateBackupWithProvider(context.Background(), store, provider, 100)
442 if err != nil {
443 t.Fatal(err)
444 }
445 raw, err := json.Marshal(archive)
446 if err != nil {
447 t.Fatal(err)
448 }
449 decoded, err := ReadBackup(bytes.NewReader(raw))
450 if err != nil {
451 t.Fatal(err)
452 }
453 if len(decoded.State.Blobs) != 1 {
454 t.Fatalf("state = %#v", decoded.State)
455 }
456 if _, err := RestoreBackupWithProvider(context.Background(), store, provider, decoded, 101); err != nil {
457 t.Fatal(err)
458 }
459 if len(provider.restored.Blobs) != 1 {
460 t.Fatal("provider restore was not called")
461 }
462}
463
464func TestRestoreRollbackRestoresProviderWhenEventWriteFails(t *testing.T) {
465 store := openReplicationStore(t)
466 ctx := context.Background()
467 _, err := store.DB().Exec(`CREATE TRIGGER fail_restore_event BEFORE INSERT ON events WHEN NEW.raw LIKE '%restore-fail%' BEGIN SELECT RAISE(ABORT, 'restore failure'); END`)
468 if err != nil {
469 t.Fatal(err)
470 }
471 good := signedEvent(t, 1)
472 failing := signedEvent(t, 1)
473 failing.Content = "restore-fail"
474 if err := event.Sign(&failing, strings.Repeat("0", 63)+"1"); err != nil {
475 t.Fatal(err)
476 }
477 provider := &fakeBackupProvider{}
478 archive := BackupArchive{Format: BackupFormat, Created: 1, Events: []event.Event{good, failing}, State: BackupState{Config: json.RawMessage(`{"owner":"new"}`)}}
479 archive, err = SealBackup(archive)
480 if err != nil {
481 t.Fatal(err)
482 }
483 if _, err := RestoreBackupWithProvider(ctx, store, provider, archive, 2); err == nil {
484 t.Fatal("restore unexpectedly succeeded")
485 }
486 if string(provider.restored.Config) != `{"owner":"owner"}` {
487 t.Fatalf("provider state was not rolled back: %s", provider.restored.Config)
488 }
489 rows, err := store.Query(ctx, event.Filter{IDs: []string{good.ID}, Tags: map[string][]string{}}, storage.QueryOptions{Now: 2, Access: storage.Access{All: true}, Limit: 1})
490 if err != nil {
491 t.Fatal(err)
492 }
493 if len(rows.Events) != 0 {
494 t.Fatal("partially inserted event survived rollback")
495 }
496}
497
498func TestReadCompatibleBackupNormalizesBindWSArchive(t *testing.T) {
499 e := signedEvent(t, 1)
500 rawEvent, err := event.Canonical(e)
501 if err != nil {
502 t.Fatal(err)
503 }
504 encodedEvent, _ := json.Marshal(string(rawEvent))
505 legacy := fmt.Sprintf(`{"format":"bind.ws/relay-backup/1","manifest":{"createdAt":1,"archiveSha256":""},"config":{"format":"bind.ws/relay-config/2"},"events":[%s],"blobs":[],"git":[]}`, string(encodedEvent))
506 legacy = legacyWithChecksum(legacy)
507 archive, err := ReadCompatibleBackup(strings.NewReader(legacy))
508 if err != nil {
509 t.Fatal(err)
510 }
511 if archive.Format != BackupFormat || len(archive.Events) != 1 || string(archive.State.Config) != `{"format":"bind.ws/relay-config/2"}` {
512 t.Fatalf("archive = %#v", archive)
513 }
514}
515
516func TestReadCompatibleBackupRequiresLegacyChecksum(t *testing.T) {
517 raw := `{"format":"bind.ws/relay-backup/1","manifest":{"createdAt":1,"archiveSha256":""},"config":{"format":"bind.ws/relay-config/2"},"events":[],"blobs":[],"git":[]}`
518 if _, err := ReadCompatibleBackup(strings.NewReader(raw)); err == nil || !strings.Contains(err.Error(), "checksum") {
519 t.Fatalf("missing checksum accepted: %v", err)
520 }
521 if _, err := ReadBackup(strings.NewReader(`{"format":"tinyrelay/backup/1"} {}`)); err == nil || !strings.Contains(err.Error(), "trailing") {
522 t.Fatalf("trailing native JSON accepted: %v", err)
523 }
524}
525
526func TestReadCompatibleBackupRetainsLegacyIdentityAndSQLGit(t *testing.T) {
527 e := signedEvent(t, 1)
528 rawEvent, err := event.Canonical(e)
529 if err != nil {
530 t.Fatal(err)
531 }
532 encodedEvent, _ := json.Marshal(string(rawEvent))
533 legacy := fmt.Sprintf(`{"format":"bind.ws/relay-backup/1","manifest":{"createdAt":1,"owner":"%s","relayIdentity":"relay-key","slug":"demo","archiveSha256":""},"config":{"format":"bind.ws/relay-config/2"},"events":[%s],"blobs":[],"git":[]}`, e.PubKey, string(encodedEvent))
534 legacy = legacyWithChecksum(legacy)
535 archive, err := ReadCompatibleBackup(strings.NewReader(legacy))
536 if err != nil {
537 t.Fatal(err)
538 }
539 var identity map[string]string
540 if err := json.Unmarshal(archive.State.Identity, &identity); err != nil || identity["owner"] != e.PubKey {
541 t.Fatalf("identity = %#v, err=%v", identity, err)
542 }
543 withSQLGit := strings.TrimSuffix(legacy, `}`) + `,"sqlGit":{"format":"bind.ws/git-sqlite/1","repositories":[]}}`
544 withSQLGit = legacyWithChecksum(withSQLGit)
545 converted, err := ReadCompatibleBackup(strings.NewReader(withSQLGit))
546 if err != nil {
547 t.Fatal(err)
548 }
549 if string(converted.State.GitSQL) == "" {
550 t.Fatal("SQL Git payload was dropped")
551 }
552}
553
554func legacyWithChecksum(raw string) string {
555 key := `"archiveSha256":"`
556 at := strings.Index(raw, key)
557 if at < 0 {
558 return raw
559 }
560 start := at + len(key)
561 end := strings.IndexByte(raw[start:], '"')
562 if end < 0 {
563 return raw
564 }
565 end += start
566 unsigned := raw[:start] + raw[end:]
567 return raw[:start] + digest([]byte(unsigned)) + raw[end:]
568}
569
570func TestPrepareIntentsDoesNotTreatImportedEventsAsLocalOutbox(t *testing.T) {
571 e := event.Event{ID: "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc", PubKey: "author", Kind: 1}
572 book := routeBook{writes: map[string][]string{"author": {"wss://write.example"}}}
573 got := PrepareIntents(context.Background(), e, OriginImport, Policy{Enabled: true}, book)
574 if len(got) != 0 {
575 t.Fatalf("import intents = %#v", got)
576 }
577 _ = storage.Intent{}
578}
579
580func TestDeliveryHandlerUsesStoredEventAndAcknowledgesDuplicate(t *testing.T) {
581 store := openReplicationStore(t)
582 e := signedEvent(t, 1)
583 if _, err := store.Save(context.Background(), e, storage.SaveOptions{Now: e.CreatedAt}); err != nil {
584 t.Fatal(err)
585 }
586 transport := &fakeTransport{result: DeliveryResult{Accepted: false, Message: "duplicate: already have this event"}}
587 handler := NewDeliveryHandler(store, transport)
588 intent := storage.Intent{Kind: "delivery", EventID: e.ID, Target: "wss://relay.example", Payload: `{ "origin": "client" }`}
589 if err := handler(context.Background(), intent); err != nil {
590 t.Fatal(err)
591 }
592 if transport.event.ID != e.ID {
593 t.Fatalf("sent event = %#v", transport.event)
594 }
595}
596
597func TestDeliveryHandlerSendsAuthorRelayListBeforePublicEvent(t *testing.T) {
598 store := openReplicationStore(t)
599 list := signedEvent(t, 1)
600 list.Kind = 10002
601 if err := event.Sign(&list, strings.Repeat("1", 64)); err != nil {
602 t.Fatal(err)
603 }
604 main := signedEvent(t, 1)
605 if _, err := store.Save(context.Background(), list, storage.SaveOptions{Now: list.CreatedAt}); err != nil {
606 t.Fatal(err)
607 }
608 if _, err := store.Save(context.Background(), main, storage.SaveOptions{Now: main.CreatedAt + 1}); err != nil {
609 t.Fatal(err)
610 }
611 transport := &recordingTransport{}
612 handler := NewDeliveryHandler(store, transport)
613 if err := handler(context.Background(), storage.Intent{Kind: "delivery", EventID: main.ID, Target: "ws://localhost:7447"}); err != nil {
614 t.Fatal(err)
615 }
616 if len(transport.events) != 2 || transport.events[0].ID != list.ID || transport.events[1].ID != main.ID {
617 t.Fatalf("delivery order = %#v", transport.events)
618 }
619}
620
621func TestSafeRelayURLDefersPrivateAddressPolicyToDialer(t *testing.T) {
622 if !safeRelayURL("ws://localhost:7447") {
623 t.Fatal("localhost relay URL rejected before operator dial policy")
624 }
625 if canonicalRelayURL("ws://LOCALHOST:7447/r/main/") != canonicalRelayURL("http://localhost:7447/r/main") {
626 t.Fatal("self relay URL aliases did not normalize")
627 }
628}
629
630func TestPushRegistrationRequiresBothOriginApprovals(t *testing.T) {
631 e := event.Event{ID: "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd", PubKey: "author", Kind: event.KIND_PUSH_REGISTRATION, Tags: [][]string{
632 {"d", "main"}, {"relay", "wss://relay.example"}, {"callback", "https://push.example"}, {"filter", `{"kinds":[1]}`}, {"include_event"},
633 }}
634 reg, err := ParsePushRegistration(e, CallbackPolicy{HostOrigins: []string{"https://push.example"}, OwnerOrigins: []string{"https://push.example"}}, "wss://relay.example")
635 if err != nil {
636 t.Fatal(err)
637 }
638 if !PushMatches(reg, event.Event{Kind: 1, Content: "hello"}) || !reg.IncludeEvent {
639 t.Fatalf("registration = %#v", reg)
640 }
641 if _, err := ParsePushRegistration(e, CallbackPolicy{HostOrigins: []string{"https://push.example"}}, "wss://relay.example"); err == nil {
642 t.Fatal("unapproved owner origin accepted")
643 }
644}
645
646func TestImportAndDumpRoundTrip(t *testing.T) {
647 store := openReplicationStore(t)
648 e := signedEvent(t, 100)
649 raw, err := event.Canonical(e)
650 if err != nil {
651 t.Fatal(err)
652 }
653 if err := ImportJSONL(context.Background(), store, bytes.NewReader(append(raw, '\n')), 101); err != nil {
654 t.Fatal(err)
655 }
656 var dump bytes.Buffer
657 if err := WriteDump(context.Background(), store, &dump, 101); err != nil {
658 t.Fatal(err)
659 }
660 if !bytes.Contains(dump.Bytes(), []byte(e.ID)) {
661 t.Fatalf("dump does not contain imported event: %s", dump.String())
662 }
663}
664
665func TestPushCursorIsDurablePerTarget(t *testing.T) {
666 store := openReplicationStore(t)
667 e := signedEvent(t, 100)
668 if _, err := store.Save(context.Background(), e, storage.SaveOptions{Now: 101}); err != nil {
669 t.Fatal(err)
670 }
671 transport := &fakePush{}
672 service, err := NewService(Config{Store: store, Push: transport})
673 if err != nil {
674 t.Fatal(err)
675 }
676 spec := JobSpec{ID: "job-1", Kind: JobPush, Relays: []string{"wss://a.example", "wss://b.example"}}
677 if err := service.push(context.Background(), spec); err != nil {
678 t.Fatal(err)
679 }
680 if len(transport.sent["wss://a.example"]) != 1 || len(transport.sent["wss://b.example"]) != 1 {
681 t.Fatalf("sent = %#v", transport.sent)
682 }
683 var first, second int64
684 if err := store.DB().QueryRow(`SELECT cursor FROM replication_job_cursors WHERE job_id=? AND target=?`, spec.ID, spec.Relays[0]).Scan(&first); err != nil {
685 t.Fatal(err)
686 }
687 if err := store.DB().QueryRow(`SELECT cursor FROM replication_job_cursors WHERE job_id=? AND target=?`, spec.ID, spec.Relays[1]).Scan(&second); err != nil {
688 t.Fatal(err)
689 }
690 if first == 0 || second == 0 {
691 t.Fatalf("cursors = %d,%d", first, second)
692 }
693}

About

ownerc01503e8035f
visibilitypublic
headd730679639bf

Maintainers

  • c01503e8035f owner
  • ba1b5beed5b5 agent hermes

Clone

  • https://012.run/npub1cq2s86qrtadf36eqq3epm38wmhdynrrcwxf84eldk3f94dkv7ttsrulrz5/tinyrelay.git

Refs