1package daemon23import (4 "context"5 "strings"6 "testing"7 "time"89 "github.com/FelineStateMachine/tinyrelay/internal/event"10 "github.com/FelineStateMachine/tinyrelay/internal/storage"11)1213// TestCallbackIntentSurvivesRestartAndHonorsRevocation verifies the daemon14// lifecycle around callback work. The callback is queued transactionally,15// survives reopening the tenant, and is consumed without delivery once its16// registration is revoked before the recovered worker runs.17func TestCallbackIntentSurvivesRestartAndHonorsRevocation(t *testing.T) {18 ctx := context.Background()19 dir := t.TempDir()20 secret := strings.Repeat("0", 63) + "1"21 owner, err := event.PublicKey(secret)22 if err != nil {23 t.Fatal(err)24 }25 config := Config{DataDir: dir, DefaultTenant: "main", PushCallbackOrigins: []string{"https://push.example"}}26 app, err := New(ctx, config)27 if err != nil {28 t.Fatal(err)29 }30 meta, err := app.Create(ctx, CreateOptions{Name: "main", Owner: owner, Template: "default"})31 if err != nil {32 t.Fatal(err)33 }34 if err := app.Start(ctx, "http://relay.test"); err != nil {35 t.Fatal(err)36 }37 tenant := app.tenants[meta.ID]38 if tenant == nil {39 t.Fatal("tenant was not started")40 }41 // Fence the production worker so the intent remains unconsumed until the42 // second App has reopened the tenant.43 tenant.workCancel()44 tenant.workWG.Wait()45 next := tenant.Policy()46 next.Features.Push = true47 next.PushCallbacks = []string{"https://push.example"}48 if err := tenant.applyPolicy(ctx, next); err != nil {49 t.Fatal(err)50 }51 registration := event.Event{Kind: event.KIND_PUSH_REGISTRATION, CreatedAt: time.Now().Unix(), Tags: [][]string{52 {"d", "restart-callback"},53 {"relay", tenant.RelayURL()},54 {"callback", "https://push.example"},55 {"filter", `{"kinds":[1]}`},56 }}57 if err := event.Sign(®istration, secret); err != nil {58 t.Fatal(err)59 }60 if _, err := tenant.store.Save(ctx, registration, storage.SaveOptions{Now: time.Now().Unix()}); err != nil {61 t.Fatal(err)62 }63 delivered := event.Event{Kind: 1, CreatedAt: time.Now().Unix(), Content: "callback restart"}64 if err := event.Sign(&delivered, secret); err != nil {65 t.Fatal(err)66 }67 intents, err := tenant.PrepareReplicationCallbacks(ctx, delivered)68 if err != nil {69 t.Fatal(err)70 }71 if len(intents) != 1 || intents[0].Kind != "callback" {72 t.Fatalf("callback intents = %#v", intents)73 }74 if _, err := tenant.store.Save(ctx, delivered, storage.SaveOptions{Now: time.Now().Unix(), Intents: intents}); err != nil {75 t.Fatal(err)76 }77 // Keep recovery deterministic: the new App must not attempt delivery in78 // the interval between starting its worker and the policy revocation.79 if _, err := tenant.store.DB().ExecContext(ctx, `UPDATE work_intents SET next_at=? WHERE kind='callback'`, time.Now().Add(time.Hour).Unix()); err != nil {80 t.Fatal(err)81 }82 if err := app.Close(ctx); err != nil {83 t.Fatal(err)84 }8586 app, err = New(ctx, config)87 if err != nil {88 t.Fatal(err)89 }90 defer app.Close(ctx)91 if err := app.Start(ctx, "http://relay.test"); err != nil {92 t.Fatal(err)93 }94 reloaded := app.tenants[meta.ID]95 if reloaded == nil {96 t.Fatal("tenant was not recovered")97 }98 // Stop the automatically recovered worker before changing policy, then99 // revoke the origin and let a fresh worker consume the recovered intent.100 reloaded.workCancel()101 reloaded.workWG.Wait()102 revoked := reloaded.Policy()103 revoked.Features.Push = true104 revoked.PushCallbacks = nil105 if err := reloaded.applyPolicy(ctx, revoked); err != nil {106 t.Fatal(err)107 }108 pending, err := reloaded.replication.Queue().Pending(ctx, "callback")109 if err != nil {110 t.Fatal(err)111 }112 if len(pending) != 1 {113 t.Fatalf("recovered callback intents = %d", len(pending))114 }115 if err := reloaded.replication.Queue().RetryNow(ctx, pending[0].ID, time.Now()); err != nil {116 t.Fatal(err)117 }118 runCtx, cancel := context.WithCancel(ctx)119 done := make(chan error, 1)120 go func() { done <- reloaded.runWork(runCtx) }()121 if !waitFor(t, 3*time.Second, func() bool {122 items, listErr := reloaded.replication.Queue().Pending(ctx, "callback")123 return listErr == nil && len(items) == 0124 }) {125 t.Fatalf("revoked callback intent remained pending: %#v", pending)126 }127 cancel()128 if err := <-done; err != nil && !strings.Contains(err.Error(), "canceled") {129 t.Fatal(err)130 }131}