package dispatcher

import (
	"context"
	"errors"
	"log/slog"
	"strings"
	"time"

	gittransporthttp "github.com/go-git/go-git/v5/plumbing/transport/http"

	"github.com/tidbcloud/AgentTeam/desktop/internal/memorywatcher"
	"github.com/tidbcloud/AgentTeam/desktop/internal/nativememory"
	"github.com/tidbcloud/AgentTeam/desktop/internal/vault"
)

const (
	idleMemoryRetryMin           = time.Second
	idleMemoryRetryMax           = time.Minute
	idleMemoryRetryMaxConcurrent = 4
)

type idleMemoryRetry struct {
	timer   *time.Timer
	attempt int
	running bool
	cancel  context.CancelFunc
}

type idleMemoryRetryContextKey struct{}

// startMemoryWatcherForAccess constructs and publishes the watcher that owns
// the serialized local commit and cloud push pipeline for an active session.
func (d *memoryService) startMemoryWatcherForAccess(ctx context.Context, owner *memoryWatcherMutationOwner, sess *interactiveSession, agentID string, access MemoryAccess, clone *vault.Clone, nativeSync *nativememory.Sync, notify memorywatcher.ChangeListener) bool {
	var watcher *memorywatcher.Watcher
	watcherStatus := func(status memorywatcher.Status) {
		// Ignore callbacks before publication and from a detached/replaced
		// watcher. The detached owner publishes its final push result instead.
		sess.mu.Lock()
		current := sess.watcher
		draining := current == nil && sess.memoryWatcherDrain != nil
		sess.mu.Unlock()
		if current == watcher || draining {
			d.setMemorySyncStatus(agentID, status)
		}
	}
	var err error
	var committer memorywatcher.Committer = memorywatcher.NewCloneCommitter(clone)
	var pusher memorywatcher.Pusher = memorywatcher.NewClonePusher(clone, access.Auth)
	var syncer memorywatcher.Syncer = memorywatcher.NewCloneSyncer(clone, access.Auth)
	if access.Protocol == MemorySyncProtocolNativeContentV1 {
		if nativeSync == nil || clone.UsesRemote() {
			d.setMemorySyncStatus(agentID, degradedMemorySyncStatus(memorywatcher.StatusReasonWatcherFailed))
			return false
		}
		nativeCommitter := newNativeMemoryCommitter(clone, nativeSync)
		if err := nativeCommitter.Recover(ctx); err != nil {
			d.setMemorySyncStatus(agentID, degradedMemorySyncStatus(memorywatcher.StatusReasonCommitFailed))
			return false
		}
		committer = nativeCommitter
		pusher = nativeMemoryPusher{sync: nativeSync}
		syncer = nil
	}
	watcher, err = memorywatcher.New(memorywatcher.Options{
		MemoryDir:    clone.Path(),
		Committer:    committer,
		Resolve:      d.buildMemoryContextResolver(sess),
		Push:         pusher,
		Sync:         syncer,
		Notify:       notify,
		Status:       watcherStatus,
		RetryOnStart: true,
	})
	if err != nil {
		d.setMemorySyncStatus(agentID, degradedMemorySyncStatus(memorywatcher.StatusReasonWatcherFailed))
		slog.WarnContext(ctx, "memory watcher: construct failed; continuing without watcher",
			"agent_id", agentID, "error", err)
		return false
	}
	if err := watcher.Start(ctx); err != nil {
		d.setMemorySyncStatus(agentID, degradedMemorySyncStatus(memorywatcher.StatusReasonWatcherFailed))
		slog.WarnContext(ctx, "memory watcher: start failed; continuing without watcher",
			"agent_id", agentID, "error", err)
		_ = watcher.Close()
		return false
	}
	if beforePublish := d.memoryWatcherBeforePublish; beforePublish != nil {
		beforePublish(ctx, sess)
	}
	if !d.publishMemoryWatcher(owner, sess, clone, access, nativeSync, watcher) {
		// Publication lost the shutdown generation fence, the session was
		// retired, or another watcher already owns the session. This watcher was
		// started, so its owner must synchronously flush/close it before finishing;
		// shutdown waits that owner before disabling the search reconciler.
		if err := watcher.FlushAndClose(); err != nil {
			slog.WarnContext(ctx, "memory watcher: rejected watcher close failed",
				"agent_id", agentID, "error", err)
		}
		return false
	}
	d.finishIdleMemoryRetry(agentID, ctx)
	d.setMemorySyncStatus(agentID, watcher.Status())
	slog.InfoContext(ctx, "memory watcher attached",
		"agent_id", agentID,
		"memory_id", access.MemoryID,
		"path", clone.Path(),
	)
	return true
}

// MemorySyncStatus returns the current process-local Cloud Memory sync state.
// It never reads or writes TiDB and does not alter the server-owned Last saved
// timestamp, which continues to represent a real upstream repository commit.
func (d *memoryService) MemorySyncStatus(agentID string) (memorywatcher.Status, bool) {
	if d == nil || d.Dispatcher == nil {
		return memorywatcher.Status{}, false
	}
	agentID = strings.TrimSpace(agentID)
	if agentID == "" {
		return memorywatcher.Status{}, false
	}
	d.memorySyncStatusMu.RLock()
	lookupStatus, lookupFailed := d.memoryLookupStatuses[agentID]
	d.memorySyncStatusMu.RUnlock()
	if lookupFailed {
		return lookupStatus, true
	}
	if sess, ok := d.sessionManager.FindByLookupID(agentID); ok && sess != nil {
		sess.mu.Lock()
		watcher := sess.watcher
		sess.mu.Unlock()
		if watcher != nil {
			return watcher.Status(), true
		}
	}
	d.memorySyncStatusMu.RLock()
	status, ok := d.memorySyncStatuses[agentID]
	d.memorySyncStatusMu.RUnlock()
	return status, ok
}

// RetryMemorySync requests immediate recovery on an active watcher. If no
// watcher exists, it reuses the authoritative access refresh path to recover
// lookup, clone, and watcher setup failures.
func (d *memoryService) RetryMemorySync(_ context.Context, agentID string) {
	if d == nil || d.Dispatcher == nil {
		return
	}
	agentID = strings.TrimSpace(agentID)
	if agentID == "" {
		return
	}
	if _, ok := d.MemorySyncStatus(agentID); !ok {
		return
	}
	var watcher *memorywatcher.Watcher
	if sess, ok := d.sessionManager.FindByLookupID(agentID); ok && sess != nil {
		sess.mu.Lock()
		blocked := sess.retired.Load() || sess.resetting
		watcher = sess.watcher
		sess.mu.Unlock()
		if blocked {
			return
		}
	}
	d.memorySyncStatusMu.RLock()
	_, lookupFailed := d.memoryLookupStatuses[agentID]
	d.memorySyncStatusMu.RUnlock()
	if lookupFailed {
		d.requestIdleMemoryRetry(agentID)
		return
	}
	if watcher != nil {
		watcher.RetrySync()
		return
	}
	d.requestIdleMemoryRetry(agentID)
}

func (d *memoryService) setMemoryLookupStatus(agentID string, status memorywatcher.Status) {
	if d == nil || d.Dispatcher == nil {
		return
	}
	agentID = strings.TrimSpace(agentID)
	if agentID == "" {
		return
	}
	d.memorySyncStatusMu.Lock()
	if d.memoryLookupStatuses == nil {
		d.memoryLookupStatuses = make(map[string]memorywatcher.Status)
	}
	d.memoryLookupStatuses[agentID] = status
	d.memorySyncStatusMu.Unlock()
}

func (d *memoryService) clearMemoryLookupStatus(agentID string) {
	if d == nil || d.Dispatcher == nil {
		return
	}
	d.memorySyncStatusMu.Lock()
	delete(d.memoryLookupStatuses, strings.TrimSpace(agentID))
	d.memorySyncStatusMu.Unlock()
}

func (d *memoryService) setMemorySyncStatus(agentID string, status memorywatcher.Status) {
	if d == nil || d.Dispatcher == nil {
		return
	}
	agentID = strings.TrimSpace(agentID)
	if agentID == "" {
		return
	}
	d.memorySyncStatusMu.Lock()
	if d.memorySyncStatuses == nil {
		d.memorySyncStatuses = make(map[string]memorywatcher.Status)
	}
	d.memorySyncStatuses[agentID] = status
	d.memorySyncStatusMu.Unlock()
}

func (d *memoryService) clearMemorySyncStatus(agentID string) {
	if d == nil || d.Dispatcher == nil {
		return
	}
	d.memorySyncStatusMu.Lock()
	delete(d.memorySyncStatuses, strings.TrimSpace(agentID))
	d.memorySyncStatusMu.Unlock()
}

func (d *memoryService) scheduleIdleMemoryRetry(agentID string) {
	if d == nil || d.Dispatcher == nil {
		return
	}
	agentID = strings.TrimSpace(agentID)
	if agentID == "" {
		return
	}
	d.memoryIdleRetryMu.Lock()
	defer d.memoryIdleRetryMu.Unlock()
	if d.memoryIdleRetryClosed || d.memoryIdleRetryBlockedLocked(agentID) {
		return
	}
	retry := d.ensureIdleMemoryRetryLocked(agentID)
	d.scheduleIdleMemoryRetryLocked(agentID, retry)
}

// requestIdleMemoryRetry coalesces explicit recovery onto the same bounded
// worker used by automatic retries without blocking the requesting API call.
func (d *memoryService) requestIdleMemoryRetry(agentID string) {
	if d == nil || d.Dispatcher == nil {
		return
	}
	agentID = strings.TrimSpace(agentID)
	if agentID == "" {
		return
	}
	d.memoryIdleRetryMu.Lock()
	defer d.memoryIdleRetryMu.Unlock()
	if d.memoryIdleRetryClosed || d.memoryIdleRetryBlockedLocked(agentID) {
		return
	}
	retry := d.ensureIdleMemoryRetryLocked(agentID)
	if retry.running {
		return
	}
	if retry.timer != nil {
		retry.timer.Stop()
	}
	retry.timer = time.AfterFunc(0, func() { d.runIdleMemoryRetry(agentID, retry) })
}

func (d *memoryService) ensureIdleMemoryRetryLocked(agentID string) *idleMemoryRetry {
	if d.memoryIdleRetries == nil {
		d.memoryIdleRetries = make(map[string]*idleMemoryRetry)
	}
	if d.memoryIdleRetrySlots == nil {
		d.memoryIdleRetrySlots = make(chan struct{}, idleMemoryRetryMaxConcurrent)
	}
	retry := d.memoryIdleRetries[agentID]
	if retry == nil {
		retry = &idleMemoryRetry{}
		d.memoryIdleRetries[agentID] = retry
	}
	return retry
}

func (d *memoryService) scheduleIdleMemoryRetryLocked(agentID string, retry *idleMemoryRetry) {
	if retry == nil || retry.timer != nil || retry.running || d.memoryIdleRetryClosed {
		return
	}
	delay := idleMemoryRetryDelay(agentID, retry.attempt)
	retry.attempt++
	retry.timer = time.AfterFunc(delay, func() { d.runIdleMemoryRetry(agentID, retry) })
}

func (d *memoryService) runIdleMemoryRetry(agentID string, retry *idleMemoryRetry) {
	retryCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	ctx := context.WithValue(retryCtx, idleMemoryRetryContextKey{}, retry)
	d.memoryIdleRetryMu.Lock()
	current := d.memoryIdleRetries[agentID]
	if d.memoryIdleRetryClosed || d.memoryIdleRetryBlockedLocked(agentID) || current != retry || retry.running {
		d.memoryIdleRetryMu.Unlock()
		cancel()
		return
	}
	retry.timer = nil
	retry.running = true
	retry.cancel = cancel
	slots := d.memoryIdleRetrySlots
	d.memoryIdleRetryMu.Unlock()

	select {
	case slots <- struct{}{}:
		d.RefreshMemoryAccess(ctx, []string{agentID})
		<-slots
	case <-ctx.Done():
	}
	cancel()

	// Lookup failures do not reach preCloneMemory, so keep the existing retry
	// alive here. Success, unbind, or watcher takeover removes the map entry.
	d.memoryIdleRetryMu.Lock()
	if d.memoryIdleRetries[agentID] == retry {
		retry.running = false
		retry.cancel = nil
		d.scheduleIdleMemoryRetryLocked(agentID, retry)
	}
	d.memoryIdleRetryMu.Unlock()
}

func (d *memoryService) cancelIdleMemoryRetry(agentID string) {
	d.removeIdleMemoryRetry(agentID, nil)
}

// blockIdleMemoryRetry makes a full reset terminal for status-driven recovery.
// The block and removal share memoryIdleRetryMu with enqueue, so a retry that
// observed stale status before reset cannot enqueue after the reset boundary.
// A later real session start clears the block before attaching its watcher.
func (d *memoryService) blockIdleMemoryRetry(agentID string) {
	if d == nil || d.Dispatcher == nil {
		return
	}
	agentID = strings.TrimSpace(agentID)
	if agentID == "" {
		return
	}
	d.memoryIdleRetryMu.Lock()
	if d.memoryIdleRetryBlocks == nil {
		d.memoryIdleRetryBlocks = make(map[string]struct{})
	}
	d.memoryIdleRetryBlocks[agentID] = struct{}{}
	retry := d.memoryIdleRetries[agentID]
	delete(d.memoryIdleRetries, agentID)
	if retry != nil && retry.timer != nil {
		retry.timer.Stop()
	}
	if retry != nil && retry.cancel != nil {
		retry.cancel()
	}
	d.memoryIdleRetryMu.Unlock()
}

func (d *memoryService) allowIdleMemoryRetry(agentID string) {
	if d == nil || d.Dispatcher == nil {
		return
	}
	d.memoryIdleRetryMu.Lock()
	delete(d.memoryIdleRetryBlocks, strings.TrimSpace(agentID))
	d.memoryIdleRetryMu.Unlock()
}

func (d *memoryService) memoryIdleRetryBlockedLocked(agentID string) bool {
	_, blocked := d.memoryIdleRetryBlocks[agentID]
	return blocked
}

func (d *memoryService) finishIdleMemoryRetry(agentID string, ctx context.Context) {
	var current *idleMemoryRetry
	if ctx != nil {
		current, _ = ctx.Value(idleMemoryRetryContextKey{}).(*idleMemoryRetry)
	}
	d.removeIdleMemoryRetry(agentID, current)
}

func (d *memoryService) removeIdleMemoryRetry(agentID string, current *idleMemoryRetry) {
	if d == nil || d.Dispatcher == nil {
		return
	}
	agentID = strings.TrimSpace(agentID)
	d.memoryIdleRetryMu.Lock()
	retry := d.memoryIdleRetries[agentID]
	delete(d.memoryIdleRetries, agentID)
	if retry != nil && retry.timer != nil {
		retry.timer.Stop()
	}
	if retry != nil && retry != current && retry.cancel != nil {
		retry.cancel()
	}
	d.memoryIdleRetryMu.Unlock()
}

func (d *memoryService) closeIdleMemoryRetries() {
	if d == nil || d.Dispatcher == nil {
		return
	}
	d.memoryIdleRetryMu.Lock()
	d.memoryIdleRetryClosed = true
	for agentID, retry := range d.memoryIdleRetries {
		if retry != nil && retry.timer != nil {
			retry.timer.Stop()
		}
		if retry != nil && retry.cancel != nil {
			retry.cancel()
		}
		delete(d.memoryIdleRetries, agentID)
	}
	d.memoryIdleRetryMu.Unlock()
}

func idleMemoryRetryDelay(agentID string, attempt int) time.Duration {
	delay := idleMemoryRetryMin
	for range attempt {
		if delay > idleMemoryRetryMax/2 {
			delay = idleMemoryRetryMax
			break
		}
		delay *= 2
	}
	if delay > idleMemoryRetryMax {
		delay = idleMemoryRetryMax
	}
	// Stable per-agent jitter spreads a catalog-wide outage without requiring
	// shared randomness or allowing the delay to exceed the configured cap.
	spread := delay / 4
	if spread <= 0 {
		return delay
	}
	hash := uint64(1469598103934665603)
	for i := 0; i < len(agentID); i++ {
		hash ^= uint64(agentID[i])
		hash *= 1099511628211
	}
	hash ^= uint64(attempt + 1)
	hash *= 1099511628211
	return delay - spread + time.Duration(hash%uint64(spread))
}

func degradedMemorySyncStatus(reason memorywatcher.StatusReason) memorywatcher.Status {
	return memorywatcher.Status{State: memorywatcher.StatusDegraded, Reason: reason}
}

func syncIdleMemoryAccess(ctx context.Context, clone *vault.Clone, nativeSync *nativememory.Sync, agentID string, auth gittransporthttp.AuthMethod) (memorywatcher.Status, error) {
	if nativeSync == nil {
		return syncIdleMemoryClone(ctx, clone, agentID, auth)
	}
	committer := newNativeMemoryCommitter(clone, nativeSync)
	if err := committer.Recover(ctx); err != nil {
		return degradedMemorySyncStatus(memorywatcher.StatusReasonCommitFailed), err
	}
	commitContext := memorywatcher.CommitContext{AgentID: agentID, ActorName: agentID}
	_, err := committer.CommitPending(ctx, memorywatcher.BuildCommitMessage(commitContext), memorywatcher.CommitOpts{AuthorName: agentID})
	if err != nil && !errors.Is(err, memorywatcher.ErrNothingToCommit) {
		return degradedMemorySyncStatus(memorywatcher.StatusReasonCommitFailed), err
	}
	_, err = (nativeMemoryPusher{sync: nativeSync}).Push(ctx, memorywatcher.PushOpts{})
	if err != nil && !errors.Is(err, memorywatcher.ErrNothingToPush) {
		return memorywatcher.Status{State: memorywatcher.StatusPendingSync, Reason: memorywatcher.StatusReasonPushFailed}, err
	}
	return memorywatcher.Status{State: memorywatcher.StatusReady}, nil
}

// syncIdleMemoryClone recovers uncommitted files and local commits while no
// interactive session owns a watcher. The caller holds the per-agent mutation
// lock, so this one-shot pipeline cannot overlap another Git writer.
func syncIdleMemoryClone(
	ctx context.Context,
	clone *vault.Clone,
	agentID string,
	auth gittransporthttp.AuthMethod,
) (memorywatcher.Status, error) {
	committer := memorywatcher.NewCloneCommitter(clone)
	commitContext := memorywatcher.CommitContext{AgentID: agentID, ActorName: agentID}
	_, err := committer.CommitPending(ctx, memorywatcher.BuildCommitMessage(commitContext), memorywatcher.CommitOpts{
		AuthorName: agentID,
	})
	if err != nil && !errors.Is(err, memorywatcher.ErrNothingToCommit) {
		return degradedMemorySyncStatus(memorywatcher.StatusReasonCommitFailed), err
	}

	pusher := memorywatcher.NewClonePusher(clone, auth)
	if _, err := pusher.Push(ctx, memorywatcher.PushOpts{}); err != nil && !errors.Is(err, memorywatcher.ErrNothingToPush) {
		return memorywatcher.Status{
			State:  memorywatcher.StatusPendingSync,
			Reason: memorywatcher.FailureReason(err, memorywatcher.StatusReasonPushFailed),
		}, err
	}
	return memorywatcher.Status{State: memorywatcher.StatusReady}, nil
}