// Package k8s provides Kubernetes client functionality
package k8s

import (
	"context"
	"fmt"
	"path/filepath"
	"sync"
	"time"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/discovery"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/homedir"
)

// PodLogsFunc is a function type for retrieving pod logs
type PodLogsFunc func(
	ctx context.Context,
	namespace, name string,
	parameters map[string]string,
) (*unstructured.Unstructured, error)

// ExecInPodFunc is a function type for executing commands in pods
type ExecInPodFunc func(
	ctx context.Context,
	namespace, name string,
	command []string,
	container string,
	timeout time.Duration,
) (*unstructured.Unstructured, error)

// Client represents a Kubernetes client with discovery and dynamic capabilities
type Client struct {
	discoveryClient discovery.DiscoveryInterface
	dynamicClient   dynamic.Interface
	clientset       kubernetes.Interface
	restConfig      *rest.Config
	kubeconfigPath  string
	mu              sync.RWMutex // Protects access to client components

	// For periodic refresh
	refreshCtx      context.Context
	refreshCancel   context.CancelFunc
	refreshInterval time.Duration
	refreshing      bool
	refreshMu       sync.Mutex // Protects refreshing state

	// function overrides for testing purposes
	getPodLogs PodLogsFunc
	execInPod  ExecInPodFunc

	// podLogReadSem bounds the number of pod-log reads in flight at once.
	// Each in-flight read can buffer up to maxPodLogLimitBytes; this caps
	// the aggregate memory the read path can hold across concurrent
	// requests. Buffered to maxConcurrentPodLogReads. See GHSA-qw5r-ppcg-f8rj.
	podLogReadSem chan struct{}
}

// maxConcurrentPodLogReads caps the number of pod-log reads that may run
// concurrently. With maxPodLogLimitBytes = 64 MiB per request, this bounds
// the aggregate transient memory of the read path at ~512 MiB.
const maxConcurrentPodLogReads = 8

// NewClient creates a new Kubernetes client
func NewClient(kubeconfigPath string) (*Client, error) {
	config, err := getConfig(kubeconfigPath)
	if err != nil {
		return nil, fmt.Errorf("failed to get Kubernetes config: %w", err)
	}

	// Create discovery client
	discoveryClient, err := discovery.NewDiscoveryClientForConfig(config)
	if err != nil {
		return nil, fmt.Errorf("failed to create discovery client: %w", err)
	}

	// Create dynamic client
	dynamicClient, err := dynamic.NewForConfig(config)
	if err != nil {
		return nil, fmt.Errorf("failed to create dynamic client: %w", err)
	}

	// Create clientset for typed API access
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		return nil, fmt.Errorf("failed to create clientset: %w", err)
	}

	client := &Client{
		discoveryClient: discoveryClient,
		dynamicClient:   dynamicClient,
		clientset:       clientset,
		restConfig:      config,
		kubeconfigPath:  kubeconfigPath,
		podLogReadSem:   make(chan struct{}, maxConcurrentPodLogReads),
	}

	// Set the default implementations
	client.getPodLogs = client.defaultGetPodLogs
	client.execInPod = client.defaultExecInPod

	return client, nil
}

// ConfigGetter is a function type for getting Kubernetes client configuration
type ConfigGetter func(kubeconfigPath string) (*rest.Config, error)

// defaultConfigGetter is the default implementation of ConfigGetter
func defaultConfigGetter(kubeconfigPath string) (*rest.Config, error) {
	if kubeconfigPath == "" {
		// Try in-cluster config first
		config, err := rest.InClusterConfig()
		if err == nil {
			return config, nil
		}

		// Fall back to default kubeconfig path
		if home := homedir.HomeDir(); home != "" {
			kubeconfigPath = filepath.Join(home, ".kube", "config")
		}
	}

	// Use the provided or default kubeconfig path
	return clientcmd.BuildConfigFromFlags("", kubeconfigPath)
}

// getConfig is the current ConfigGetter implementation
var getConfig ConfigGetter = defaultConfigGetter

// ListAPIResources returns all API resources supported by the Kubernetes API server
func (c *Client) ListAPIResources(_ context.Context) ([]*metav1.APIResourceList, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	_, resourcesList, err := c.discoveryClient.ServerGroupsAndResources()
	if err != nil {
		return nil, fmt.Errorf("failed to get server resources: %w", err)
	}
	return resourcesList, nil
}

// ListClusteredResources returns all clustered resources of the specified group/version/kind
// with optional pagination support via limit and continueToken parameters
func (c *Client) ListClusteredResources(
	ctx context.Context,
	gvr schema.GroupVersionResource,
	labelSelector string,
	limit int64,
	continueToken string,
) (*unstructured.UnstructuredList, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	listOptions := metav1.ListOptions{
		LabelSelector: labelSelector,
	}

	// Add pagination options if provided
	if limit > 0 {
		listOptions.Limit = limit
	}
	if continueToken != "" {
		listOptions.Continue = continueToken
	}

	return c.dynamicClient.Resource(gvr).List(ctx, listOptions)
}

// ListNamespacedResources returns all namespaced resources of the specified group/version/kind in the given namespace
// with optional pagination support via limit and continueToken parameters
func (c *Client) ListNamespacedResources(
	ctx context.Context,
	gvr schema.GroupVersionResource,
	namespace string,
	labelSelector string,
	limit int64,
	continueToken string,
) (*unstructured.UnstructuredList, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	listOptions := metav1.ListOptions{
		LabelSelector: labelSelector,
	}

	// Add pagination options if provided
	if limit > 0 {
		listOptions.Limit = limit
	}
	if continueToken != "" {
		listOptions.Continue = continueToken
	}

	return c.dynamicClient.Resource(gvr).Namespace(namespace).List(ctx, listOptions)
}

// ApplyClusteredResource creates or updates a clustered resource
func (c *Client) ApplyClusteredResource(
	ctx context.Context,
	gvr schema.GroupVersionResource,
	obj *unstructured.Unstructured,
) (*unstructured.Unstructured, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	name := obj.GetName()

	// Check if resource exists
	existing, err := c.dynamicClient.Resource(gvr).Get(ctx, name, metav1.GetOptions{})
	if err == nil {
		// Resource exists, update it
		// Set the resource version to ensure we're updating the latest version
		obj.SetResourceVersion(existing.GetResourceVersion())
		return c.dynamicClient.Resource(gvr).Update(ctx, obj, metav1.UpdateOptions{})
	}

	// Resource doesn't exist or error occurred, create it
	return c.dynamicClient.Resource(gvr).Create(ctx, obj, metav1.CreateOptions{})
}

// GetClusteredResource gets a clustered resource
func (c *Client) GetClusteredResource(ctx context.Context, gvr schema.GroupVersionResource, name string) (interface{}, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	return c.dynamicClient.Resource(gvr).Get(ctx, name, metav1.GetOptions{})
}

// GetNamespacedResource gets a namespaced resource
func (c *Client) GetNamespacedResource(
	ctx context.Context,
	gvr schema.GroupVersionResource,
	namespace, name string,
) (interface{}, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	return c.dynamicClient.Resource(gvr).Namespace(namespace).Get(ctx, name, metav1.GetOptions{})
}

// ApplyNamespacedResource creates or updates a namespaced resource
func (c *Client) ApplyNamespacedResource(
	ctx context.Context,
	gvr schema.GroupVersionResource,
	namespace string,
	obj *unstructured.Unstructured,
) (*unstructured.Unstructured, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	name := obj.GetName()

	// Check if resource exists
	existing, err := c.dynamicClient.Resource(gvr).Namespace(namespace).Get(ctx, name, metav1.GetOptions{})
	if err == nil {
		// Resource exists, update it
		// Set the resource version to ensure we're updating the latest version
		obj.SetResourceVersion(existing.GetResourceVersion())
		return c.dynamicClient.Resource(gvr).Namespace(namespace).Update(ctx, obj, metav1.UpdateOptions{})
	}

	// Resource doesn't exist or error occurred, create it
	return c.dynamicClient.Resource(gvr).Namespace(namespace).Create(ctx, obj, metav1.CreateOptions{})
}

// DeleteClusteredResource deletes a clustered resource
func (c *Client) DeleteClusteredResource(ctx context.Context, gvr schema.GroupVersionResource, name string) error {
	c.mu.RLock()
	defer c.mu.RUnlock()

	return c.dynamicClient.Resource(gvr).Delete(ctx, name, metav1.DeleteOptions{})
}

// DeleteNamespacedResource deletes a namespaced resource
func (c *Client) DeleteNamespacedResource(ctx context.Context, gvr schema.GroupVersionResource, namespace, name string) error {
	c.mu.RLock()
	defer c.mu.RUnlock()

	return c.dynamicClient.Resource(gvr).Namespace(namespace).Delete(ctx, name, metav1.DeleteOptions{})
}

// WithImpersonation returns a new Client that impersonates the given user and groups.
// The underlying rest.Config is cloned with ImpersonationConfig set, and new
// discovery, dynamic, and clientset clients are created from it.
//
// The returned Client is independent of the receiver — mutations to one do not
// affect the other. The returned Client does NOT inherit periodic refresh or
// testing overrides (getPodLogs, execInPod) from the base client; it uses the
// default implementations. This is intentional: impersonated clients are
// short-lived, per-request objects.
func (c *Client) WithImpersonation(user string, groups []string) (*Client, error) {
	c.mu.RLock()
	baseConfig := c.restConfig
	c.mu.RUnlock()

	if baseConfig == nil {
		return nil, fmt.Errorf("base rest config is nil")
	}

	cfg := rest.CopyConfig(baseConfig)
	cfg.Impersonate = rest.ImpersonationConfig{
		UserName: user,
		Groups:   groups,
	}

	discoveryClient, err := discovery.NewDiscoveryClientForConfig(cfg)
	if err != nil {
		return nil, fmt.Errorf("failed to create impersonated discovery client: %w", err)
	}

	dynamicClient, err := dynamic.NewForConfig(cfg)
	if err != nil {
		return nil, fmt.Errorf("failed to create impersonated dynamic client: %w", err)
	}

	clientset, err := kubernetes.NewForConfig(cfg)
	if err != nil {
		return nil, fmt.Errorf("failed to create impersonated clientset: %w", err)
	}

	impersonated := &Client{
		discoveryClient: discoveryClient,
		dynamicClient:   dynamicClient,
		clientset:       clientset,
		restConfig:      cfg,
		// Share the parent's semaphore: the memory budget is per-process,
		// not per-impersonation, so independent semaphores would let N
		// concurrent users each hold maxConcurrentPodLogReads slots.
		podLogReadSem: c.podLogReadSem,
	}

	// Set default implementations for pod logs and exec
	impersonated.getPodLogs = impersonated.defaultGetPodLogs
	impersonated.execInPod = impersonated.defaultExecInPod

	return impersonated, nil
}

// SetDynamicClient sets the dynamic client (for testing purposes)
func (c *Client) SetDynamicClient(dynamicClient dynamic.Interface) {
	c.mu.Lock()
	defer c.mu.Unlock()

	c.dynamicClient = dynamicClient
}

// SetDiscoveryClient sets the discovery client (for testing purposes)
func (c *Client) SetDiscoveryClient(discoveryClient discovery.DiscoveryInterface) {
	c.mu.Lock()
	defer c.mu.Unlock()

	c.discoveryClient = discoveryClient
}

// SetClientset sets the clientset (for testing purposes)
func (c *Client) SetClientset(clientset kubernetes.Interface) {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Store the interface directly, we'll use it through the interface methods
	c.clientset = clientset
}

// SetRestConfigForTest sets the rest config (for testing purposes)
func (c *Client) SetRestConfigForTest(config *rest.Config) {
	c.mu.Lock()
	defer c.mu.Unlock()

	c.restConfig = config
}

// SetPodLogsFunc sets the function used to get pod logs (for testing purposes)
func (c *Client) SetPodLogsFunc(getPodLogs PodLogsFunc) {
	c.mu.Lock()
	defer c.mu.Unlock()

	c.getPodLogs = getPodLogs
}

// GetPodLogs returns the current pod logs function
func (c *Client) GetPodLogs() PodLogsFunc {
	c.mu.RLock()
	defer c.mu.RUnlock()

	return c.getPodLogs
}

// SetExecInPodFunc sets the function used to execute commands in pods (for testing purposes)
func (c *Client) SetExecInPodFunc(execInPodFunc ExecInPodFunc) {
	c.mu.Lock()
	defer c.mu.Unlock()

	c.execInPod = execInPodFunc
}

// GetExecInPodFunc returns the current exec in pod function
func (c *Client) GetExecInPodFunc() ExecInPodFunc {
	c.mu.RLock()
	defer c.mu.RUnlock()

	return c.execInPod
}

// ExecInPod executes a command in a pod and returns the result
// The command will be killed if it exceeds the timeout
// If timeout is 0 or negative, the default timeout (15s) will be used
// If timeout exceeds MaxExecTimeout, it will be capped at MaxExecTimeout
func (c *Client) ExecInPod(
	ctx context.Context,
	namespace, name string,
	command []string,
	container string,
	timeout time.Duration,
) (*unstructured.Unstructured, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	return c.execInPod(ctx, namespace, name, command, container, timeout)
}

// IsReady returns true if the client is ready to use
func (c *Client) IsReady() bool {
	c.mu.RLock()
	defer c.mu.RUnlock()

	return c.discoveryClient != nil && c.dynamicClient != nil && c.clientset != nil
}

// RefreshClient re-reads the kubeconfig and updates the client's components
func (c *Client) RefreshClient() error {
	// Get the updated config
	config, err := getConfig(c.kubeconfigPath)
	if err != nil {
		return fmt.Errorf("failed to get updated Kubernetes config: %w", err)
	}

	// Create new discovery client
	discoveryClient, err := discovery.NewDiscoveryClientForConfig(config)
	if err != nil {
		return fmt.Errorf("failed to create discovery client: %w", err)
	}

	// Create new dynamic client
	dynamicClient, err := dynamic.NewForConfig(config)
	if err != nil {
		return fmt.Errorf("failed to create dynamic client: %w", err)
	}

	// Create new clientset for typed API access
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		return fmt.Errorf("failed to create clientset: %w", err)
	}

	// Update the client's components with proper locking
	c.mu.Lock()
	defer c.mu.Unlock()

	c.discoveryClient = discoveryClient
	c.dynamicClient = dynamicClient
	c.clientset = clientset
	c.restConfig = config

	return nil
}

// StartPeriodicRefresh starts a goroutine that periodically refreshes the client's configuration
// The interval specifies how often to refresh the configuration
// Returns an error if refresh is already running
func (c *Client) StartPeriodicRefresh(interval time.Duration) error {
	c.refreshMu.Lock()
	defer c.refreshMu.Unlock()

	if c.refreshing {
		return fmt.Errorf("periodic refresh is already running")
	}

	// Create a cancellable context for the refresh goroutine
	ctx, cancel := context.WithCancel(context.Background())
	c.refreshCtx = ctx
	c.refreshCancel = cancel
	c.refreshInterval = interval
	c.refreshing = true

	// Start the refresh goroutine
	go func() {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()

		for {
			select {
			case <-ticker.C:
				// Refresh the client
				if err := c.RefreshClient(); err != nil {
					// Log the error but continue refreshing
					fmt.Printf("Error refreshing Kubernetes client: %v\n", err)
				}
			case <-ctx.Done():
				// Context cancelled, stop refreshing
				return
			}
		}
	}()

	return nil
}

// StopPeriodicRefresh stops the periodic refresh goroutine
// Returns an error if refresh is not running
func (c *Client) StopPeriodicRefresh() error {
	c.refreshMu.Lock()
	defer c.refreshMu.Unlock()

	if !c.refreshing {
		return fmt.Errorf("periodic refresh is not running")
	}

	// Cancel the refresh context to stop the goroutine
	c.refreshCancel()
	c.refreshing = false

	return nil
}

// IsRefreshing returns true if the client is periodically refreshing
func (c *Client) IsRefreshing() bool {
	c.refreshMu.Lock()
	defer c.refreshMu.Unlock()

	return c.refreshing
}

// GetRefreshInterval returns the current refresh interval
// Returns 0 if refresh is not running
func (c *Client) GetRefreshInterval() time.Duration {
	c.refreshMu.Lock()
	defer c.refreshMu.Unlock()

	if !c.refreshing {
		return 0
	}

	return c.refreshInterval
}
