1package replication23import (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"1617 "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)2223type routeBook struct {24 writes map[string][]string25 reads map[string][]string26}2728type countingDiscovery struct{ calls int }2930func (d *countingDiscovery) DiscoverRelays(context.Context, string) ([]string, error) {31 d.calls++32 return []string{"wss://remote.example"}, nil33}3435type fakeTransport struct {36 result DeliveryResult37 event event.Event38}3940type recordingTransport struct {41 events []event.Event42}4344func (r *recordingTransport) Send(_ context.Context, _ string, e event.Event) (DeliveryResult, error) {45 r.events = append(r.events, e)46 return DeliveryResult{Accepted: true}, nil47}4849type fakePush struct{ sent map[string][]string }5051type countingRoundTripper struct{ calls int }5253type fixedResolver []net.IPAddr5455func (r fixedResolver) LookupIPAddr(context.Context, string) ([]net.IPAddr, error) {56 return []net.IPAddr(r), nil57}5859func 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}6566func 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 return71 }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.Host81 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}8788func 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}9596func 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}105106func (r *countingRoundTripper) RoundTrip(*http.Request) (*http.Response, error) {107 r.calls++108 return &http.Response{StatusCode: 200, Body: http.NoBody, Header: make(http.Header)}, nil109}110111type fakeBackupProvider struct{ restored BackupState }112113func (f *fakeBackupProvider) Snapshot(context.Context) (BackupState, error) {114 return BackupState{Config: json.RawMessage(`{"owner":"owner"` + `}`), Blobs: []BackupObject{{Name: "blob", SHA256: "hash", Content: []byte("data")}}}, nil115}116func (f *fakeBackupProvider) Restore(_ context.Context, state BackupState) error {117 f.restored = state118 return nil119}120121func (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}, nil127}128129func (f *fakeTransport) Send(_ context.Context, _ string, e event.Event) (DeliveryResult, error) {130 f.event = e131 return f.result, nil132}133134func 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 store142}143144func 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 e152}153154func (r routeBook) WriteRelays(string) []string { return r.writes["author"] }155func (r routeBook) ReadRelays(string) []string { return r.reads["recipient"] }156157func 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}178179func 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]any187 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}194195func 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}202203func 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}211212func 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}227228func 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}244245func 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 <-done278}279280func TestCallbackHandlerSkipsRevokedRegistration(t *testing.T) {281 store := openReplicationStore(t)282 registration := signedEvent(t, 100)283 registration.Kind = event.KIND_PUSH_REGISTRATION284 registration.Tags = [][]string{{"d", "main"}, {"relay", "wss://relay.example"}, {"callback", "https://push.example"}, {"filter", `{"kinds":[1]}`}}285 if err := event.Sign(®istration, "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 PushRegistration296 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}315316func TestCallbackHandlerRechecksDeliveredEventVisibility(t *testing.T) {317 store := openReplicationStore(t)318 registration := signedEvent(t, 200)319 registration.Kind = event.KIND_PUSH_REGISTRATION320 registration.Tags = [][]string{{"d", "main"}, {"relay", "wss://relay.example"}, {"callback", "https://push.example"}, {"filter", `{"kinds":[4]}`}}321 if err := event.Sign(®istration, "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 = 4329 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}353354func TestSafeArtifactPathRejectsTraversal(t *testing.T) {355 if _, err := safeArtifactPath("/tmp/tenant", "../other.json"); err == nil {356 t.Fatal("path traversal accepted")357 }358}359360func 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.RawMessage372 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}392393func 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.Filter413 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}423424type testRelayDirectory struct {425 reads []string426 all []string427}428429func (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}437438func 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}463464func 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}497498func 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}515516func 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}525526func 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]string540 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}553554func legacyWithChecksum(raw string) string {555 key := `"archiveSha256":"`556 at := strings.Index(raw, key)557 if at < 0 {558 return raw559 }560 start := at + len(key)561 end := strings.IndexByte(raw[start:], '"')562 if end < 0 {563 return raw564 }565 end += start566 unsigned := raw[:start] + raw[end:]567 return raw[:start] + digest([]byte(unsigned)) + raw[end:]568}569570func 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}579580func 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}596597func TestDeliveryHandlerSendsAuthorRelayListBeforePublicEvent(t *testing.T) {598 store := openReplicationStore(t)599 list := signedEvent(t, 1)600 list.Kind = 10002601 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}620621func 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}629630func 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}645646func 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.Buffer657 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}664665func 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 int64684 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}