| 1 | package mail |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "strings" |
| 9 | "sync" |
| 10 | "testing" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/carlosframework/platform/internal/dnsrec" |
| 14 | "github.com/carlosframework/platform/internal/hostops" |
| 15 | "github.com/carlosframework/platform/internal/release" |
| 16 | ) |
| 17 | |
| 18 | const claimTestAccount = "bag" |
| 19 | const claimTestDomain = "shared.com" |
| 20 | |
| 21 | // claimLogStore records every Get and Put. It embeds only release.Store, so it is |
| 22 | // not a ClaimMailStore and behaves as an unmarked store. |
| 23 | type claimLogStore struct { |
| 24 | release.Store |
| 25 | mu sync.Mutex |
| 26 | gets, puts []string |
| 27 | } |
| 28 | |
| 29 | func (l *claimLogStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 30 | l.mu.Lock() |
| 31 | l.gets = append(l.gets, key) |
| 32 | l.mu.Unlock() |
| 33 | return l.Store.Get(ctx, key) |
| 34 | } |
| 35 | func (l *claimLogStore) Put(ctx context.Context, key string, body []byte, tag string) (string, error) { |
| 36 | l.mu.Lock() |
| 37 | l.puts = append(l.puts, key) |
| 38 | l.mu.Unlock() |
| 39 | return l.Store.Put(ctx, key, body, tag) |
| 40 | } |
| 41 | func (l *claimLogStore) reset() { l.mu.Lock(); l.gets, l.puts = nil, nil; l.mu.Unlock() } |
| 42 | func (l *claimLogStore) admissionGets() int { |
| 43 | l.mu.Lock() |
| 44 | defer l.mu.Unlock() |
| 45 | n := 0 |
| 46 | for _, k := range l.gets { |
| 47 | if strings.HasSuffix(k, "/admission.json") { |
| 48 | n++ |
| 49 | } |
| 50 | } |
| 51 | return n |
| 52 | } |
| 53 | func (l *claimLogStore) putsTo(key string) int { |
| 54 | l.mu.Lock() |
| 55 | defer l.mu.Unlock() |
| 56 | n := 0 |
| 57 | for _, k := range l.puts { |
| 58 | if k == key { |
| 59 | n++ |
| 60 | } |
| 61 | } |
| 62 | return n |
| 63 | } |
| 64 | |
| 65 | // countingProv counts every Provisioner call, reads included. |
| 66 | type countingProv struct { |
| 67 | *MemProvisioner |
| 68 | cmu sync.Mutex |
| 69 | hits map[string]int |
| 70 | } |
| 71 | |
| 72 | func (c *countingProv) hit(name string) { c.cmu.Lock(); c.hits[name]++; c.cmu.Unlock() } |
| 73 | func (c *countingProv) total() int { |
| 74 | c.cmu.Lock() |
| 75 | defer c.cmu.Unlock() |
| 76 | n := 0 |
| 77 | for _, v := range c.hits { |
| 78 | n += v |
| 79 | } |
| 80 | return n |
| 81 | } |
| 82 | func (c *countingProv) reset() { c.cmu.Lock(); c.hits = map[string]int{}; c.cmu.Unlock() } |
| 83 | func (c *countingProv) EnsureTenant(ctx context.Context, name string) error { |
| 84 | c.hit("EnsureTenant") |
| 85 | return c.MemProvisioner.EnsureTenant(ctx, name) |
| 86 | } |
| 87 | func (c *countingProv) EnsureConfigSet(ctx context.Context, region, name, topic string) error { |
| 88 | c.hit("EnsureConfigSet") |
| 89 | return c.MemProvisioner.EnsureConfigSet(ctx, region, name, topic) |
| 90 | } |
| 91 | func (c *countingProv) SetIdentityConfigSet(ctx context.Context, region, domain, set string) error { |
| 92 | c.hit("SetIdentityConfigSet") |
| 93 | return c.MemProvisioner.SetIdentityConfigSet(ctx, region, domain, set) |
| 94 | } |
| 95 | func (c *countingProv) AssociateTenantResource(ctx context.Context, tenant, resource string) error { |
| 96 | c.hit("AssociateTenantResource") |
| 97 | return c.MemProvisioner.AssociateTenantResource(ctx, tenant, resource) |
| 98 | } |
| 99 | func (c *countingProv) DisassociateTenantResource(ctx context.Context, tenant, resource string) error { |
| 100 | c.hit("DisassociateTenantResource") |
| 101 | return c.MemProvisioner.DisassociateTenantResource(ctx, tenant, resource) |
| 102 | } |
| 103 | func (c *countingProv) EnsureIdentity(ctx context.Context, region, domain, from string, adopt bool) (Identity, error) { |
| 104 | c.hit("EnsureIdentity") |
| 105 | return c.MemProvisioner.EnsureIdentity(ctx, region, domain, from, adopt) |
| 106 | } |
| 107 | func (c *countingProv) IdentityStatus(ctx context.Context, region, domain string) (Status, error) { |
| 108 | c.hit("IdentityStatus") |
| 109 | return c.MemProvisioner.IdentityStatus(ctx, region, domain) |
| 110 | } |
| 111 | func (c *countingProv) DeleteIdentity(ctx context.Context, region, domain string) error { |
| 112 | c.hit("DeleteIdentity") |
| 113 | return c.MemProvisioner.DeleteIdentity(ctx, region, domain) |
| 114 | } |
| 115 | func (c *countingProv) EnsureUser(ctx context.Context, name, path string, identities, sets, domains []string) (string, error) { |
| 116 | c.hit("EnsureUser") |
| 117 | return c.MemProvisioner.EnsureUser(ctx, name, path, identities, sets, domains) |
| 118 | } |
| 119 | func (c *countingProv) MintKey(ctx context.Context, user string) (string, string, error) { |
| 120 | c.hit("MintKey") |
| 121 | return c.MemProvisioner.MintKey(ctx, user) |
| 122 | } |
| 123 | func (c *countingProv) SetKeyActive(ctx context.Context, user, key string, active bool) error { |
| 124 | c.hit("SetKeyActive") |
| 125 | return c.MemProvisioner.SetKeyActive(ctx, user, key, active) |
| 126 | } |
| 127 | func (c *countingProv) DeleteKey(ctx context.Context, user, key string) error { |
| 128 | c.hit("DeleteKey") |
| 129 | return c.MemProvisioner.DeleteKey(ctx, user, key) |
| 130 | } |
| 131 | func (c *countingProv) ReapKeys(ctx context.Context, user string, keep []string) error { |
| 132 | c.hit("ReapKeys") |
| 133 | return c.MemProvisioner.ReapKeys(ctx, user, keep) |
| 134 | } |
| 135 | func (c *countingProv) DeleteUser(ctx context.Context, user string) error { |
| 136 | c.hit("DeleteUser") |
| 137 | return c.MemProvisioner.DeleteUser(ctx, user) |
| 138 | } |
| 139 | |
| 140 | type claimDomainFixture struct { |
| 141 | t *testing.T |
| 142 | ctx context.Context |
| 143 | base release.Store |
| 144 | log *claimLogStore |
| 145 | marked ClaimMailStore |
| 146 | mem *MemProvisioner |
| 147 | prov *countingProv |
| 148 | now time.Time |
| 149 | mg *memGate |
| 150 | adm *claimFixtureAdmitter |
| 151 | states map[string]string |
| 152 | } |
| 153 | |
| 154 | var claimTestOwner = ClaimMailOwner{Console: "c", Enrollment: "e", Controller: "one"} |
| 155 | |
| 156 | // memGate is a test-only ClaimMailGate holding {state,id,revision} in memory. |
| 157 | type memGate struct { |
| 158 | mu sync.Mutex |
| 159 | State string |
| 160 | id string |
| 161 | Revision int |
| 162 | } |
| 163 | type memGateKey struct{} |
| 164 | |
| 165 | func (g *memGate) AcquireClaimMailGate(ctx context.Context, _ ClaimMailOwner, _ ClaimMailScope) (context.Context, string, error) { |
| 166 | g.mu.Lock() |
| 167 | defer g.mu.Unlock() |
| 168 | if g.State == "held" { |
| 169 | return nil, "", ErrClaimMailPending |
| 170 | } |
| 171 | g.Revision++ |
| 172 | g.State = "held" |
| 173 | g.id = fmt.Sprintf("gate-%d", g.Revision) |
| 174 | return context.WithValue(ctx, memGateKey{}, g.id), g.id, nil |
| 175 | } |
| 176 | func (g *memGate) owns(ctx context.Context) bool { |
| 177 | id, _ := ctx.Value(memGateKey{}).(string) |
| 178 | return g.State == "held" && id == g.id |
| 179 | } |
| 180 | func (g *memGate) CheckClaimMailGate(ctx context.Context) error { |
| 181 | g.mu.Lock() |
| 182 | defer g.mu.Unlock() |
| 183 | if !g.owns(ctx) { |
| 184 | return ErrClaimMailPending |
| 185 | } |
| 186 | return nil |
| 187 | } |
| 188 | func (g *memGate) ReleaseClaimMailGate(ctx context.Context) error { |
| 189 | g.mu.Lock() |
| 190 | defer g.mu.Unlock() |
| 191 | if !g.owns(ctx) { |
| 192 | return ErrClaimMailPending |
| 193 | } |
| 194 | g.Revision++ |
| 195 | g.State = "free" |
| 196 | return nil |
| 197 | } |
| 198 | func (g *memGate) snapshot() memGate { |
| 199 | g.mu.Lock() |
| 200 | defer g.mu.Unlock() |
| 201 | return memGate{State: g.State, id: g.id, Revision: g.Revision} |
| 202 | } |
| 203 | |
| 204 | // fixtureAdmission is the admitter's result: it adds no authority of its own. |
| 205 | type fixtureAdmission struct { |
| 206 | ctx context.Context |
| 207 | released *int |
| 208 | mu *sync.Mutex |
| 209 | } |
| 210 | |
| 211 | func (a fixtureAdmission) Context() context.Context { return a.ctx } |
| 212 | func (a fixtureAdmission) Settle(context.Context, error, []hostops.MailAttempt) error { return nil } |
| 213 | func (a fixtureAdmission) Release(context.Context) error { |
| 214 | a.mu.Lock() |
| 215 | defer a.mu.Unlock() |
| 216 | *a.released++ |
| 217 | return nil |
| 218 | } |
| 219 | |
| 220 | // claimFixtureAdmitter refuses an app whose fixture state is "purging". Each |
| 221 | // app it checks is logged as an admission read so the call-count assertions |
| 222 | // keep their meaning. |
| 223 | type claimFixtureAdmitter struct { |
| 224 | f *claimDomainFixture |
| 225 | mu sync.Mutex |
| 226 | released int |
| 227 | } |
| 228 | |
| 229 | func (a *claimFixtureAdmitter) AdmitClaimMailApps(ctx context.Context, _ ClaimMailHold, apps []ClaimMailApp) (ClaimMailAdmission, error) { |
| 230 | for _, app := range apps { |
| 231 | a.f.log.mu.Lock() |
| 232 | a.f.log.gets = append(a.f.log.gets, "apps/"+app.Account+"/"+app.App+"/admission.json") |
| 233 | a.f.log.mu.Unlock() |
| 234 | if a.f.states[app.App] == "purging" { |
| 235 | return nil, ErrClaimMailPending |
| 236 | } |
| 237 | } |
| 238 | return fixtureAdmission{ctx: ctx, released: &a.released, mu: &a.mu}, nil |
| 239 | } |
| 240 | |
| 241 | func (f *claimDomainFixture) useAdmitter(a ClaimMailAdmitter) { |
| 242 | f.t.Helper() |
| 243 | m, err := NewClaimMailStore(f.log, f.mg, a, claimTestOwner) |
| 244 | if err != nil { |
| 245 | f.t.Fatal(err) |
| 246 | } |
| 247 | f.marked = m |
| 248 | } |
| 249 | |
| 250 | // newClaimDomainFixture seeds one shared domain whose Apps are members, through the |
| 251 | // unmarked store, then opens an admission for alpha, beta and gamma. |
| 252 | func newClaimDomainFixture(t *testing.T, members ...string) *claimDomainFixture { |
| 253 | t.Helper() |
| 254 | ctx := context.Background() |
| 255 | base := release.NewDirStore(t.TempDir()) |
| 256 | mem := NewMemProvisioner("1") |
| 257 | now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC) |
| 258 | for _, app := range members { |
| 259 | if _, _, err := EnsureDomain(ctx, base, mem, dnsrec.NewMemWriter(), testDeployment(), claimTestAccount, app, claimTestDomain, now); err != nil { |
| 260 | t.Fatalf("seed %s: %v", app, err) |
| 261 | } |
| 262 | } |
| 263 | mem.SetIdentityStatus("eu-west-1", claimTestDomain, Status{DKIM: "SUCCESS", MailFrom: "SUCCESS", Sending: "SUCCESS"}) |
| 264 | log := &claimLogStore{Store: base} |
| 265 | f := &claimDomainFixture{t: t, ctx: ctx, base: base, log: log, mem: mem, now: now, mg: &memGate{State: "free"}, states: map[string]string{}, |
| 266 | prov: &countingProv{MemProvisioner: mem, hits: map[string]int{}}} |
| 267 | f.adm = &claimFixtureAdmitter{f: f} |
| 268 | for _, app := range []string{"alpha", "beta", "gamma"} { |
| 269 | f.setAdmission(app, "open") |
| 270 | } |
| 271 | f.useAdmitter(f.adm) |
| 272 | f.reset() |
| 273 | return f |
| 274 | } |
| 275 | func (f *claimDomainFixture) reset() { f.log.reset(); f.prov.reset() } |
| 276 | |
| 277 | // setAdmission sets the state the fixture admitter reports: "open" or |
| 278 | // "purging". |
| 279 | func (f *claimDomainFixture) setAdmission(app, state string) { f.t.Helper(); f.states[app] = state } |
| 280 | func (f *claimDomainFixture) domainRaw() ([]byte, string) { |
| 281 | f.t.Helper() |
| 282 | raw, etag, err := f.base.Get(f.ctx, DomainKey(claimTestAccount, claimTestDomain)) |
| 283 | if err != nil { |
| 284 | f.t.Fatal(err) |
| 285 | } |
| 286 | return raw, etag |
| 287 | } |
| 288 | func (f *claimDomainFixture) requireUntouched(raw []byte, etag string) { |
| 289 | f.t.Helper() |
| 290 | gotRaw, gotEtag := f.domainRaw() |
| 291 | if !bytes.Equal(gotRaw, raw) || gotEtag != etag { |
| 292 | f.t.Fatalf("domain record changed: etag %s -> %s", etag, gotEtag) |
| 293 | } |
| 294 | if n := f.log.putsTo(DomainKey(claimTestAccount, claimTestDomain)); n != 0 { |
| 295 | f.t.Fatalf("domain puts=%d", n) |
| 296 | } |
| 297 | if n := f.prov.total(); n != 0 { |
| 298 | f.t.Fatalf("backend calls=%d %v", n, f.prov.hits) |
| 299 | } |
| 300 | } |
| 301 | func (f *claimDomainFixture) gate() memGate { f.t.Helper(); return f.mg.snapshot() } |
| 302 | |
| 303 | // requireGateFree: one acquire and one release since the gate was first used. |
| 304 | func (f *claimDomainFixture) requireGateFree() { |
| 305 | f.t.Helper() |
| 306 | if g := f.gate(); g.State != "free" || g.Revision != 2 { |
| 307 | f.t.Fatalf("gate not released cleanly: %+v", &g) |
| 308 | } |
| 309 | } |
| 310 | |
| 311 | func TestClaimMailPutDomainPreflightsEveryAffectedApp(t *testing.T) { |
| 312 | cases := []struct { |
| 313 | name string |
| 314 | members []string |
| 315 | purging string |
| 316 | proposed []string |
| 317 | }{ |
| 318 | {"proposal drops the purging member", []string{"alpha", "beta"}, "beta", []string{"alpha"}}, |
| 319 | {"proposal adds a purging app", []string{"alpha"}, "gamma", []string{"alpha", "gamma"}}, |
| 320 | {"proposal is reordered", []string{"alpha", "beta", "gamma"}, "gamma", []string{"gamma", "beta", "alpha"}}, |
| 321 | } |
| 322 | for _, tc := range cases { |
| 323 | t.Run(tc.name, func(t *testing.T) { |
| 324 | f := newClaimDomainFixture(t, tc.members...) |
| 325 | rec, etag, err := GetDomain(f.ctx, f.base, claimTestAccount, claimTestDomain) |
| 326 | if err != nil { |
| 327 | t.Fatal(err) |
| 328 | } |
| 329 | beforeRaw, beforeEtag := f.domainRaw() |
| 330 | f.setAdmission(tc.purging, "purging") |
| 331 | f.reset() |
| 332 | rec.Apps = tc.proposed |
| 333 | if _, err = PutDomain(f.ctx, f.marked, claimTestAccount, rec, etag); !errors.Is(err, ErrClaimMailPending) { |
| 334 | t.Fatalf("PutDomain with a purging affected app: %v", err) |
| 335 | } |
| 336 | f.requireUntouched(beforeRaw, beforeEtag) |
| 337 | f.requireGateFree() |
| 338 | if f.log.admissionGets() == 0 { |
| 339 | t.Fatal("preflight never read admissions") |
| 340 | } |
| 341 | // The gate is free and the refusal was about the purge only: reopening it lets the same write through. |
| 342 | f.setAdmission(tc.purging, "open") |
| 343 | if _, err = PutDomain(f.ctx, f.marked, claimTestAccount, rec, etag); err != nil { |
| 344 | t.Fatalf("retry after refusal: %v", err) |
| 345 | } |
| 346 | if _, after := f.domainRaw(); after == beforeEtag { |
| 347 | t.Fatal("control write did not change the domain record") |
| 348 | } |
| 349 | }) |
| 350 | } |
| 351 | } |
| 352 | |
| 353 | func TestClaimMailVerifyDomainPreflightsEveryMemberBeforeBackend(t *testing.T) { |
| 354 | verify := func(f *claimDomainFixture) (*DomainRecord, error) { |
| 355 | return VerifyDomain(f.ctx, f.marked, f.prov, testDeployment(), claimTestAccount, claimTestDomain, "eu-west-1", f.now) |
| 356 | } |
| 357 | t.Run("one member purging", func(t *testing.T) { |
| 358 | f := newClaimDomainFixture(t, "alpha", "beta", "gamma") |
| 359 | beforeRaw, beforeEtag := f.domainRaw() |
| 360 | f.setAdmission("beta", "purging") |
| 361 | f.reset() |
| 362 | rec, err := verify(f) |
| 363 | if !errors.Is(err, ErrClaimMailPending) || rec != nil { |
| 364 | t.Fatalf("VerifyDomain: rec=%v err=%v", rec, err) |
| 365 | } |
| 366 | f.requireUntouched(beforeRaw, beforeEtag) |
| 367 | f.requireGateFree() |
| 368 | if f.log.admissionGets() == 0 { |
| 369 | t.Fatal("preflight never read admissions") |
| 370 | } |
| 371 | f.setAdmission("beta", "open") |
| 372 | f.reset() |
| 373 | rec, err = verify(f) |
| 374 | if err != nil { |
| 375 | t.Fatalf("VerifyDomain after reopen: %v", err) |
| 376 | } |
| 377 | if rec.Regions["eu-west-1"].Status != StatusVerified { |
| 378 | t.Fatalf("status=%q", rec.Regions["eu-west-1"].Status) |
| 379 | } |
| 380 | if f.prov.total() == 0 { |
| 381 | t.Fatal("control run never reached the backend, so the zero count above proves nothing") |
| 382 | } |
| 383 | }) |
| 384 | t.Run("membership is read fresh at call time", func(t *testing.T) { |
| 385 | f := newClaimDomainFixture(t, "alpha") |
| 386 | // beta joins after the fixture exists and is already purging when the call starts. |
| 387 | if _, _, err := EnsureDomain(f.ctx, f.base, f.mem, dnsrec.NewMemWriter(), testDeployment(), claimTestAccount, "beta", claimTestDomain, f.now); err != nil { |
| 388 | t.Fatal(err) |
| 389 | } |
| 390 | f.mem.SetIdentityStatus("eu-west-1", claimTestDomain, Status{DKIM: "SUCCESS", MailFrom: "SUCCESS", Sending: "SUCCESS"}) |
| 391 | beforeRaw, beforeEtag := f.domainRaw() |
| 392 | f.setAdmission("beta", "purging") |
| 393 | f.reset() |
| 394 | if _, err := verify(f); !errors.Is(err, ErrClaimMailPending) { |
| 395 | t.Fatalf("VerifyDomain: %v", err) |
| 396 | } |
| 397 | f.requireUntouched(beforeRaw, beforeEtag) |
| 398 | f.requireGateFree() |
| 399 | }) |
| 400 | } |
| 401 | |
| 402 | func TestClaimMailDomainWrappersUnmarkedDoNoNewReads(t *testing.T) { |
| 403 | f := newClaimDomainFixture(t, "alpha", "beta") |
| 404 | f.setAdmission("beta", "purging") |
| 405 | rec, etag, err := GetDomain(f.ctx, f.base, claimTestAccount, claimTestDomain) |
| 406 | if err != nil { |
| 407 | t.Fatal(err) |
| 408 | } |
| 409 | f.reset() |
| 410 | // f.log is not a ClaimMailStore: the purging member must not matter. |
| 411 | if _, err = PutDomain(f.ctx, f.log, claimTestAccount, rec, etag); err != nil { |
| 412 | t.Fatalf("unmarked PutDomain: %v", err) |
| 413 | } |
| 414 | if len(f.log.gets) != 0 { |
| 415 | t.Fatalf("unmarked PutDomain read %v", f.log.gets) |
| 416 | } |
| 417 | f.reset() |
| 418 | if _, err = VerifyDomain(f.ctx, f.log, f.prov, testDeployment(), claimTestAccount, claimTestDomain, "eu-west-1", f.now); err != nil { |
| 419 | t.Fatalf("unmarked VerifyDomain: %v", err) |
| 420 | } |
| 421 | for _, k := range f.log.gets { |
| 422 | if strings.HasSuffix(k, "/admission.json") { |
| 423 | t.Fatalf("unmarked VerifyDomain read %s", k) |
| 424 | } |
| 425 | } |
| 426 | if g := f.gate(); g.Revision != 0 { |
| 427 | t.Fatalf("unmarked call used the gate: %+v", &g) |
| 428 | } |
| 429 | } |
| 430 | |
amadan