package connector_outboundintent

import (
	"context"
	"fmt"
	"time"

	"github.com/pingcap-inc/nexuscrm-connectors/internal/dependencyretry"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

const (
	defaultGenerationProbeInterval   = 500 * time.Millisecond
	defaultRecoveryReconcileInterval = 30 * time.Second
	defaultLegacyPollInterval        = 250 * time.Millisecond
	maxWorkerBurst                   = 8
)

// Service owns the complete Integration-side lifecycle without creating a new
// process or network service. The Runtime generation is only a wake cursor;
// ProcessBatch and the durable Integration queue remain the authoritative
// handoff and delivery paths.
type Service struct {
	acceptor                    *Acceptor
	worker                      *Worker
	generationProbeInterval     time.Duration
	recoveryReconcileInterval   time.Duration
	legacyPollInterval          time.Duration
	dependencyReconcileInterval time.Duration
}

func NewService(acceptor *Acceptor, worker *Worker) (*Service, error) {
	if acceptor == nil || worker == nil {
		return nil, ErrInvalidInput
	}
	return &Service{
		acceptor: acceptor, worker: worker,
		generationProbeInterval:     defaultGenerationProbeInterval,
		recoveryReconcileInterval:   defaultRecoveryReconcileInterval,
		legacyPollInterval:          defaultLegacyPollInterval,
		dependencyReconcileInterval: defaultDependencyReconcileInterval,
	}, nil
}

func (s *Service) supportsGenerationScheduling() bool {
	if s == nil || s.acceptor == nil || s.worker == nil {
		return false
	}
	_, generation := s.acceptor.source.(intentGenerationSource)
	_, schedule := s.worker.store.(commandScheduleStore)
	return generation && schedule
}

// runLegacy preserves the established public Service contract for adapters
// that have not adopted both additive scheduling capabilities. Production
// repositories use generation scheduling; compatibility adapters retain the
// previous bounded poll instead of failing startup or waiting 30 seconds.
func (s *Service) runLegacy(ctx context.Context, started func()) error {
	// Legacy adapters lack the separate authority/schedule probes. Preserve
	// their established first-turn startup contract instead of reporting ready
	// before Runtime receipts or the worker path have been checked.
	if _, err := s.processAcceptanceBatch(ctx); err != nil {
		return fmt.Errorf("connector outbound intent acceptance startup probe: %w", err)
	}
	if _, err := s.worker.processWithRecovery(ctx, true); err != nil {
		return fmt.Errorf("connector outbound intent worker startup probe: %w", err)
	}
	started()
	var dependencyTicker *time.Ticker
	var dependencyTick <-chan time.Time
	if _, ok := s.worker.store.(dependencyReconcileStore); ok {
		dependencyTicker = time.NewTicker(s.effectiveDependencyReconcileInterval())
		dependencyTick = dependencyTicker.C
		defer dependencyTicker.Stop()
	}
	dependencyCursor := ""
	for {
		if err := ctx.Err(); err != nil {
			return err
		}
		select {
		case <-dependencyTick:
			nextCursor, more, _, reconcileErr := s.reconcileDependencies(ctx, dependencyCursor)
			if reconcileErr != nil {
				return reconcileErr
			}
			if more {
				dependencyCursor = nextCursor
			} else {
				dependencyCursor = ""
			}
		default:
		}
		accepted := 0
		err := dependencyretry.Do(ctx, "outbound acceptance", func() error {
			var err error
			accepted, err = s.acceptor.ProcessBatch(ctx)
			return err
		})
		if err != nil {
			return err
		}
		worked, err := s.worker.processWithRecovery(ctx, true)
		if err != nil {
			return err
		}
		if accepted != 0 || worked {
			continue
		}
		timer := time.NewTimer(s.legacyPollInterval)
		select {
		case <-ctx.Done():
			if !timer.Stop() {
				<-timer.C
			}
			return ctx.Err()
		case <-timer.C:
		}
	}
}

func (s *Service) effectiveDependencyReconcileInterval() time.Duration {
	if s != nil && s.dependencyReconcileInterval > 0 {
		return s.dependencyReconcileInterval
	}
	return defaultDependencyReconcileInterval
}

func (s *Service) reconcileDependencies(
	ctx context.Context,
	cursor string,
) (string, bool, int, error) {
	if s == nil || s.worker == nil {
		return "", false, 0, ErrDependency
	}
	return s.worker.reconcileDependencies(ctx, cursor)
}

func (s *Service) processAcceptanceBatch(ctx context.Context) (bool, error) {
	if err := ctx.Err(); err != nil {
		return false, err
	}
	accepted := 0
	err := dependencyretry.Do(ctx, "outbound acceptance", func() error {
		var err error
		accepted, err = s.acceptor.ProcessBatch(ctx)
		return err
	})
	if err != nil {
		return false, err
	}
	// A full response may have more Runtime-owned intents behind it. Continue
	// through an in-memory wake only after the provider worker gets its bounded
	// turn; never drain an unbounded producer queue ahead of delivery.
	return accepted == outboundintentv1.MaxIntentsPerList, nil
}

func (s *Service) drainWorker(ctx context.Context) (time.Time, bool, error) {
	for processed := 0; processed < maxWorkerBurst; processed++ {
		worked, err := s.worker.processWithRecovery(ctx, false)
		if err != nil {
			return time.Time{}, false, err
		}
		if !worked {
			next, found, scheduleErr := s.worker.nextWorkAt(ctx)
			if scheduleErr != nil {
				// The exact schedule is an optimization hint, not work
				// authority. Preserve liveness through the established bounded
				// poll and retry the schedule read on that next turn.
				return time.Now().Add(s.legacyPollInterval), true, nil
			}
			return next, found, nil
		}
	}
	// Yield after a bounded positive burst so a slow provider backlog cannot
	// starve Runtime generation probes. The immediate local wake performs the
	// next authoritative claim without polling either queue table.
	return time.Now(), true, nil
}

func resetScheduleTimer(timer **time.Timer, wake *<-chan time.Time, next time.Time, found bool) {
	if *timer != nil {
		if !(*timer).Stop() {
			select {
			case <-(*timer).C:
			default:
			}
		}
		*timer = nil
		*wake = nil
	}
	if !found {
		return
	}
	delay := time.Until(next)
	if delay < 0 {
		delay = 0
	}
	*timer = time.NewTimer(delay)
	*wake = (*timer).C
}

func requestAcceptanceContinuation(wake chan<- struct{}, more bool) {
	if !more {
		return
	}
	select {
	case wake <- struct{}{}:
	default:
	}
}

func (s *Service) processAcceptanceAndWorker(ctx context.Context) (time.Time, bool, bool, error) {
	more, err := s.processAcceptanceBatch(ctx)
	if err != nil {
		return time.Time{}, false, false, err
	}
	nextWorkAt, scheduled, err := s.drainWorker(ctx)
	return nextWorkAt, scheduled, more, err
}

func (s *Service) Run(ctx context.Context, started func()) error {
	if s == nil || s.acceptor == nil || s.worker == nil || ctx == nil || started == nil ||
		s.generationProbeInterval <= 0 || s.recoveryReconcileInterval <= 0 || s.legacyPollInterval <= 0 {
		return ErrDependency
	}
	if err := dependencyretry.Do(ctx, "outbound schema readiness", func() error { return s.acceptor.VerifyReady(ctx) }); err != nil {
		return fmt.Errorf("connector outbound intent schema readiness: %w", err)
	}
	if !s.supportsGenerationScheduling() {
		return s.runLegacy(ctx, started)
	}
	generation, err := s.readGeneration(ctx, 0)
	if err != nil {
		return fmt.Errorf("connector outbound intent generation startup probe: %w", err)
	}
	// Readiness establishes schema and Runtime authority, not provider/backlog
	// progress. Business work starts below and remains subject to every durable
	// claim, identity and dispatch checkpoint.
	started()

	generationTicker := time.NewTicker(s.generationProbeInterval)
	defer generationTicker.Stop()
	reconcileTicker := time.NewTicker(s.recoveryReconcileInterval)
	defer reconcileTicker.Stop()
	var dependencyTicker *time.Ticker
	var dependencyTick <-chan time.Time
	if _, ok := s.worker.store.(dependencyReconcileStore); ok {
		dependencyTicker = time.NewTicker(s.effectiveDependencyReconcileInterval())
		dependencyTick = dependencyTicker.C
		defer dependencyTicker.Stop()
	}
	dependencyCursor := ""
	var scheduleTimer *time.Timer
	var scheduleWake <-chan time.Time
	acceptanceWake := make(chan struct{}, 1)
	// Always consume an initial turn even if the generation never changes.
	requestAcceptanceContinuation(acceptanceWake, true)
	defer func() {
		if scheduleTimer != nil {
			scheduleTimer.Stop()
		}
	}()

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-generationTicker.C:
			current, readErr := s.readGeneration(ctx, generation)
			if readErr != nil {
				return readErr
			}
			if current == generation {
				continue
			}
			generation = current
			nextWorkAt, scheduled, more, drainErr := s.processAcceptanceAndWorker(ctx)
			if drainErr != nil {
				return drainErr
			}
			resetScheduleTimer(&scheduleTimer, &scheduleWake, nextWorkAt, scheduled)
			requestAcceptanceContinuation(acceptanceWake, more)
		case <-acceptanceWake:
			nextWorkAt, scheduled, more, drainErr := s.processAcceptanceAndWorker(ctx)
			if drainErr != nil {
				return drainErr
			}
			resetScheduleTimer(&scheduleTimer, &scheduleWake, nextWorkAt, scheduled)
			requestAcceptanceContinuation(acceptanceWake, more)
		case <-scheduleWake:
			nextWorkAt, scheduled, drainErr := s.drainWorker(ctx)
			if drainErr != nil {
				return drainErr
			}
			resetScheduleTimer(&scheduleTimer, &scheduleWake, nextWorkAt, scheduled)
		case <-reconcileTicker.C:
			nextWorkAt, scheduled, more, reconcileErr := s.processAcceptanceAndWorker(ctx)
			if reconcileErr != nil {
				return reconcileErr
			}
			resetScheduleTimer(&scheduleTimer, &scheduleWake, nextWorkAt, scheduled)
			requestAcceptanceContinuation(acceptanceWake, more)
		case <-dependencyTick:
			nextCursor, more, woken, reconcileErr := s.reconcileDependencies(ctx, dependencyCursor)
			if reconcileErr != nil {
				return reconcileErr
			}
			if more {
				dependencyCursor = nextCursor
			} else {
				dependencyCursor = ""
			}
			if woken == 0 && !more {
				continue
			}
			nextWorkAt, scheduled, drainErr := s.drainWorker(ctx)
			if drainErr != nil {
				return drainErr
			}
			resetScheduleTimer(&scheduleTimer, &scheduleWake, nextWorkAt, scheduled)
		}
	}
}

func (s *Service) readGeneration(ctx context.Context, after uint64) (uint64, error) {
	var generation uint64
	err := dependencyretry.Do(ctx, "outbound generation read", func() error {
		var err error
		generation, err = s.acceptor.ReadGeneration(ctx, after)
		return err
	})
	return generation, err
}