package host

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log"
	"os"
	"sync/atomic"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/controlname"
	"github.com/carlosframework/platform/internal/layout"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/seeds"
)

// storeSeedObjects reads seed snapshots from this box's data bucket.
type storeSeedObjects struct{ s release.Store }

// Open streams seeds/<host>/<id>/snapshot.db. A store that cannot stream is
// refused rather than read whole: a seed can be gigabytes.
func (o storeSeedObjects) Open(ctx context.Context, host, id string) (io.ReadCloser, error) {
	if err := seeds.Check(host, id); err != nil {
		return nil, fmt.Errorf("seed %q of %s: %w", id, host, err)
	}
	op, ok := o.s.(release.Opener)
	if !ok {
		return nil, errors.New("the data bucket store cannot stream")
	}
	rc, _, err := op.Open(ctx, seeds.SnapshotKey(host, id))
	return rc, err
}

// seedPlacedHook is ExecBackend.OnSeedPlaced (seeded-instances spec §3.4
// step 5): the record's CAS to placed, then the delete of this seed's
// snapshot, never another's. The activator calls it only once the host's
// own replica holds the seed.
//
// It is called more than once for one placement: twice on a finishing wake
// (Restore's finishPlacement and doWake's step 5b), and by any box that
// wakes the host before the first call's delete lands. A record already
// placed (or committed) with this id is that case: a success, and the
// delete still runs, because the earlier call's delete may be the one that
// failed. A CAS refused for any other reason deletes nothing.
func seedPlacedHook(deploy, data release.Store) func(ctx context.Context, acct, app, host, id string) error {
	return func(ctx context.Context, acct, app, host, id string) error {
		if deploy == nil || data == nil {
			return errors.New("seed recording is not wired on this box")
		}
		if err := seeds.Check(host, id); err != nil {
			return fmt.Errorf("seed %q of %s: %w", id, host, err)
		}
		_, err := provision.SetRecordSeed(ctx, deploy, acct, app, host, provision.SeedPlace(id))
		if errors.Is(err, provision.ErrSeedConflict) {
			rec, gerr := provision.GetRecord(ctx, deploy, acct, app, host)
			if gerr != nil {
				return gerr
			}
			if !placedWith(rec, id) {
				return fmt.Errorf("seed %s of %s: %w", id, host, err)
			}
		} else if err != nil {
			return err
		}
		if err := data.Delete(ctx, seeds.SnapshotKey(host, id)); err != nil && !errors.Is(err, os.ErrNotExist) {
			return fmt.Errorf("delete seed %s of %s: %w", id, host, err)
		}
		return nil
	}
}

// placedWith reports whether rec's seed is id and has been placed: placed,
// or committed, which only follows placed.
func placedWith(rec *provision.Record, id string) bool {
	return rec != nil && rec.Seed != nil && rec.Seed.ID == id &&
		(rec.Seed.State == seeds.StatePlaced || rec.Seed.State == seeds.StateCommitted)
}

// armedSeedHooks is what reconcileSerialized hands the provision pass
// (ReconcileConfig.Seeds). Nil until the activator is armed with a data
// bucket: the pass then mints and finishes no seeded record.
var armedSeedHooks atomic.Pointer[provision.SeedHooks]

// armSeedHooks arms the provision pass's seed hooks. Hooks without a data
// bucket are never armed: the pass's aborted-seed finish deletes through
// it.
func armSeedHooks(h *provision.SeedHooks) {
	if h != nil && h.Data == nil {
		h = nil
	}
	armedSeedHooks.Store(h)
}

// seedDataOpener opens this region's data bucket, where seeds/ lives.
func seedDataOpener(bucket, region string) func() (release.Store, error) {
	return func() (release.Store, error) {
		s, err := release.NewS3StoreIn(context.Background(), bucket, region)
		if err != nil {
			return nil, err
		}
		return s, nil
	}
}

// seedBox is what the import swap and rollback need from this box beyond
// the two stores (preflight M7): the activator's fence, the key prefix of
// the replicas in the data bucket (cfg.Prefix), whether the startup unit
// sweep succeeded, and the fence generation source.
type seedBox struct {
	Fence        layout.Fence
	Prefix       string
	Swept        func() bool
	NextFenceGen func() (uint64, error)
}

// wireSeedPlacement arms seeded instances on this box (seeded-instances
// spec §3.4): the claim flag that makes the seed fallback refuse, the data
// bucket for snapshots, the record store for the placed CAS, and the
// provision pass's hooks. The claim flag is set whatever else happens. The
// rest is best-effort like the staging-fallback mirror, and all or nothing:
// a box that cannot build both stores places no seed, and a seeded row's
// wake fails instead of starting empty.
//
// The import swap's and rollback's hooks (spec §5.4, §5.6) are armed only
// with claim control off: a claim-armed box refuses seeds until #250, so it
// never swaps, and an import journaled before claims came on stays gated
// and untouched there. The rollback's seven hooks are armed together or not
// at all (provision runs no rollback missing one); without a fence from the
// activator none is.
//
// The rollback's fence is known by the generation its journal stores, so
// box.NextFenceGen must never repeat one across a restart
// (fenceGenerations). The hooks set no clock: the swap and the rollback
// read the pass's own (ReconcileConfig.Now).
func wireSeedPlacement(eb *activator.ExecBackend, claimArmed bool, openData, openDeploy func() (release.Store, error), box seedBox) {
	eb.ClaimArmed = claimArmed
	data, err := openData()
	if err != nil {
		log.Printf("agent: seed reader: %v (no seed can be placed on this box)", err)
		return
	}
	deploy, err := openDeploy()
	if err != nil {
		log.Printf("agent: seed recorder: %v (no seed can be placed on this box)", err)
		return
	}
	eb.Seeds = storeSeedObjects{data}
	eb.OnSeedPlaced = seedPlacedHook(deploy, data)
	h := &provision.SeedHooks{Data: data}
	if !claimArmed {
		h.NestedOK = eb.NestedOK
		h.LegacyOwners = func(ctx context.Context) (*seeds.LegacyOwners, error) {
			o, _, err := seeds.ReadLegacyOwners(ctx, deploy)
			return o, err
		}
		h.EnvBefore = activator.SeedEnvBefore
		if box.Fence != nil && box.Swept != nil && box.NextFenceGen != nil {
			fence := box.Fence
			h.FenceForRollback = fenceForRollback(fence, box.Swept)
			h.LiftFence = func(host string, gen uint64) { fence.LiftFence(host, gen) }
			h.RemoveLocal = func(rt registry.Route) error { return activator.PurgeSeedLocal(rt.DBPath) }
			h.LocalSeedID = activator.ReadSeedMarker
			h.RetireKey = activator.RetireSeedKey
			h.DeleteReplica = deleteReplica(data, box.Prefix)
			h.NextFenceGen = box.NextFenceGen
		}
	}
	armSeedHooks(h)
}

// seedSweptAtStart is whether the agent's startup unit sweep succeeded,
// for the import rollback's stop proof: the same fact the layout pass is
// built with. It stays false with claim control on, where the activator
// starts without a sweep (and no rollback is armed).
var seedSweptAtStart atomic.Bool

// fenceGenerations is this box's fence generation source, the registry's:
// never 0, and never one it handed out before, across agent restarts. The
// layout pass draws from it too (newLayoutPass), so a rollback resumed
// after a restart can never meet the pass's fence under its own number.
func fenceGenerations(reg *registry.Registry) func() (uint64, error) {
	return func() (uint64, error) { return reg.NextFenceGeneration(time.Now()) }
}

// rollbackFenceTimeout bounds one fence-and-stop, as the layout pass's.
const rollbackFenceTimeout = 30 * time.Second

// fenceForRollback is the import rollback's step 2 (seeded-instances spec
// §5.6), by internal/layout/pass.go's rule (Pass.quiesce) for the same
// activator API. A fence already holding gen is this rollback's own, from
// an earlier pass of this process: proceed. Otherwise fence and stop under
// a 30-second deadline. A proven stop proceeds; another operation's fence
// defers, touching nothing; an unproven stop proceeds only when the box
// was swept at start, the host is quiesced and its stop is not uncertain.
// Anything else lifts this attempt's fence and defers.
//
// Generation 0 is no fence's: FenceGeneration answers 0 for a host with
// none, which must never read as ours.
func fenceForRollback(f layout.Fence, swept func() bool) func(ctx context.Context, host string, gen uint64) (bool, error) {
	return func(ctx context.Context, host string, gen uint64) (bool, error) {
		if gen == 0 {
			return false, activator.ErrFenceGeneration
		}
		if f.FenceGeneration(host) == gen {
			return true, nil
		}
		fctx, cancel := context.WithTimeout(ctx, rollbackFenceTimeout)
		defer cancel()
		_, err := f.FenceAndStopIfUnfenced(fctx, host, gen)
		switch {
		case err == nil:
			return true, nil
		case errors.Is(err, activator.ErrFenceHeld):
			return false, err
		case errors.Is(err, activator.ErrStopUnproven) && swept() && f.Quiesced(host) && !f.StopUncertain(host):
			return true, nil
		}
		f.LiftFence(host, gen)
		return false, err
	}
}

// deleteReplica is the rollback's step 6 for the host's replica: every key
// under <prefix>/<host>/ in the region data bucket, the prefix the
// activator replicates to (activator.HibernatePrefix). The trailing slash
// keeps a host whose name only begins with this one out of it. Nothing
// there is not an error.
func deleteReplica(data release.Store, prefix string) func(ctx context.Context, host string) error {
	return func(ctx context.Context, host string) error {
		if !controlname.ValidHostShape(host) {
			return fmt.Errorf("delete the replica of %q: not a host name", host)
		}
		keys, err := data.List(ctx, activator.HibernatePrefix(prefix, host)+"/")
		if err != nil {
			return err
		}
		for _, k := range keys {
			if err := data.Delete(ctx, k); err != nil && !errors.Is(err, os.ErrNotExist) {
				return err
			}
		}
		return nil
	}
}