| 1 | package hostbox |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "time" |
| 6 | |
| 7 | "github.com/carlosframework/platform/internal/hostclaim" |
| 8 | "github.com/carlosframework/platform/internal/park" |
| 9 | ) |
| 10 | |
| 11 | type Regional struct { |
| 12 | Engine park.Engine |
| 13 | // Owner checks complete deployment membership and this worker's assignment. |
| 14 | Owner func(context.Context, hostclaim.Observed) error |
| 15 | } |
| 16 | |
| 17 | func (r Regional) observe(ctx context.Context, host string) (hostclaim.Observed, error) { |
| 18 | e := r.Engine |
| 19 | if e.Claims == nil || e.Store == nil || e.Leases == nil || e.View == nil || e.Authorize == nil || e.Now == nil || e.NewID == nil || e.Bucket == "" || e.Region == "" || r.Owner == nil { |
| 20 | return hostclaim.Observed{}, ErrUnsupported |
| 21 | } |
| 22 | obs, err := hostclaim.Read(ctx, e.Claims, host) |
| 23 | if err != nil { |
| 24 | return hostclaim.Observed{}, err |
| 25 | } |
| 26 | if obs.Absent || obs.Claim == nil { |
| 27 | return hostclaim.Observed{}, ErrUnsupported |
| 28 | } |
| 29 | if err := r.Owner(ctx, obs); err != nil { |
| 30 | return hostclaim.Observed{}, err |
| 31 | } |
| 32 | return obs, nil |
| 33 | } |
| 34 | |
| 35 | func (r Regional) Run(ctx context.Context, work park.Work) error { |
| 36 | obs, err := r.observe(ctx, work.Host) |
| 37 | if err != nil { |
| 38 | return err |
| 39 | } |
| 40 | for _, ob := range obs.Claim.Obligations { |
| 41 | if ob.ID != work.ID || ob.Op != work.Op { |
| 42 | continue |
| 43 | } |
| 44 | if !r.Engine.OwnsRegion(ob.Region) || ob.Owner != "region:"+ob.Region || (ob.Kind != "park-replica" && ob.Kind != "park-generation") { |
| 45 | return ErrUnsupported |
| 46 | } |
| 47 | return r.Engine.Run(ctx, work) |
| 48 | } |
| 49 | return nil // Already acknowledged; no mutation. |
| 50 | } |
| 51 | |
| 52 | func (r Regional) Sweep(ctx context.Context, host, legacyIncarnation string, evidenceAt time.Time, requestLimit time.Duration) (park.Decision, error) { |
| 53 | if _, err := r.observe(ctx, host); err != nil { |
| 54 | return park.Decision{}, err |
| 55 | } |
| 56 | if evidenceAt.IsZero() || requestLimit <= 0 { |
| 57 | return park.Decision{}, ErrUnsupported |
| 58 | } |
| 59 | return r.Engine.Sweep(ctx, host, legacyIncarnation, evidenceAt, requestLimit) |
| 60 | } |
| 61 | |
amadan