| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "log" |
| 7 | "time" |
| 8 | |
| 9 | "github.com/carlosframework/platform/internal/plans" |
| 10 | "github.com/carlosframework/platform/internal/release" |
| 11 | ) |
| 12 | |
| 13 | const planLimitsPassInterval = time.Minute |
| 14 | |
| 15 | // planLimitsPassBatch is how many rows one query reads; the pass pages |
| 16 | // through every due row regardless. A var so a test can shrink it. |
| 17 | var planLimitsPassBatch = 200 |
| 18 | |
| 19 | // StartPlanLimitsPass keeps every account's limits record in the bucket in |
| 20 | // step with its plan row and the configured catalog: now, then every |
| 21 | // minute (StartExpiryPass's shape). With no catalog it does nothing at all. |
| 22 | func (s *Server) StartPlanLimitsPass(ctx context.Context) { |
| 23 | if s.cfg.Releases == nil || s.cfg.PlanCatalog == nil { |
| 24 | return |
| 25 | } |
| 26 | t := time.NewTicker(planLimitsPassInterval) |
| 27 | defer t.Stop() |
| 28 | for { |
| 29 | s.planLimitsPass(ctx, time.Now()) |
| 30 | select { |
| 31 | case <-ctx.Done(): |
| 32 | return |
| 33 | case <-t.C: |
| 34 | } |
| 35 | } |
| 36 | } |
| 37 | |
| 38 | // planLimitsPass is one pass at the given clock: give unassigned accounts |
| 39 | // the default plan, then publish every row the bucket is behind on. One |
| 40 | // account's failure never stops the others. |
| 41 | func (s *Server) planLimitsPass(ctx context.Context, now time.Time) { |
| 42 | cat := s.cfg.PlanCatalog |
| 43 | if s.cfg.Releases == nil || cat == nil { |
| 44 | return |
| 45 | } |
| 46 | if _, err := s.store.EnsureDefaultPlans(cat.DefaultPlan, now); err != nil { |
| 47 | log.Printf("console: plan limits: default plans: %v", err) |
| 48 | return |
| 49 | } |
| 50 | // Page by account id: a row that cannot be published (an orphaned |
| 51 | // assignment) stays due, but never holds a later account back. |
| 52 | var after int64 |
| 53 | for { |
| 54 | rows, err := s.store.PlansToPublish(cat.ID(), after, planLimitsPassBatch) |
| 55 | if err != nil { |
| 56 | log.Printf("console: plan limits: list: %v", err) |
| 57 | return |
| 58 | } |
| 59 | for _, row := range rows { |
| 60 | if ctx.Err() != nil { |
| 61 | return |
| 62 | } |
| 63 | if err := s.publishAccountLimits(ctx, row, now); err != nil { |
| 64 | log.Printf("console: plan limits: %s: %v", row.Sqid, err) |
| 65 | } |
| 66 | after = row.AccountID |
| 67 | } |
| 68 | if len(rows) < planLimitsPassBatch { |
| 69 | return |
| 70 | } |
| 71 | } |
| 72 | } |
| 73 | |
| 74 | // publishAccountLimits writes row's effective limits to its account's |
| 75 | // record. An assignment the catalog cannot price (a plan or pack it no |
| 76 | // longer has) writes nothing, so boxes keep the last good record. A record |
| 77 | // already at this revision or newer, under this catalog, is left alone. |
| 78 | func (s *Server) publishAccountLimits(ctx context.Context, row AccountPlanRow, now time.Time) error { |
| 79 | cat := s.cfg.PlanCatalog |
| 80 | if cat == nil { |
| 81 | return errors.New("no plan catalog") |
| 82 | } |
| 83 | limits, err := cat.Effective(row.Plan, row.Packs, row.Sponsored) |
| 84 | if err != nil { |
| 85 | return err |
| 86 | } |
| 87 | rec := &plans.Record{ |
| 88 | Account: row.Sqid, |
| 89 | CatalogID: cat.ID(), |
| 90 | Plan: row.Plan, |
| 91 | Source: row.Source, |
| 92 | Revision: row.Revision, |
| 93 | Limits: limits, |
| 94 | IssuedAt: now.UTC().Truncate(time.Second), |
| 95 | } |
| 96 | for attempt := 0; attempt < 2; attempt++ { |
| 97 | cur, etag, err := plans.GetLimits(ctx, s.cfg.Releases, row.Sqid) |
| 98 | if err != nil { |
| 99 | return err |
| 100 | } |
| 101 | if cur != nil && cur.CatalogID == rec.CatalogID && cur.Revision >= rec.Revision { |
| 102 | return s.store.MarkPlanPublished(row.AccountID, row.Revision, cat.ID()) |
| 103 | } |
| 104 | _, err = plans.PutLimits(ctx, s.cfg.Releases, rec, etag) |
| 105 | if errors.Is(err, release.ErrCASConflict) { |
| 106 | continue |
| 107 | } |
| 108 | if err != nil { |
| 109 | return err |
| 110 | } |
| 111 | return s.store.MarkPlanPublished(row.AccountID, row.Revision, cat.ID()) |
| 112 | } |
| 113 | return errors.New("lost two write races; the next pass retries") |
| 114 | } |
| 115 | |
| 116 | // EffectiveLimits is acct's limits under the configured catalog, or nil |
| 117 | // when plans are off. An account with no row is on the default plan. |
| 118 | func (s *Server) EffectiveLimits(acct *Account) (*plans.Limits, error) { |
| 119 | cat := s.cfg.PlanCatalog |
| 120 | if cat == nil { |
| 121 | return nil, nil |
| 122 | } |
| 123 | row, err := s.store.AccountPlan(acct.ID) |
| 124 | if err != nil { |
| 125 | return nil, err |
| 126 | } |
| 127 | plan, packs := cat.DefaultPlan, map[string]int(nil) |
| 128 | var sponsored *plans.Limits |
| 129 | if row != nil { |
| 130 | plan, packs, sponsored = row.Plan, row.Packs, row.Sponsored |
| 131 | } |
| 132 | l, err := cat.Effective(plan, packs, sponsored) |
| 133 | if err != nil { |
| 134 | return nil, err |
| 135 | } |
| 136 | return &l, nil |
| 137 | } |
| 138 | |
amadan