1package daemon23import (4 "context"5 "encoding/json"6 "net/http"7 "net/http/httptest"8 "strings"9 "sync"10 "testing"11 "time"1213 "github.com/FelineStateMachine/tinyrelay/internal/event"14 "github.com/FelineStateMachine/tinyrelay/internal/replication"15 "github.com/FelineStateMachine/tinyrelay/internal/storage"16 "github.com/coder/websocket"17)1819type acceptanceRelay struct {20 mu sync.Mutex21 seen []event.Event22 failNext bool23}2425func (r *acceptanceRelay) ServeHTTP(w http.ResponseWriter, req *http.Request) {26 conn, err := websocket.Accept(w, req, nil)27 if err != nil {28 return29 }30 defer conn.Close(websocket.StatusNormalClosure, "done")31 _, data, err := conn.Read(req.Context())32 if err != nil {33 return34 }35 var message []json.RawMessage36 if json.Unmarshal(data, &message) != nil || len(message) < 2 {37 return38 }39 var kind string40 _ = json.Unmarshal(message[0], &kind)41 if kind != "EVENT" {42 return43 }44 var item event.Event45 if len(message) < 2 || json.Unmarshal(message[1], &item) != nil {46 return47 }48 r.mu.Lock()49 r.seen = append(r.seen, item)50 reject := r.failNext51 r.failNext = false52 r.mu.Unlock()53 accepted := !reject54 reason := ""55 if reject {56 reason = "temporary upstream failure"57 }58 response, _ := json.Marshal([]any{"OK", item.ID, accepted, reason})59 _ = conn.Write(req.Context(), websocket.MessageText, response)60}6162func TestDaemonDeliveryPersistsAcrossRestartAndExcludesImports(t *testing.T) {63 ctx := context.Background()64 root := t.TempDir()65 remote := &acceptanceRelay{failNext: true}66 remoteServer := httptest.NewServer(remote)67 t.Cleanup(remoteServer.Close)68 remoteTarget := "ws" + strings.TrimPrefix(remoteServer.URL, "http")6970 app, err := New(ctx, Config{DataDir: root, DefaultTenant: "main", AllowPrivateRelays: true})71 if err != nil {72 t.Fatal(err)73 }74 ownerSecret := strings.Repeat("0", 63) + "1"75 owner, err := event.PublicKey(ownerSecret)76 if err != nil {77 t.Fatal(err)78 }79 meta, err := app.Create(ctx, CreateOptions{Name: "main", Owner: owner, Template: "outbox"})80 if err != nil {81 t.Fatal(err)82 }83 sourceServer := httptest.NewServer(app)84 t.Cleanup(sourceServer.Close)85 tenant, err := app.tenant(ctx, meta, sourceServer.URL+"/r/main")86 if err != nil {87 t.Fatal(err)88 }89 if tenant == nil {90 t.Fatal("created tenant was not loaded")91 }92 // The real daemon websocket endpoint is used for source admission; its93 // durable worker then delivers through a second real websocket server.94 sourceTarget := "ws" + strings.TrimPrefix(sourceServer.URL, "http") + "/r/main"9596 announcement := event.Event{Kind: 10002, CreatedAt: time.Now().Unix(), Tags: [][]string{{"r", remoteTarget}}, Content: ""}97 if err := event.Sign(&announcement, ownerSecret); err != nil {98 t.Fatal(err)99 }100 transport := &replication.NostrTransport{Dialer: replication.WebsocketDialer{AllowPrivate: true}}101 if _, err := tenant.store.Save(ctx, announcement, storage.SaveOptions{Now: announcement.CreatedAt, SearchMode: tenant.Policy().Features.Search}); err != nil {102 t.Fatalf("store relay directory announcement: %v", err)103 }104105 eventToDeliver := event.Event{Kind: 1, CreatedAt: time.Now().Unix(), Tags: [][]string{}, Content: "durable outbox"}106 if err := event.Sign(&eventToDeliver, ownerSecret); err != nil {107 t.Fatal(err)108 }109 if result, err := transport.Send(ctx, sourceTarget, eventToDeliver); err != nil || !result.Accepted {110 t.Fatalf("source relay rejected publish: %+v %v", result, err)111 }112 if !waitFor(t, 5*time.Second, func() bool {113 pending, pendingErr := tenant.replication.Queue().Pending(ctx, "delivery")114 return pendingErr == nil && len(pending) > 0 && pending[0].State == "pending" && pending[0].Attempts > 0 && remoteCount(remote) >= 1115 }) {116 t.Fatalf("failed delivery was not durable; remote attempts=%d", remoteCount(remote))117 }118 items, err := tenant.replication.Queue().Pending(ctx, "delivery")119 if err != nil {120 t.Fatal(err)121 }122 if len(items) != 1 {123 t.Fatalf("failed delivery was not durable: %d pending intents", len(items))124 }125 if err := app.Close(ctx); err != nil {126 t.Fatal(err)127 }128129 app, err = New(ctx, Config{DataDir: root, DefaultTenant: "main", AllowPrivateRelays: true})130 if err != nil {131 t.Fatal(err)132 }133 t.Cleanup(func() { _ = app.Close(ctx) })134 if err := app.Start(ctx, sourceServer.URL); err != nil {135 t.Fatal(err)136 }137 reloaded, err := app.tenant(ctx, meta, sourceServer.URL+"/r/main")138 if err != nil {139 t.Fatal(err)140 }141 if reloaded == nil {142 t.Fatal("tenant did not reopen")143 }144 if !waitFor(t, 8*time.Second, func() bool {145 pending, pendingErr := reloaded.replication.Queue().Pending(ctx, "delivery")146 return pendingErr == nil && len(pending) == 0 && remoteCount(remote) >= 3147 }) {148 t.Fatalf("restart retry did not complete: remote=%d", remoteCount(remote))149 }150 if got := remoteCount(remote); got != 3 {151 t.Fatalf("restart delivery count = %d, want failed list plus accepted list and event", got)152 }153154 for _, kind := range []int{1, event.KIND_WRAP} {155 imported := event.Event{Kind: kind, CreatedAt: time.Now().Unix(), Tags: [][]string{}, Content: "imported"}156 if kind == event.KIND_WRAP {157 imported.Tags = [][]string{{"p", owner}}158 }159 if err := event.Sign(&imported, ownerSecret); err != nil {160 t.Fatal(err)161 }162 if err := reloaded.ingest(ctx, imported, replication.OriginImport); err != nil {163 t.Fatalf("import kind %d: %v", kind, err)164 }165 }166 items, err = reloaded.replication.Queue().Pending(ctx, "delivery")167 if err != nil {168 t.Fatal(err)169 }170 if len(items) != 0 || remoteCount(remote) != 3 {171 t.Fatalf("imported events echoed to outbox: pending=%d remote=%d", len(items), remoteCount(remote))172 }173}174175func remoteCount(remote *acceptanceRelay) int {176 remote.mu.Lock()177 defer remote.mu.Unlock()178 return len(remote.seen)179}180181func waitFor(t *testing.T, timeout time.Duration, condition func() bool) bool {182 t.Helper()183 deadline := time.Now().Add(timeout)184 for time.Now().Before(deadline) {185 if condition() {186 return true187 }188 time.Sleep(25 * time.Millisecond)189 }190 return false191}