/******************************************************************************
 * Copyright (c)  2022 PingCAP, Inc.                                          *
 * Licensed under the Apache License, Version 2.0 (the "License");            *
 * you may not use this file except in compliance with the License.           *
 * You may obtain a copy of the License at                                    *
 *                                                                            *
 * http://www.apache.org/licenses/LICENSE-2.0                                 *
 *                                                                            *
 * Unless required by applicable law or agreed to in writing, software        *
 * distributed under the License is distributed on an "AS IS" BASIS,          *
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.   *
 * See the License for the specific language governing permissions and        *
 * limitations under the License.                                             *
 ******************************************************************************/

/*******************************************************************************
 * @File: manager.go
 * @Description:
 * @Author: jiangxunyu@pingcap.com
 * @Version: 1.0.0
 * @Date: 2022/6/7 17:41
*******************************************************************************/

package globalservice

//go:generate mockgen -destination ../../test/mockmetamanager/mock_logical_region_router.go -package mock_meta_manager -source ./logical_router.go LogicalRegionRouter

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"time"

	"github.com/tidbcloud/observability/pkg/common/metrics"

	"github.com/pingcap/log"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/errors"
	"github.com/tidbcloud/observability/pkg/globalservice/structs"
	"github.com/tidbcloud/observability/pkg/metamanager"
	metaservice "github.com/tidbcloud/observability/pkg/metamanager/etcd"
	"go.uber.org/zap"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/kubernetes"
)

// globalServerNsAndCmName define global server namespace and configmap name
const globalServerNsAndCmName = constants.ObservabilityBaseServiceName

var defaultGlobalManager *GlobalManager
var managerOnce sync.Once

type MetaCacheGetter interface {
	GetGlobalMetaCache() *structs.GlobalServiceMeta
}

type GlobalManager struct {
	globalMetaService   metamanager.MetaService
	globalMetaCache     *structs.GlobalServiceMeta
	logicalRegionRouter LogicalRegionRouter
}

func NewManager(ctx context.Context, kubeClient kubernetes.Interface) (err error) {
	defer func() {
		if e := recover(); e != nil {
			err = fmt.Errorf("NewManager panic recovered: %v", e)
		}
	}()

	managerOnce.Do(func() {
		if defaultGlobalManager != nil {
			return
		}

		defaultGlobalManager = &GlobalManager{
			globalMetaService: metaservice.NewConfigMapMetaService(kubeClient),
		}

		// Initialize logical region router with the MetaCacheGetter interface
		// This eliminates circular dependency by only passing the interface, not the full manager
		defaultGlobalManager.logicalRegionRouter = NewLogicalRegionRouter(defaultGlobalManager)

		// initial load of global meta cache
		if err := defaultGlobalManager.loadGlobalMetaCache(ctx); err != nil {
			panic(fmt.Sprintf("initial load of global cache failed. error: %v", err.Error()))
		}

		// start global server metadata change listening
		go func() {
			if err := defaultGlobalManager.WatchGlobalMetaCache(ctx); err != nil {
				log.Error("start metadata change listen failed", zap.Error(err))
			}
		}()
	})

	return nil
}

// GetGlobalManager
// @Description: get default global manager instances
// @return *GlobalManager
func GetGlobalManager() *GlobalManager {
	return defaultGlobalManager
}

// GetGlobalMetaCache
// @Description: get metadata cache for global services
// @Receiver m
// @return *structs.GlobalServiceMeta
func (m *GlobalManager) GetGlobalMetaCache() *structs.GlobalServiceMeta {
	return m.globalMetaCache
}

// GetRegionalRouteInfo returns the regional router info
func (m *GlobalManager) GetRegionalRouteInfo(ctx context.Context, routingCtx structs.RoutingContext) (*structs.RegionalRouter, error) {
	if routingCtx.LogicalRegion != "" {
		log.Debug("Routing by X-Logical-Region", zap.String("logical_region", routingCtx.LogicalRegion))
		return m.getExplicitLogicalRegionRouteInfo(routingCtx)
	}

	if m.isLogicalRegionEnabled() && routingCtx.BizType != "" && routingCtx.TenantID != "" {
		log.Debug("Using logical region routing", zap.Any("routing_context", routingCtx))

		routeInfo, err := m.logicalRegionRouter.GetLogicalRegionRouteInfo(ctx, routingCtx)
		if err == nil {
			return routeInfo, nil
		}
		log.Info("Logical region routing failed, fallback to physical region routing", zap.Error(err))
	}

	return m.getRegionalRouteInfo(ctx, routingCtx)
}

func (m *GlobalManager) getExplicitLogicalRegionRouteInfo(routingCtx structs.RoutingContext) (*structs.RegionalRouter, error) {
	if m.globalMetaCache == nil {
		return nil, errors.New(errors.GetRegionalRouteAddressError, "global meta cache is nil")
	}

	if _, ok := m.globalMetaCache.Vendors[routingCtx.Vendor]; ok {
		for _, regionInfo := range m.globalMetaCache.Vendors[routingCtx.Vendor] {
			if routingCtx.Region != "" && regionInfo.Region != routingCtx.Region {
				continue
			}
			for _, router := range regionInfo.Routers {
				if router.LogicalRegionName == routingCtx.LogicalRegion {
					return &router, nil
				}
			}
		}
	}

	return nil, errors.NewErrorf(errors.GetRegionalRouteAddressError,
		"failed to get the routing address of vendor `%s`, region `%s`, logicalRegion `%s`",
		routingCtx.Vendor, routingCtx.Region, routingCtx.LogicalRegion)
}

func (m *GlobalManager) getRegionalRouteInfo(ctx context.Context, routingCtx structs.RoutingContext) (*structs.RegionalRouter, error) {
	if _, ok := m.globalMetaCache.Vendors[routingCtx.Vendor]; ok {
		for _, regionInfo := range m.globalMetaCache.Vendors[routingCtx.Vendor] {
			if regionInfo.Region == routingCtx.Region {
				// if there is only one region route, get the dns address directly
				if len(regionInfo.Routers) == 1 {
					return &regionInfo.Routers[0], nil
				}
				// if there are multiple region routes, get the dns address of the primary tag
				for _, router := range regionInfo.Routers {
					if router.Primary {
						return &router, nil
					}
				}
			}
		}
	}
	return nil, errors.NewErrorf(errors.GetRegionalRouteAddressError,
		"failed to get the routing address of vendor `%s` and region `%s`", routingCtx.Vendor, routingCtx.Region)
}

// IsLogicalRegionEnabled
// @Description: check if logical region routing is enabled (convenience method)
// @return bool
func (m *GlobalManager) isLogicalRegionEnabled() bool {
	return m.logicalRegionRouter != nil && m.logicalRegionRouter.IsLogicalRegionEnabled()
}

// EditGlobalMetaCache
// @Description: edit metadata cache for global services
// @Receiver m
// @Parameter ctx
// @Parameter globalMeta
// @return error
func (m *GlobalManager) EditGlobalMetaCache(ctx context.Context, globalMeta structs.GlobalServiceMeta) error {
	b, err := json.Marshal(globalMeta)
	if err != nil {
		log.Error("json marshal global meta failed", zap.Error(err))
		return errors.New(errors.MarshalError, err.Error())
	}
	if err = m.globalMetaService.UpdateMeta(ctx, globalServerNsAndCmName, string(b)); err != nil {
		log.Error("update global meta failed", zap.Error(err))
		return errors.New(errors.UpdateGlobalMetaError, err.Error())
	}
	// update successfully loaded with the latest cache information
	if err = m.loadGlobalMetaCache(ctx); err != nil {
		return errors.New(errors.LoadGlobalMetaError, err.Error())
	}
	return nil
}

// CreateGlobalMetaCache
// @Description: create metadata cache for global services
// @Receiver m
// @Parameter ctx
// @Parameter globalMeta
// @return error
func (m *GlobalManager) CreateGlobalMetaCache(ctx context.Context, globalMeta structs.GlobalServiceMeta) error {
	b, err := json.Marshal(globalMeta)
	if err != nil {
		log.Error("json marshal global meta failed", zap.Error(err))
		return errors.New(errors.MarshalError, err.Error())
	}
	if err = m.globalMetaService.CreateMeta(ctx, globalServerNsAndCmName, string(b)); err != nil {
		log.Error("create global meta failed", zap.Error(err))
		return errors.New(errors.CreateGlobalMetaError, err.Error())
	}
	// update successfully loaded with the latest cache information
	if err = m.loadGlobalMetaCache(ctx); err != nil {
		return errors.New(errors.LoadGlobalMetaError, err.Error())
	}
	return nil
}

// WatchGlobalMetaCache
// @Description: metadata change listening for global services
// @Receiver m
// @Parameter ctx
// @return error
func (m *GlobalManager) WatchGlobalMetaCache(ctx context.Context) error {
	if m.globalMetaService == nil {
		return fmt.Errorf("globalMetaService is nil")
	}

	for {
		select {
		case <-ctx.Done():
			log.Info("watch global meta cache context cancelled")
			return ctx.Err()
		default:
			if err := m.watchWithRetry(ctx); err != nil {
				log.Error("watch failed, will retry", zap.Error(err))
				// Add a small delay before retrying to avoid tight loop
				select {
				case <-ctx.Done():
					return ctx.Err()
				case <-time.After(60 * time.Second):
					// Continue to retry after delay
				}
			}
		}
	}
}

// watchWithRetry handles a single watch session with proper cleanup
func (m *GlobalManager) watchWithRetry(ctx context.Context) error {
	watcher, err := m.globalMetaService.WatchMeta(ctx, globalServerNsAndCmName)
	if err != nil {
		return fmt.Errorf("failed to create watch: %w", err)
	}

	defer watcher.Stop()

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case event, ok := <-watcher.ResultChan():
			if !ok {
				log.Info("watch channel closed, will restart")
				return fmt.Errorf("watch channel closed")
			}

			// Process only relevant events
			switch event.Type {
			case watch.Modified, watch.Added:
				log.Info("detected configmap change",
					zap.String("event_type", string(event.Type)),
					zap.String("configmap", globalServerNsAndCmName))

				// Reload cache with the latest information
				if err := m.loadGlobalMetaCache(ctx); err != nil {
					log.Error("failed to reload global meta cache", zap.Error(err))
					// Don't return error here, continue watching
				}
			case watch.Deleted:
				log.Warn("configmap was deleted", zap.String("configmap", globalServerNsAndCmName))
			case watch.Error:
				// Handle different types of watch errors
				log.Error("watch error event received", zap.Any("event", event))
				return fmt.Errorf("watch error event")
			default:
				log.Debug("ignoring watch event", zap.String("event_type", string(event.Type)))
			}
		}
	}
}

// loadGlobalMetaCache
// @Description: load global routing information metadata configuration
// @Receiver m
// @Parameter ctx
// @return error
func (m *GlobalManager) loadGlobalMetaCache(ctx context.Context) error {
	metadata, err := m.globalMetaService.DetailMeta(ctx, globalServerNsAndCmName)
	if err != nil {
		return errors.New(errors.DetailGlobalMetaError, err.Error())
	}
	var serviceMeta structs.GlobalServiceMeta
	if err = json.Unmarshal([]byte(metadata), &serviceMeta); err != nil {
		log.Error("json unmarshal global meta failed", zap.Error(err))
		return errors.New(errors.UnmarshalError, err.Error())
	}
	log.Debug("load global meta cache", zap.Any("Meta_Cache", serviceMeta))
	m.globalMetaCache = &serviceMeta

	// Build logical region pattern caches
	if m.logicalRegionRouter != nil && serviceMeta.Routes != nil {
		m.logicalRegionRouter.BuildPatternCache(serviceMeta.Routes)
	}

	// monitoring buried points
	for vendor, regionInfos := range serviceMeta.Vendors {
		total := 0
		for _, region := range regionInfos {
			total++
			metrics.HandleGlobalRoutingRegionalInfoMetrics(string(vendor), region.Region)
		}
		metrics.HandleGlobalRoutingRegionalTotalMetrics(string(vendor), total)
	}
	return nil
}

func (m *GlobalManager) LoadGlobalMetaCache(ctx context.Context) error {
	if m == nil {
		return fmt.Errorf("GlobalManager instance m is nil")
	}

	return m.loadGlobalMetaCache(ctx)
}