package case_driver

import (
	"context"
	"fmt"
	"regexp"
	"sync"

	"github.com/tidbcloud/dataflow-service-e2e/cases"
	"github.com/tidbcloud/dataflow-service-e2e/internal/log"
	"github.com/tidbcloud/dataflow-service-e2e/internal/misc"
	. "github.com/tidbcloud/dataflow-service-e2e/internal/model"
)

const deleteErrorSensitiveKey = `password|passwd|token|secret([_-]?((access)[_-]?)?key)?|client[_-]?secret|sas[_-]?token|api[_-]?key|access[_-]?key|authorization|user(name)?|ca[_-]?cert|certificate|private[_-]?key`

var (
	deleteErrorSensitiveValue = regexp.MustCompile(`(?i)(["']?\b(` + deleteErrorSensitiveKey + `)\b["']?\s*[:=]\s*)("(\\.|[^"\\])*"|'(\\.|[^'\\])*'|[^\s,;}]+)`)
	deleteErrorBearerValue    = regexp.MustCompile(`(?i)\b(bearer|basic)\s+[^\s,;]+`)
)

type dependenciesManager struct {
	lgr log.SimpleLogger

	// CaseName:Case
	cases map[string]cases.E2ECase
	// DependencyType:Dependency
	dependencies map[string]Dependency
	// DependencyType:DependencyEntities
	dependenciesStateMu  sync.Mutex
	dependenciesEntities map[string][]DependencyEntity
	// CaseName:RequirementName:DependencyEntity
	dependenciesEntitiesAssignment map[string]map[string]DependencyEntity
	// All concrete DependencyEntity implementations are pointers, so interface
	// values are comparable and can serve as the entity identity key here.
	entityUsages map[DependencyEntity]*dependencyEntityUsage

	cleanupQueue chan DependencyEntity
	cleanupWG    sync.WaitGroup
	cleanupMu    sync.Mutex
	cleaning     map[DependencyEntity]struct{}

	caseDepResult    map[string]<-chan error
	dispatcher       *depDispatcher
	allProvisionDone chan struct{}
}

type dependencyEntityUsage struct {
	cases map[string]bool
}

func NewDependenciesManager(lgr log.SimpleLogger, cases map[string]cases.E2ECase, deps ...Dependency) *dependenciesManager {
	mgr := &dependenciesManager{
		lgr:                            lgr,
		cases:                          cases,
		dependencies:                   initDependencies(deps...),
		dependenciesEntities:           map[string][]DependencyEntity{},
		dependenciesEntitiesAssignment: map[string]map[string]DependencyEntity{},
		entityUsages:                   map[DependencyEntity]*dependencyEntityUsage{},
		cleanupQueue:                   make(chan DependencyEntity),
		cleaning:                       map[DependencyEntity]struct{}{},

		caseDepResult:    make(map[string]<-chan error, len(cases)),
		allProvisionDone: make(chan struct{}),
	}
	mgr.dispatcher = newDepDispatcher(lgr, mgr.noteAssignment)
	for _, c := range mgr.cases {
		mgr.caseDepResult[c.Name()] = mgr.dispatcher.RegisterCase(c.Name(), c)
	}
	go mgr.runCleanupLoop()

	return mgr
}

func initDependencies(deps ...Dependency) map[string]Dependency {
	dependencies := map[string]Dependency{}
	for _, d := range deps {
		if _, existed := dependencies[d.Type()]; existed {
			panic("duplicated dependency: " + d.Type())
		}
		dependencies[d.Type()] = d
	}
	return dependencies
}

func (t *dependenciesManager) ProvisionAsync(ctx context.Context) {
	reqGapMap := t.prepareProvisioning()

	var wg sync.WaitGroup
	for depType, gap := range reqGapMap {
		if len(gap) == 0 {
			continue
		}

		wg.Add(1)
		go func(depType string, gap EntitiesStats) {
			defer wg.Done()
			t.lgr.Info("Start async provisioning for dependency %s, gap: %s", depType, misc.PrettyJSON(gap))
			dep := t.dependencies[depType]
			ch, err := dep.ProvisionEntities(ctx, gap)
			if err != nil {
				t.lgr.Error("Provision failed for dependency %s, error: %s", depType, err.Error())
				t.dispatcher.SetDependencyError(depType, err)
				return
			}
			var provided EntitiesStats
			for {
				select {
				case <-ctx.Done():
					err := fmt.Errorf("context canceled while provisioning dependency %s: %w", depType, ctx.Err())
					t.lgr.Warn("Stop receiving provisioned entities for dependency %s due to context cancellation: %v", depType, ctx.Err())
					t.dispatcher.SetDependencyError(depType, err)
					return
				case entity, ok := <-ch:
					if !ok {
						missing := gap.Gap(provided)
						if len(missing) > 0 {
							err := fmt.Errorf("unsatisfied dependency %s, missing entities: %s", depType, misc.PrettyJSON(missing))
							t.lgr.Error("Provision incomplete for dependency %s, gap: %s, provided: %s", depType, misc.PrettyJSON(gap), misc.PrettyJSON(provided))
							t.dispatcher.SetDependencyError(depType, err)
						}
						return
					}
					provided.AddEntity(entity)
					t.addDependenciesEntities(entity)
				}
			}
		}(depType, gap)
	}
	go func() {
		wg.Wait()
		close(t.allProvisionDone)
		t.lgr.Info("All async provisioning goroutines finished")
	}()
}

func (t *dependenciesManager) WaitForDeps(ctx context.Context, caseName string) error {
	select {
	case <-ctx.Done():
		return fmt.Errorf("context canceled while waiting for dependencies of case %s", caseName)
	case err := <-t.caseDepResult[caseName]:
		return err
	}
}

func (t *dependenciesManager) prepareProvisioning() map[string]EntitiesStats {
	reqGapMap := map[string]EntitiesStats{}
	reqStatsMap := map[string]EntitiesStats{}
	for ty := range t.dependencies {
		reqStatsMap[ty] = t.calculateOverallRequirement(ty)
	}
	t.lgr.Info("All dependency requirements:\n%s", misc.PrettyJSON(reqStatsMap))

	t.initExistedDependenciesEntities()
	t.lgr.Info("All existed dependency entities:\n%s", misc.PrettyJSON(safeDependencyEntities(t.dependenciesEntities)))

	for ty := range t.dependencies {
		reqGapMap[ty] = t.calculateRequirementGap(ty, reqStatsMap[ty])
	}
	t.lgr.Info("Gaps of dependency requirements:\n%s", misc.PrettyJSON(reqGapMap))
	return reqGapMap
}

func (t *dependenciesManager) initExistedDependenciesEntities() {
	for _, d := range t.dependencies {
		entities, err := d.GetExistingUsableEntities()
		if err != nil {
			panic(fmt.Sprintf("GetExistingUsableEntities failed, entity type: %s, error: %s", d.Type(), err.Error()))
		}
		t.addDependenciesEntities(entities...)
	}
}

func (t *dependenciesManager) addDependenciesEntities(entities ...DependencyEntity) {
	t.dependenciesStateMu.Lock()
	for _, e := range entities {
		if _, existed := t.dependenciesEntities[e.Type()]; !existed {
			t.dependenciesEntities[e.Type()] = []DependencyEntity{}
		}
		t.dependenciesEntities[e.Type()] = append(t.dependenciesEntities[e.Type()], e)
		t.entityUsages[e] = &dependencyEntityUsage{cases: map[string]bool{}}
	}
	t.dependenciesStateMu.Unlock()

	t.dispatcher.Dispatch(entities...)
}

func (t *dependenciesManager) Cleanup() error {
	// Parent test cleanup runs after every subtest cleanup. Wait for all queued
	// case-scoped deletes before taking the final fallback inventory snapshot.
	t.cleanupWG.Wait()
	close(t.cleanupQueue)

	t.lgr.Info("dependency cleanup started: dependencyTypes=%d", len(t.dependenciesEntities))
	t.dependenciesStateMu.Lock()
	var entities []DependencyEntity
	for _, candidates := range t.dependenciesEntities {
		for _, entity := range candidates {
			if entity.IsCreatedByDriver() {
				entities = append(entities, entity)
			}
		}
	}
	t.dependenciesStateMu.Unlock()

	t.lgr.Info("suite teardown cleanup sweep start, driver-created dependencies remaining: %d (types: %s)", len(entities), entityTypes(entities))
	failed := 0
	for _, entity := range entities {
		if err := t.dependencies[entity.Type()].Delete(entity); err != nil {
			t.logDeleteFailure(entity, err, false)
			failed++
		}
	}
	if failed != 0 {
		t.lgr.Error("suite teardown cleanup incomplete, types: %s, swept: %d, failed: %d", entityTypes(entities), len(entities), failed)
		return fmt.Errorf("some dependencies cannot be deleted")
	}
	t.lgr.Info("suite teardown cleanup finished, swept: %d driver-created dependencies", len(entities))
	return nil
}

// entityTypes returns the distinct dependency types among the given entities,
// used only for summarising cleanup progress in logs.
func entityTypes(entities []DependencyEntity) string {
	seen := map[string]struct{}{}
	out := ""
	for _, e := range entities {
		if _, ok := seen[e.Type()]; ok {
			continue
		}
		seen[e.Type()] = struct{}{}
		if out != "" {
			out += ","
		}
		out += e.Type()
	}
	return out
}

func (t *dependenciesManager) ReleaseCase(caseName string) {
	t.dependenciesStateMu.Lock()
	var ready []DependencyEntity
	seen := map[DependencyEntity]struct{}{}
	for _, entity := range t.dependenciesEntitiesAssignment[caseName] {
		if _, duplicate := seen[entity]; duplicate {
			continue
		}
		seen[entity] = struct{}{}

		usage := t.entityUsages[entity]
		if usage == nil || usage.cases[caseName] {
			continue
		}
		usage.cases[caseName] = true
		if !entity.IsCreatedByDriver() {
			t.lgr.Info("release case, entity not created by driver, type: %s, dimensions: %s, case: %s", entity.Type(), misc.PrettyJSON(entity.Dimensions()), caseName)
			continue
		}
		if !allUsingCasesCompleted(usage.cases) {
			t.lgr.Info("release case, entity not yet eligible for cleanup, type: %s, dimensions: %s, case: %s, remainingUsingCases: %v", entity.Type(), misc.PrettyJSON(entity.Dimensions()), caseName, remainingUsingCases(usage.cases))
			continue
		}
		ready = append(ready, entity)
	}
	t.dependenciesStateMu.Unlock()

	for _, entity := range ready {
		t.lgr.Info("release case, entity eligible for cleanup, type: %s, dimensions: %s, case: %s", entity.Type(), misc.PrettyJSON(entity.Dimensions()), caseName)
		t.enqueueCleanup(entity)
	}
}

// remainingUsingCases returns the case names assigned to an entity that are not
// yet marked completed. Used only for logging in the cleanup decision path.
func remainingUsingCases(cases map[string]bool) []string {
	remaining := []string{}
	for caseName, completed := range cases {
		if !completed {
			remaining = append(remaining, caseName)
		}
	}
	return remaining
}

type dependencyEntityLog struct {
	Type            string            `json:"type"`
	CreatedByDriver bool              `json:"createdByDriver"`
	Dimensions      map[string]string `json:"dimensions"`
}

func safeDependencyEntity(entity DependencyEntity) dependencyEntityLog {
	return dependencyEntityLog{
		Type:            entity.Type(),
		CreatedByDriver: entity.IsCreatedByDriver(),
		Dimensions:      entity.Dimensions(),
	}
}

func safeDependencyEntities(entities map[string][]DependencyEntity) map[string][]dependencyEntityLog {
	safe := make(map[string][]dependencyEntityLog, len(entities))
	for dependencyType, values := range entities {
		safe[dependencyType] = make([]dependencyEntityLog, 0, len(values))
		for _, entity := range values {
			safe[dependencyType] = append(safe[dependencyType], safeDependencyEntity(entity))
		}
	}
	return safe
}

func (t *dependenciesManager) calculateOverallRequirement(depType string) EntitiesStats {
	var stats RequirementsStats
	for _, ca := range t.cases {
		for _, req := range ca.DeclareDependencyRequirements() {
			if req.Type() != depType {
				continue
			}
			stats.AddRequirement(req)
		}
	}
	return stats.ToEntitiesStats()
}

func (t *dependenciesManager) calculateRequirementGap(depType string, reqStats EntitiesStats) EntitiesStats {
	var existedStats EntitiesStats

	t.dependenciesStateMu.Lock()
	defer t.dependenciesStateMu.Unlock()

	for _, it := range t.dependenciesEntities[depType] {
		existedStats.AddEntity(it)
	}
	return reqStats.Gap(existedStats)
}

func (t *dependenciesManager) noteAssignment(caseName, reqName string, entity DependencyEntity) {
	t.dependenciesStateMu.Lock()
	defer t.dependenciesStateMu.Unlock()

	if t.dependenciesEntitiesAssignment[caseName] == nil {
		t.dependenciesEntitiesAssignment[caseName] = map[string]DependencyEntity{}
	}
	t.dependenciesEntitiesAssignment[caseName][reqName] = entity
	usage := t.entityUsages[entity]
	if usage == nil {
		t.lgr.Error("assigned entity has no usage record, type: %s, dimensions: %s, case: %s, requirement: %s; cleanup for this entity may not run", entity.Type(), misc.PrettyJSON(entity.Dimensions()), caseName, reqName)
		usage = &dependencyEntityUsage{cases: map[string]bool{}}
		t.entityUsages[entity] = usage
	}
	usage.cases[caseName] = false
	t.lgr.Info("assigned dependency, type: %s, dimensions: %s, case: %s, requirement: %s, createdByDriver: %t", entity.Type(), misc.PrettyJSON(entity.Dimensions()), caseName, reqName, entity.IsCreatedByDriver())
}

func allUsingCasesCompleted(cases map[string]bool) bool {
	if len(cases) == 0 {
		return false
	}
	for _, completed := range cases {
		if !completed {
			return false
		}
	}
	return true
}

func (t *dependenciesManager) enqueueCleanup(entity DependencyEntity) {
	t.cleanupWG.Add(1)
	t.cleanupQueue <- entity
}

func (t *dependenciesManager) runCleanupLoop() {
	for entity := range t.cleanupQueue {
		t.cleanupMu.Lock()
		if _, alreadyCleaning := t.cleaning[entity]; alreadyCleaning {
			t.cleanupMu.Unlock()
			t.cleanupWG.Done()
			t.lgr.Info("cleanup already queued for dependency, type: %s, dimensions: %s, duplicate enqueue suppressed", entity.Type(), misc.PrettyJSON(entity.Dimensions()))
			continue
		}
		t.cleaning[entity] = struct{}{}
		t.cleanupMu.Unlock()

		go t.deleteEntityAfterCase(entity)
	}
}

func (t *dependenciesManager) deleteEntityAfterCase(entity DependencyEntity) {
	defer t.cleanupWG.Done()
	defer func() {
		t.cleanupMu.Lock()
		delete(t.cleaning, entity)
		t.cleanupMu.Unlock()
	}()

	t.lgr.Info("deleting dependency after case completion, type: %s, dimensions: %s", entity.Type(), misc.PrettyJSON(entity.Dimensions()))
	if err := t.dependencies[entity.Type()].Delete(entity); err != nil {
		t.logDeleteFailure(entity, err, true)
		return
	}

	t.dependenciesStateMu.Lock()
	t.removeEntityLocked(entity)
	t.dependenciesStateMu.Unlock()
	t.lgr.Info("deleted dependency, type: %s, dimensions: %s", entity.Type(), misc.PrettyJSON(entity.Dimensions()))
}

func (t *dependenciesManager) logDeleteFailure(entity DependencyEntity, err error, retryAtTeardown bool) {
	deleteErr := redactDeleteError(err)
	if retryAtTeardown {
		t.lgr.Error("failed to delete dependency after case completion, type: %s, dimensions: %s, error: %s; retrying at suite teardown", entity.Type(), misc.PrettyJSON(entity.Dimensions()), deleteErr)
		return
	}
	t.lgr.Error("failed to delete dependency during suite teardown, type: %s, dimensions: %s, error: %s", entity.Type(), misc.PrettyJSON(entity.Dimensions()), deleteErr)
}

// redactDeleteError is the logging boundary for all dependency Delete errors.
// Dedicated, Premium, and Serverless Delete implementations include failed
// entities in their returned errors, so mask their JSON fields as well as the
// ordinary key/value and Authorization forms used by client errors.
func redactDeleteError(err error) string {
	if err == nil {
		return "<nil>"
	}
	message := deleteErrorBearerValue.ReplaceAllString(err.Error(), "$1 <redacted>")
	return deleteErrorSensitiveValue.ReplaceAllString(message, "$1<redacted>")
}

func (t *dependenciesManager) removeEntityLocked(entity DependencyEntity) {
	entities := t.dependenciesEntities[entity.Type()]
	kept := entities[:0]
	for _, candidate := range entities {
		if candidate != entity {
			kept = append(kept, candidate)
		}
	}
	if len(kept) == 0 {
		delete(t.dependenciesEntities, entity.Type())
	} else {
		t.dependenciesEntities[entity.Type()] = kept
	}

	delete(t.entityUsages, entity)

	for caseName, assignments := range t.dependenciesEntitiesAssignment {
		for reqName, assigned := range assignments {
			if assigned == entity {
				delete(assignments, reqName)
			}
		}
		if len(assignments) == 0 {
			delete(t.dependenciesEntitiesAssignment, caseName)
		}
	}
}