package console

import (
	"context"
	"errors"
	"log"
	"time"

	"github.com/carlosframework/platform/internal/plans"
	"github.com/carlosframework/platform/internal/release"
)

const planLimitsPassInterval = time.Minute

// planLimitsPassBatch is how many rows one query reads; the pass pages
// through every due row regardless. A var so a test can shrink it.
var planLimitsPassBatch = 200

// StartPlanLimitsPass keeps every account's limits record in the bucket in
// step with its plan row and the configured catalog: now, then every
// minute (StartExpiryPass's shape). With no catalog it does nothing at all.
func (s *Server) StartPlanLimitsPass(ctx context.Context) {
	if s.cfg.Releases == nil || s.cfg.PlanCatalog == nil {
		return
	}
	t := time.NewTicker(planLimitsPassInterval)
	defer t.Stop()
	for {
		s.planLimitsPass(ctx, time.Now())
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
	}
}

// planLimitsPass is one pass at the given clock: give unassigned accounts
// the default plan, then publish every row the bucket is behind on. One
// account's failure never stops the others.
func (s *Server) planLimitsPass(ctx context.Context, now time.Time) {
	cat := s.cfg.PlanCatalog
	if s.cfg.Releases == nil || cat == nil {
		return
	}
	if _, err := s.store.EnsureDefaultPlans(cat.DefaultPlan, now); err != nil {
		log.Printf("console: plan limits: default plans: %v", err)
		return
	}
	// Page by account id: a row that cannot be published (an orphaned
	// assignment) stays due, but never holds a later account back.
	var after int64
	for {
		rows, err := s.store.PlansToPublish(cat.ID(), after, planLimitsPassBatch)
		if err != nil {
			log.Printf("console: plan limits: list: %v", err)
			return
		}
		for _, row := range rows {
			if ctx.Err() != nil {
				return
			}
			if err := s.publishAccountLimits(ctx, row, now); err != nil {
				log.Printf("console: plan limits: %s: %v", row.Sqid, err)
			}
			after = row.AccountID
		}
		if len(rows) < planLimitsPassBatch {
			return
		}
	}
}

// publishAccountLimits writes row's effective limits to its account's
// record. An assignment the catalog cannot price (a plan or pack it no
// longer has) writes nothing, so boxes keep the last good record. A record
// already at this revision or newer, under this catalog, is left alone.
func (s *Server) publishAccountLimits(ctx context.Context, row AccountPlanRow, now time.Time) error {
	cat := s.cfg.PlanCatalog
	if cat == nil {
		return errors.New("no plan catalog")
	}
	limits, err := cat.Effective(row.Plan, row.Packs, row.Sponsored)
	if err != nil {
		return err
	}
	rec := &plans.Record{
		Account:   row.Sqid,
		CatalogID: cat.ID(),
		Plan:      row.Plan,
		Source:    row.Source,
		Revision:  row.Revision,
		Limits:    limits,
		IssuedAt:  now.UTC().Truncate(time.Second),
	}
	for attempt := 0; attempt < 2; attempt++ {
		cur, etag, err := plans.GetLimits(ctx, s.cfg.Releases, row.Sqid)
		if err != nil {
			return err
		}
		if cur != nil && cur.CatalogID == rec.CatalogID && cur.Revision >= rec.Revision {
			return s.store.MarkPlanPublished(row.AccountID, row.Revision, cat.ID())
		}
		_, err = plans.PutLimits(ctx, s.cfg.Releases, rec, etag)
		if errors.Is(err, release.ErrCASConflict) {
			continue
		}
		if err != nil {
			return err
		}
		return s.store.MarkPlanPublished(row.AccountID, row.Revision, cat.ID())
	}
	return errors.New("lost two write races; the next pass retries")
}

// EffectiveLimits is acct's limits under the configured catalog, or nil
// when plans are off. An account with no row is on the default plan.
func (s *Server) EffectiveLimits(acct *Account) (*plans.Limits, error) {
	cat := s.cfg.PlanCatalog
	if cat == nil {
		return nil, nil
	}
	row, err := s.store.AccountPlan(acct.ID)
	if err != nil {
		return nil, err
	}
	plan, packs := cat.DefaultPlan, map[string]int(nil)
	var sponsored *plans.Limits
	if row != nil {
		plan, packs, sponsored = row.Plan, row.Packs, row.Sponsored
	}
	l, err := cat.Effective(plan, packs, sponsored)
	if err != nil {
		return nil, err
	}
	return &l, nil
}