| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "os" |
| 7 | |
| 8 | "github.com/carlosframework/platform/internal/hostops" |
| 9 | "github.com/carlosframework/platform/internal/relay" |
| 10 | "github.com/carlosframework/platform/internal/release" |
| 11 | ) |
| 12 | |
| 13 | // The relay's app gate. Every per-app relay walk (enable, disable, adopted, |
| 14 | // reissue, converge, mark done) runs its whole body inside one app mutation |
| 15 | // of that app, and every bucket write below it uses the permitted context. |
| 16 | // The public handlers have already authenticated and authorized the caller; |
| 17 | // these helpers only open or join the operation. With HostClaim off the |
| 18 | // deployment bucket implements no release.AppMutationRunner, so each helper |
| 19 | // runs its callback unchanged. |
| 20 | |
| 21 | // relayAppMutation runs fn as an app mutation of acct/app. A caller already |
| 22 | // holding a live permit for the same app joins it. |
| 23 | func (s *Server) relayAppMutation(ctx context.Context, acct, app, action string, fn func(context.Context) error) error { |
| 24 | return release.WithAppMutation(ctx, s.cfg.Releases, acct, app, action, fn) |
| 25 | } |
| 26 | |
| 27 | // relayGated reports whether the deployment bucket is the enabled boundary. |
| 28 | func (s *Server) relayGated() bool { |
| 29 | _, ok := s.cfg.Releases.(release.AppMutationRunner) |
| 30 | return ok |
| 31 | } |
| 32 | |
| 33 | // relayWithScopes acquires (or joins) an app mutation for every scope, in |
| 34 | // the sorted, deduplicated order hostops.RelayAuthorityScopes returns, and |
| 35 | // runs fn under the innermost context only after all of them are held. A |
| 36 | // failed acquisition returns before fn, so nothing was issued and every |
| 37 | // enclosing operation may release as unapplied. |
| 38 | func (s *Server) relayWithScopes(ctx context.Context, scopes []hostops.AppScope, action string, fn func(context.Context) error) error { |
| 39 | if len(scopes) == 0 { |
| 40 | return fn(ctx) |
| 41 | } |
| 42 | first := scopes[0] |
| 43 | return s.relayAppMutation(ctx, first.Account, first.App, action, func(ctx context.Context) error { |
| 44 | return s.relayWithScopes(ctx, scopes[1:], action, fn) |
| 45 | }) |
| 46 | } |
| 47 | |
| 48 | // relayPublishGated is relayPublishOnce's write on the enabled boundary. It |
| 49 | // reads the aggregate at the caller's expected etag, computes exactly which |
| 50 | // app entries the new body adds, removes or changes, acquires a permit for |
| 51 | // each, and then writes with that same etag. An aggregate that moved since |
| 52 | // the caller read its etag is a conflict before any permit is taken. A body |
| 53 | // whose every entry is identical needs no permit: the boundary classifies it |
| 54 | // as the written_at-only rewrite it is. |
| 55 | func (s *Server) relayPublishGated(ctx context.Context, etag string, body []byte) error { |
| 56 | before, cur, err := s.cfg.Releases.Get(ctx, relay.AuthorityKey) |
| 57 | switch { |
| 58 | case errors.Is(err, os.ErrNotExist): |
| 59 | before, cur = nil, "" |
| 60 | case err != nil: |
| 61 | return err |
| 62 | } |
| 63 | if cur != etag { |
| 64 | return release.ErrCASConflict |
| 65 | } |
| 66 | scopes, err := hostops.RelayAuthorityScopes(before, body) |
| 67 | if err != nil { |
| 68 | return err |
| 69 | } |
| 70 | return s.relayWithScopes(ctx, scopes, "relay-aggregate", func(ctx context.Context) error { |
| 71 | if err := s.requireRelayLease(); err != nil { |
| 72 | return err |
| 73 | } |
| 74 | _, err := s.cfg.Releases.Put(ctx, relay.AuthorityKey, body, etag) |
| 75 | return err |
| 76 | }) |
| 77 | } |
| 78 | |
amadan