| 1 | package replication |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 13 | "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 14 | ) |
| 15 | |
| 16 | func TestJobLifecyclePublishesArtifactAndSurvivesRestart(t *testing.T) { |
| 17 | ctx := context.Background() |
| 18 | dir := t.TempDir() |
| 19 | dbPath := filepath.Join(dir, "relay.db" ) |
| 20 | store, err := storage.Open(ctx, dbPath) |
| 21 | if err != nil { |
| 22 | t.Fatal(err) |
| 23 | } |
| 24 | e := signedEvent(t, 100) |
| 25 | if _, err := store.Save(ctx, e, storage.SaveOptions{Now: 101}); err != nil { |
| 26 | t.Fatal(err) |
| 27 | } |
| 28 | service, err := NewService(Config{Store: store, DataDir: dir, Workers: 1}) |
| 29 | if err != nil { |
| 30 | t.Fatal(err) |
| 31 | } |
| 32 | if err := service.AddJob(ctx, JobSpec{ID: "dump-lifecycle" , Kind: JobDump, Relays: []string{"events.jsonl" }, Every: 1}); err != nil { |
| 33 | t.Fatal(err) |
| 34 | } |
| 35 | runCtx, stop := context.WithCancel(ctx) |
| 36 | done := make(chan error, 1) |
| 37 | go func() { done <- service.Run(runCtx) }() |
| 38 | status := waitJobStatus(t, service, "dump-lifecycle" , func(status JobStatus) bool { return status.Phase == "complete" }) |
| 39 | deadline := time.Now().Add(time.Second) |
| 40 | for { |
| 41 | pending, listErr := service.Queue().List(ctx, "pending" ) |
| 42 | if listErr != nil { |
| 43 | t.Fatal(listErr) |
| 44 | } |
| 45 | if len(pending) == 1 && pending[0].EventID == "dump-lifecycle" { |
| 46 | break |
| 47 | } |
| 48 | if time.Now().After(deadline) { |
| 49 | t.Fatalf("recurring run was not enqueued: %#v" , pending) |
| 50 | } |
| 51 | time.Sleep(10 * time.Millisecond) |
| 52 | } |
| 53 | stop() |
| 54 | if err := <-done; !errors.Is(err, context.Canceled) { |
| 55 | t.Fatalf("worker exit = %v" , err) |
| 56 | } |
| 57 | if status.Started == 0 || status.Finished == 0 { |
| 58 | t.Fatalf("missing lifecycle timestamps: %#v" , status) |
| 59 | } |
| 60 | raw, err := os.ReadFile(filepath.Join(dir, "events.jsonl" )) |
| 61 | if err != nil { |
| 62 | t.Fatal(err) |
| 63 | } |
| 64 | var decoded map[string]any |
| 65 | if err := json.Unmarshal(raw, &decoded); err != nil { |
| 66 | t.Fatalf("artifact is not JSON: %v" , err) |
| 67 | } |
| 68 | if decoded["id" ] != e.ID { |
| 69 | t.Fatalf("artifact event id = %v, want %s" , decoded["id" ], e.ID) |
| 70 | } |
| 71 | if err := store.Close(); err != nil { |
| 72 | t.Fatal(err) |
| 73 | } |
| 74 | |
| 75 | reopened, err := storage.Open(ctx, dbPath) |
| 76 | if err != nil { |
| 77 | t.Fatal(err) |
| 78 | } |
| 79 | defer reopened.Close() |
| 80 | resumed, err := NewService(Config{Store: reopened, DataDir: dir}) |
| 81 | if err != nil { |
| 82 | t.Fatal(err) |
| 83 | } |
| 84 | statuses, err := resumed.ListJobStatus(ctx) |
| 85 | if err != nil { |
| 86 | t.Fatal(err) |
| 87 | } |
| 88 | if len(statuses) != 1 || statuses[0].Phase != "complete" { |
| 89 | t.Fatalf("reopened status = %#v" , statuses) |
| 90 | } |
| 91 | } |
| 92 | |
| 93 | func TestFailedJobRecordsFailureAndRemainsRetryable(t *testing.T) { |
| 94 | ctx := context.Background() |
| 95 | store := openReplicationStore(t) |
| 96 | service, err := NewService(Config{Store: store, DataDir: t.TempDir(), Workers: 1}) |
| 97 | if err != nil { |
| 98 | t.Fatal(err) |
| 99 | } |
| 100 | spec := JobSpec{ID: "missing-pull" , Kind: JobPull, Relays: []string{"wss://relay.example" }} |
| 101 | if err := service.AddJob(ctx, spec); err != nil { |
| 102 | t.Fatal(err) |
| 103 | } |
| 104 | runCtx, stop := context.WithCancel(ctx) |
| 105 | done := make(chan error, 1) |
| 106 | go func() { done <- service.Run(runCtx) }() |
| 107 | status := waitJobStatus(t, service, spec.ID, func(status JobStatus) bool { return status.Phase == "failed" }) |
| 108 | if status.Error == "" { |
| 109 | t.Fatal("failed job did not record an error" ) |
| 110 | } |
| 111 | pending, err := service.Queue().List(ctx, "pending,running" ) |
| 112 | if err != nil { |
| 113 | t.Fatal(err) |
| 114 | } |
| 115 | if len(pending) != 1 || pending[0].Attempts < 1 { |
| 116 | t.Fatalf("retryable intent = %#v" , pending) |
| 117 | } |
| 118 | stop() |
| 119 | if err := <-done; !errors.Is(err, context.Canceled) { |
| 120 | t.Fatalf("worker exit = %v" , err) |
| 121 | } |
| 122 | } |
| 123 | |
| 124 | type integrationPull struct{ events []event.Event } |
| 125 | |
| 126 | type ownerReadDiscovery struct{ relays []string } |
| 127 | |
| 128 | func (d ownerReadDiscovery) DiscoverRelays(context.Context, string) ([]string, error) { |
| 129 | return nil, nil |
| 130 | } |
| 131 | func (d ownerReadDiscovery) DiscoverReadRelays(context.Context, string) ([]string, error) { |
| 132 | return d.relays, nil |
| 133 | } |
| 134 | |
| 135 | func TestInboxPullDiscoversOwnerReadRelaysWithoutStaticTargets(t *testing.T) { |
| 136 | ctx := context.Background() |
| 137 | store := openReplicationStore(t) |
| 138 | defer store.Close() |
| 139 | item := signedEvent(t, 400) |
| 140 | service, err := NewService(Config{ |
| 141 | Store: store, DataDir: t.TempDir(), Discovery: ownerReadDiscovery{relays: []string{"ws://owner-read.example" }}, |
| 142 | Pull: integrationPull{events: []event.Event{item}}, |
| 143 | Ingest: func(context.Context, event.Event, Origin) error { return nil }, |
| 144 | }) |
| 145 | if err != nil { |
| 146 | t.Fatal(err) |
| 147 | } |
| 148 | spec := JobSpec{ID: "inbox-discovery" , Kind: JobPull, DiscoverPubKey: item.PubKey, Filter: `{"kinds":[1],"#p":["` + item.PubKey + `"]}` , Every: 1} |
| 149 | if err := ValidateJob(spec); err != nil { |
| 150 | t.Fatal(err) |
| 151 | } |
| 152 | if err := service.pull(ctx, spec); err != nil { |
| 153 | t.Fatal(err) |
| 154 | } |
| 155 | } |
| 156 | |
| 157 | func (p integrationPull) Query(context.Context, string, event.Filter) ([]event.Event, error) { |
| 158 | return p.events, nil |
| 159 | } |
| 160 | |
| 161 | func TestPullAndPushJobsPersistSyncStats(t *testing.T) { |
| 162 | ctx := context.Background() |
| 163 | store := openReplicationStore(t) |
| 164 | pullEvent := signedEvent(t, 200) |
| 165 | pull, err := NewService(Config{Store: store, DataDir: t.TempDir(), Pull: integrationPull{events: []event.Event{pullEvent}}, Workers: 1}) |
| 166 | if err != nil { |
| 167 | t.Fatal(err) |
| 168 | } |
| 169 | pullSpec := JobSpec{ID: "pull-stats" , Kind: JobPull, Relays: []string{"wss://source.example" }} |
| 170 | if err := pull.AddJob(ctx, pullSpec); err != nil { |
| 171 | t.Fatal(err) |
| 172 | } |
| 173 | pullCtx, stopPull := context.WithCancel(ctx) |
| 174 | pullDone := make(chan error, 1) |
| 175 | go func() { pullDone <- pull.Run(pullCtx) }() |
| 176 | pullStatus := waitJobStatus(t, pull, pullSpec.ID, func(status JobStatus) bool { return status.Phase == "complete" }) |
| 177 | if pullStatus.Stored != 1 || pullStatus.Rounds != 1 { |
| 178 | t.Fatalf("pull status = %#v" , pullStatus) |
| 179 | } |
| 180 | stopPull() |
| 181 | if err := <-pullDone; !errors.Is(err, context.Canceled) { |
| 182 | t.Fatalf("pull worker exit = %v" , err) |
| 183 | } |
| 184 | |
| 185 | pushEvent := signedEvent(t, 201) |
| 186 | if _, err := store.Save(ctx, pushEvent, storage.SaveOptions{Now: 202}); err != nil { |
| 187 | t.Fatal(err) |
| 188 | } |
| 189 | transport := &fakePush{} |
| 190 | push, err := NewService(Config{Store: store, DataDir: t.TempDir(), Push: transport, Workers: 1}) |
| 191 | if err != nil { |
| 192 | t.Fatal(err) |
| 193 | } |
| 194 | pushSpec := JobSpec{ID: "push-stats" , Kind: JobPush, Relays: []string{"wss://destination.example" }} |
| 195 | if err := push.AddJob(ctx, pushSpec); err != nil { |
| 196 | t.Fatal(err) |
| 197 | } |
| 198 | pushCtx, stopPush := context.WithCancel(ctx) |
| 199 | pushDone := make(chan error, 1) |
| 200 | go func() { pushDone <- push.Run(pushCtx) }() |
| 201 | pushStatus := waitJobStatus(t, push, pushSpec.ID, func(status JobStatus) bool { return status.Phase == "complete" }) |
| 202 | if pushStatus.Sent < 1 || pushStatus.Cursor == 0 || pushStatus.Rounds != 1 { |
| 203 | t.Fatalf("push status = %#v" , pushStatus) |
| 204 | } |
| 205 | stopPush() |
| 206 | if err := <-pushDone; !errors.Is(err, context.Canceled) { |
| 207 | t.Fatalf("push worker exit = %v" , err) |
| 208 | } |
| 209 | } |
| 210 | |
| 211 | func TestValidateJobRejectsInvalidFilter(t *testing.T) { |
| 212 | err := ValidateJob(JobSpec{ID: "bad-filter" , Kind: JobPull, Relays: []string{"wss://relay.example" }, Filter: "[]" }) |
| 213 | if err == nil { |
| 214 | t.Fatal("invalid filter accepted" ) |
| 215 | } |
| 216 | } |
| 217 | |
| 218 | func waitJobStatus(t *testing.T, service *Service, id string, ready func(JobStatus) bool) JobStatus { |
| 219 | t.Helper() |
| 220 | deadline := time.Now().Add(3 * time.Second) |
| 221 | for time.Now().Before(deadline) { |
| 222 | statuses, err := service.ListJobStatus(context.Background()) |
| 223 | if err != nil { |
| 224 | t.Fatal(err) |
| 225 | } |
| 226 | for _, status := range statuses { |
| 227 | if status.Spec.ID == id && ready(status) { |
| 228 | return status |
| 229 | } |
| 230 | } |
| 231 | time.Sleep(10 * time.Millisecond) |
| 232 | } |
| 233 | t.Fatalf("job %q did not reach expected state" , id) |
| 234 | return JobStatus{} |
| 235 | } |