package hostops

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"sort"
	"strings"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/release"
)

type PurgeObject struct {
	Key       string          `json:"key"`
	ETag      string          `json:"etag"`
	VersionID string          `json:"version_id,omitempty"`
	Body      json.RawMessage `json:"body"`
}
type PurgeHost struct {
	ClaimETag     string                     `json:"claim_etag,omitempty"`
	ClaimGen      uint64                     `json:"claim_gen,omitempty"`
	Holder        *hostclaim.Holder          `json:"holder,omitempty"`
	Registrations map[string]json.RawMessage `json:"registrations,omitempty"`
	RecordETag    string                     `json:"record_etag,omitempty"`
	RecordVersion string                     `json:"record_version,omitempty"`
	FenceState    string                     `json:"fence_state,omitempty"`
	FenceGen      uint64                     `json:"fence_gen,omitempty"`
	FenceETag     string                     `json:"fence_etag,omitempty"`
	Retired       bool                       `json:"retired"`
	FreeFromGen   uint64                     `json:"free_from_gen,omitempty"`
}
type PurgeSweep struct {
	Batch    uint64                `json:"batch"`
	Bucket   string                `json:"bucket"`
	Versions []lease.ObjectVersion `json:"versions"`
	Prepared bool                  `json:"prepared"`
	Deleted  bool                  `json:"deleted"`
	Verified bool                  `json:"verified"`
}
type PurgeCleanup struct {
	Sweep     PurgeSweep `json:"sweep"`
	Tombstone bool       `json:"tombstone"`
	AppRow    bool       `json:"app_row"`
	Fences    bool       `json:"fences"`
	Admission bool       `json:"admission"`
}
type PurgeJournal struct {
	IngestRegistrations map[string]Registration `json:"ingest_registrations,omitempty"`
	Purge
	V              int                        `json:"v"`
	AppID          string                     `json:"app_id"`
	Account        string                     `json:"account"`
	App            string                     `json:"app"`
	RequestedBy    string                     `json:"requested_by"`
	RequestedAt    string                     `json:"requested_at"`
	Reason         string                     `json:"reason"`
	Tombstone      PurgeObject                `json:"tombstone"`
	AdmissionEpoch uint64                     `json:"admission_epoch"`
	Hosts          map[string]PurgeHost       `json:"hosts"`
	Effects        map[string]json.RawMessage `json:"effects"`
	Cleanup        PurgeCleanup               `json:"cleanup"`
}

func ReadPurgeJournal(ctx context.Context, s release.Store, account, app, id, appID string) (PurgeJournal, string, error) {
	key, err := PurgeKey(account, app)
	if err != nil {
		return PurgeJournal{}, "", err
	}
	body, tag, err := s.Get(ctx, key)
	if err != nil {
		return PurgeJournal{}, "", err
	}
	var j PurgeJournal
	if err = decode(body, &j); err != nil {
		return j, "", err
	}
	if tag == "" || j.V != 1 || j.ID != id || j.AppID != appID || j.Account != account || j.App != app || phaseIndex(j.Phase) < 0 || j.Hosts == nil {
		return j, "", ErrIdentity
	}
	if j.Tombstone.Key != release.AppTombstoneKey(account, app) || j.Tombstone.ETag == "" || !json.Valid(j.Tombstone.Body) {
		return j, "", ErrIdentity
	}
	for id, reg := range j.IngestRegistrations {
		if !validIngestRegistration(id, reg, j.AppID, j.AdmissionEpoch) {
			return j, "", ErrIdentity
		}
	}
	return j, tag, nil
}
func purgeExcluded(j PurgeJournal, key string) bool {
	prefix := "apps/" + j.Account + "/" + j.App + "/"
	return key == prefix+"admission.json" || key == prefix+"purge.json" || key == prefix+"purge-request.json" || key == j.Tombstone.Key
}
func PreparePurgeSweep(ctx context.Context, s release.Store, w *ControlWriter, v lease.VersionedStore, bucket, account, app, id, appID string) error {
	if v == nil || bucket == "" {
		return ErrContract
	}
	j, tag, err := ReadPurgeJournal(ctx, s, account, app, id, appID)
	if err != nil {
		return err
	}
	if j.Phase != "retired" {
		return ErrPending
	}
	if j.Cleanup.Sweep.Prepared {
		if j.Cleanup.Sweep.Bucket != bucket {
			return ErrIdentity
		}
		if !j.Cleanup.Sweep.Deleted || j.Cleanup.Sweep.Verified {
			return nil
		}
		// A late old-app auxiliary write needs another persisted exact-version
		// batch. Only the still-retired matching reservation can save it; a
		// stale worker cannot enumerate/delete under done or a recreated app.
	}
	prefix := "apps/" + account + "/" + app + "/"
	versions, err := v.ListVersions(ctx, bucket, prefix)
	if err != nil {
		return err
	}
	if j.Cleanup.Sweep.Batch == ^uint64(0) {
		return ErrIdentity
	}
	batch := PurgeSweep{Batch: j.Cleanup.Sweep.Batch + 1, Bucket: bucket, Prepared: true, Versions: []lease.ObjectVersion{}}
	for _, o := range versions {
		if !strings.HasPrefix(o.Key, prefix) || o.VersionID == "" || o.VersionID == "null" {
			return ErrIdentity
		}
		if purgeExcluded(j, o.Key) {
			continue
		}
		tail := strings.TrimPrefix(o.Key, prefix)
		if strings.HasPrefix(tail, "instances/") || strings.HasPrefix(tail, "sidecars/") {
			// Historical record versions may be swept only after current records
			// have been retired. The generic sweep never removes a live record.
			if _, _, e := s.Get(ctx, o.Key); !errors.Is(e, os.ErrNotExist) {
				if e != nil {
					return e
				}
				return ErrPending
			}
		}
		batch.Versions = append(batch.Versions, o)
	}
	j.Cleanup.Sweep = batch
	_, err = w.SavePurgeJournal(ctx, j, tag)
	return err
}

type retiredRecordSweepKey struct{}

func DeletePurgeSweep(ctx context.Context, s release.Store, w *ControlWriter, v lease.VersionedStore, bucket, account, app, id, appID string) error {
	if v == nil {
		return ErrContract
	}
	j, _, err := ReadPurgeJournal(ctx, s, account, app, id, appID)
	if err != nil {
		return err
	}
	if j.Phase != "retired" || !j.Cleanup.Sweep.Prepared || j.Cleanup.Sweep.Bucket != bucket {
		return ErrPending
	}
	if j.Cleanup.Sweep.Deleted {
		return nil
	}
	batch := append([]lease.ObjectVersion(nil), j.Cleanup.Sweep.Versions...)
	// Replay old journals safely too: retire record data before markers.
	sort.SliceStable(batch, func(i, k int) bool {
		a := provisioningKey(batch[i].Key) && batch[i].DeleteMarker
		b := provisioningKey(batch[k].Key) && batch[k].DeleteMarker
		return !a && b
	})
	for _, o := range batch {
		fresh, _, err := ReadPurgeJournal(ctx, s, account, app, id, appID)
		if err != nil {
			return err
		}
		if fresh.Phase != "retired" || fresh.Cleanup.Sweep.Bucket != bucket || fresh.Cleanup.Sweep.Batch != j.Cleanup.Sweep.Batch {
			return ErrPending
		}
		if o.VersionID == "" || o.VersionID == "null" || purgeExcluded(j, o.Key) {
			return ErrIdentity
		}
		deleteCtx := ctx
		if provisioningKey(o.Key) {
			// Recheck after preparation, before every version deletion. The
			// existing exclusive purge authority excludes recreation.
			if err := retiredRecordVersion(ctx, s, v, bucket, o.Key, o.VersionID); err != nil {
				return err
			}
			deleteCtx = context.WithValue(ctx, retiredRecordSweepKey{}, o.Key)
		}
		if err = v.DeleteVersion(deleteCtx, bucket, o.Key, o.VersionID); err != nil {
			return err
		}
	}
	fresh, tag, err := ReadPurgeJournal(ctx, s, account, app, id, appID)
	if err != nil {
		return err
	}
	if fresh.Phase != "retired" || fresh.Cleanup.Sweep.Batch != j.Cleanup.Sweep.Batch {
		return ErrPending
	}
	fresh.Cleanup.Sweep.Deleted = true
	_, err = w.SavePurgeJournal(ctx, fresh, tag)
	return err
}
func VerifyPurgeSweep(ctx context.Context, s release.Store, w *ControlWriter, v lease.VersionedStore, bucket, account, app, id, appID string) error {
	j, tag, err := ReadPurgeJournal(ctx, s, account, app, id, appID)
	if err != nil {
		return err
	}
	if j.Phase != "retired" || !j.Cleanup.Sweep.Deleted || j.Cleanup.Sweep.Bucket != bucket {
		return ErrPending
	}
	if j.Cleanup.Sweep.Verified {
		return nil
	}
	prefix := "apps/" + account + "/" + app + "/"
	versions, err := v.ListVersions(ctx, bucket, prefix)
	if err != nil {
		return err
	}
	for _, o := range versions {
		if !strings.HasPrefix(o.Key, prefix) || !purgeExcluded(j, o.Key) {
			return ErrPending
		}
	}
	keys, err := s.List(ctx, prefix)
	if err != nil {
		return err
	}
	for _, key := range keys {
		if !purgeExcluded(j, key) {
			return ErrPending
		}
	}
	j.Cleanup.Sweep.Verified = true
	_, err = w.SavePurgeJournal(ctx, j, tag)
	return err
}