| 1 | package mail |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "slices" |
| 9 | "sort" |
| 10 | "sync" |
| 11 | "sync/atomic" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/carlosframework/platform/internal/hostops" |
| 15 | "github.com/carlosframework/platform/internal/release" |
| 16 | ) |
| 17 | |
| 18 | const releaseTimeout = 30 * time.Second |
| 19 | |
| 20 | var ErrClaimMailPending = errors.New("mail: lifecycle resource operation pending") |
| 21 | |
| 22 | type ClaimMailOwner struct{ Console, Enrollment, Controller string } |
| 23 | type ClaimMailScope struct{ Account, App, AppID, Operation string } |
| 24 | |
| 25 | // ClaimMailApp names one app a gated mail call can affect. |
| 26 | type ClaimMailApp struct{ Account, App string } |
| 27 | |
| 28 | // ClaimMailHold identifies the held shared gate an admission belongs to. |
| 29 | type ClaimMailHold struct { |
| 30 | Gate string |
| 31 | Owner ClaimMailOwner |
| 32 | } |
| 33 | |
| 34 | // ClaimMailGate is the shared resource gate. The production implementation |
| 35 | // is the admission engine's typed control writer; this package never sees |
| 36 | // the gate's storage key and cannot write it. |
| 37 | type ClaimMailGate interface { |
| 38 | // AcquireClaimMailGate takes the free gate and returns a context that |
| 39 | // carries the lease (the backend and bucket boundaries check it). |
| 40 | AcquireClaimMailGate(ctx context.Context, owner ClaimMailOwner, scope ClaimMailScope) (held context.Context, gateID string, err error) |
| 41 | // CheckClaimMailGate proves held still owns the exact held revision. |
| 42 | CheckClaimMailGate(held context.Context) error |
| 43 | // ReleaseClaimMailGate frees exactly the revision held wrote. |
| 44 | ReleaseClaimMailGate(held context.Context) error |
| 45 | } |
| 46 | |
| 47 | // ClaimMailAdmission is one outermost gated call's durable app authority. |
| 48 | type ClaimMailAdmission interface { |
| 49 | // Context carries a live permit for every admitted app. |
| 50 | Context() context.Context |
| 51 | // Settle runs once, after the call joined: it revokes the permits and |
| 52 | // records each app effect complete (no error, or no attempt reached a |
| 53 | // mutating surface) or uncertain (an error after an attempt). attempts |
| 54 | // is exactly what THIS admission's call reached, never another |
| 55 | // admission's under the same gate (review P2-9); an uncertain effect |
| 56 | // records them for the fenced reconciler. |
| 57 | Settle(ctx context.Context, callErr error, attempts []hostops.MailAttempt) error |
| 58 | // Release runs only after the shared gate's release CAS succeeded. It |
| 59 | // ends each app's mutation operation. An uncertain effect is never |
| 60 | // released; it waits for fenced reconciliation. |
| 61 | Release(ctx context.Context) error |
| 62 | } |
| 63 | |
| 64 | // ClaimMailAdmitter is called at most once per outermost gated call, while |
| 65 | // the gate is held, before the call's first backend, SES, DNS or store |
| 66 | // mutation. apps is complete, sorted and de-duplicated. A refusal leaves |
| 67 | // every admission unchanged. |
| 68 | type ClaimMailAdmitter interface { |
| 69 | AdmitClaimMailApps(ctx context.Context, hold ClaimMailHold, apps []ClaimMailApp) (ClaimMailAdmission, error) |
| 70 | } |
| 71 | |
| 72 | // ClaimMailStore marks the enabled mail path. Its inner store is the |
| 73 | // console's DATA boundary, never a raw bucket. Named private fields only. |
| 74 | type ClaimMailStore struct { |
| 75 | data release.Store |
| 76 | gate ClaimMailGate |
| 77 | admitter ClaimMailAdmitter |
| 78 | owner ClaimMailOwner |
| 79 | } |
| 80 | |
| 81 | var ( |
| 82 | _ release.Store = ClaimMailStore{} |
| 83 | _ release.ConditionalDeleter = ClaimMailStore{} |
| 84 | _ release.Opener = ClaimMailStore{} |
| 85 | _ release.ModLister = ClaimMailStore{} |
| 86 | _ release.PageLister = ClaimMailStore{} |
| 87 | _ release.SizeLister = ClaimMailStore{} |
| 88 | _ release.BucketListDirs = ClaimMailStore{} |
| 89 | ) |
| 90 | |
| 91 | func NewClaimMailStore(data release.Store, gate ClaimMailGate, admitter ClaimMailAdmitter, owner ClaimMailOwner) (ClaimMailStore, error) { |
| 92 | if data == nil || gate == nil || admitter == nil || owner.Console == "" || owner.Enrollment == "" || owner.Controller == "" { |
| 93 | return ClaimMailStore{}, ErrClaimMailPending |
| 94 | } |
| 95 | return ClaimMailStore{data: data, gate: gate, admitter: admitter, owner: owner}, nil |
| 96 | } |
| 97 | |
| 98 | func (s ClaimMailStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 99 | return s.data.Get(ctx, key) |
| 100 | } |
| 101 | func (s ClaimMailStore) List(ctx context.Context, prefix string) ([]string, error) { |
| 102 | return s.data.List(ctx, prefix) |
| 103 | } |
| 104 | func (s ClaimMailStore) ListDirs(ctx context.Context, prefix string) ([]string, error) { |
| 105 | return s.data.ListDirs(ctx, prefix) |
| 106 | } |
| 107 | func (s ClaimMailStore) BucketShapedListDirs() bool { return release.HasBucketListDirs(s.data) } |
| 108 | func (s ClaimMailStore) Open(ctx context.Context, key string) (io.ReadCloser, string, error) { |
| 109 | return release.OpenObject(ctx, s.data, key) |
| 110 | } |
| 111 | func (s ClaimMailStore) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) { |
| 112 | if l, ok := s.data.(release.ModLister); ok { |
| 113 | return l.ListMod(ctx, prefix) |
| 114 | } |
| 115 | return nil, ErrClaimMailPending |
| 116 | } |
| 117 | func (s ClaimMailStore) ListPage(ctx context.Context, prefix, after string, limit int) ([]string, error) { |
| 118 | if l, ok := s.data.(release.PageLister); ok { |
| 119 | return l.ListPage(ctx, prefix, after, limit) |
| 120 | } |
| 121 | return nil, ErrClaimMailPending |
| 122 | } |
| 123 | func (s ClaimMailStore) ListSizes(ctx context.Context, prefix string) ([]release.KeySize, error) { |
| 124 | if l, ok := s.data.(release.SizeLister); ok { |
| 125 | return l.ListSizes(ctx, prefix) |
| 126 | } |
| 127 | return nil, ErrClaimMailPending |
| 128 | } |
| 129 | func (s ClaimMailStore) Put(ctx context.Context, key string, body []byte, tag string) (string, error) { |
| 130 | etag, err := s.data.Put(ctx, key, body, tag) |
| 131 | markClaimMailEffect(ctx, hostops.MailAttempt{Surface: "store", Method: "Put", Key: key}, err) |
| 132 | return etag, err |
| 133 | } |
| 134 | func (s ClaimMailStore) Delete(ctx context.Context, key string) error { |
| 135 | err := s.data.Delete(ctx, key) |
| 136 | markClaimMailEffect(ctx, hostops.MailAttempt{Surface: "store", Method: "Delete", Key: key}, err) |
| 137 | return err |
| 138 | } |
| 139 | func (s ClaimMailStore) DeleteIfMatch(ctx context.Context, key, etag string) error { |
| 140 | err := release.DeleteIfMatch(ctx, s.data, key, etag) |
| 141 | markClaimMailEffect(ctx, hostops.MailAttempt{Surface: "store", Method: "DeleteIfMatch", Key: key}, err) |
| 142 | return err |
| 143 | } |
| 144 | |
| 145 | type claimMailToken struct{} |
| 146 | type claimMailScopeKey struct{} |
| 147 | type claimMailCapsKey struct{} |
| 148 | type claimMailAttemptsKey struct{} |
| 149 | type claimMailCaps map[ClaimMailApp]struct{} |
| 150 | |
| 151 | type claimMailHeld struct { |
| 152 | ID string |
| 153 | Owner ClaimMailOwner |
| 154 | // Unadmitted: a mutating surface was reached under the gate outside any |
| 155 | // admission (no durable effect records it). The gate is then retained |
| 156 | // on failure, since nothing else can say what happened. |
| 157 | Unadmitted atomic.Bool |
| 158 | Retain atomic.Bool // an effect is unsettled: keep the gate |
| 159 | mu sync.Mutex |
| 160 | releases []func(context.Context) error |
| 161 | } |
| 162 | |
| 163 | // claimMailAttempts is ONE admission's record of the surfaces its call |
| 164 | // reached. It travels in the admitted context, so nested gated calls add to |
| 165 | // the outermost admission and a later admission starts empty. |
| 166 | type claimMailAttempts struct { |
| 167 | mu sync.Mutex |
| 168 | list []hostops.MailAttempt |
| 169 | err error |
| 170 | } |
| 171 | |
| 172 | func (a *claimMailAttempts) snapshot() ([]hostops.MailAttempt, error) { |
| 173 | a.mu.Lock() |
| 174 | defer a.mu.Unlock() |
| 175 | return slices.Clone(a.list), a.err |
| 176 | } |
| 177 | |
| 178 | // markClaimMailDirty records an attempt on an unnamed surface (kept for the |
| 179 | // existing claim_gate_test.go fixture). Outside an admission it marks the |
| 180 | // hold Unadmitted; inside one it makes that admission's failure uncertain |
| 181 | // with an attempt the reconciler cannot read back. |
| 182 | func markClaimMailDirty(ctx context.Context) { |
| 183 | markClaimMailEffect(ctx, hostops.MailAttempt{Surface: "unknown", Method: "unknown"}, nil) |
| 184 | } |
| 185 | |
| 186 | // markClaimMailEffect records that a mutation reached its surface, unless a |
| 187 | // boundary refused it before any effect. An unknown error may have acted. |
| 188 | func markClaimMailEffect(ctx context.Context, at hostops.MailAttempt, err error) { |
| 189 | if err != nil && hostops.RefusedBeforeEffect(err) { |
| 190 | return |
| 191 | } |
| 192 | if rec, ok := ctx.Value(claimMailAttemptsKey{}).(*claimMailAttempts); ok { |
| 193 | rec.mu.Lock() |
| 194 | rec.list = append(rec.list, at) |
| 195 | rec.err = errors.Join(rec.err, err) |
| 196 | rec.mu.Unlock() |
| 197 | return |
| 198 | } |
| 199 | if held, ok := ctx.Value(claimMailToken{}).(*claimMailHeld); ok { |
| 200 | held.Unadmitted.Store(true) |
| 201 | } |
| 202 | } |
| 203 | |
| 204 | func WithClaimMailScope(ctx context.Context, scope ClaimMailScope) context.Context { |
| 205 | return context.WithValue(ctx, claimMailScopeKey{}, scope) |
| 206 | } |
| 207 | func ClaimMailScopeFrom(ctx context.Context) (ClaimMailScope, bool) { |
| 208 | scope, ok := ctx.Value(claimMailScopeKey{}).(ClaimMailScope) |
| 209 | return scope, ok |
| 210 | } |
| 211 | |
| 212 | // claimMailPlan says which apps one gated call can affect. Apps come from the |
| 213 | // arguments (the proposed members). Domain adds the persisted members of |
| 214 | // that domain (fresh read) and the app named on this account's global claim |
| 215 | // for it. AppDomains adds the same for every domain that app refers to. |
| 216 | // |
| 217 | // The claim's app is part of the persisted set because the claim is what |
| 218 | // the backend boundary binds an SES identity and its DNS to, and the claim |
| 219 | // write is scoped to it: an EnsureDomain re-run by a sibling app, a |
| 220 | // VerifyDomain stamp or a RemoveDomain release all touch it. Resolving it |
| 221 | // here admits it before any SES, DNS or store call, instead of letting the |
| 222 | // boundary refuse halfway through a walk. A claim held by another account |
| 223 | // adds nothing: the walk refuses or lapses it, and a takeover's write needs |
| 224 | // that account's authority, which no admission here grants. |
| 225 | // |
| 226 | // Refresh adds what a downstream RefreshPolicies or EnsureUser will need. A |
| 227 | // policy is written from the app's granted domains, and the backend boundary |
| 228 | // authorizes each granted domain against ALL its members. So every verified |
| 229 | // domain naming a refreshed app (and Domain itself, which may verify during |
| 230 | // the call) contributes its members and its claim app, and those members are |
| 231 | // refreshed in turn: the closure is the TRANSITIVE fixed point across shared |
| 232 | // domains (D1/A, D2/A+B, D3/B+C admits A, B and C). An unverified domain other |
| 233 | // than Domain grants nothing and adds nothing. Without it a nested call |
| 234 | // refuses after the SES and domain effects of the outer call, retaining the |
| 235 | // shared gate. |
| 236 | type claimMailPlan struct { |
| 237 | Apps []string |
| 238 | Domain string |
| 239 | AppDomains string |
| 240 | Refresh bool |
| 241 | } |
| 242 | |
| 243 | func claimPutDomainPlan(rec *DomainRecord) claimMailPlan { |
| 244 | if rec == nil { |
| 245 | return claimMailPlan{} |
| 246 | } |
| 247 | return claimMailPlan{Apps: rec.Apps, Domain: rec.Domain} |
| 248 | } |
| 249 | |
| 250 | func (p claimMailPlan) resolve(ctx context.Context, s release.Store, acct string) ([]ClaimMailApp, error) { |
| 251 | if !release.ValidAccount(acct) { |
| 252 | return nil, fmt.Errorf("mail: invalid account %q", acct) |
| 253 | } |
| 254 | set := map[string]struct{}{} |
| 255 | for _, app := range p.Apps { |
| 256 | set[app] = struct{}{} |
| 257 | } |
| 258 | claimApp := func(domain string) error { |
| 259 | c, err := getDomainClaim(ctx, s, domain) |
| 260 | if err != nil { |
| 261 | return err |
| 262 | } |
| 263 | if c != nil && c.Account == acct && c.App != "" { |
| 264 | set[c.App] = struct{}{} |
| 265 | } |
| 266 | return nil |
| 267 | } |
| 268 | if p.Domain != "" { |
| 269 | if !ValidDomain(p.Domain) { |
| 270 | return nil, fmt.Errorf("mail: invalid domain %q", p.Domain) |
| 271 | } |
| 272 | rec, _, err := GetDomain(ctx, s, acct, p.Domain) |
| 273 | if err != nil { |
| 274 | return nil, err |
| 275 | } |
| 276 | if rec != nil { |
| 277 | for _, app := range rec.Apps { |
| 278 | set[app] = struct{}{} |
| 279 | } |
| 280 | } |
| 281 | if err = claimApp(p.Domain); err != nil { |
| 282 | return nil, err |
| 283 | } |
| 284 | } |
| 285 | if p.AppDomains != "" { |
| 286 | // A complete listing: a malformed record aborts the census. |
| 287 | recs, err := StrictListDomains(ctx, s, acct) |
| 288 | if err != nil { |
| 289 | return nil, err |
| 290 | } |
| 291 | for _, rec := range recs { |
| 292 | if slices.Contains(rec.Apps, p.AppDomains) { |
| 293 | for _, app := range rec.Apps { |
| 294 | set[app] = struct{}{} |
| 295 | } |
| 296 | if err = claimApp(rec.Domain); err != nil { |
| 297 | return nil, err |
| 298 | } |
| 299 | } |
| 300 | } |
| 301 | } |
| 302 | if p.Refresh { |
| 303 | refreshed := map[string]struct{}{} |
| 304 | for _, app := range p.Apps { |
| 305 | refreshed[app] = struct{}{} |
| 306 | } |
| 307 | if p.Domain != "" { |
| 308 | rec, _, err := GetDomain(ctx, s, acct, p.Domain) |
| 309 | if err != nil { |
| 310 | return nil, err |
| 311 | } |
| 312 | if rec != nil { |
| 313 | for _, app := range rec.Apps { |
| 314 | refreshed[app] = struct{}{} |
| 315 | } |
| 316 | } |
| 317 | } |
| 318 | // A complete listing: an unreadable or malformed record aborts. |
| 319 | recs, err := StrictListDomains(ctx, s, acct) |
| 320 | if err != nil { |
| 321 | return nil, err |
| 322 | } |
| 323 | // Fixed point: a domain in the closure pulls all its members in, and a |
| 324 | // newly pulled member can in turn use another relevant domain, so |
| 325 | // repeat until a pass adds no app. Each domain is taken at most once. |
| 326 | taken := map[string]struct{}{} |
| 327 | for changed := true; changed; { |
| 328 | changed = false |
| 329 | for _, rec := range recs { |
| 330 | if _, done := taken[rec.Domain]; done { |
| 331 | continue |
| 332 | } |
| 333 | if len(rec.VerifiedARNs()) == 0 && rec.Domain != p.Domain { |
| 334 | continue |
| 335 | } |
| 336 | touched := false |
| 337 | for _, app := range rec.Apps { |
| 338 | if _, ok := refreshed[app]; ok { |
| 339 | touched = true |
| 340 | break |
| 341 | } |
| 342 | } |
| 343 | if !touched { |
| 344 | continue |
| 345 | } |
| 346 | taken[rec.Domain] = struct{}{} |
| 347 | for _, app := range rec.Apps { |
| 348 | set[app] = struct{}{} |
| 349 | if _, ok := refreshed[app]; !ok { |
| 350 | refreshed[app] = struct{}{} |
| 351 | changed = true |
| 352 | } |
| 353 | } |
| 354 | if err = claimApp(rec.Domain); err != nil { |
| 355 | return nil, err |
| 356 | } |
| 357 | } |
| 358 | } |
| 359 | } |
| 360 | names := make([]string, 0, len(set)) |
| 361 | for app := range set { |
| 362 | if !release.ValidApp(app) { |
| 363 | return nil, fmt.Errorf("mail: invalid app %q", app) |
| 364 | } |
| 365 | names = append(names, app) |
| 366 | } |
| 367 | sort.Strings(names) |
| 368 | out := make([]ClaimMailApp, len(names)) |
| 369 | for i, app := range names { |
| 370 | out[i] = ClaimMailApp{Account: acct, App: app} |
| 371 | } |
| 372 | return out, nil |
| 373 | } |
| 374 | |
| 375 | // claimMailAdmitted must run inside WithClaimMailResource on a marked store. |
| 376 | func claimMailAdmitted(ctx context.Context, s release.Store, acct string, plan claimMailPlan, call func(context.Context) error) error { |
| 377 | held, ok := ctx.Value(claimMailToken{}).(*claimMailHeld) |
| 378 | if !ok { |
| 379 | return ErrClaimMailPending |
| 380 | } |
| 381 | marked, ok := s.(ClaimMailStore) |
| 382 | if !ok || marked.admitter == nil { |
| 383 | return ErrClaimMailPending |
| 384 | } |
| 385 | apps, err := plan.resolve(ctx, s, acct) |
| 386 | if err != nil { |
| 387 | return err |
| 388 | } |
| 389 | if installed, nested := ctx.Value(claimMailCapsKey{}).(claimMailCaps); nested { |
| 390 | // Nested: only the preflighted set is usable. No admission read, no |
| 391 | // second hook call, no widening. |
| 392 | for _, a := range apps { |
| 393 | if _, covered := installed[a]; !covered { |
| 394 | return ErrClaimMailPending |
| 395 | } |
| 396 | } |
| 397 | return call(ctx) |
| 398 | } |
| 399 | if len(apps) == 0 { |
| 400 | return ErrClaimMailPending |
| 401 | } |
| 402 | caps := claimMailCaps{} |
| 403 | for _, a := range apps { |
| 404 | caps[a] = struct{}{} |
| 405 | } |
| 406 | admission, err := marked.admitter.AdmitClaimMailApps(ctx, ClaimMailHold{Gate: held.ID, Owner: held.Owner}, slices.Clone(apps)) |
| 407 | if err != nil { |
| 408 | return err |
| 409 | } |
| 410 | if admission == nil || admission.Context() == nil { |
| 411 | return ErrClaimMailPending |
| 412 | } |
| 413 | // The admitted context descends from ctx, so it still carries the held |
| 414 | // gate token and the gate lease. |
| 415 | rec := &claimMailAttempts{} |
| 416 | admitted := context.WithValue(context.WithValue(admission.Context(), claimMailCapsKey{}, caps), claimMailAttemptsKey{}, rec) |
| 417 | result := call(admitted) |
| 418 | attempts, effectErr := rec.snapshot() |
| 419 | result = errors.Join(result, effectErr) |
| 420 | if err := admission.Settle(ctx, result, attempts); err != nil { |
| 421 | // An unsettled effect retains both the app authority and the gate. |
| 422 | held.Retain.Store(true) |
| 423 | return errors.Join(result, err) |
| 424 | } |
| 425 | if result != nil && len(attempts) != 0 { |
| 426 | held.Retain.Store(true) |
| 427 | return result |
| 428 | } |
| 429 | held.mu.Lock() |
| 430 | held.releases = append(held.releases, admission.Release) |
| 431 | held.mu.Unlock() |
| 432 | return result |
| 433 | } |
| 434 | |
| 435 | // WithClaimMailResource serializes one enabled mail mutation under the |
| 436 | // shared gate. An unmarked store runs the original body with no gate, |
| 437 | // admission or domain read. The gate is released only when no effect is |
| 438 | // retained; the app authorities are released only after that release CAS. |
| 439 | func WithClaimMailResource(ctx context.Context, s release.Store, call func(context.Context) error) error { |
| 440 | marked, enabled := s.(ClaimMailStore) |
| 441 | if !enabled { |
| 442 | return call(ctx) |
| 443 | } |
| 444 | if marked.gate == nil || marked.admitter == nil { |
| 445 | return ErrClaimMailPending |
| 446 | } |
| 447 | if nested, ok := ctx.Value(claimMailToken{}).(*claimMailHeld); ok { |
| 448 | if nested.Owner != marked.owner { |
| 449 | return ErrClaimMailPending |
| 450 | } |
| 451 | if err := marked.gate.CheckClaimMailGate(ctx); err != nil { |
| 452 | return err |
| 453 | } |
| 454 | return call(ctx) |
| 455 | } |
| 456 | scope, _ := ClaimMailScopeFrom(ctx) |
| 457 | heldCtx, gateID, err := marked.gate.AcquireClaimMailGate(ctx, marked.owner, scope) |
| 458 | if err != nil { |
| 459 | return err |
| 460 | } |
| 461 | held := &claimMailHeld{ID: gateID, Owner: marked.owner} |
| 462 | result := call(context.WithValue(heldCtx, claimMailToken{}, held)) |
| 463 | if held.Retain.Load() || (result != nil && held.Unadmitted.Load()) { |
| 464 | // Retained for fenced reconciliation; no timer or heartbeat frees it. |
| 465 | return result |
| 466 | } |
| 467 | saveCtx, cancel := context.WithTimeout(context.WithoutCancel(heldCtx), releaseTimeout) |
| 468 | defer cancel() |
| 469 | if err = marked.gate.ReleaseClaimMailGate(saveCtx); err != nil { |
| 470 | // The gate stays held, so its app authorities stay held too. |
| 471 | return errors.Join(result, err) |
| 472 | } |
| 473 | held.mu.Lock() |
| 474 | releases := held.releases |
| 475 | held.mu.Unlock() |
| 476 | var errs []error |
| 477 | for _, end := range releases { |
| 478 | errs = append(errs, end(saveCtx)) |
| 479 | } |
| 480 | return errors.Join(append(errs, result)...) |
| 481 | } |
| 482 | |
amadan