| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/hostops" |
| 11 | "github.com/carlosframework/platform/internal/mail" |
| 12 | "github.com/carlosframework/platform/internal/release" |
| 13 | ) |
| 14 | |
| 15 | // claimMailGate adapts the typed control writer's shared gate to the mail |
| 16 | // package. The lease travels in the context; possession is its token. |
| 17 | type claimMailGate struct{ w *hostops.ControlWriter } |
| 18 | |
| 19 | var _ mail.ClaimMailGate = claimMailGate{} |
| 20 | |
| 21 | func (g claimMailGate) AcquireClaimMailGate(ctx context.Context, owner mail.ClaimMailOwner, scope mail.ClaimMailScope) (context.Context, string, error) { |
| 22 | l, err := g.w.AcquireMailGate(ctx, hostops.MailGateOwner(owner), hostops.MailGateScope(scope)) |
| 23 | if err != nil { |
| 24 | if errors.Is(err, hostops.ErrPending) { |
| 25 | return nil, "", errors.Join(mail.ErrClaimMailPending, err) |
| 26 | } |
| 27 | return nil, "", err |
| 28 | } |
| 29 | return hostops.WithMailGateLease(ctx, l), l.ID, nil |
| 30 | } |
| 31 | func (g claimMailGate) CheckClaimMailGate(held context.Context) error { |
| 32 | l, ok := hostops.MailGateLeaseFrom(held) |
| 33 | if !ok { |
| 34 | return mail.ErrClaimMailPending |
| 35 | } |
| 36 | return g.w.CheckMailGate(held, l) |
| 37 | } |
| 38 | func (g claimMailGate) ReleaseClaimMailGate(held context.Context) error { |
| 39 | l, ok := hostops.MailGateLeaseFrom(held) |
| 40 | if !ok { |
| 41 | return mail.ErrClaimMailPending |
| 42 | } |
| 43 | return g.w.ReleaseMailGate(held, l) |
| 44 | } |
| 45 | |
| 46 | // claimMailAdmitter is the required production admission provider. Each |
| 47 | // affected app gets a durable effect under a NON-EXPIRING operation: its |
| 48 | // own app-mutation operation, or the purge operation named by the gate |
| 49 | // scope for the purging app itself. A failed call after an effect leaves |
| 50 | // that effect uncertain and the operation held, so purge cannot pass it |
| 51 | // (review P2-4). Operations end only after the gate release CAS. |
| 52 | type claimMailAdmitter struct { |
| 53 | localAppID func(context.Context, string, string) (string, error) |
| 54 | data release.Store |
| 55 | control *hostops.ControlWriter |
| 56 | owner hostops.RegistrationOwner |
| 57 | } |
| 58 | |
| 59 | var _ mail.ClaimMailAdmitter = claimMailAdmitter{} |
| 60 | |
| 61 | type claimMailMember struct { |
| 62 | handle *hostops.OperationHandle |
| 63 | began bool // this admission began the operation and must end it |
| 64 | settle func(string, error) error |
| 65 | } |
| 66 | |
| 67 | type claimMailAdmission struct { |
| 68 | ctx context.Context |
| 69 | members []claimMailMember |
| 70 | settled bool |
| 71 | failed bool |
| 72 | } |
| 73 | |
| 74 | func (m claimMailAdmitter) AdmitClaimMailApps(ctx context.Context, hold mail.ClaimMailHold, apps []mail.ClaimMailApp) (mail.ClaimMailAdmission, error) { |
| 75 | if m.control == nil || m.data == nil || m.localAppID == nil || hold.Gate == "" || len(apps) == 0 { |
| 76 | return nil, mail.ErrClaimMailPending |
| 77 | } |
| 78 | lease, ok := hostops.MailGateLeaseFrom(ctx) |
| 79 | if !ok || lease.ID != hold.Gate { |
| 80 | return nil, mail.ErrClaimMailPending |
| 81 | } |
| 82 | scope, scoped := mail.ClaimMailScopeFrom(ctx) |
| 83 | // Resolve every local incarnation before recording any member or admitting |
| 84 | // any app. Recovery of the exact purge member uses its journal-bound ID. |
| 85 | expected := make(map[mail.ClaimMailApp]string, len(apps)) |
| 86 | for _, app := range apps { |
| 87 | var id string |
| 88 | var err error |
| 89 | if scoped && scope.Account == app.Account && scope.App == app.App && scope.Operation != "" { |
| 90 | id = scope.AppID |
| 91 | } else { |
| 92 | id, err = m.localAppID(ctx, app.Account, app.App) |
| 93 | } |
| 94 | if err != nil { |
| 95 | return nil, err |
| 96 | } |
| 97 | if id == "" { |
| 98 | return nil, hostops.ErrIdentity |
| 99 | } |
| 100 | if _, err = hostops.ReadAdmission(ctx, m.data, app.Account, app.App, id); err != nil { |
| 101 | return nil, err |
| 102 | } |
| 103 | expected[app] = id |
| 104 | } |
| 105 | adm := &claimMailAdmission{ctx: ctx} |
| 106 | abandon := func(cause error) (mail.ClaimMailAdmission, error) { |
| 107 | // Nothing was called yet: every issued effect settles "not-applied" |
| 108 | // and every operation this admission began ends. |
| 109 | var errs []error |
| 110 | for i := len(adm.members) - 1; i >= 0; i-- { |
| 111 | mem := adm.members[i] |
| 112 | if mem.settle != nil { |
| 113 | if err := mem.settle("", hostops.ErrNotApplied); err != nil && !errors.Is(err, hostops.ErrNotApplied) { |
| 114 | errs = append(errs, err) |
| 115 | } |
| 116 | } |
| 117 | if mem.began { |
| 118 | if err := mem.handle.FinishMutation(context.WithoutCancel(ctx)); err != nil { |
| 119 | errs = append(errs, err) |
| 120 | } |
| 121 | } |
| 122 | } |
| 123 | return nil, errors.Join(append(errs, cause)...) |
| 124 | } |
| 125 | for _, app := range apps { |
| 126 | a, err := hostops.ReadAdmission(ctx, m.data, app.Account, app.App, expected[app]) |
| 127 | if err != nil { |
| 128 | return abandon(err) |
| 129 | } |
| 130 | appID := a.Value.AppID |
| 131 | var mem claimMailMember |
| 132 | purgeMember := scoped && scope.Account == app.Account && scope.App == app.App && scope.AppID == appID && scope.Operation != "" |
| 133 | // Review P1-2: an app-mutation member's effect ID IS its operation ID, |
| 134 | // so RecoverMutation finds it. A purge member's effect is named by the |
| 135 | // gate (the purge op owns its own ID). |
| 136 | var opID, effectID string |
| 137 | if purgeMember { |
| 138 | opID, effectID = scope.Operation, "mail-"+hold.Gate |
| 139 | } else { |
| 140 | if opID, err = claimEffectID(); err != nil { |
| 141 | return abandon(err) |
| 142 | } |
| 143 | effectID = opID |
| 144 | } |
| 145 | // Durable membership BEFORE the op or effect exists: the gate |
| 146 | // reconciler can then find every effect this hold could issue. |
| 147 | if err = m.control.RecordMailGateMember(ctx, lease, hostops.MailGateMember{Account: app.Account, App: app.App, AppID: appID, Operation: opID, Effect: effectID}); err != nil { |
| 148 | return abandon(err) |
| 149 | } |
| 150 | if purgeMember { |
| 151 | if mem.handle, err = m.control.ResumeExclusive(ctx, app.Account, app.App, appID, opID, hostops.OpPurging, m.owner); err != nil { |
| 152 | return abandon(err) |
| 153 | } |
| 154 | } else { |
| 155 | op := hostops.AppOperation{ID: opID, Kind: hostops.OpAppMutation, AppID: appID, Phase: "issued", Intent: map[string]string{"action": "mail", "gate": hold.Gate}} |
| 156 | if mem.handle, err = m.control.BeginExclusive(ctx, app.Account, app.App, appID, op, m.owner); err != nil { |
| 157 | return abandon(errors.Join(mail.ErrClaimMailPending, err)) |
| 158 | } |
| 159 | mem.began = true |
| 160 | } |
| 161 | adm.members = append(adm.members, mem) |
| 162 | issued, settle, err := mem.handle.Runner(m.now).Issue(adm.ctx, hostops.Effect{ID: effectID, Resource: "mail-gate:" + hold.Gate, Action: "mail", Inputs: map[string]string{"gate": hold.Gate}}) |
| 163 | if err != nil { |
| 164 | return abandon(err) |
| 165 | } |
| 166 | adm.members[len(adm.members)-1].settle = settle |
| 167 | adm.ctx = issued // permits accumulate across the whole app set |
| 168 | } |
| 169 | return adm, nil |
| 170 | } |
| 171 | |
| 172 | func (m claimMailAdmitter) now() time.Time { return time.Now() } |
| 173 | |
| 174 | func (a *claimMailAdmission) Context() context.Context { return a.ctx } |
| 175 | |
| 176 | // Settle records this admission's outcome only. attempts are the surfaces |
| 177 | // THIS admission's call reached (review P2-9); a failure after any attempt |
| 178 | // settles uncertain WITH that record, which is the reconciler's readback |
| 179 | // list. No caller-supplied boolean decides anything. |
| 180 | func (a *claimMailAdmission) Settle(ctx context.Context, callErr error, attempts []hostops.MailAttempt) error { |
| 181 | if a.settled { |
| 182 | return hostops.ErrContract |
| 183 | } |
| 184 | a.settled = true |
| 185 | result, cause := "joined", error(nil) |
| 186 | switch { |
| 187 | case callErr == nil: |
| 188 | case len(attempts) == 0: |
| 189 | cause = fmt.Errorf("%w: %v", hostops.ErrNotApplied, callErr) |
| 190 | default: |
| 191 | cause = callErr |
| 192 | result = hostops.MailAttemptsResult(attempts) |
| 193 | a.failed = true |
| 194 | } |
| 195 | var errs []error |
| 196 | for i := len(a.members) - 1; i >= 0; i-- { |
| 197 | err := a.members[i].settle(result, cause) |
| 198 | if err != nil && !errors.Is(err, hostops.ErrNotApplied) && (cause == nil || !errors.Is(err, cause)) { |
| 199 | errs = append(errs, err) |
| 200 | a.failed = true |
| 201 | } |
| 202 | } |
| 203 | return errors.Join(errs...) |
| 204 | } |
| 205 | |
| 206 | // Release is called only after the gate release CAS succeeded. |
| 207 | func (a *claimMailAdmission) Release(ctx context.Context) error { |
| 208 | if !a.settled || a.failed { |
| 209 | return mail.ErrClaimMailPending |
| 210 | } |
| 211 | var errs []error |
| 212 | for _, mem := range a.members { |
| 213 | if mem.began { |
| 214 | errs = append(errs, mem.handle.FinishMutation(ctx)) |
| 215 | } |
| 216 | } |
| 217 | return errors.Join(errs...) |
| 218 | } |
| 219 | |
| 220 | // claimMailStore is the only store enabled mail mutations receive. Flag-off |
| 221 | // returns the original handle. A misconfigured enabled console gets a store |
| 222 | // that refuses every call, never the bare data handle. |
| 223 | func (s *Server) claimMailStore() release.Store { |
| 224 | if !s.cfg.HostClaim { |
| 225 | return s.cfg.Releases |
| 226 | } |
| 227 | if s.cfg.ClaimAuthority == nil || s.cfg.ClaimAuthority.Control == nil || s.cfg.ClaimPurge == nil { |
| 228 | return refusingStore{hostops.ErrContract} |
| 229 | } |
| 230 | o := s.cfg.ClaimPurge.Owner |
| 231 | admitter := claimMailAdmitter{data: s.cfg.Releases, control: s.cfg.ClaimAuthority.Control, owner: o, localAppID: s.mutationAppID} |
| 232 | store, err := mail.NewClaimMailStore(s.cfg.Releases, claimMailGate{w: s.cfg.ClaimAuthority.Control}, admitter, |
| 233 | mail.ClaimMailOwner{Console: o.Console, Enrollment: o.Enrollment, Controller: o.Controller}) |
| 234 | if err != nil { |
| 235 | return refusingStore{err} |
| 236 | } |
| 237 | return store |
| 238 | } |
| 239 | |
| 240 | // claimMailReconcileMinHeld only avoids fencing a holder that is probably |
| 241 | // still inside its call. It is not a safety bound: issued effects are never |
| 242 | // settled by ReconcileMailGate whatever the age. |
| 243 | const claimMailReconcileMinHeld = 10 * time.Minute |
| 244 | |
| 245 | // claimMailIdentityReader is the reconciler's typed SES readback: a READ |
| 246 | // through the console's mail provisioner (IdentityStatus is unguarded). |
| 247 | type claimMailIdentityReader struct{ p mail.Provisioner } |
| 248 | |
| 249 | var _ hostops.MailIdentityReader = claimMailIdentityReader{} |
| 250 | |
| 251 | func (r claimMailIdentityReader) MailIdentityState(ctx context.Context, region, domain string) (json.RawMessage, error) { |
| 252 | if r.p == nil { |
| 253 | return nil, hostops.ErrMailUnproven |
| 254 | } |
| 255 | st, err := r.p.IdentityStatus(ctx, region, domain) |
| 256 | if err != nil { |
| 257 | return nil, err |
| 258 | } |
| 259 | return json.Marshal(st) |
| 260 | } |
| 261 | |
| 262 | // ReconcileClaimMailGate is the operator/sweep entry for a retained gate |
| 263 | // (review P1-2): fence, read back, settle, free; then end the member |
| 264 | // app-mutation operations through RecoverMutation (gate no longer held). |
| 265 | // ErrMailUnproven names surfaces with no readback; the gate stays held. |
| 266 | func (s *Server) ReconcileClaimMailGate(ctx context.Context, gateID string) (hostops.MailGateReconciliation, error) { |
| 267 | if !s.cfg.HostClaim || s.cfg.ClaimAuthority == nil || s.cfg.ClaimAuthority.Control == nil { |
| 268 | return hostops.MailGateReconciliation{}, hostops.ErrContract |
| 269 | } |
| 270 | w := s.cfg.ClaimAuthority.Control |
| 271 | rec, err := w.ReconcileMailGate(ctx, gateID, claimMailReconcileMinHeld, claimMailIdentityReader{p: s.cfg.MailProvisioner}) |
| 272 | if err != nil { |
| 273 | return rec, err |
| 274 | } |
| 275 | var errs []error |
| 276 | for _, m := range rec.Members { |
| 277 | if m.Operation != m.Effect { |
| 278 | continue // a purge member: the purge op continues on its own |
| 279 | } |
| 280 | if err := w.RecoverMutation(ctx, m.Account, m.App, m.AppID, m.Operation); err != nil && !errors.Is(err, hostops.ErrPending) { |
| 281 | errs = append(errs, err) |
| 282 | } |
| 283 | } |
| 284 | return rec, errors.Join(errs...) |
| 285 | } |
| 286 | |
| 287 | type refusingStore struct{ err error } |
| 288 | |
| 289 | func (r refusingStore) Get(context.Context, string) ([]byte, string, error) { return nil, "", r.err } |
| 290 | func (r refusingStore) Put(context.Context, string, []byte, string) (string, error) { return "", r.err } |
| 291 | func (r refusingStore) List(context.Context, string) ([]string, error) { return nil, r.err } |
| 292 | func (r refusingStore) ListDirs(context.Context, string) ([]string, error) { return nil, r.err } |
| 293 | func (r refusingStore) Delete(context.Context, string) error { return r.err } |
| 294 | |
amadan