package runtime

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/gogf/gf/v2/container/gvar"

	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/contract"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
)

// SimpleReplay hands every source failure to a separate Run. The source hash
// is the checkpoint: a record is removed only after the target's task-state
// confirms that EnqueueRunBatch accepted it. No cursor is retained across
// mutations of the source hash.
//
// The target's retry-round budget is taken from the caller when supplied and
// falls back to the configured default otherwise. It is persisted on the
// source reservation because a worker restart re-enters through reconcile(),
// which no longer has the caller's arguments.
type SimpleReplay struct {
	stores []*zset.Store
	redis  zset.RedisClient
	// defaultMaxRounds applies when a replay request does not specify one.
	defaultMaxRounds int
}

// Replay round bounds. MaxReplayRounds is a server-side guard, not a capacity
// limit: the observed chain of 10-09 needed 74 rounds to reach 98% recovery,
// so a low ceiling would cut off a legitimate convergence path.
//
// DefaultReplayMaxRounds is only the fallback for a request that omits a
// budget; callers may ask for more, up to MaxReplayRounds.
const (
	DefaultReplayMaxRounds = 3
	MaxReplayRounds        = 100
	legacyReplayMaxRounds  = 1
)

type ReplayReceipt struct {
	ReplayRunID string `json:"replayRunId"`
	Status      string `json:"status"`
}

type ReplayStatus struct {
	ReplayRunID string `json:"replayRunId"`
	SourceRunID string `json:"sourceRunId"`
	Status      string `json:"status"`
	Remaining   int64  `json:"remaining"`
	Enqueued    int64  `json:"enqueued"`
	RunStatus   string `json:"runStatus,omitempty"`
	Error       string `json:"error,omitempty"`
}

var ErrNoFailures = errors.New("run has no failures to replay")

func NewSimpleReplay(stores ...*zset.Store) (*SimpleReplay, error) {
	return NewSimpleReplayWithDefault(DefaultReplayMaxRounds, stores...)
}

// NewSimpleReplayWithDefault builds a replay service whose unspecified replay
// requests inherit defaultMaxRounds. A non-positive value falls back to
// DefaultReplayMaxRounds so a missing config key cannot disable the budget.
func NewSimpleReplayWithDefault(defaultMaxRounds int, stores ...*zset.Store) (*SimpleReplay, error) {
	if len(stores) == 0 || stores[0] == nil {
		return nil, fmt.Errorf("replay requires Redis stores")
	}
	for _, store := range stores {
		if store == nil {
			return nil, fmt.Errorf("replay store is nil")
		}
	}
	if defaultMaxRounds <= 0 {
		defaultMaxRounds = DefaultReplayMaxRounds
	}
	if defaultMaxRounds > MaxReplayRounds {
		defaultMaxRounds = MaxReplayRounds
	}
	return &SimpleReplay{stores: stores, redis: stores[0].RedisClient(), defaultMaxRounds: defaultMaxRounds}, nil
}

func (r *SimpleReplay) find(ctx context.Context, id string) (*zset.Store, zset.RunMeta, error) {
	if err := contract.ValidateID(id); err != nil || len(id) > 128 {
		return nil, zset.RunMeta{}, contract.ErrInvalidArgument
	}
	var found *zset.Store
	var meta zset.RunMeta
	for _, store := range r.stores {
		key, _ := store.RunMetaKey(id)
		v, err := r.redis.Do(ctx, "EXISTS", key)
		if err != nil {
			return nil, meta, err
		}
		if v.Int() == 0 {
			continue
		}
		if found != nil {
			return nil, meta, contract.ErrConflict
		}
		found = store
		meta, err = store.GetRunMeta(ctx, id)
		if err != nil {
			return nil, meta, err
		}
	}
	if found == nil {
		return nil, meta, contract.ErrNotFound
	}
	return found, meta, nil
}

// Request accepts a replay of the source run's failures. maxRounds == 0 means
// "use the configured default"; requests above MaxReplayRounds are rejected.
func (r *SimpleReplay) Request(ctx context.Context, sourceID, requestedBy string, maxRounds int) (ReplayReceipt, error) {
	if len(requestedBy) == 0 || len(requestedBy) > 64 {
		return ReplayReceipt{}, contract.ErrInvalidArgument
	}
	if maxRounds < 0 || maxRounds > MaxReplayRounds {
		return ReplayReceipt{}, contract.ErrInvalidArgument
	}
	if maxRounds == 0 {
		maxRounds = r.defaultMaxRounds
	}
	store, _, err := r.find(ctx, sourceID)
	if err != nil {
		return ReplayReceipt{}, err
	}
	metaKey, _ := store.RunMetaKey(sourceID)
	failedKey, _ := store.FailedKey(sourceID)
	var random [16]byte
	if _, err := rand.Read(random[:]); err != nil {
		return ReplayReceipt{}, err
	}
	targetID := "replay-" + hex.EncodeToString(random[:])
	// An index written before the reservation makes the pending source
	// discoverable even if the process stops immediately after the Lua call.
	if _, err = r.redis.Do(ctx, "SADD", store.Prefix()+"replay:pending", sourceID); err != nil {
		return ReplayReceipt{}, err
	}
	v, err := r.redis.Eval(ctx, reserveReplay, 2, []string{metaKey, failedKey}, []any{targetID, requestedBy, maxRounds})
	if err != nil {
		return ReplayReceipt{}, err
	}
	result := v.String()
	// An unsuccessful reservation leaves an inert discovery entry. Never
	// remove it here: a concurrent caller might have reserved the source
	// after this script returned, and that reservation needs recovery.
	switch result {
	case "missing":
		return ReplayReceipt{}, contract.ErrNotFound
	case "running":
		return ReplayReceipt{}, contract.ErrConflict
	case "empty":
		return ReplayReceipt{}, ErrNoFailures
	}
	if !strings.HasPrefix(result, "replay-") {
		return ReplayReceipt{}, fmt.Errorf("unexpected replay reservation result")
	}
	go r.process(store, sourceID)
	return ReplayReceipt{ReplayRunID: result, Status: "accepted"}, nil
}

// Start reconciles pending sources after restarts and missed goroutine wakes.
// Every process may run it: the per-source Redis lease elects one writer.
func (r *SimpleReplay) Start(ctx context.Context) {
	go func() {
		ticker := time.NewTicker(10 * time.Second)
		defer ticker.Stop()
		for {
			r.reconcile(ctx)
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
			}
		}
	}()
}

func (r *SimpleReplay) reconcile(ctx context.Context) {
	for _, store := range r.stores {
		v, err := r.redis.Do(ctx, "SMEMBERS", store.Prefix()+"replay:pending")
		if err != nil {
			continue
		}
		for _, id := range v.Strings() {
			if ctx.Err() != nil {
				return
			}
			key, keyErr := store.RunMetaKey(id)
			if keyErr != nil {
				continue
			}
			reservation, getErr := r.redis.Do(ctx, "HGET", key, "replayTargetId")
			// SADD happens before the reservation so even a crash during
			// acceptance cannot lose work. Unreserved entries are inert.
			if getErr != nil || reservation == nil || reservation.String() == "" {
				continue
			}
			state, stateErr := r.redis.Do(ctx, "HGET", key, "replayState")
			if stateErr != nil || state == nil || state.String() == "needs_attention" {
				continue
			}
			go r.process(store, id)
		}
	}
}

func (r *SimpleReplay) Status(ctx context.Context, targetID string) (ReplayStatus, error) {
	store, target, err := r.find(ctx, targetID)
	if err != nil {
		// The target may not exist yet: locate its durable source reservation.
		if !errors.Is(err, contract.ErrNotFound) {
			return ReplayStatus{}, err
		}
		for _, candidate := range r.stores {
			v, e := r.redis.Do(ctx, "SMEMBERS", candidate.Prefix()+"replay:pending")
			if e != nil {
				return ReplayStatus{}, e
			}
			for _, sourceID := range v.Strings() {
				key, _ := candidate.RunMetaKey(sourceID)
				value, e := r.redis.Do(ctx, "HGET", key, "replayTargetId")
				if e == nil && value != nil && value.String() == targetID {
					store = candidate
					target.SourceRunID = sourceID
					break
				}
			}
		}
		if store == nil {
			return ReplayStatus{}, contract.ErrNotFound
		}
	}
	if target.SourceRunID == "" {
		return ReplayStatus{}, contract.ErrNotFound
	}
	sourceKey, _ := store.RunMetaKey(target.SourceRunID)
	failureKey, _ := store.FailedKey(target.SourceRunID)
	values, err := r.redis.Do(ctx, "HMGET", sourceKey, "replayTargetId", "replayState", "replayError", "replayEnqueued")
	if err != nil {
		return ReplayStatus{}, err
	}
	fields := values.Vars()
	if len(fields) != 4 || fields[0].String() != targetID {
		return ReplayStatus{}, contract.ErrNotFound
	}
	remaining, err := r.redis.Do(ctx, "HLEN", failureKey)
	if err != nil {
		return ReplayStatus{}, err
	}
	return ReplayStatus{ReplayRunID: targetID, SourceRunID: target.SourceRunID, Status: fields[1].String(), Error: fields[2].String(), Enqueued: fields[3].Int64(), Remaining: remaining.Int64(), RunStatus: target.Status}, nil
}

func (r *SimpleReplay) process(store *zset.Store, sourceID string) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Hour)
	defer cancel()
	// Keep one process per source; batches are bounded so a short lease is
	// renewed before it expires. A lost lease stops cleanup at the next batch.
	lockKey := store.Prefix() + "{" + sourceID + "}:replay-lock"
	var token [16]byte
	if _, err := rand.Read(token[:]); err != nil {
		return
	}
	owner := hex.EncodeToString(token[:])
	locked, err := store.AcquireLock(ctx, lockKey, owner, time.Minute)
	if err != nil || !locked {
		return
	}
	defer func() { _, _ = store.ReleaseLock(context.Background(), lockKey, owner) }()
	if err := r.handoff(ctx, store, sourceID, lockKey, owner); err != nil {
		// The time slice ended, not the replay: the reconciler will resume
		// from remaining source records under a new lease.
		if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
			return
		}
		sourceKey, _ := store.RunMetaKey(sourceID)
		// A stale owner must not overwrite a newer worker's status.
		_, _ = r.redis.Eval(context.Background(), replayAttention, 2, []string{sourceKey, lockKey}, []any{owner})
	}
}

func (r *SimpleReplay) handoff(ctx context.Context, store *zset.Store, sourceID, lockKey, owner string) error {
	sourceKey, _ := store.RunMetaKey(sourceID)
	failedKey, _ := store.FailedKey(sourceID)
	meta, err := store.GetRunMeta(ctx, sourceID)
	if err != nil {
		return err
	}
	v, err := r.redis.Do(ctx, "HGET", sourceKey, "replayTargetId")
	if err != nil {
		return err
	}
	if v == nil || v.String() == "" {
		return fmt.Errorf("replay reservation missing")
	}
	targetID := v.String()
	// New reservations persist their budget. Reservations made before this
	// field existed used one round; a restart must not apply today's default
	// to those already accepted runs.
	storedMaxRounds, err := r.redis.Do(ctx, "HGET", sourceKey, "replayMaxRound")
	if err != nil {
		return err
	}
	maxRounds := legacyReplayMaxRounds
	if storedMaxRounds != nil && storedMaxRounds.String() != "" {
		maxRounds = storedMaxRounds.Int()
	}
	if maxRounds <= 0 {
		return fmt.Errorf("replay reservation has invalid round budget")
	}
	if maxRounds > MaxReplayRounds {
		return fmt.Errorf("replay reservation exceeds round budget limit")
	}
	// CreateRun is idempotent for the main hash but writes extension fields
	// again. Read and validate an existing target BEFORE CreateRun, otherwise
	// recovery could silently replace its accepted maxRound.
	existing, err := store.GetRunMeta(ctx, targetID)
	if err != nil && !errors.Is(err, zset.ErrRunNotFound) {
		return err
	}
	if err == nil {
		if storedMaxRounds == nil || storedMaxRounds.String() == "" {
			maxRounds = existing.MaxRound
		}
		if maxRounds <= 0 || maxRounds > MaxReplayRounds || existing.SourceRunID != sourceID || existing.SyncType != "replay" || existing.MaxRound != maxRounds || existing.Operation != meta.Operation || existing.QueryMonth != meta.QueryMonth {
			return fmt.Errorf("replay target metadata is inconsistent")
		}
	}
	target := zset.RunMeta{RunID: targetID, Operation: meta.Operation, QueryMonth: meta.QueryMonth, SyncType: "replay", SourceRunID: sourceID, ReplayCycle: meta.ReplayCycle + 1, TriggerSource: "replay", MaxRound: maxRounds}
	if err := store.CreateRun(ctx, target); err != nil {
		return err
	}
	actual, err := store.GetRunMeta(ctx, targetID)
	if err != nil || actual.SourceRunID != sourceID || actual.SyncType != "replay" || actual.MaxRound != maxRounds || actual.Operation != meta.Operation || actual.QueryMonth != meta.QueryMonth {
		return fmt.Errorf("replay target metadata is inconsistent")
	}
	// The target task-state is the admission receipt during recovery. It may
	// not expire while source failures are still being transferred.
	targetKey, _ := store.RunMetaKey(targetID)
	stateKey, _ := store.TaskStateKey(targetID)
	if !actual.ProducerDone {
		if _, err := r.redis.Do(ctx, "PERSIST", targetKey); err != nil {
			return err
		}
		if _, err := r.redis.Do(ctx, "PERSIST", stateKey); err != nil {
			return err
		}
	}
	_, err = r.redis.Do(ctx, "HSET", sourceKey, "replayState", "processing", "replayError", "")
	if err != nil {
		return err
	}
	for {
		if ctx.Err() != nil {
			return ctx.Err()
		}
		valid, err := store.RenewLock(ctx, lockKey, owner, time.Minute)
		if err != nil || !valid {
			return fmt.Errorf("replay lease lost: %w", err)
		}
		// HSCAN is not a snapshot. Complete a bounded cursor pass, then
		// recheck HLEN and repeat from zero if deletion changed the scan.
		var cursor uint64
		for {
			v, e := r.redis.Do(ctx, "HSCAN", failedKey, cursor, "COUNT", 100)
			if e != nil {
				return e
			}
			var scanPairs []*gvar.Var
			cursor, scanPairs, e = parseScanResult(v)
			if e != nil {
				return e
			}
			if len(scanPairs) > 0 {
				for i := 0; i < len(scanPairs); i += 2 {
					valid, renewErr := store.RenewLock(ctx, lockKey, owner, time.Minute)
					if renewErr != nil || !valid {
						return fmt.Errorf("replay lease lost")
					}
					field, raw := scanPairs[i].String(), scanPairs[i+1].String()
					var rec zset.FailureRecord
					if json.Unmarshal([]byte(raw), &rec) != nil || rec.TaskID == "" || rec.CardID <= 0 || rec.ICCID == "" || rec.ChanKey == "" || rec.TempKey == "" || field != rec.TaskID || (rec.LatestRunID != "" && rec.LatestRunID != sourceID) {
						return fmt.Errorf("source failure record cannot be replayed")
					}
					operation := rec.Operation
					if operation == "" {
						operation = meta.Operation
					}
					// Source task identity (not just card ID) prevents collisions
					// between distinct failures for the same card.
					taskID := targetID + ":" + rec.TaskID
					queryMonth := rec.QueryMonth
					if queryMonth == "" {
						queryMonth = meta.QueryMonth
					}
					task := zset.Task{RunID: targetID, TaskID: taskID, Operation: operation, QueryMonth: queryMonth, CardID: rec.CardID, ICCID: rec.ICCID, MSISDN: rec.MSISDN, IMSI: rec.IMSI, ChanKey: rec.ChanKey, TempKey: rec.TempKey, PackageCode: rec.PackageCode}
					if _, e = store.EnqueueRunBatch(ctx, targetID, rec.ChanKey, []zset.Task{task}, time.Now()); e != nil {
						return e
					}
					confirmed, e := r.redis.Do(ctx, "HEXISTS", stateKey, taskID)
					if e != nil || confirmed == nil || confirmed.Int() != 1 {
						return fmt.Errorf("target enqueue unconfirmed: %w", e)
					}
					removed, e := r.redis.Eval(ctx, cleanupReplayFailure, 3, []string{sourceKey, failedKey, lockKey}, []any{field, raw, targetID, owner})
					if e != nil {
						return e
					}
					if removed.Int() == -1 {
						return fmt.Errorf("source failure changed during replay")
					}
				}
			}
			if cursor == 0 {
				break
			}
		}
		remaining, err := r.redis.Do(ctx, "HLEN", failedKey)
		if err != nil {
			return err
		}
		if remaining.Int64() == 0 {
			break
		}
	}
	// Persist producerDone only after every source record was confirmed.
	if err := store.MarkProducerDone(ctx, targetID); err != nil {
		return err
	}
	if _, err := r.redis.Do(ctx, "HSET", sourceKey, "replayState", "completed", "replayError", ""); err != nil {
		return err
	}
	// The source summary has already been archived. Keep the small source
	// metadata for lookup but restore a bounded retention after handoff.
	if _, err := r.redis.Do(ctx, "EXPIRE", sourceKey, 30*24*3600); err != nil {
		return err
	}
	_, err = r.redis.Do(ctx, "SREM", store.Prefix()+"replay:pending", sourceID)
	return err
}

const replayAttention = `
if redis.call('GET',KEYS[2])~=ARGV[1] then return 0 end
redis.call('HSET',KEYS[1],'replayState','needs_attention','replayError','Replay handoff stopped; check Worker logs')
return 1
`

const reserveReplay = `
local id=redis.call('HGET',KEYS[1],'replayTargetId')
if id then return id end
local status=redis.call('HGET',KEYS[1],'status')
if not status then return 'missing' end
if status~='completed' and status~='completed_with_failures' then return 'running' end
if redis.call('HLEN',KEYS[2])==0 then return 'empty' end
redis.call('HSET',KEYS[1],'replayTargetId',ARGV[1],'replayState','accepted','replayRequestedBy',ARGV[2],'replayEnqueued',0,'replayMaxRound',ARGV[3])
redis.call('PERSIST',KEYS[1]); redis.call('PERSIST',KEYS[2])
return ARGV[1]
`
const cleanupReplayFailure = `
if redis.call('HGET',KEYS[1],'replayTargetId')~=ARGV[3] or redis.call('GET',KEYS[3])~=ARGV[4] then return -1 end
local value=redis.call('HGET',KEYS[2],ARGV[1])
if not value then return 0 end
if value~=ARGV[2] then return -1 end
redis.call('HDEL',KEYS[2],ARGV[1]); redis.call('HINCRBY',KEYS[1],'replayEnqueued',1)
redis.call('HSET',KEYS[1],'pendingFailureCount',redis.call('HLEN',KEYS[2]))
return 1
`