1package replication23import (4 "context"5 "encoding/json"6 "errors"7 "os"8 "path/filepath"9 "testing"10 "time"1112 "github.com/FelineStateMachine/tinyrelay/internal/event"13 "github.com/FelineStateMachine/tinyrelay/internal/storage"14)1516func 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 break47 }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]any65 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 }7475 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}9293func 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}123124type integrationPull struct{ events []event.Event }125126type ownerReadDiscovery struct{ relays []string }127128func (d ownerReadDiscovery) DiscoverRelays(context.Context, string) ([]string, error) {129 return nil, nil130}131func (d ownerReadDiscovery) DiscoverReadRelays(context.Context, string) ([]string, error) {132 return d.relays, nil133}134135func 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}156157func (p integrationPull) Query(context.Context, string, event.Filter) ([]event.Event, error) {158 return p.events, nil159}160161func 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 }184185 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}210211func 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}217218func 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 status229 }230 }231 time.Sleep(10 * time.Millisecond)232 }233 t.Fatalf("job %q did not reach expected state", id)234 return JobStatus{}235}