// The console's finish for an abandoned seed (seeded-instances spec §5.6,
// codex R6-1). Once a box has rolled a seed back, or the delete door has
// moved an import no box swapped straight to rolled-back, the record still
// holds the hostname. finishSeed is what frees it: the store the seed made,
// the seed's snapshot, a --from seed's replica, and the record last.

package console

import (
	"context"
	"errors"
	"fmt"
	"log"
	"os"
	"slices"
	"time"

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

// finishSeed finishes seedID on host's record, only while that record
// still names seedID and says rolled-back. Any other record (another seed,
// another state, none) is left alone and answers nil. In order:
//
//  1. The store of the record's environment, under the store's walk lock
//     and re-read there with the sharing check (final review F3), is torn
//     down only while its Declared.Seed is seedID (preflight M9), and only
//     when no other instance names that environment (M10), read strictly:
//     a record of the app that will not decode counts as one (Plan 8 M7).
//     A store with another seed, or none, is never touched.
//  2. The seed's snapshot, seeds/<host>/<id>/snapshot.db.
//  3. For a --from seed, every key under the host's replica prefix in the
//     data bucket (M11). It is this seed's: the record is create-only and
//     the create door's leftover-replica rail refused the host if anything
//     was there, so only this seed's instance has written there since.
//     That holds only while the record still names this seed, so the
//     record is re-read before the listing and before every batch of
//     deletes (review I2); and nothing is deleted while the host's lease is
//     held, because a row may still be writing (review I1).
//  4. The record, by CAS on the etag of the last read that saw this seed
//     in rolled-back. It goes last because until then it holds the
//     hostname, so a re-import of the host never meets a leftover store.
//
// Every step is idempotent: a finish that fails part way answers an error
// with the record still rolled-back, and the next finish (the caller's
// retry, or the sweep) resumes it. seeds.Check runs before any key is
// built (M8). ctx bounds the whole finish; the teardown also has its own
// storeTeardownTimeout. Callers that finish inline on a request's behalf
// pass context.WithoutCancel of it, so a departing client cannot cut a
// teardown midway.
//
// A console with host claims armed refuses: its record deletes go through
// the claim (retireClaimHostRecord), and the seed doors refuse such a
// console anyway (seedRegionReady).
func (s *Server) finishSeed(ctx context.Context, sqid, app, host, seedID string) error {
	if s.cfg.HostClaim {
		return fmt.Errorf("finish seed %s of %s: this console has host claims armed, and finishes no seed", seedID, host)
	}
	if err := seeds.Check(host, seedID); err != nil {
		return fmt.Errorf("finish seed %q of %q: %w", seedID, host, err)
	}
	err := s.finishSeedOnce(ctx, sqid, app, host, seedID)
	if errors.Is(err, release.ErrCASConflict) {
		// Another writer changed the record between the read and the
		// delete. Every step is idempotent, so read it again once.
		err = s.finishSeedOnce(ctx, sqid, app, host, seedID)
	}
	return err
}

func (s *Server) finishSeedOnce(ctx context.Context, sqid, app, host, seedID string) error {
	rec, etag, err := provision.GetRecordETag(ctx, s.cfg.Releases, sqid, app, host)
	if errors.Is(err, os.ErrNotExist) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("read the record of %s: %w", host, err)
	}
	if rec.Seed == nil || rec.Seed.ID != seedID || rec.Seed.State != seeds.StateRolledBack {
		return nil
	}
	// Checked before anything is destroyed: without it the store would be
	// torn down and every later finish fail with the record holding the
	// host.
	if s.cfg.SeedData == nil {
		return fmt.Errorf("finish seed %s of %s: this console has no data bucket to delete its snapshot from", seedID, host)
	}
	if err := s.finishSeedStore(ctx, sqid, app, host, seedID, rec.Environment); err != nil {
		return err
	}
	if err := s.cfg.SeedData.Delete(ctx, seeds.SnapshotKey(host, seedID)); err != nil && !errors.Is(err, os.ErrNotExist) {
		return fmt.Errorf("delete the snapshot of seed %s of %s: %w", seedID, host, err)
	}
	if rec.Seed.Kind == seeds.KindReplica {
		var gone bool
		if etag, gone, err = s.deleteSeedReplica(ctx, rec, etag); err != nil {
			return err
		}
		if gone {
			return nil // another finish deleted the record
		}
	}
	if err := release.DeleteIfMatch(ctx, s.cfg.Releases, provision.RecordKey(sqid, app, host), etag); err != nil && !errors.Is(err, os.ErrNotExist) {
		return fmt.Errorf("delete the record of %s: %w", host, err)
	}
	log.Printf("console: finished seed %s of %s/%s/%s", seedID, sqid, app, host)
	s.bumpSerial(ctx)
	return nil
}

// finishSeedStore is the finish's first step: the store of env, torn down
// only while seedID made it and no other instance uses it. A first look
// runs without the lock, and a store it keeps ends the step there. The
// teardown itself runs under the store's walk lock, after the store record
// and the sharing check are read again there (final review F3): a grant (a
// begin's, or carlos store create --environment) and an instance's
// environment binding (seedenvbind.go) take that lock too, and a binding
// clears the seed's mark, so whatever bound since the first look keeps the
// store.
func (s *Server) finishSeedStore(ctx context.Context, sqid, app, host, seedID, env string) error {
	store, err := s.seedStoreToTear(ctx, sqid, app, host, seedID, env)
	if store == nil || err != nil {
		return err
	}
	if finishSeedCheckedHook != nil {
		finishSeedCheckedHook(host)
	}
	defer s.storeWalks.lock(sqid, app, env)()
	if store, err = s.seedStoreToTear(ctx, sqid, app, host, seedID, env); store == nil || err != nil {
		return err
	}
	if s.cfg.StoreProvisioner == nil {
		return fmt.Errorf("finish seed %s of %s: this console has no storage backend to tear its store down with", seedID, host)
	}
	tctx, cancel := context.WithTimeout(ctx, storeTeardownTimeout)
	defer cancel()
	if err := objstore.TeardownIn(tctx, s.cfg.Releases, s.cfg.StoreProvisioner, s.cfg.StoreScanner, sqid, app, env); err != nil {
		return fmt.Errorf("tear down the store of %s: %w", host, err)
	}
	log.Printf("console: finish seed %s of %s/%s/%s: tore down its store (bucket %s)", seedID, sqid, app, host, storeBucketName(store))
	return nil
}

// finishSeedCheckedHook, when set (tests), runs between the finish's first
// look at the store and its locked re-read.
var finishSeedCheckedHook func(host string)

// seedStoreToTear reads env's store and answers it when seedID made it and
// no other instance names env; nil when the store is to be kept (logged
// for an operator when another instance may use it).
func (s *Server) seedStoreToTear(ctx context.Context, sqid, app, host, seedID, env string) (*objstore.Record, error) {
	store, _, err := objstore.GetIn(ctx, s.cfg.Releases, sqid, app, env)
	if err != nil {
		return nil, fmt.Errorf("read the store of %s: %w", host, err)
	}
	if store == nil || store.Declared == nil || store.Declared.Seed != seedID {
		return nil, nil
	}
	other, unread, err := s.otherInstanceInEnvironment(ctx, sqid, app, host, env)
	if err != nil {
		return nil, err
	}
	if unread {
		log.Printf("console: finish seed %s of %s/%s/%s: OPERATOR: the record of instance %s could not be read, so it may use environment %q, and the store seed %s made (bucket %s) is left standing; tear it down by hand if it is not wanted",
			seedID, sqid, app, host, other, env, seedID, storeBucketName(store))
		return nil, nil
	}
	if other != "" {
		log.Printf("console: finish seed %s of %s/%s/%s: OPERATOR: instance %s also uses environment %q, so the store seed %s made (bucket %s) is left standing; tear it down by hand if it is not wanted",
			seedID, sqid, app, host, other, env, seedID, storeBucketName(store))
		return nil, nil
	}
	return store, nil
}

// otherInstanceInEnvironment names an instance of app other than host
// whose record names env, or "" when there is none. It reads strictly
// (provision.ListRecordsStrict, preflight M7): it runs unattended from the
// seeds sweep, and its "none" tears a store down. A listing it cannot
// complete, or one with more than instanceListMax records, is an error:
// the finish cannot tell whether the store is shared, so the record stays
// rolled-back for a later finish. A record that reads but will not decode,
// or disagrees with its path, counts as sharing: it is named, with unread
// true, since what it says about env is not known.
func (s *Server) otherInstanceInEnvironment(ctx context.Context, sqid, app, host, env string) (other string, unread bool, err error) {
	recs, unreadable, err := provision.ListRecordsStrict(ctx, s.cfg.Releases, sqid, app, instanceListMax)
	if errors.Is(err, provision.ErrRecordsTruncated) {
		return "", false, fmt.Errorf("list the instances of %s/%s: more than %d, so the finish cannot tell who uses environment %q", sqid, app, instanceListMax, env)
	}
	if err != nil {
		return "", false, fmt.Errorf("list the instances of %s/%s: %w", sqid, app, err)
	}
	if len(unreadable) > 0 {
		hosts := make([]string, 0, len(unreadable))
		for h := range unreadable {
			hosts = append(hosts, h)
		}
		slices.Sort(hosts)
		return hosts[0], true, nil
	}
	for _, r := range recs {
		if r.Host != host && r.Environment == env {
			return r.Host, false, nil
		}
	}
	return "", false, nil
}

// seedReplicaBatch is how many replica keys are deleted between two
// re-reads of the record.
const seedReplicaBatch = 100

// deleteSeedReplica deletes every key under a --from seed's replica prefix
// in the data bucket, and answers the etag the record delete is to be
// conditional on (the last read that saw this seed rolled-back), or gone
// when the record was deleted meanwhile (another finish).
//
// The rail that makes the delete safe (createleftovers.go) only checked a
// record this console can see; a record outside that (another region, a
// pool, a fleet) was never checked, so its prefix here is not known to be
// this seed's and is left alone.
//
// Nothing is deleted while the host's lease (lock.json under the prefix) is
// held by the lease package's own rule (lease.LockState): a row this seed
// started may still be running and writing there. A lease that is absent,
// released or expired past its margin does not block, so a box that died
// does not hold the host for ever. A held lease is an error, and the
// record stays rolled-back for the next finish.
func (s *Server) deleteSeedReplica(ctx context.Context, rec *provision.Record, etag string) (string, bool, error) {
	sqid, app, host, seedID := rec.Account, rec.App, rec.Host, rec.Seed.ID
	if !s.leftoverCheckApplies(rec.Region, rec.Pool) {
		log.Printf("console: finish seed %s of %s: its replica is outside this console's data bucket; not deleting it", seedID, host)
		return etag, false, nil
	}
	// The read before the teardown can be a minute old: the host may have
	// been finished and created again since, by a new instance that is
	// writing here now.
	etag, gone, err := s.seedRecordETag(ctx, sqid, app, host, seedID)
	if gone || err != nil {
		return "", gone, err
	}
	location := activator.HibernatePrefix(s.cfg.HibernatePrefix, host)
	held, owner, expires, err := lease.LockState(ctx, s.cfg.HibernateBucket, location, s.cfg.HibernateRegion)
	if err != nil {
		return "", false, fmt.Errorf("read the lease of %s: %w", host, err)
	}
	if held {
		return "", false, fmt.Errorf("finish seed %s of %s: its lease is held by %q until %s, so its replica is not deleted yet",
			seedID, host, owner, expires.UTC().Format(time.RFC3339))
	}
	// The trailing slash keeps app.example.co from covering
	// app.example.com's keys.
	prefix := location + "/"
	keys, err := s.cfg.SeedData.List(ctx, prefix)
	if err != nil {
		return "", false, fmt.Errorf("list the replica of %s: %w", host, err)
	}
	for start := 0; start < len(keys); start += seedReplicaBatch {
		now, gone, err := s.seedRecordETag(ctx, sqid, app, host, seedID)
		if gone || err != nil {
			return "", gone, err
		}
		if now != etag {
			return "", false, fmt.Errorf("finish seed %s of %s: the record changed while its replica was being deleted", seedID, host)
		}
		for _, key := range keys[start:min(start+seedReplicaBatch, len(keys))] {
			if err := s.cfg.SeedData.Delete(ctx, key); err != nil && !errors.Is(err, os.ErrNotExist) {
				return "", false, fmt.Errorf("delete the replica of %s: %w", host, err)
			}
		}
	}
	if len(keys) > 0 {
		log.Printf("console: finish seed %s of %s: deleted %d replica objects under %s", seedID, host, len(keys), prefix)
	}
	return etag, false, nil
}

// seedRecordETag re-reads host's record and answers its etag while it
// still names seedID in rolled-back, or gone when there is no record. A
// record that names anything else is an error, not "done": the finish only
// says done when the seed's own record is gone.
func (s *Server) seedRecordETag(ctx context.Context, sqid, app, host, seedID string) (etag string, gone bool, err error) {
	rec, etag, err := provision.GetRecordETag(ctx, s.cfg.Releases, sqid, app, host)
	switch {
	case errors.Is(err, os.ErrNotExist):
		return "", true, nil
	case err != nil:
		return "", false, fmt.Errorf("read the record of %s: %w", host, err)
	case rec.Seed == nil || rec.Seed.ID != seedID || rec.Seed.State != seeds.StateRolledBack:
		return "", false, fmt.Errorf("finish seed %s of %s: the record no longer names this seed rolled-back", seedID, host)
	}
	return etag, false, nil
}