// The seeds sweep (seeded-instances spec §3.1, §5.6): the console's
// cleanup of abandoned seed state. It abandons, by a guarded CAS, a seed
// stuck where no box will ever move it; it finishes a rolled-back seed;
// and it deletes a snapshot under seeds/ that no live seed needs. It runs
// unattended and deletes customer data, so every deletion rests on
// positive knowledge: two complete listings, taken in the order that makes
// "no record names it" exact, and a record set with no gap in it.

package console

import (
	"context"
	"errors"
	"log"
	"os"
	"sort"
	"time"

	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/seeds"
)

const (
	seedsPassInterval = 5 * time.Minute
	// seedReserveTTL is how long a --from seed may stay reserving. The
	// door's copy is bounded at about 52 minutes (store walk, a 30-minute
	// restore, 60s, a 20-minute upload, the publish), so an hour is past
	// any live copy. That bound leaves out the private-store path's wait
	// for its store's walk lock (storeWalks), which has no deadline and
	// queues behind holders of up to storeTeardownTimeout each, this
	// sweep's own finishes among them. A copy that overruns the hour that
	// way is aborted by a CAS with one winner: the user retries, and no
	// data is lost.
	seedReserveTTL = time.Hour
	// seedImportTTL is how long an import may wait for its upload, or for
	// the app to say its process is stopped.
	seedImportTTL = 7 * 24 * time.Hour
	// seedOrphanAge is how old a snapshot no record names must be before
	// it is deleted.
	seedOrphanAge = time.Hour
)

// seedsPassTimeout bounds one pass, so a hung list, read or delete cannot
// stall the sweep for good (the S3 store sets no client timeout). It is
// three intervals: enough for the two listings and a dozen finishes that
// each run to storeTeardownTimeout (60s), the slowest step. A pass cut
// short is resumed by the next tick. A variable so a test can shorten it.
var seedsPassTimeout = 15 * time.Minute

// The tests' seams, nil in production. seedsPassBetweenListings runs after
// the snapshot listing and before the record listing.
// seedsPassBeforeAbort runs inside each abandonment's mutation, after the
// record is read and before it is written, with the call's number (1, then
// 2 on SetRecordSeed's retry).
var (
	seedsPassBetweenListings func()
	seedsPassBeforeAbort     func(call int)
)

// StartSeedsPass runs the seeds sweep now and every five minutes, until
// ctx is done. A console with host claims armed runs none: its record
// writes go through the claim, every seed door refuses it, and finishSeed
// refuses it too.
func (s *Server) StartSeedsPass(ctx context.Context) {
	if s.cfg.HostClaim {
		log.Printf("console: seeds pass: off on a host-claim console, where every seed door refuses")
		return
	}
	if s.cfg.Releases == nil || s.cfg.SeedData == nil {
		return
	}
	if _, ok := s.cfg.SeedData.(release.ModLister); !ok {
		log.Printf("console: seeds pass: the data bucket cannot list last-modified times, so this sweep deletes no unreferenced snapshot")
	}
	t := time.NewTicker(seedsPassInterval)
	defer t.Stop()
	for {
		s.seedsPass(ctx, time.Now())
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
	}
}

// seedStale reports a seed stuck where no box moves it: a --from seed
// reserving for over seedReserveTTL, or an import waiting for its upload or
// its stop for over seedImportTTL. Both times are a console's: At is
// stamped by the console whose door reserved or began the seed, now is the
// sweeping console's. With one console per deployment they are one clock;
// with more (region HA) their skew comes off the reservation's margin of
// about 8 minutes, and an overrun costs a copy, never data.
func seedStale(seed *provision.SeedSource, now time.Time) bool {
	switch {
	case seed == nil:
		return false
	case seed.Kind == seeds.KindReplica && seed.State == seeds.StateReserving:
		return now.Sub(seed.At) > seedReserveTTL
	case seed.Import() && (seed.State == seeds.StateAwaitingUpload || seed.State == seeds.StateUploaded):
		return now.Sub(seed.At) > seedImportTTL
	}
	return false
}

// seedsPass is one sweep. In order:
//
//  1. The snapshots under seeds/, with their last-modified times. Listed
//     first: every door writes its record before its snapshot (spec §4
//     step 7, §5.1), so a snapshot in this listing whose seed is live has
//     its record in the next one. A seed begun after this listing has no
//     snapshot in it, so nothing below can reach it.
//  2. Every record, strictly: from the bucket, or, with the record view
//     armed and step 1 empty (so nothing can be deleted), from the view.
//     Any listing error, here or in step 1, ends the pass: nothing is
//     decided on a partial set. A host with a record that would not
//     decode is unreadable; a host two records hold is ambiguous. Nothing
//     of either is touched: no snapshot deleted, no seed abandoned or
//     finished.
//  3. The decisions, one per snapshot, from the two listings alone.
//  4. Each stale seed abandoned by a CAS from the exact state read
//     (provision.SeedAbortStale); the box rolls back its part.
//  5. The snapshots deleted: those decided in step 3, and each abandoned
//     seed's own once its CAS has landed.
//  6. Each rolled-back seed finished (finishSeed re-reads its record).
//
// Every delete names one exact seeds/<host>/<id>/snapshot.db, so it never
// reaches another seed's.
//
// The pass is bounded by seedsPassTimeout. Cutting it anywhere is safe and
// the next pass resumes: a listing cut short is an error, so nothing is
// decided; an abandonment's CAS either landed (its snapshot then goes next
// pass, as an aborted seed's) or did not (the seed is still stale next
// pass); a snapshot not yet deleted is decided again from fresh listings;
// a finish cut short leaves its record rolled-back, and every finish step
// is idempotent.
func (s *Server) seedsPass(ctx context.Context, now time.Time) {
	if s.cfg.HostClaim || s.cfg.Releases == nil || s.cfg.SeedData == nil {
		return
	}
	ctx, cancel := context.WithTimeout(ctx, seedsPassTimeout)
	defer cancel()
	defer func() {
		if ctx.Err() != nil {
			log.Printf("console: seeds pass: cut short: %v (the next pass resumes)", ctx.Err())
		}
	}()
	var snaps []release.KeyMod
	ml, walk := s.cfg.SeedData.(release.ModLister)
	if walk {
		var err error
		if snaps, err = ml.ListMod(ctx, seeds.Prefix); err != nil {
			log.Printf("console: seeds pass: list snapshots: %v (nothing swept this pass)", err)
			return
		}
	}
	if seedsPassBetweenListings != nil {
		seedsPassBetweenListings()
	}
	// Step 2's source. A snapshot listing with nothing in it gives the pass
	// nothing to delete, so with the record view armed and able to vouch the
	// records come from it (S3 change-index design §4.6): abandonments and
	// finishes each act by a CAS on, or a re-read of, the bucket's record.
	// With any snapshot listed the walk is the bucket's, strict, after the
	// snapshot listing, because a delete is proved by a record's absence.
	src := s.cfg.Releases
	if len(snaps) == 0 {
		src, _ = s.recordWalkStore(ctx, s.cfg.Releases)
	}
	recs, unreadable, err := provision.ListAllRecordsStrict(ctx, src)
	if err != nil {
		log.Printf("console: seeds pass: list records: %v (nothing swept this pass)", err)
		return
	}
	byHost := map[string][]provision.Record{}
	for _, rec := range recs {
		byHost[rec.Host] = append(byHost[rec.Host], rec)
	}
	// held reports a host the pass must not touch, logging why once.
	logged := map[string]bool{}
	held := func(host string) bool {
		switch {
		case unreadable[host]:
			if !logged[host] {
				log.Printf("console: seeds pass: a record of %s could not be read; nothing of %s is swept this pass", host, host)
			}
		case len(byHost[host]) > 1:
			if !logged[host] {
				log.Printf("console: seeds pass: OPERATOR: %d records hold %s; nothing of %s is swept until one remains", len(byHost[host]), host, host)
			}
		default:
			return false
		}
		logged[host] = true
		return true
	}

	// Step 3: the decisions.
	var condemned []string
	waiting := map[[2]string]string{} // host, id -> key: deleted only if this pass abandons the seed
	for _, o := range snaps {
		host, id, ok := seeds.ParseSnapshotKey(o.Key)
		if !ok || held(host) {
			continue
		}
		var rec *provision.Record
		if rs := byHost[host]; len(rs) == 1 {
			rec = &rs[0]
		}
		switch snapshotDecision(rec, id, now, o.Mod) {
		case snapshotDelete:
			condemned = append(condemned, o.Key)
		case snapshotIfAbandoned:
			waiting[[2]string{host, id}] = o.Key
		}
	}

	// Step 4: the abandonments.
	for _, rec := range recs {
		if ctx.Err() != nil {
			return
		}
		if !seedStale(rec.Seed, now) || held(rec.Host) {
			continue
		}
		if s.abandonSeed(ctx, rec) {
			if key, ok := waiting[[2]string{rec.Host, rec.Seed.ID}]; ok {
				condemned = append(condemned, key)
			}
		}
	}

	// Step 5: the snapshot deletes.
	sort.Strings(condemned)
	for _, key := range condemned {
		if ctx.Err() != nil {
			return
		}
		if err := s.cfg.SeedData.Delete(ctx, key); err != nil && !errors.Is(err, os.ErrNotExist) {
			log.Printf("console: seeds pass: delete %s: %v", key, err)
			continue
		}
		log.Printf("console: seeds pass: deleted %s", key)
	}

	// Step 6: the finishes.
	for _, rec := range recs {
		if ctx.Err() != nil {
			return
		}
		if rec.Seed == nil || rec.Seed.State != seeds.StateRolledBack || held(rec.Host) {
			continue
		}
		if err := s.finishSeed(ctx, rec.Account, rec.App, rec.Host, rec.Seed.ID); err != nil {
			log.Printf("console: seeds pass: finish seed %s of %s: %v", rec.Seed.ID, rec.Host, err)
		}
	}
}

type snapshotVerdict int

const (
	snapshotKeep snapshotVerdict = iota
	snapshotDelete
	snapshotIfAbandoned // keep, unless this pass abandons the seed by CAS
)

// snapshotDecision decides one snapshot of seed id from the one record that
// holds its host (nil for none), and the snapshot's last-modified time. Its
// age is the console's clock less the bucket's stamp, so skew between
// the two moves it; it only ever decides a snapshot no record names, which
// the listing order has already shown is no live seed's, so the hour is a
// margin and not the proof.
//
//   - The record names id: deleted when the seed is placed or committed
//     (the host's own replica exists, so the box's delete was lost) or
//     aborted or rolled-back (terminal: nothing places it, and the box's
//     own rollback deletes the same key; this also covers a trashed app's
//     aborted seed, which no box finishes). A seed stale at now waits on
//     its abandonment. Any other state is a seed still wanted: kept.
//   - The record names another seed, placed or committed: that seed
//     superseded this one (spec §3.1).
//   - Otherwise no record names it (spec §3.1): deleted once it is over
//     seedOrphanAge old.
func snapshotDecision(rec *provision.Record, id string, now, mod time.Time) snapshotVerdict {
	if rec != nil && rec.Seed != nil {
		state := rec.Seed.State
		placed := state == seeds.StatePlaced || state == seeds.StateCommitted
		if rec.Seed.ID == id {
			switch {
			case placed || rec.Seed.Abandoned():
				return snapshotDelete
			case seedStale(rec.Seed, now):
				return snapshotIfAbandoned
			}
			return snapshotKeep
		}
		if placed {
			return snapshotDelete
		}
	}
	if now.Sub(mod) > seedOrphanAge {
		return snapshotDelete
	}
	return snapshotKeep
}

// abandonSeed moves rec's stale seed to aborted by a CAS from exactly the
// state the pass read, and answers whether that CAS landed. A seed that
// moved on meanwhile (the door published, the upload landed, the app said
// stopped) is left alone.
func (s *Server) abandonSeed(ctx context.Context, rec provision.Record) bool {
	seed := *rec.Seed
	abort := provision.SeedAbortStale(seed.ID, seed.State, seed.At)
	call := 0
	_, err := provision.SetRecordSeed(ctx, s.cfg.Releases, rec.Account, rec.App, rec.Host, func(r *provision.Record) error {
		call++
		if seedsPassBeforeAbort != nil {
			seedsPassBeforeAbort(call)
		}
		return abort(r)
	})
	if err != nil {
		if !errors.Is(err, provision.ErrSeedConflict) {
			log.Printf("console: seeds pass: abandon seed %s of %s: %v", seed.ID, rec.Host, err)
		}
		return false
	}
	log.Printf("console: seeds pass: abandoned seed %s of %s/%s/%s (%s since %s)", seed.ID, rec.Account, rec.App, rec.Host,
		seed.State, seed.At.UTC().Format(time.RFC3339))
	s.bumpSerial(ctx)
	return true
}