package connector_webhooks

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

	connector_inbound_routing "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	"golang.org/x/sync/singleflight"
)

const defaultMaterializerRouteCacheTTL = 30 * time.Second

type materializerRouteCacheKey struct {
	namespace     string
	packInstallID string
	eventKind     string
}

type materializerRouteCacheEntry struct {
	routeVersion      int64
	flowDeveloperName string
	expiresAt         time.Time
}

type materializerRouteCache struct {
	mu      sync.Mutex
	group   singleflight.Group
	ttl     time.Duration
	entries map[materializerRouteCacheKey]materializerRouteCacheEntry
	now     func() time.Time
}

func newMaterializerRouteCache(ttl time.Duration) *materializerRouteCache {
	if ttl <= 0 {
		return nil
	}
	return &materializerRouteCache{
		ttl:     ttl,
		entries: map[materializerRouteCacheKey]materializerRouteCacheEntry{},
		now:     time.Now,
	}
}

func (d *Dispatcher) resolveMaterializerRoute(ctx context.Context, namespace, packInstallID, eventKind string) (int64, string, error) {
	if d == nil || d.routeCache == nil {
		return d.routes.ResolveFlow(ctx, namespace, packInstallID, eventKind)
	}
	return d.routeCache.resolve(ctx, d.routes, namespace, packInstallID, eventKind)
}

func (d *Dispatcher) evictMaterializerRoute(namespace, packInstallID, eventKind string) {
	if d == nil || d.routeCache == nil {
		return
	}
	d.routeCache.evict(namespace, packInstallID, eventKind)
}

func (c *materializerRouteCache) resolve(ctx context.Context, routes connector_inbound_routing.RouteResolver, namespace, packInstallID, eventKind string) (int64, string, error) {
	key := normalizeMaterializerRouteCacheKey(namespace, packInstallID, eventKind)
	now := c.now()
	c.mu.Lock()
	if entry, ok := c.entries[key]; ok {
		if now.Before(entry.expiresAt) {
			c.mu.Unlock()
			recordConnectorMaterializerRouteCache("hit")
			return entry.routeVersion, entry.flowDeveloperName, nil
		}
		delete(c.entries, key)
	}
	c.mu.Unlock()

	recordConnectorMaterializerRouteCache("miss")
	value, err, shared := c.group.Do(materializerRouteCacheSingleflightKey(key), func() (any, error) {
		now := c.now()
		c.mu.Lock()
		if entry, ok := c.entries[key]; ok {
			if now.Before(entry.expiresAt) {
				c.mu.Unlock()
				return entry, nil
			}
			delete(c.entries, key)
		}
		c.mu.Unlock()

		recordConnectorMaterializerRouteCache("load")
		routeVersion, flowDeveloperName, err := routes.ResolveFlow(ctx, namespace, packInstallID, eventKind)
		if err != nil {
			return nil, err
		}
		entry := materializerRouteCacheEntry{
			routeVersion:      routeVersion,
			flowDeveloperName: flowDeveloperName,
			expiresAt:         c.now().Add(c.ttl),
		}

		c.mu.Lock()
		c.entries[key] = entry
		c.mu.Unlock()
		return entry, nil
	})
	if shared {
		recordConnectorMaterializerRouteCache("shared")
	}
	if err != nil {
		return 0, "", err
	}
	entry, ok := value.(materializerRouteCacheEntry)
	if !ok {
		return 0, "", fmt.Errorf("connector_webhooks: materializer route cache returned unexpected value")
	}
	return entry.routeVersion, entry.flowDeveloperName, nil
}

func (c *materializerRouteCache) evict(namespace, packInstallID, eventKind string) {
	c.mu.Lock()
	delete(c.entries, normalizeMaterializerRouteCacheKey(namespace, packInstallID, eventKind))
	c.mu.Unlock()
	recordConnectorMaterializerRouteCache("evict")
}

func normalizeMaterializerRouteCacheKey(namespace, packInstallID, eventKind string) materializerRouteCacheKey {
	return materializerRouteCacheKey{
		namespace:     strings.ToLower(strings.TrimSpace(namespace)),
		packInstallID: strings.TrimSpace(packInstallID),
		eventKind:     strings.ToLower(strings.TrimSpace(eventKind)),
	}
}

func materializerRouteCacheSingleflightKey(key materializerRouteCacheKey) string {
	return key.namespace + "\x00" + key.packInstallID + "\x00" + key.eventKind
}