1package daemon23import (4 "context"5 "testing"6 "time"78 "github.com/FelineStateMachine/tinyrelay/internal/event"9 "github.com/FelineStateMachine/tinyrelay/internal/relay"10 "github.com/FelineStateMachine/tinyrelay/internal/storage"11)1213func enablePushPlanning(t *testing.T, tenant *Tenant) {14 t.Helper()15 tenant.app.cfg.PushCallbackOrigins = []string{"https://push.example"}16 p := tenant.Policy()17 p.Features.Push = true18 p.PushCallbacks = []string{"https://push.example"}19 if err := tenant.applyPolicy(context.Background(), p); err != nil {20 t.Fatal(err)21 }22}2324func makeReplicationPushRegistration(t *testing.T, tenant *Tenant, d string, created int64, kinds ...int) event.Event {25 t.Helper()26 if len(kinds) == 0 {27 kinds = []int{1}28 }29 filter := `{"kinds":[1]}`30 if kinds[0] != 1 {31 filter = `{"kinds":[30618]}`32 }33 registration := event.Event{Kind: event.KIND_PUSH_REGISTRATION, CreatedAt: created, Tags: [][]string{34 {"d", d}, {"relay", tenant.RelayURL()}, {"callback", "https://push.example"}, {"filter", filter},35 }}36 if err := event.Sign(®istration, testOwnerSecret); err != nil {37 t.Fatal(err)38 }39 if _, err := tenant.store.Save(context.Background(), registration, storage.SaveOptions{Now: created}); err != nil {40 t.Fatal(err)41 }42 return registration43}4445func TestReplicationPushFollowupUsesInsertionCutoff(t *testing.T) {46 _, tenant := loopbackTenant(t)47 enablePushPlanning(t, tenant)48 ctx := context.Background()49 first := makeReplicationPushRegistration(t, tenant, "first", 100)50 e := signedEvent(t, testOwnerSecret, 1, 101, nil, "push cutoff")51 payload, err := tenant.prepareReplicationPushFollowup(ctx, e, 102)52 if err != nil {53 t.Fatal(err)54 }55 if len(payload.RegistrationIDs) != 1 || payload.RegistrationIDs[0] != first.ID {56 t.Fatalf("captured registrations = %#v", payload.RegistrationIDs)57 }58 if payload.RegistrationCutoff == 0 {59 t.Fatal("missing insertion cutoff")60 }61 if _, err := tenant.store.Save(ctx, e, storage.SaveOptions{Now: 102}); err != nil {62 t.Fatal(err)63 }64 makeReplicationPushRegistration(t, tenant, "later", 1)65 payload.Discover = true66 if err := tenant.recoverReplicationPushFollowup(ctx, &payload); err != nil {67 t.Fatal(err)68 }69 intents, err := tenant.prepareReplicationPushIntents(ctx, payload)70 if err != nil {71 t.Fatal(err)72 }73 if len(intents) != 1 || intents[0].Target != first.ID {74 t.Fatalf("recovered callback intents = %#v", intents)75 }76}7778func TestReplicationPushFollowupSkipsSupersededRegistration(t *testing.T) {79 _, tenant := loopbackTenant(t)80 enablePushPlanning(t, tenant)81 ctx := context.Background()82 old := makeReplicationPushRegistration(t, tenant, "same", 100)83 e := signedEvent(t, testOwnerSecret, 1, 101, nil, "push replacement")84 payload, err := tenant.prepareReplicationPushFollowup(ctx, e, 102)85 if err != nil {86 t.Fatal(err)87 }88 if _, err := tenant.store.Save(ctx, e, storage.SaveOptions{Now: time.Now().Unix()}); err != nil {89 t.Fatal(err)90 }91 newer := event.Event{Kind: event.KIND_PUSH_REGISTRATION, CreatedAt: 200, Tags: [][]string{92 {"d", "same"}, {"relay", tenant.RelayURL()}, {"callback", "https://push.example"}, {"filter", `{"kinds":[1]}`},93 }}94 if err := event.Sign(&newer, testOwnerSecret); err != nil {95 t.Fatal(err)96 }97 if _, err := tenant.store.Save(ctx, newer, storage.SaveOptions{Now: 200}); err != nil {98 t.Fatal(err)99 }100 intents, err := tenant.prepareReplicationPushIntents(ctx, payload)101 if err != nil {102 t.Fatal(err)103 }104 if len(intents) != 0 {105 t.Fatalf("superseded callback intents = %#v (old registration %s)", intents, old.ID)106 }107}108109func TestReplicationPushFollowupFailureKeepsClientEventDurable(t *testing.T) {110 _, tenant := loopbackTenant(t)111 enablePushPlanning(t, tenant)112 ctx := context.Background()113 registration := makeReplicationPushRegistration(t, tenant, "durable", 100)114 if _, err := tenant.store.DB().ExecContext(ctx, `CREATE TRIGGER reject_replication_push_work BEFORE INSERT ON work_intents WHEN NEW.kind='callback' BEGIN SELECT RAISE(ABORT,'push work unavailable'); END`); err != nil {115 t.Fatal(err)116 }117 e := signedEvent(t, testOwnerSecret, 1, 101, nil, "durable push")118 if _, err := tenant.Publish(ctx, e, relay.Session{PubKeys: []string{e.PubKey}}); err != nil {119 t.Fatalf("publish rejected after optional push planning failure: %v", err)120 }121 var stored int122 if err := tenant.store.DB().QueryRowContext(ctx, `SELECT count(*) FROM events WHERE id=?`, e.ID).Scan(&stored); err != nil {123 t.Fatal(err)124 }125 if stored != 1 {126 t.Fatalf("stored events=%d, want 1", stored)127 }128 job := followupJob(t, tenant, e.ID)129 if job.Kind != eventFollowupKind {130 t.Fatalf("follow-up kind=%q", job.Kind)131 }132 if err := tenant.store.DB().QueryRowContext(ctx, `SELECT state FROM work_intents WHERE id=?`, job.ID).Scan(&job.State); err != nil {133 t.Fatal(err)134 }135 if job.State != "pending" {136 t.Fatalf("parent state=%q, want pending", job.State)137 }138 if _, err := tenant.store.DB().ExecContext(ctx, `DROP TRIGGER reject_replication_push_work`); err != nil {139 t.Fatal(err)140 }141 parent := storage.Intent{Kind: job.Kind, EventID: job.EventID, Target: job.Target, Payload: job.Payload}142 tenant.followups.try(ctx, &parent)143 var parentState string144 if err := tenant.store.DB().QueryRowContext(ctx, `SELECT state FROM work_intents WHERE id=?`, job.ID).Scan(&parentState); err != nil {145 t.Fatal(err)146 }147 if parentState != "completed" {148 t.Fatalf("recovered parent state=%q, want completed", parentState)149 }150 var callbackState string151 if err := tenant.store.DB().QueryRowContext(ctx, `SELECT state FROM work_intents WHERE kind='callback' AND event_id=? AND target=?`, e.ID, registration.ID).Scan(&callbackState); err != nil {152 t.Fatal(err)153 }154 if callbackState != "pending" {155 t.Fatalf("derived callback state=%q, want pending", callbackState)156 }157 if _, err := tenant.store.DB().ExecContext(ctx, `UPDATE work_intents SET state='completed' WHERE kind='callback' AND event_id=? AND target=?`, e.ID, registration.ID); err != nil {158 t.Fatal(err)159 }160 tenant.followups.try(ctx, &parent)161 if err := tenant.store.DB().QueryRowContext(ctx, `SELECT state FROM work_intents WHERE kind='callback' AND event_id=? AND target=?`, e.ID, registration.ID).Scan(&callbackState); err != nil {162 t.Fatal(err)163 }164 if callbackState != "completed" {165 t.Fatalf("recovered completed callback state=%q, want completed", callbackState)166 }167}168169func TestReplicationPushFollowupSkipsRevokedPolicy(t *testing.T) {170 _, tenant := loopbackTenant(t)171 enablePushPlanning(t, tenant)172 ctx := context.Background()173 registration := makeReplicationPushRegistration(t, tenant, "revoked", 100)174 e := signedEvent(t, testOwnerSecret, 1, 101, nil, "revoked push")175 payload, err := tenant.prepareReplicationPushFollowup(ctx, e, 102)176 if err != nil {177 t.Fatal(err)178 }179 if _, err := tenant.store.Save(ctx, e, storage.SaveOptions{Now: 102}); err != nil {180 t.Fatal(err)181 }182 payload.RegistrationIDs = []string{registration.ID}183 p := tenant.Policy()184 p.Features.Push = false185 p.PushCallbacks = nil186 if err := tenant.applyPolicy(ctx, p); err != nil {187 t.Fatal(err)188 }189 intents, err := tenant.planReplicationPushFollowup(ctx, payload)190 if err != nil {191 t.Fatal(err)192 }193 if len(intents) != 0 {194 t.Fatalf("revoked push intents=%d, want 0", len(intents))195 }196}197198func TestReplicationPushFollowupGitReleaseUsesCapturedRegistrations(t *testing.T) {199 // A pending source retains its client snapshot in the parent follow-up;200 // releasedPush must recover that snapshot instead of rediscovering later201 // registrations.202 _, tenant := loopbackTenant(t)203 enablePushPlanning(t, tenant)204 ctx := context.Background()205 first := makeReplicationPushRegistration(t, tenant, "git-first", 100, event.KIND_REPO_STATE)206 e := signedEvent(t, testOwnerSecret, event.KIND_REPO_STATE, 101, nil, "pending git")207 payload, err := tenant.prepareReplicationPushFollowup(ctx, e, 102)208 if err != nil {209 t.Fatal(err)210 }211 followup := tenant.followups.prepareWithPush(ctx, e, 102, false, &payload)212 if followup == nil {213 t.Fatal("missing pending git follow-up")214 }215 if _, err := tenant.store.Save(ctx, e, storage.SaveOptions{Now: 102, Intents: []storage.Intent{*followup}}); err != nil {216 t.Fatal(err)217 }218 makeReplicationPushRegistration(t, tenant, "git-later", 1, event.KIND_REPO_STATE)219 released, err := tenant.followups.releasedPush(ctx, e.ID)220 if err != nil {221 t.Fatal(err)222 }223 if released == nil || len(released.RegistrationIDs) != 1 || released.RegistrationIDs[0] != first.ID {224 t.Fatalf("released push snapshot=%#v", released)225 }226 intents, err := tenant.prepareReplicationPushIntents(ctx, *released)227 if err != nil {228 t.Fatal(err)229 }230 if len(intents) != 1 || intents[0].Target != first.ID {231 t.Fatalf("released push intents=%#v", intents)232 }233}