package service

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

	"github.com/pingcap/agent-git-service/internal/db"
	"gorm.io/gorm/clause"
)

const (
	wikiReconcilePollInterval      = 5 * time.Second
	wikiProjectionRecoveryInterval = time.Minute
	wikiReconcileLeaseDuration     = 5 * time.Minute
	wikiReconcileTimeout           = 4 * time.Minute
)

type wikiReconcileLeaseKey struct{}

// StartWikiReconcileWorker consumes durable requests and discovers interrupted
// publication from catalog heads. No in-memory enqueue is needed for recovery
// after a crash between catalog commit and Git publication.
func (s *Service) StartWikiReconcileWorker(ctx context.Context) error {
	if s == nil {
		return ErrTenantResourcesUnavailable
	}
	workerCtx, err := s.DetachContext(ctx)
	if err != nil {
		return err
	}
	key, err := s.tenantStateKey(workerCtx, "wiki-reconcile")
	if err != nil {
		return err
	}
	if _, exists := s.wikiReconcileWorkers.LoadOrStore(key, true); exists {
		return nil
	}
	s.Wg.Add(1)
	go func() {
		defer s.Wg.Done()
		defer s.wikiReconcileWorkers.Delete(key)
		poll := time.NewTicker(wikiReconcilePollInterval)
		defer poll.Stop()
		recovery := time.NewTicker(wikiProjectionRecoveryInterval)
		defer recovery.Stop()
		s.runWikiRecoveryPass(workerCtx, true)
		for {
			select {
			case <-workerCtx.Done():
				return
			case <-poll.C:
				s.runWikiRecoveryPass(workerCtx, false)
			case <-recovery.C:
				s.runWikiRecoveryPass(workerCtx, true)
			}
		}
	}()
	return nil
}

func (s *Service) runWikiRecoveryPass(ctx context.Context, discoverPublications bool) {
	if err := s.recoverPendingWikiReconciles(ctx); err != nil && ctx.Err() == nil {
		slog.WarnContext(ctx, "wiki reconcile requests failed", "error", err)
	}
	if discoverPublications {
		if err := s.recoverWikiPublications(ctx); err != nil && ctx.Err() == nil {
			slog.WarnContext(ctx, "wiki publication recovery failed", "error", err)
		}
	}
}

// Walk bounded database batches so one failed repository neither hides later
// work nor causes an unbounded scan allocation. Tenant DB context is retained.
func (s *Service) recoverPendingWikiReconciles(ctx context.Context) error {
	var failures []error
	var after uint
	for {
		var repos []db.Repository
		if err := s.DBForCtx(ctx).Table("repositories").
			Select("repositories.id, repositories.full_name").
			Joins("JOIN wiki_index_state AS state ON state.repository_id = repositories.id").
			Where("repositories.id > ? AND state.reconcile_requested_at IS NOT NULL", after).
			Where("state.reconciler_lease_until IS NULL OR state.reconciler_lease_until < ?", time.Now().UTC()).
			Order("repositories.id ASC").Limit(100).Scan(&repos).Error; err != nil {
			return errors.Join(append(failures, err)...)
		}
		if len(repos) == 0 {
			return errors.Join(failures...)
		}
		for _, repo := range repos {
			after = repo.ID
			if err := s.runWikiReconcileRequest(ctx, repo); err != nil {
				failures = append(failures, fmt.Errorf("%s: %w", repo.FullName, err))
			}
		}
	}
}

func (s *Service) runWikiReconcileRequest(ctx context.Context, repo db.Repository) error {
	return s.withWikiReconcileLease(ctx, repo, true, func(jobCtx context.Context) error {
		result, err := s.ReconcileWikiV2(jobCtx, repo.FullName)
		if err == nil && !result.Reconciled {
			err = fmt.Errorf("%w: wiki changed during reconcile; request retained", ErrConflict)
		}
		return err
	})
}

func (s *Service) withWikiReconcileLease(ctx context.Context, repo db.Repository, requested bool, run func(context.Context) error) error {
	if !requested {
		if err := s.DBForCtx(ctx).Clauses(clause.OnConflict{DoNothing: true}).Create(&db.WikiIndexState{RepositoryID: repo.ID}).Error; err != nil {
			return err
		}
	}
	now := time.Now().UTC().Truncate(time.Millisecond)
	lease := now.Add(wikiReconcileLeaseDuration)
	query := s.DBForCtx(ctx).Model(&db.WikiIndexState{}).Where("repository_id = ?", repo.ID)
	if requested {
		query = query.Where("reconcile_requested_at IS NOT NULL")
	}
	claim := query.Where("reconciler_lease_until IS NULL OR reconciler_lease_until < ?", now).UpdateColumn("reconciler_lease_until", lease)
	if claim.Error != nil || claim.RowsAffected == 0 {
		return claim.Error
	}
	jobCtx, cancel := context.WithTimeout(context.WithValue(ctx, wikiReconcileLeaseKey{}, lease), wikiReconcileTimeout)
	defer cancel()
	err := run(jobCtx)
	// A short retry delay also prevents a broken repository from dominating
	// every poll. Release is fenced, so an expired owner cannot clear a new lease.
	var next any
	if err != nil {
		next = time.Now().UTC().Add(wikiProjectionRecoveryInterval).Truncate(time.Millisecond)
	}
	releaseErr := s.DBForCtx(ctx).Model(&db.WikiIndexState{}).
		Where("repository_id = ? AND reconciler_lease_until = ?", repo.ID, lease).
		UpdateColumn("reconciler_lease_until", next).Error
	return errors.Join(err, releaseErr)
}

func (s *Service) recoverWikiPublications(ctx context.Context) error {
	var failures []error
	var after uint
	for {
		var repos []db.Repository
		if err := s.DBForCtx(ctx).Table("repositories").
			Select("repositories.id, repositories.full_name").
			Joins("JOIN wiki_repo_heads AS heads ON heads.repository_id = repositories.id").
			Where("repositories.id > ?", after).
			Order("repositories.id ASC").Limit(100).Scan(&repos).Error; err != nil {
			return errors.Join(append(failures, err)...)
		}
		if len(repos) == 0 {
			return errors.Join(failures...)
		}
		for _, repo := range repos {
			after = repo.ID
			err := s.withWikiReconcileLease(ctx, repo, false, func(jobCtx context.Context) error {
				return s.ReconcileWikiProjection(jobCtx, repo.FullName)
			})
			if err != nil {
				failures = append(failures, fmt.Errorf("%s: %w", repo.FullName, err))
			}
		}
	}
}