package grpc_service

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net/http"

	"github.com/tidbcloud/config-center/internal/config"
	"github.com/tidbcloud/config-center/internal/db"
	"github.com/tidbcloud/config-center/internal/service"
	ccch "github.com/tidbcloud/config-center/pkg/http_client"
	pb "github.com/tidbcloud/config-center/proto/config_center"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/types/known/emptypb"
)

type ServiceMode string

const (
	ModeGlobal   ServiceMode = "global"
	ModeRegional ServiceMode = "regional"
)

type proxyable interface {
	GetClusterType() pb.ClusterType
	GetLogicalRegion() string
}

type ConfigCenterService struct {
	pb.UnimplementedConfigCenterServiceServer
	DB             db.DB
	ServiceMode    ServiceMode
	UnifiedService *service.ConfigCenterService
}

func NewGlobalConfigCenterService(database db.DB) (*ConfigCenterService, error) {
	routers, err := database.ListRegionalServices()
	if err != nil {
		return nil, fmt.Errorf("failed to list regional services on startup: %w", err)
	}
	for _, r := range routers {
		slog.Info("discovered regional config-center client",
			"cluster_type", r.ClusterType,
			"logical_region", r.LogicalRegion,
			"endpoint", r.Endpoint,
			"sync_status", r.SyncStatus)
	}

	return &ConfigCenterService{
		DB:          database,
		ServiceMode: ModeGlobal,
		UnifiedService: &service.ConfigCenterService{
			DB: database,
		},
	}, nil
}

func NewRegionalConfigCenterService(db db.DB, cfg *config.RootConfig) *ConfigCenterService {
	service.ServerRegion = cfg.Regional.LogicalRegion
	return &ConfigCenterService{
		DB:          db,
		ServiceMode: ModeRegional,
		UnifiedService: &service.ConfigCenterService{
			DB:           db,
			ServerRegion: cfg.Regional.LogicalRegion,
		},
	}
}

func (c ConfigCenterService) getRegionalClientIfRoutable(req proxyable) *ccch.ConfigCenterHTTPClient {
	if c.ServiceMode != ModeGlobal || !req.GetClusterType().UseRegionalMode() {
		return nil
	}
	return c.UnifiedService.GetRegionalClient(req.GetClusterType(), req.GetLogicalRegion())
}

func (c ConfigCenterService) RegisterRegionalServer(ctx context.Context, req *pb.RegisterRegionalServerRequest) (*pb.RegionalServer, error) {
	if c.ServiceMode != ModeGlobal {
		return nil, status.Errorf(codes.InvalidArgument, "service mode is not global")
	}
	if !req.GetClusterType().UseRegionalMode() {
		return nil, status.Errorf(codes.InvalidArgument, "cluster type %s does not support regional mode", req.GetClusterType().String())
	}

	slog.Info("UpsertRegionalService", "cluster_type", req.GetClusterType(), "logical_region", req.GetLogicalRegion(), "server_address", req.GetServerAddress(), "is_byoc", req.GetIsByoc())
	err := c.DB.UpsertRegionalService(req.GetClusterType().String(), req.GetLogicalRegion(), req.GetServerAddress())
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to upsert regional service: %v", err)
	}

	// Invalidate cache so all subsequent calls on this pod pick up the change immediately.
	// Other pods will pick it up when their TTL expires.
	c.UnifiedService.InvalidateRegionalCache()

	// workaround: insert region level INITSQL default template
	if req.GetIsByoc() && req.GetClusterType() == pb.ClusterType_SHARED_POOL {
		c.insertByocDefaultTemplateBinding(ctx, req.GetLogicalRegion())
	}

	go c.doSync(req.GetClusterType(), req.GetLogicalRegion(), req.GetServerAddress())

	return regionRouteToProto(&db.RegionRoute{
		ClusterType:   req.GetClusterType().String(),
		LogicalRegion: req.GetLogicalRegion(),
		Endpoint:      req.GetServerAddress(),
		SyncStatus:    db.SyncStatusPending,
	}), nil
}

func (c ConfigCenterService) ListRegionalServers(ctx context.Context, req *pb.ListRegionalServersRequest) (*pb.ListRegionalServersResponse, error) {
	if c.ServiceMode != ModeGlobal {
		return nil, status.Errorf(codes.InvalidArgument, "service mode is not global")
	}
	if !req.GetClusterType().UseRegionalMode() {
		return nil, status.Errorf(codes.InvalidArgument, "cluster type %s does not support regional mode", req.GetClusterType().String())
	}

	routers, err := c.DB.ListRegionalServices()
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to get regional config-center endpoints: %v", err)
	}
	servers := make([]*pb.RegionalServer, 0)
	for _, router := range routers {
		if router.ClusterType == req.GetClusterType().String() {
			servers = append(servers, regionRouteToProto(router))
		}
	}
	return &pb.ListRegionalServersResponse{
		Servers: servers,
	}, nil
}

func (c ConfigCenterService) GetRegionalServer(ctx context.Context, req *pb.GetRegionalServerRequest) (*pb.RegionalServer, error) {
	if c.ServiceMode != ModeGlobal {
		return nil, status.Errorf(codes.InvalidArgument, "service mode is not global")
	}

	route, err := c.DB.GetRegionalService(req.GetClusterType().String(), req.GetLogicalRegion())
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to get regional service: %v", err)
	}
	return regionRouteToProto(route), nil
}

func (c ConfigCenterService) SyncRegionalServer(ctx context.Context, req *pb.SyncRegionalServerRequest) (*pb.RegionalServer, error) {
	if c.ServiceMode != ModeGlobal {
		return nil, status.Errorf(codes.InvalidArgument, "service mode is not global")
	}

	route, err := c.DB.GetRegionalService(req.GetClusterType().String(), req.GetLogicalRegion())
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to get regional service: %v", err)
	}

	// Mark syncing before returning so an immediate GET sees the new state.
	if err := c.DB.UpdateRegionalServiceSyncStatus(route.ClusterType, route.LogicalRegion, db.SyncStatusSyncing); err != nil {
		return nil, status.Errorf(codes.Internal, "failed to mark sync status as syncing: %v", err)
	}
	route.SyncStatus = db.SyncStatusSyncing

	go c.doSync(req.GetClusterType(), route.LogicalRegion, route.Endpoint)

	return regionRouteToProto(route), nil
}

func (c ConfigCenterService) OfflineRegionalServer(ctx context.Context, req *pb.OfflineRegionalServerRequest) (*emptypb.Empty, error) {
	if c.ServiceMode != ModeGlobal {
		return nil, status.Errorf(codes.InvalidArgument, "service mode is not global")
	}

	slog.Info("OfflineRegionalServer", "cluster_type", req.GetClusterType(), "logical_region", req.GetLogicalRegion())
	err := c.DB.DeleteRegionalService(req.GetClusterType().String(), req.GetLogicalRegion())
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to delete regional service record: %v", err)
	}

	c.UnifiedService.InvalidateRegionalCache()

	return &emptypb.Empty{}, nil
}

func (c ConfigCenterService) insertByocDefaultTemplateBinding(ctx context.Context, region string) {
	clusterTypeStr := pb.ClusterType_SHARED_POOL.String()
	version, err := c.DB.ListVersionsFromComponentImage(clusterTypeStr, "tikv")
	if err != nil {
		slog.Error("failed to list tikv versions for BYOC", "error", err)
		return
	} else {
		for _, v := range version {
			templateID, err := c.DB.GetDefaultTemplate(clusterTypeStr, pb.SHARED_POOL_CONFIG_INITSQL, v, "", "", pb.BYOC_EXAMPLE_REGION, "")
			if err != nil {
				slog.Error("failed to get default INITSQL template for BYOC", "error", err)
			} else {
				c.UpdateDefaultConfigurationTemplate(ctx, &pb.UpdateDefaultConfigurationTemplateRequest{
					ClusterType:       pb.ClusterType_SHARED_POOL,
					ConfigurationType: pb.SHARED_POOL_CONFIG_INITSQL,
					Version:           v,
					Regions:           []string{region},
					TemplateId:        templateID,
				})
			}
		}
	}
}

func regionRouteToProto(route *db.RegionRoute) *pb.RegionalServer {
	return &pb.RegionalServer{
		ClusterType:   pb.ClusterType(pb.ClusterType_value[route.ClusterType]),
		LogicalRegion: route.LogicalRegion,
		ServerAddress: route.Endpoint,
		SyncStatus:    route.SyncStatus,
	}
}

func (c ConfigCenterService) doSync(clusterType pb.ClusterType, region, endpoint string) {
	clusterTypeStr := clusterType.String()

	if err := c.DB.UpdateRegionalServiceSyncStatus(clusterTypeStr, region, db.SyncStatusSyncing); err != nil {
		slog.Error("failed to update sync status to syncing", "region", region, "error", err)
		return
	}

	err := c.syncAllDataToRegion(context.Background(), clusterType, region, endpoint)
	if err != nil {
		slog.Error("failed to sync all data to regional server", "region", region, "error", err)
		if dbErr := c.DB.UpdateRegionalServiceSyncStatus(clusterTypeStr, region, db.SyncStatusFailed); dbErr != nil {
			slog.Error("failed to update sync status to failed", "region", region, "error", dbErr)
		}
		return
	}

	if err := c.DB.UpdateRegionalServiceSyncStatus(clusterTypeStr, region, db.SyncStatusSynced); err != nil {
		slog.Error("failed to update sync status to synced", "region", region, "error", err)
	}
}

func (c ConfigCenterService) syncAllDataToRegion(ctx context.Context, clusterType pb.ClusterType, region, endpoint string) error {
	client := &ccch.ConfigCenterHTTPClient{
		HTTPClient: http.DefaultClient,
		BaseURL:    endpoint,
	}

	slog.Info("Start to sync templates to regional server", "cluster_type", clusterType, "region", region, "endpoint", endpoint)
	if err := c.syncAllTemplatesToRegion(ctx, client, clusterType); err != nil {
		return fmt.Errorf("failed to sync templates: %w", err)
	}

	slog.Info("Start to sync default versions to regional server", "cluster_type", clusterType, "region", region, "endpoint", endpoint)
	if err := c.syncAllDefaultVersionsToRegion(ctx, client, clusterType); err != nil {
		return fmt.Errorf("failed to sync default versions: %w", err)
	}

	slog.Info("Start to sync default templates to regional server", "cluster_type", clusterType, "region", region, "endpoint", endpoint)
	if err := c.syncAllDefaultTemplatesToRegion(ctx, client, clusterType); err != nil {
		return fmt.Errorf("failed to sync default templates: %w", err)
	}

	slog.Info("Start to sync component images to regional server", "cluster_type", clusterType, "region", region, "endpoint", endpoint)
	if err := c.syncAllComponentImagesToRegion(ctx, client, clusterType); err != nil {
		return fmt.Errorf("failed to sync component images: %w", err)
	}

	slog.Info("Start to sync tenant initial custom params to regional server", "cluster_type", clusterType, "region", region, "endpoint", endpoint)
	if err := c.syncAllTenantInitialCustomParamsToRegion(ctx, client, clusterType); err != nil {
		return fmt.Errorf("failed to sync tenant initial custom params: %w", err)
	}

	return nil
}

func (c ConfigCenterService) syncAllTemplatesToRegion(ctx context.Context, client *ccch.ConfigCenterHTTPClient, clusterType pb.ClusterType) error {
	templates, err := service.ListConfigurationTemplates(c.DB, &pb.ListConfigurationTemplatesRequest{
		ClusterType: clusterType,
	})
	if err != nil {
		return fmt.Errorf("failed to list templates: %w", err)
	}

	var errs []error
	for _, template := range templates.Templates {
		slog.Info("Create template on regional server", "template_id", template.TemplateId, "cluster_type", clusterType)
		_, err := client.CreateConfigurationTemplate(ctx, &pb.CreateConfigurationTemplateRequest{
			ClusterType:       template.ClusterType,
			TemplateId:        &template.TemplateId,
			ConfigurationType: template.ConfigurationType,
			TemplateName:      template.TemplateName,
			Data:              template.Data,
			Tags:              template.Tags,
		})
		if err != nil {
			slog.Error("failed to sync template to region", "template_id", template.TemplateId, "error", err)
			errs = append(errs, fmt.Errorf("template %d: %w", template.TemplateId, err))
		}
	}

	return errors.Join(errs...)
}

func (c ConfigCenterService) syncAllDefaultVersionsToRegion(ctx context.Context, client *ccch.ConfigCenterHTTPClient, clusterType pb.ClusterType) error {
	rules, err := service.ListDefaultVersionRules(c.DB, &pb.ListDefaultVersionRulesRequest{
		ClusterType: clusterType,
	})
	if err != nil {
		return fmt.Errorf("failed to list default version rules: %w", err)
	}

	var errs []error
	for _, rule := range rules.Rules {
		req := &pb.UpdateDefaultVersionRequest{
			ClusterType:    rule.ClusterType,
			Component:      rule.Component,
			ClusterVersion: &rule.ClusterVersion,
			CloudProvider:  &rule.CloudProvider,
			Region:         &rule.Region,
			Version:        rule.Version,
		}
		slog.Info("Update default version rule on regional server", "rule", protojson.Format(req))
		_, err := client.UpdateDefaultVersion(ctx, req)
		if err != nil {
			slog.Error("failed to sync default version to region", "component", rule.Component, "version", rule.Version, "error", err)
			errs = append(errs, fmt.Errorf("%s/%s: %w", rule.Component, rule.Version, err))
		}
	}

	return errors.Join(errs...)
}

func (c ConfigCenterService) syncAllDefaultTemplatesToRegion(ctx context.Context, client *ccch.ConfigCenterHTTPClient, clusterType pb.ClusterType) error {
	pageSize := int32(200)
	pageNum := int32(1)

	var errs []error
	for {
		bindings, err := c.DB.ListDefaultTemplateBindings(clusterType.String(), "", "", "", "", "", pageSize, pageNum, false)
		if err != nil {
			return fmt.Errorf("failed to list default template bindings: %w", err)
		}

		if len(bindings) == 0 {
			break
		}

		// sync default template binding to region
		for _, binding := range bindings {
			req := &pb.UpdateDefaultConfigurationTemplateRequest{
				ClusterType:       clusterType,
				ConfigurationType: binding.ConfigType,
				Version:           binding.Version,
				CloudProvider:     binding.CloudProvider,
				NodeInstance:      binding.NodeInstance,
				Regions:           []string{binding.Region},
				TemplateId:        binding.TemplateID,
			}
			slog.Info("Update default template binding on regional server", "req", protojson.Format(req))
			_, err := client.UpdateDefaultConfigurationTemplate(ctx, req)
			if err != nil {
				slog.Error("failed to sync default template binding to region",
					"template_id", binding.TemplateID,
					"config_type", binding.ConfigType,
					"version", binding.Version,
					"cloud_provider", binding.CloudProvider,
					"node_instance", binding.NodeInstance,
					"region", binding.Region,
					"error", err)
				errs = append(errs, fmt.Errorf("binding template=%d %s/%s/%s/%s/%s: %w",
					binding.TemplateID, binding.ConfigType, binding.Version, binding.CloudProvider, binding.NodeInstance, binding.Region, err))
			}
		}

		if len(bindings) < int(pageSize) {
			break
		}
		pageNum++
	}

	return errors.Join(errs...)
}

func (c ConfigCenterService) syncAllComponentImagesToRegion(ctx context.Context, client *ccch.ConfigCenterHTTPClient, clusterType pb.ClusterType) error {
	// Get all component images for this cluster type by calling the database directly
	rows, err := c.DB.ListComponentImages(clusterType.String(), "", "")
	if err != nil {
		return fmt.Errorf("failed to list component images: %w", err)
	}

	var errs []error
	// Sync each component image to the regional server
	for _, row := range rows {
		req := &pb.UpdateComponentImageRequest{
			ClusterType:      clusterType,
			ComponentName:    row.ComponentName,
			ComponentVersion: row.Version,
			BaseImage:        row.BaseImage,
			Tag:              row.Tag,
		}
		slog.Info("Update component image on regional server", "component", row.ComponentName, "version", row.Version)
		_, err := client.UpdateComponentImage(ctx, req)
		if err != nil {
			slog.Error("failed to sync component image to region",
				"component", row.ComponentName,
				"version", row.Version,
				"base_image", row.BaseImage,
				"tag", row.Tag,
				"error", err)
			errs = append(errs, fmt.Errorf("component %s/%s: %w", row.ComponentName, row.Version, err))
		}
	}

	return errors.Join(errs...)
}

func (c ConfigCenterService) syncAllTenantInitialCustomParamsToRegion(ctx context.Context, client *ccch.ConfigCenterHTTPClient, clusterType pb.ClusterType) error {
	rows, err := c.DB.ListAllTenantInitialCustomParams(clusterType.String())
	if err != nil {
		return fmt.Errorf("failed to list all tenant initial custom params: %w", err)
	}

	// Group rows by (tenant_id, config_type). The region applies the batch as a
	// full-replace per (tenant, config_type), so sending a partial set would
	// silently delete the failing key on the region side. If any value fails to
	// convert, mark the whole group as poisoned and skip it — better to leave
	// the region state untouched and report the error than to drop a key.
	type key struct {
		TenantID   uint64
		ConfigType string
	}
	grouped := make(map[key]map[string]*pb.ParamItem)
	poisoned := make(map[key]bool)
	var errs []error
	for _, row := range rows {
		k := key{TenantID: row.TenantID, ConfigType: row.ConfigType}
		if grouped[k] == nil {
			grouped[k] = make(map[string]*pb.ParamItem)
		}
		v, err := pb.NewValue(row.Value.Raw)
		if err != nil {
			slog.Error("failed to convert tenant initial custom param value, dropping group to avoid partial overwrite",
				"tenant_id", row.TenantID,
				"config_type", row.ConfigType,
				"key", row.Key,
				"error", err)
			poisoned[k] = true
			errs = append(errs, fmt.Errorf("tenant %d config_type %s key %q: %w", row.TenantID, row.ConfigType, row.Key, err))
			continue
		}
		grouped[k][row.Key] = &pb.ParamItem{
			Value:   v,
			Comment: row.Comment,
		}
	}

	// Re-group clean entries by tenant so one batch carries all config types.
	byTenant := make(map[uint64][]*pb.UpdateTenantInitialCustomParamsRequest)
	for k, params := range grouped {
		if poisoned[k] {
			continue
		}
		byTenant[k.TenantID] = append(byTenant[k.TenantID], &pb.UpdateTenantInitialCustomParamsRequest{
			ClusterType:       clusterType,
			ConfigurationType: k.ConfigType,
			TenantId:          k.TenantID,
			InitialParams:     params,
		})
	}

	for tenantID, requests := range byTenant {
		req := &pb.BatchUpdateTenantInitialCustomParamsRequest{
			ClusterType: clusterType,
			TenantId:    tenantID,
			Requests:    requests,
		}
		slog.Info("Update tenant initial custom params on regional server",
			"tenant_id", tenantID,
			"config_types", len(requests))
		if _, err := client.BatchUpdateTenantInitialCustomParams(ctx, req); err != nil {
			slog.Error("failed to sync tenant initial custom params to region",
				"tenant_id", tenantID,
				"error", err)
			errs = append(errs, fmt.Errorf("tenant %d: %w", tenantID, err))
		}
	}

	return errors.Join(errs...)
}