| 1 | package daemon |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "strconv" |
| 9 | "strings" |
| 10 | "testing" |
| 11 | "time" |
| 12 | |
| 13 | "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 | ) |
| 18 | |
| 19 | func 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 error |
| 32 | 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 int |
| 41 | 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 int |
| 48 | 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 | } |
| 57 | |
| 58 | func 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 | } |
| 91 | |
| 92 | func 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 | } |
| 125 | |
| 126 | func 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 = followupMaxAttempts |
| 145 | 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 | } |
| 150 | |
| 151 | func 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 int |
| 180 | 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 | } |
| 187 | |
| 188 | func callbackDeliveryState(t *testing.T, tenant *Tenant, target, eventID string) string { |
| 189 | t.Helper() |
| 190 | var state string |
| 191 | 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 state |
| 199 | } |
| 200 | |
| 201 | func 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 | } |
| 207 | |
| 208 | func 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 | } |