/*
Copyright 2024.

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.
*/

package providers

import (
	"context"
	"errors"
	"fmt"
	"maps"
	"net/http"
	"os"
	"strings"

	container "google.golang.org/api/container/v1"
	"google.golang.org/api/googleapi"
	"google.golang.org/api/option"
	corev1 "k8s.io/api/core/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"

	o11yv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
)

// GKECloudProvider implements CloudProviderInterface for Google GKE.
//
// Notes:
//   - This implementation uses Application Default Credentials.
//   - Cluster identifier supports either full resource name:
//     "projects/<p>/locations/<loc>/clusters/<name>"
//     or short cluster name with env var GCP_PROJECT set.
type GKECloudProvider struct {
	svc       *container.Service
	region    string
	k8sClient client.Client
}

func NewGKECloudProvider(region string, k8sClient client.Client) (*GKECloudProvider, error) {
	ctx := context.Background()
	svc, err := container.NewService(ctx, option.WithScopes(container.CloudPlatformScope))
	if err != nil {
		return nil, fmt.Errorf("failed to create GKE client: %w", err)
	}
	return &GKECloudProvider{svc: svc, region: region, k8sClient: k8sClient}, nil
}

func (g *GKECloudProvider) DescribeCluster(ctx context.Context, clusterName string) (any, error) {
	name, err := g.clusterResourceName(clusterName)
	if err != nil {
		return nil, err
	}
	cluster, err := g.svc.Projects.Locations.Clusters.Get(name).Context(ctx).Do()
	if err != nil {
		return nil, fmt.Errorf("failed to describe GKE cluster %q: %w", clusterName, err)
	}
	return cluster, nil
}

func (g *GKECloudProvider) UpgradeControlPlane(ctx context.Context, clusterName, targetVersion string) error {
	name, err := g.clusterResourceName(clusterName)
	if err != nil {
		return err
	}
	req := &container.UpdateClusterRequest{
		Update: &container.ClusterUpdate{
			DesiredMasterVersion: targetVersion,
		},
	}
	_, err = g.svc.Projects.Locations.Clusters.Update(name, req).Context(ctx).Do()
	if err != nil {
		return fmt.Errorf("failed to upgrade GKE control plane: %w", err)
	}
	return nil
}

func (_ *GKECloudProvider) UpgradeClusterAddons(_ context.Context, _ string, _ []o11yv1.AddonUpgrade) error {
	// GKE addon upgrades are not modeled the same way as EKS (no "addon version" per name in the same sense).
	// Keep this step as a no-op for now to avoid blocking the upgrade workflow.
	return nil
}

func (_ *GKECloudProvider) GetClusterAddonsStatus(_ context.Context, _ string, addons []o11yv1.AddonUpgrade) ([]AddonStatusInfo, error) {
	out := make([]AddonStatusInfo, len(addons))
	for i, a := range addons {
		out[i] = AddonStatusInfo{Name: a.Name, Version: a.Version, Status: "ACTIVE"}
	}
	return out, nil
}

func (_ *GKECloudProvider) ListClusterAddons(_ context.Context, _ string) ([]AddonStatusInfo, error) {
	// GKE addons are not modeled the same way as EKS; return empty.
	return nil, nil
}

func (g *GKECloudProvider) GetNodeGroupsByLabels(ctx context.Context, clusterName string, labelSelector map[string]string) ([]NodeGroup, error) {
	clusterRes, err := g.clusterResourceName(clusterName)
	if err != nil {
		return nil, err
	}

	resp, err := g.svc.Projects.Locations.Clusters.NodePools.List(clusterRes).Context(ctx).Do()
	if err != nil {
		return nil, fmt.Errorf("failed to list GKE nodepools: %w", err)
	}

	out := make([]NodeGroup, 0, len(resp.NodePools))
	for _, np := range resp.NodePools {
		labels := map[string]string{}
		if np.Config != nil && np.Config.Labels != nil {
			for k, v := range np.Config.Labels {
				labels[k] = v
			}
		}

		matches := true
		for k, v := range labelSelector {
			if labels[k] != v {
				matches = false
				break
			}
		}
		if !matches {
			continue
		}

		out = append(out, g.convertGKENodePoolToNodeGroup(ctx, clusterName, np))
	}
	return out, nil
}

func (g *GKECloudProvider) GetNodeGroupByName(ctx context.Context, clusterId string, nodeGroupName string) (*NodeGroup, error) {
	npRes, err := g.nodePoolResourceName(clusterId, nodeGroupName)
	if err != nil {
		return nil, err
	}
	np, err := g.svc.Projects.Locations.Clusters.NodePools.Get(npRes).Context(ctx).Do()
	if err != nil {
		var ae *googleapi.Error
		if errors.As(err, &ae) && ae.Code == http.StatusNotFound {
			return nil, ErrNodeGroupNotFound
		}
		return nil, fmt.Errorf("failed to describe GKE nodepool %q: %w", nodeGroupName, err)
	}
	ng := g.convertGKENodePoolToNodeGroup(ctx, clusterId, np)
	return &ng, nil
}

func (g *GKECloudProvider) CreateNodeGroup(ctx context.Context, clusterName string, spec NodeGroup) (string, error) {
	clusterRes, err := g.clusterResourceName(clusterName)
	if err != nil {
		return "", err
	}

	req := &container.CreateNodePoolRequest{
		NodePool: &container.NodePool{
			Name:             spec.Name,
			InitialNodeCount: int64(spec.DesiredSize),
			Version:          spec.Version,
			Config: &container.NodeConfig{
				MachineType: spec.InstanceType[0],
				Labels:      spec.Labels,
			},
		},
	}
	// Preserve the same availability zones as the original node pool.
	if len(spec.Locations) > 0 {
		req.NodePool.Locations = append([]string(nil), spec.Locations...)
	}

	// Autoscaling
	if spec.MinSize >= 0 && spec.MaxSize >= 0 {
		req.NodePool.Autoscaling = &container.NodePoolAutoscaling{
			Enabled:      true,
			MinNodeCount: int64(spec.MinSize),
			MaxNodeCount: int64(spec.MaxSize),
		}
	}

	_, err = g.svc.Projects.Locations.Clusters.NodePools.Create(clusterRes, req).Context(ctx).Do()
	if err != nil {
		return "", fmt.Errorf("failed to create GKE nodepool: %w", err)
	}
	return spec.Name, nil
}

func (g *GKECloudProvider) UpdateNodeGroupMinSize(ctx context.Context, clusterName, nodeGroupID string, minSize int) error {
	npRes, err := g.nodePoolResourceName(clusterName, nodeGroupID)
	if err != nil {
		return err
	}
	np, err := g.svc.Projects.Locations.Clusters.NodePools.Get(npRes).Context(ctx).Do()
	if err != nil {
		return fmt.Errorf("failed to describe GKE nodepool %q: %w", nodeGroupID, err)
	}
	if np.Autoscaling == nil {
		return fmt.Errorf("GKE nodepool %q does not have autoscaling configured", nodeGroupID)
	}
	autoscaling := *np.Autoscaling
	autoscaling.Enabled = true
	autoscaling.MinNodeCount = int64(minSize)
	autoscaling.ForceSendFields = append(autoscaling.ForceSendFields, "Enabled", "MinNodeCount")
	_, err = g.svc.Projects.Locations.Clusters.NodePools.SetAutoscaling(npRes, &container.SetNodePoolAutoscalingRequest{
		Name:        npRes,
		Autoscaling: &autoscaling,
	}).Context(ctx).Do()
	if err != nil {
		return fmt.Errorf("failed to update GKE nodepool %q min size: %w", nodeGroupID, err)
	}
	return nil
}

func (g *GKECloudProvider) UpgradeNodeGroup(ctx context.Context, clusterName, nodeGroupID, targetVersion string, _ *o11yv1.NodeGroupUpgradeOptions) (string, error) {
	npRes, err := g.nodePoolResourceName(clusterName, nodeGroupID)
	if err != nil {
		return "", err
	}
	req := &container.UpdateNodePoolRequest{
		Name:        npRes,
		NodeVersion: targetVersion,
	}
	op, err := g.svc.Projects.Locations.Clusters.NodePools.Update(npRes, req).Context(ctx).Do()
	if err != nil {
		return "", fmt.Errorf("failed to upgrade GKE nodepool %q: %w", nodeGroupID, err)
	}
	if op != nil && op.Name != "" {
		return op.Name, nil
	}
	return "", nil
}

func (g *GKECloudProvider) DeleteNodeGroup(ctx context.Context, clusterName, nodeGroupID string) error {
	npRes, err := g.nodePoolResourceName(clusterName, nodeGroupID)
	if err != nil {
		return err
	}
	_, err = g.svc.Projects.Locations.Clusters.NodePools.Delete(npRes).Context(ctx).Do()
	if err != nil {
		return fmt.Errorf("failed to delete GKE nodepool %q: %w", nodeGroupID, err)
	}
	return nil
}

// operationResourceName returns the full GKE operation resource name.
// GKE API expects "projects/{project}/locations/{location}/operations/{operationId}".
// If operationID is already a full path (starts with "projects/"), return as-is.
func (g *GKECloudProvider) operationResourceName(clusterName, operationID string) (string, error) {
	if operationID == "" {
		return "", fmt.Errorf("operationID is empty")
	}
	if strings.HasPrefix(operationID, "projects/") {
		return operationID, nil
	}
	clusterRes, err := g.clusterResourceName(clusterName)
	if err != nil {
		return "", err
	}
	// clusterRes: projects/{project}/locations/{location}/clusters/{cluster}
	// Extract projects/{project}/locations/{location} for operations parent.
	parts := strings.Split(clusterRes, "/")
	if len(parts) < 4 {
		return "", fmt.Errorf("invalid cluster resource name: %s", clusterRes)
	}
	parent := strings.Join(parts[:4], "/")
	return fmt.Sprintf("%s/operations/%s", parent, operationID), nil
}

// GetNodeGroupUpgradeOperationStatus polls the GKE Operation to check if the node pool upgrade is done.
func (g *GKECloudProvider) GetNodeGroupUpgradeOperationStatus(ctx context.Context, clusterName, nodeGroupID, operationID string) (bool, error) {
	if operationID == "" {
		return true, nil
	}
	opName, err := g.operationResourceName(clusterName, operationID)
	if err != nil {
		return false, fmt.Errorf("failed to resolve operation name: %w", err)
	}
	op, err := g.svc.Projects.Locations.Operations.Get(opName).Context(ctx).Do()
	if err != nil {
		return false, fmt.Errorf("failed to get GKE operation %q: %w", operationID, err)
	}
	switch op.Status {
	case "DONE":
		if op.Error != nil {
			return true, fmt.Errorf("GKE nodepool upgrade failed: %s", op.Error.Message)
		}
		return true, nil
	case "PENDING", "RUNNING", "ABORTING":
		return false, nil
	default:
		return false, fmt.Errorf("unknown GKE operation status: %s", op.Status)
	}
}

func (g *GKECloudProvider) GetNodeGroupStatus(ctx context.Context, clusterName, nodeGroupID string) (NodeGroupStatus, error) {
	np, err := g.GetNodeGroupByName(ctx, clusterName, nodeGroupID)
	if err != nil {
		return "", err
	}
	return np.Status, nil
}

func (g *GKECloudProvider) GetNodeGroupLabelSelector(_ context.Context, _ string, nodeGroupID string) (map[string]string, error) {
	// GKE labels nodes with their nodepool name under cloud.google.com/gke-nodepool.
	return map[string]string{
		"cloud.google.com/gke-nodepool": nodeGroupID,
	}, nil
}

func (g *GKECloudProvider) GetControlPlaneVersion(ctx context.Context, clusterName string) (string, error) {
	anyCluster, err := g.DescribeCluster(ctx, clusterName)
	if err != nil {
		return "", err
	}
	cluster, ok := anyCluster.(*container.Cluster)
	if !ok || cluster == nil {
		return "", fmt.Errorf("unexpected cluster type returned for GKE cluster %q", clusterName)
	}
	if cluster.CurrentMasterVersion == "" {
		return "", fmt.Errorf("GKE cluster %q current master version is empty", clusterName)
	}
	return cluster.CurrentMasterVersion, nil
}

// GetControlPlaneStatus returns the GKE cluster control plane status (ACTIVE or UPDATING).
func (g *GKECloudProvider) GetControlPlaneStatus(ctx context.Context, clusterName string) (ControlPlaneStatus, error) {
	anyCluster, err := g.DescribeCluster(ctx, clusterName)
	if err != nil {
		return "", err
	}
	cluster, ok := anyCluster.(*container.Cluster)
	if !ok || cluster == nil {
		return "", fmt.Errorf("unexpected cluster type returned for GKE cluster %q", clusterName)
	}
	// RECONCILING means cluster is being updated (e.g. master upgrade).
	if cluster.Status == "RECONCILING" {
		return ControlPlaneStatusUpdating, nil
	}
	if cluster.Status == "RUNNING" {
		return ControlPlaneStatusActive, nil
	}
	return ControlPlaneStatusOther, nil
}

// GetInstanceTypeByTemplateVersion is not applicable to GKE (no EC2-style launch templates). Returns empty string, nil.
func (_ *GKECloudProvider) GetInstanceTypeByTemplateVersion(_ context.Context, _, _, _ string) (string, error) {
	return "", nil
}

func (g *GKECloudProvider) clusterResourceName(clusterName string) (string, error) {
	if strings.HasPrefix(clusterName, "projects/") {
		return clusterName, nil
	}
	project := os.Getenv("GCP_PROJECT")
	if project == "" {
		project = "gcp-autonomous-prod"
	}
	location := g.region
	if location == "" {
		location = "-"
	}
	return fmt.Sprintf("projects/%s/locations/%s/clusters/%s", project, location, clusterName), nil
}

func (g *GKECloudProvider) nodePoolResourceName(clusterName, nodePoolName string) (string, error) {
	clusterRes, err := g.clusterResourceName(clusterName)
	if err != nil {
		return "", err
	}
	return fmt.Sprintf("%s/nodePools/%s", clusterRes, nodePoolName), nil
}

func (g *GKECloudProvider) convertGKENodePoolToNodeGroup(ctx context.Context, clusterName string, np *container.NodePool) NodeGroup {
	ng := NodeGroup{
		ID:      "",
		Name:    "",
		Labels:  map[string]string{},
		Version: "",
		Status:  "",
	}
	if np == nil {
		return ng
	}
	ng.ID = np.Name
	ng.Name = np.Name
	ng.Version = np.Version
	if np.Config != nil && np.Config.Labels != nil {
		maps.Copy(ng.Labels, np.Config.Labels)
	}
	if np.Config != nil {
		ng.InstanceType = []string{np.Config.MachineType}
	}
	if np.Autoscaling != nil && np.Autoscaling.Enabled {
		ng.MinSize = int(np.Autoscaling.MinNodeCount)
		ng.MaxSize = int(np.Autoscaling.MaxNodeCount)
	}
	ng.DesiredSize = int(np.InitialNodeCount)
	ng.NodeCount = ng.DesiredSize
	// Get actual node count from K8s (InitialNodeCount may differ for autoscaling nodepools).
	if g.k8sClient != nil {
		labelSelector, err := g.GetNodeGroupLabelSelector(ctx, clusterName, np.Name)
		if err == nil {
			nodeList := &corev1.NodeList{}
			if err := g.k8sClient.List(ctx, nodeList, client.MatchingLabels(labelSelector)); err == nil {
				ng.NodeCount = len(nodeList.Items)
				ng.DesiredSize = ng.NodeCount
			}
		}
	}
	if len(np.Locations) > 0 {
		ng.Locations = append([]string(nil), np.Locations...)
	}

	switch np.Status {
	case "PROVISIONING":
		ng.Status = NodeGroupStatusCreating
	case "RUNNING":
		ng.Status = NodeGroupStatusActive
	case "RECONCILING":
		ng.Status = NodeGroupStatusUpdating
	case "STOPPING":
		ng.Status = NodeGroupStatusDeleting
	case "ERROR":
		ng.Status = NodeGroupStatusCreateFailed
	default:
		ng.Status = NodeGroupStatus(np.Status)
	}

	return ng
}