package connector_webhooks

import (
	"context"
	"fmt"
	"strings"
	"sync"
	"time"

	"golang.org/x/sync/singleflight"
)

const defaultMaterializerCompanionCacheTTL = 30 * time.Second

type materializerCompanionSingleflight struct {
	group    singleflight.Group
	mu       sync.Mutex
	cache    map[materializerCompanionKey]materializerCompanionCacheEntry
	cacheTTL time.Duration
	now      func() time.Time
}

type materializerCompanionKey struct {
	namespace         string
	flowDeveloperName string
}

type materializerCompanionResult struct {
	incarnation CompanionFlowIncarnation
	isCrossPack bool
}

type materializerCompanionCacheEntry struct {
	result    materializerCompanionResult
	expiresAt time.Time
}

func newMaterializerCompanionSingleflight() *materializerCompanionSingleflight {
	return &materializerCompanionSingleflight{
		cacheTTL: defaultMaterializerCompanionCacheTTL,
		now:      time.Now,
	}
}

// resolveMaterializerCompanion collapses concurrent identical companion
// resolutions and briefly reuses successful incarnation reads for a burst of
// receipt materializers. Errors are never cached. The short TTL keeps pack/flow
// changes authoritative in the DB while preventing an 800-event webhook burst
// from hammering FlowDefinition with the same lookup.
func (d *Dispatcher) resolveMaterializerCompanion(ctx context.Context, namespace, flowDeveloperName string) (CompanionFlowIncarnation, bool, error) {
	if d == nil || d.resolveCompanion == nil {
		return CompanionFlowIncarnation{}, false, nil
	}
	if d.companionSingleflight == nil {
		return d.resolveCompanion(ctx, namespace, flowDeveloperName)
	}
	return d.companionSingleflight.resolve(ctx, d.resolveCompanion, namespace, flowDeveloperName)
}

func (s *materializerCompanionSingleflight) resolve(
	ctx context.Context,
	resolve func(context.Context, string, string) (CompanionFlowIncarnation, bool, error),
	namespace, flowDeveloperName string,
) (CompanionFlowIncarnation, bool, error) {
	key := normalizeMaterializerCompanionKey(namespace, flowDeveloperName)
	if result, ok := s.cachedResult(key); ok {
		recordConnectorMaterializerCompanionSingleflight("cache_hit")
		return result.incarnation, result.isCrossPack, nil
	}
	value, err, shared := s.group.Do(materializerCompanionSingleflightKey(key), func() (any, error) {
		if result, ok := s.cachedResult(key); ok {
			recordConnectorMaterializerCompanionSingleflight("cache_hit")
			return result, nil
		}
		recordConnectorMaterializerCompanionSingleflight("load")
		incarnation, isCrossPack, err := resolve(ctx, namespace, flowDeveloperName)
		if err != nil {
			return nil, err
		}
		result := materializerCompanionResult{incarnation: incarnation, isCrossPack: isCrossPack}
		s.cacheResult(key, result)
		return result, nil
	})
	if shared {
		recordConnectorMaterializerCompanionSingleflight("shared")
	}
	if err != nil {
		recordConnectorMaterializerCompanionSingleflight("error")
		return CompanionFlowIncarnation{}, false, err
	}
	result, ok := value.(materializerCompanionResult)
	if !ok {
		recordConnectorMaterializerCompanionSingleflight("error")
		return CompanionFlowIncarnation{}, false, fmt.Errorf("connector_webhooks: materializer companion singleflight returned unexpected value")
	}
	return result.incarnation, result.isCrossPack, nil
}

func (s *materializerCompanionSingleflight) cachedResult(key materializerCompanionKey) (materializerCompanionResult, bool) {
	if s == nil {
		return materializerCompanionResult{}, false
	}
	now := s.cacheNow()
	s.mu.Lock()
	defer s.mu.Unlock()
	entry, ok := s.cache[key]
	if !ok || !entry.expiresAt.After(now) {
		if ok {
			delete(s.cache, key)
		}
		return materializerCompanionResult{}, false
	}
	return entry.result, true
}

func (s *materializerCompanionSingleflight) cacheResult(key materializerCompanionKey, result materializerCompanionResult) {
	if s == nil {
		return
	}
	ttl := s.cacheTTL
	if ttl <= 0 {
		ttl = defaultMaterializerCompanionCacheTTL
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.cache == nil {
		s.cache = map[materializerCompanionKey]materializerCompanionCacheEntry{}
	}
	s.cache[key] = materializerCompanionCacheEntry{result: result, expiresAt: s.cacheNow().Add(ttl)}
}

func (s *materializerCompanionSingleflight) cacheNow() time.Time {
	if s != nil && s.now != nil {
		return s.now()
	}
	return time.Now()
}

func normalizeMaterializerCompanionKey(namespace, flowDeveloperName string) materializerCompanionKey {
	return materializerCompanionKey{
		namespace:         strings.ToLower(strings.TrimSpace(namespace)),
		flowDeveloperName: strings.ToLower(strings.TrimSpace(flowDeveloperName)),
	}
}

func materializerCompanionSingleflightKey(key materializerCompanionKey) string {
	return key.namespace + "\x00" + key.flowDeveloperName
}

type materializerConsentSingleflight struct {
	group singleflight.Group
}

type materializerConsentKey struct {
	companionPackName      string
	companionPackInstallID string
	namespace              string
	providerPackVersion    string
}

func newMaterializerConsentSingleflight() *materializerConsentSingleflight {
	return &materializerConsentSingleflight{}
}

// checkMaterializerConsent is singleflight-only, not a TTL cache. Enqueue guards
// companion install activity, but consent/dependency rows are still authoritative
// only in the DB, so each non-overlapping check re-reads them.
func (d *Dispatcher) checkMaterializerConsent(ctx context.Context, companionPackName, companionPackInstallID, namespace, providerPackVersion string) error {
	if d == nil || d.checkConsent == nil {
		return nil
	}
	if d.consentSingleflight == nil {
		return d.checkConsent(ctx, companionPackName, companionPackInstallID, namespace, providerPackVersion)
	}
	return d.consentSingleflight.check(ctx, d.checkConsent, companionPackName, companionPackInstallID, namespace, providerPackVersion)
}

func (s *materializerConsentSingleflight) check(
	ctx context.Context,
	check func(context.Context, string, string, string, string) error,
	companionPackName, companionPackInstallID, namespace, providerPackVersion string,
) error {
	key := normalizeMaterializerConsentKey(companionPackName, companionPackInstallID, namespace, providerPackVersion)
	_, err, shared := s.group.Do(materializerConsentSingleflightKey(key), func() (any, error) {
		recordConnectorMaterializerConsentSingleflight("load")
		return struct{}{}, check(ctx, companionPackName, companionPackInstallID, namespace, providerPackVersion)
	})
	if shared {
		recordConnectorMaterializerConsentSingleflight("shared")
	}
	if err != nil {
		recordConnectorMaterializerConsentSingleflight("error")
	}
	return err
}

func normalizeMaterializerConsentKey(companionPackName, companionPackInstallID, namespace, providerPackVersion string) materializerConsentKey {
	return materializerConsentKey{
		companionPackName:      strings.ToLower(strings.TrimSpace(companionPackName)),
		companionPackInstallID: strings.TrimSpace(companionPackInstallID),
		namespace:              strings.ToLower(strings.TrimSpace(namespace)),
		providerPackVersion:    strings.TrimSpace(providerPackVersion),
	}
}

func materializerConsentSingleflightKey(key materializerConsentKey) string {
	return key.companionPackName + "\x00" + key.companionPackInstallID + "\x00" + key.namespace + "\x00" + key.providerPackVersion
}