1package daemon23import (4 "context"5 "database/sql"6 "encoding/json"7 "errors"8 "strconv"9 "strings"10 "testing"11 "time"1213 "github.com/FelineStateMachine/tinyrelay/internal/event"14 "github.com/FelineStateMachine/tinyrelay/internal/relay"15 "github.com/FelineStateMachine/tinyrelay/internal/replication"16 "github.com/FelineStateMachine/tinyrelay/internal/storage"17)1819func TestCallbackFollowupPlanningFailureDoesNotRejectEvents(t *testing.T) {20 for _, source := range []string{"client", "import"} {21 t.Run(source, func(t *testing.T) {22 _, tenant := loopbackTenant(t)23 ctx := context.Background()24 member, _ := event.PublicKey(testMemberSecret)25 setRole(t, tenant, member, "member")26 addCallback(t, tenant, member, "https://127.0.0.1:1/wake", `{"kinds":[1]}`)27 if _, err := tenant.store.DB().ExecContext(ctx, `CREATE TRIGGER reject_callback_work BEFORE INSERT ON work_intents WHEN NEW.kind='callback-delivery' BEGIN SELECT RAISE(ABORT,'callback work unavailable'); END`); err != nil {28 t.Fatal(err)29 }30 e := signedEvent(t, testOwnerSecret, 1, time.Now().Unix(), [][]string{{"p", member}}, "callback planning")31 var err error32 if source == "import" {33 err = tenant.commitImported(ctx, e, replication.OriginImport)34 } else {35 _, err = tenant.Publish(ctx, e, relay.Session{PubKeys: []string{e.PubKey}})36 }37 if err != nil {38 t.Fatalf("%s event rejected after optional planning failure: %v", source, err)39 }40 var stored int41 if err := tenant.store.DB().QueryRowContext(ctx, `SELECT count(*) FROM events WHERE id=?`, e.ID).Scan(&stored); err != nil {42 t.Fatal(err)43 }44 if stored != 1 {45 t.Fatalf("stored events=%d, want 1", stored)46 }47 var pending int48 if err := tenant.store.DB().QueryRowContext(ctx, `SELECT count(*) FROM work_intents WHERE kind=? AND event_id=? AND state='pending'`, eventFollowupKind, e.ID).Scan(&pending); err != nil {49 t.Fatal(err)50 }51 if pending != 1 {52 t.Fatalf("pending follow-up jobs=%d, want 1", pending)53 }54 })55 }56}5758func TestCallbackFollowupRecoveryPreservesCompletedDelivery(t *testing.T) {59 _, tenant := loopbackTenant(t)60 ctx := context.Background()61 member, _ := event.PublicKey(testMemberSecret)62 setRole(t, tenant, member, "member")63 first := addCallback(t, tenant, member, "https://127.0.0.1:1/first", `{"kinds":[1]}`)64 second := addCallback(t, tenant, member, "https://127.0.0.1:1/second", `{"kinds":[1]}`)65 e := signedEvent(t, testOwnerSecret, 1, time.Now().Unix(), [][]string{{"p", member}}, "recover callback work")66 followup := tenant.followups.prepare(ctx, e, time.Now().Unix(), false)67 if followup == nil {68 t.Fatal("missing callback follow-up")69 }70 if _, err := tenant.store.Save(ctx, e, storage.SaveOptions{Now: time.Now().Unix(), Intents: []storage.Intent{*followup}}); err != nil {71 t.Fatal(err)72 }73 job := followupJob(t, tenant, e.ID)74 if err := tenant.followups.handle(ctx, job); err != nil {75 t.Fatal(err)76 }77 firstID := first["id"].(string)78 secondID := second["id"].(string)79 markCallbackDeliveryCompleted(t, tenant, firstID, e.ID)80 deleteCallbackDelivery(t, tenant, secondID, e.ID)81 if err := tenant.followups.handle(ctx, job); err != nil {82 t.Fatal(err)83 }84 if state := callbackDeliveryState(t, tenant, firstID, e.ID); state != "completed" {85 t.Fatalf("completed delivery state=%q, want completed", state)86 }87 if state := callbackDeliveryState(t, tenant, secondID, e.ID); state != "pending" {88 t.Fatalf("recovered delivery state=%q, want pending", state)89 }90}9192func TestCallbackFollowupSkipsPausedRemovedAndLaterRegistrations(t *testing.T) {93 for _, action := range []string{"pausecallback", "removecallback"} {94 t.Run(action, func(t *testing.T) {95 _, tenant := loopbackTenant(t)96 ctx := context.Background()97 member, _ := event.PublicKey(testMemberSecret)98 setRole(t, tenant, member, "member")99 captured := addCallback(t, tenant, member, "https://127.0.0.1:1/captured", `{"kinds":[1]}`)100 e := signedEvent(t, testOwnerSecret, 1, time.Now().Unix(), [][]string{{"p", member}}, "captured callback")101 followup := tenant.followups.prepare(ctx, e, time.Now().Unix(), false)102 if followup == nil {103 t.Fatal("missing callback follow-up")104 }105 id := captured["id"].(string)106 if _, err := tenant.Execute(ctx, member, action, []json.RawMessage{json.RawMessage(strconv.Quote(id))}); err != nil {107 t.Fatal(err)108 }109 later := addCallback(t, tenant, member, "https://127.0.0.1:1/later", `{"kinds":[1]}`)110 if _, err := tenant.store.Save(ctx, e, storage.SaveOptions{Now: time.Now().Unix(), Intents: []storage.Intent{*followup}}); err != nil {111 t.Fatal(err)112 }113 if err := tenant.followups.handle(ctx, followupJob(t, tenant, e.ID)); err != nil {114 t.Fatal(err)115 }116 if got := callbackDeliveryState(t, tenant, id, e.ID); got != "" {117 t.Fatalf("%s callback state=%q, want absent", action, got)118 }119 if got := callbackDeliveryState(t, tenant, later["id"].(string), e.ID); got != "" {120 t.Fatalf("later callback state=%q, want absent", got)121 }122 })123 }124}125126func TestCallbackFollowupPlanningExhaustionStops(t *testing.T) {127 _, tenant := loopbackTenant(t)128 ctx := context.Background()129 member, _ := event.PublicKey(testMemberSecret)130 setRole(t, tenant, member, "member")131 addCallback(t, tenant, member, "https://127.0.0.1:1/wake", `{"kinds":[1]}`)132 e := signedEvent(t, testOwnerSecret, 1, time.Now().Unix(), [][]string{{"p", member}}, "exhaust planning")133 followup := tenant.followups.prepare(ctx, e, time.Now().Unix(), false)134 if followup == nil {135 t.Fatal("missing callback follow-up")136 }137 if _, err := tenant.store.Save(ctx, e, storage.SaveOptions{Now: time.Now().Unix(), Intents: []storage.Intent{*followup}}); err != nil {138 t.Fatal(err)139 }140 if _, err := tenant.store.DB().ExecContext(ctx, `CREATE TRIGGER reject_callback_work BEFORE INSERT ON work_intents WHEN NEW.kind='callback-delivery' BEGIN SELECT RAISE(ABORT,'callback work unavailable'); END`); err != nil {141 t.Fatal(err)142 }143 job := followupJob(t, tenant, e.ID)144 job.Attempts = followupMaxAttempts145 err := tenant.followups.handle(ctx, job)146 if err == nil || !strings.Contains(err.Error(), "callback work unavailable") || errors.Unwrap(err) == nil {147 t.Fatalf("exhausted planning error=%v, want terminal wrapped error", err)148 }149}150151func TestCallbackFollowupRecoversWhenViewLookupIsUnavailable(t *testing.T) {152 tenant, _, _ := viewTenant(t, []int{1}, nil)153 ctx := context.Background()154 member, _ := event.PublicKey(testMemberSecret)155 setRole(t, tenant, member, "member")156 callback := addCallback(t, tenant, member, "https://127.0.0.1:1/wake", `{"kinds":[1]}`)157 e := signedEvent(t, testOwnerSecret, 1, time.Now().Unix(), [][]string{{"p", member}}, "```mermaid\ngraph LR; A-->B\n```")158 if _, err := tenant.store.DB().ExecContext(ctx, `ALTER TABLE custom_views RENAME TO custom_views_unavailable`); err != nil {159 t.Fatal(err)160 }161 if _, err := tenant.Publish(ctx, e, relay.Session{PubKeys: []string{e.PubKey}}); err != nil {162 t.Fatalf("publish with unavailable optional view lookup: %v", err)163 }164 job := followupJob(t, tenant, e.ID)165 callbackID := callback["id"].(string)166 if state := callbackDeliveryState(t, tenant, callbackID, e.ID); state != "pending" {167 t.Fatalf("callback state after partial planning=%q, want pending", state)168 }169 markCallbackDeliveryCompleted(t, tenant, callbackID, e.ID)170 if _, err := tenant.store.DB().ExecContext(ctx, `ALTER TABLE custom_views_unavailable RENAME TO custom_views`); err != nil {171 t.Fatal(err)172 }173 if err := tenant.followups.handle(ctx, job); err != nil {174 t.Fatalf("recover follow-up planning: %v", err)175 }176 if state := callbackDeliveryState(t, tenant, callbackID, e.ID); state != "completed" {177 t.Fatalf("callback state after recovery=%q, want completed", state)178 }179 var callbackCount int180 if err := tenant.store.DB().QueryRowContext(ctx, `SELECT count(*) FROM work_intents WHERE kind=? AND target=? AND event_id=?`, callbackDelivery, callbackID, e.ID).Scan(&callbackCount); err != nil {181 t.Fatal(err)182 }183 if callbackCount != 1 {184 t.Fatalf("callback intent count=%d, want 1", callbackCount)185 }186}187188func callbackDeliveryState(t *testing.T, tenant *Tenant, target, eventID string) string {189 t.Helper()190 var state string191 err := tenant.store.DB().QueryRow(`SELECT state FROM work_intents WHERE kind=? AND target=? AND event_id=?`, callbackDelivery, target, eventID).Scan(&state)192 if errors.Is(err, sql.ErrNoRows) {193 return ""194 }195 if err != nil {196 t.Fatal(err)197 }198 return state199}200201func markCallbackDeliveryCompleted(t *testing.T, tenant *Tenant, target, eventID string) {202 t.Helper()203 if _, err := tenant.store.DB().Exec(`UPDATE work_intents SET state='completed' WHERE kind=? AND target=? AND event_id=?`, callbackDelivery, target, eventID); err != nil {204 t.Fatal(err)205 }206}207208func deleteCallbackDelivery(t *testing.T, tenant *Tenant, target, eventID string) {209 t.Helper()210 if _, err := tenant.store.DB().Exec(`DELETE FROM work_intents WHERE kind=? AND target=? AND event_id=?`, callbackDelivery, target, eventID); err != nil {211 t.Fatal(err)212 }213}