1package daemon23import (4 "context"5 "encoding/json"6 "net/http"7 "net/http/httptest"8 "os"9 "path/filepath"10 "strings"11 "sync"12 "testing"13 "time"1415 "github.com/FelineStateMachine/tinyrelay/internal/event"16 "github.com/FelineStateMachine/tinyrelay/internal/replication"17 "github.com/FelineStateMachine/tinyrelay/internal/storage"18 "github.com/coder/websocket"19)2021// TestDaemonRestartResumesUnconsumedJobAndViewCheckpoint exercises the daemon22// boundary: a tenant is stopped with durable work still pending, then opened23// by a new App. The job worker and view scheduler must consume the old rows.24func TestDaemonRestartResumesUnconsumedJobAndViewCheckpoint(t *testing.T) {25 ctx := context.Background()26 dir := t.TempDir()27 ownerSecret := strings.Repeat("0", 63) + "1"28 owner, err := event.PublicKey(ownerSecret)29 if err != nil {30 t.Fatal(err)31 }32 app, err := New(ctx, Config{DataDir: dir, DefaultTenant: "main"})33 if err != nil {34 t.Fatal(err)35 }36 meta, err := app.Create(ctx, CreateOptions{Name: "main", Owner: owner, Template: "default"})37 if err != nil {38 t.Fatal(err)39 }40 if err := app.Start(ctx, "http://relay.test"); err != nil {41 t.Fatal(err)42 }43 tenant := app.tenants[meta.ID]44 if tenant == nil {45 t.Fatal("tenant was not started")46 }47 // Stop only the tenant workers. This lets us enqueue work at a known48 // unconsumed point while still using the production queue and service.49 tenant.workCancel()50 tenant.workWG.Wait()51 e := event.Event{Kind: 1, CreatedAt: time.Now().Unix(), Content: "restart durable"}52 if err := event.Sign(&e, ownerSecret); err != nil {53 t.Fatal(err)54 }55 if _, err := tenant.store.Save(ctx, e, storage.SaveOptions{Now: time.Now().Unix()}); err != nil {56 t.Fatal(err)57 }58 if err := tenant.replication.AddJob(ctx, replication.JobSpec{ID: "restart-dump", Kind: replication.JobDump, Relays: []string{"restart.jsonl"}}); err != nil {59 t.Fatal(err)60 }61 if err := tenant.records.MarkView(ctx, "articles", time.Now().Unix()-9); err != nil {62 t.Fatal(err)63 }64 if err := app.Close(ctx); err != nil {65 t.Fatal(err)66 }6768 app, err = New(ctx, Config{DataDir: dir, DefaultTenant: "main"})69 if err != nil {70 t.Fatal(err)71 }72 defer app.Close(ctx)73 if err := app.Start(ctx, "http://relay.test"); err != nil {74 t.Fatal(err)75 }76 reloaded := app.tenants[meta.ID]77 if reloaded == nil {78 t.Fatal("tenant was not recovered")79 }80 if !waitFor(t, 8*time.Second, func() bool {81 statuses, statusErr := reloaded.replication.ListJobStatus(ctx)82 return statusErr == nil && len(statuses) == 1 && statuses[0].Phase == "complete"83 }) {84 statuses, _ := reloaded.replication.ListJobStatus(ctx)85 t.Fatalf("restart job did not resume: %#v", statuses)86 }87 if _, err := os.Stat(filepath.Join(meta.Paths.Root, "restart.jsonl")); err != nil {88 t.Fatalf("restart artifact missing: %v", err)89 }90 var checkpoint int6491 if err := reloaded.store.GetSetting(ctx, "records.view.articles.dirty", &checkpoint); err != nil || checkpoint == 0 {92 t.Fatalf("unconsumed view checkpoint was lost across restart: %d %v", checkpoint, err)93 }94}9596// TestDaemonRestartDrainsDurableNotificationIntent verifies that generated97// notification work is persisted before shutdown and handled after recovery.98func TestDaemonRestartDrainsDurableNotificationIntent(t *testing.T) {99 ctx := context.Background()100 dir := t.TempDir()101 secret := strings.Repeat("0", 63) + "1"102 owner, err := event.PublicKey(secret)103 if err != nil {104 t.Fatal(err)105 }106 var mu sync.Mutex107 var received int108 remote := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {109 conn, err := websocket.Accept(w, r, nil)110 if err != nil {111 return112 }113 defer conn.Close(websocket.StatusNormalClosure, "done")114 _, raw, err := conn.Read(r.Context())115 if err != nil {116 return117 }118 var message []json.RawMessage119 if json.Unmarshal(raw, &message) == nil && len(message) > 1 {120 var kind string121 _ = json.Unmarshal(message[0], &kind)122 if kind == "EVENT" {123 var item event.Event124 if json.Unmarshal(message[1], &item) == nil {125 mu.Lock()126 received++127 mu.Unlock()128 response, _ := json.Marshal([]any{"OK", item.ID, true, ""})129 _ = conn.Write(r.Context(), websocket.MessageText, response)130 }131 }132 }133 }))134 defer remote.Close()135 remoteURL := "ws" + strings.TrimPrefix(remote.URL, "http")136 app, err := New(ctx, Config{DataDir: dir, DefaultTenant: "main", AllowPrivateRelays: true})137 if err != nil {138 t.Fatal(err)139 }140 meta, err := app.Create(ctx, CreateOptions{Name: "main", Owner: owner, Template: "default"})141 if err != nil {142 t.Fatal(err)143 }144 if err := app.Start(ctx, "http://relay.test"); err != nil {145 t.Fatal(err)146 }147 tenant := app.tenants[meta.ID]148 tenant.workCancel()149 tenant.workWG.Wait()150 inbox := event.Event{Kind: 10050, CreatedAt: time.Now().Unix(), Tags: [][]string{{"relay", remoteURL}}}151 if err := event.Sign(&inbox, secret); err != nil {152 t.Fatal(err)153 }154 if _, err := tenant.store.Save(ctx, inbox, storage.SaveOptions{Now: time.Now().Unix()}); err != nil {155 t.Fatal(err)156 }157 wrap := event.Event{Kind: event.KIND_WRAP, CreatedAt: time.Now().Unix(), Tags: [][]string{{"p", owner}}, Content: "cipher"}158 if err := event.Sign(&wrap, secret); err != nil {159 t.Fatal(err)160 }161 if err := tenant.deliverNotification(ctx, wrap, owner); err != nil {162 t.Fatal(err)163 }164 if err := app.Close(ctx); err != nil {165 t.Fatal(err)166 }167 app, err = New(ctx, Config{DataDir: dir, DefaultTenant: "main", AllowPrivateRelays: true})168 if err != nil {169 t.Fatal(err)170 }171 defer app.Close(ctx)172 if err := app.Start(ctx, "http://relay.test"); err != nil {173 t.Fatal(err)174 }175 if !waitFor(t, 8*time.Second, func() bool { mu.Lock(); defer mu.Unlock(); return received == 1 }) {176 t.Fatal("durable notification intent was not delivered after restart")177 }178}