// Licensed to Apache Software Foundation (ASF) under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Apache Software Foundation (ASF) licenses this file to you 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 testhelper provides test helpers for FODC agent components.
package testhelper

import (
	"context"
	"fmt"
	"time"

	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
	fodcv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/fodc/v1"
	agentcluster "github.com/apache/skywalking-banyandb/fodc/agent/internal/cluster"
	"github.com/apache/skywalking-banyandb/fodc/agent/internal/crashcollector"
	"github.com/apache/skywalking-banyandb/fodc/agent/internal/flightrecorder"
	agentlifecycle "github.com/apache/skywalking-banyandb/fodc/agent/internal/lifecycle"
	agentmetrics "github.com/apache/skywalking-banyandb/fodc/agent/internal/metrics"
	"github.com/apache/skywalking-banyandb/fodc/agent/internal/proxy"
	"github.com/apache/skywalking-banyandb/pkg/logger"
)

// NewFlightRecorder creates a new FlightRecorder instance for testing.
func NewFlightRecorder(capacitySize int64) *flightrecorder.FlightRecorder {
	return flightrecorder.NewFlightRecorder(capacitySize)
}

// RawMetric represents a metric for testing.
type RawMetric struct {
	Desc   string
	Name   string
	Type   string
	Labels []Label
	Value  float64
}

// Label represents a metric label.
type Label struct {
	Name  string
	Value string
}

// UpdateMetrics updates the FlightRecorder with metrics.
// fr should be a *flightrecorder.FlightRecorder instance.
func UpdateMetrics(fr interface{}, rawMetrics []RawMetric) error {
	frTyped, ok := fr.(*flightrecorder.FlightRecorder)
	if !ok {
		return fmt.Errorf("invalid FlightRecorder type")
	}
	// Convert test RawMetric to internal RawMetric
	internalMetrics := make([]agentmetrics.RawMetric, len(rawMetrics))
	for idx := range rawMetrics {
		internalLabels := make([]agentmetrics.Label, len(rawMetrics[idx].Labels))
		for labelIdx := range rawMetrics[idx].Labels {
			internalLabels[labelIdx] = agentmetrics.Label{
				Name:  rawMetrics[idx].Labels[labelIdx].Name,
				Value: rawMetrics[idx].Labels[labelIdx].Value,
			}
		}
		internalMetrics[idx] = agentmetrics.RawMetric{
			Name:   rawMetrics[idx].Name,
			Value:  rawMetrics[idx].Value,
			Desc:   rawMetrics[idx].Desc,
			Type:   rawMetrics[idx].Type,
			Labels: internalLabels,
		}
	}
	return frTyped.Update(internalMetrics)
}

// ValidateMetricsBufferAlignment validates that the timestamps buffer length equals each metric buffer length.
func ValidateMetricsBufferAlignment(fr interface{}) error {
	frTyped, ok := fr.(*flightrecorder.FlightRecorder)
	if !ok {
		return fmt.Errorf("invalid FlightRecorder type")
	}

	datasources := frTyped.GetDatasources()
	if len(datasources) == 0 {
		return fmt.Errorf("no datasources")
	}

	ds := datasources[0]
	timestamps := ds.GetTimestamps()
	if timestamps == nil {
		return fmt.Errorf("timestamps ring buffer is nil")
	}

	timestampValues := timestamps.GetAllValues()
	timestampValuesLen := len(timestampValues)
	if timestampValuesLen == 0 {
		return fmt.Errorf("timestamps ring buffer has no values")
	}

	metricsMap := ds.GetMetrics()
	if len(metricsMap) == 0 {
		return fmt.Errorf("metrics map is empty")
	}

	for metricKey, metricBuffer := range metricsMap {
		if metricBuffer == nil {
			return fmt.Errorf("metric buffer is nil for key %q", metricKey)
		}
		metricValuesLen := len(metricBuffer.GetAllValues())
		if metricValuesLen != timestampValuesLen {
			return fmt.Errorf("metric buffer %q length %d != timestamps length %d", metricKey, metricValuesLen, timestampValuesLen)
		}
	}

	return nil
}

// NewProxyClient creates a new ProxyClient instance for testing.
func NewProxyClient(
	proxyAddr string,
	nodeRole string,
	podName string,
	containerNames []string,
	labels map[string]string,
	heartbeatInterval time.Duration,
	reconnectInterval time.Duration,
	flightRecorder interface{},
	logger *logger.Logger,
) *proxy.Client {
	return NewProxyClientWithCollector(
		proxyAddr,
		nodeRole,
		podName,
		containerNames,
		labels,
		heartbeatInterval,
		reconnectInterval,
		flightRecorder,
		nil,
		logger,
	)
}

// NewProxyClientWithCollector creates a new ProxyClient instance for testing with optional cluster collector.
func NewProxyClientWithCollector(
	proxyAddr string,
	nodeRole string,
	podName string,
	containerNames []string,
	labels map[string]string,
	heartbeatInterval time.Duration,
	reconnectInterval time.Duration,
	flightRecorder interface{},
	collector *agentcluster.Collector,
	logger *logger.Logger,
) *proxy.Client {
	frTyped, ok := flightRecorder.(*flightrecorder.FlightRecorder)
	if !ok {
		// This should not happen in normal usage, but handle gracefully
		return nil
	}
	return proxy.NewClient(
		proxyAddr,
		nodeRole,
		podName,
		containerNames,
		labels,
		heartbeatInterval,
		reconnectInterval,
		frTyped,
		collector,
		nil,
		nil,
		nil,
		logger,
	)
}

// ProxyClientWrapper wraps ProxyClient methods for testing.
type ProxyClientWrapper struct {
	client             *proxy.Client
	collector          *agentcluster.Collector
	crashWatcher       *crashcollector.DirectoryWatcher
	lifecycleCollector *agentlifecycle.Collector
	ctx                context.Context
	connMgrActive      bool
}

// StartConnManager starts the connection manager.
func (w *ProxyClientWrapper) StartConnManager(ctx context.Context) {
	w.ctx = ctx
	w.client.StartConnManager(ctx)
	w.connMgrActive = true
}

// Connect establishes a gRPC connection to Proxy.
func (w *ProxyClientWrapper) Connect(ctx context.Context) error {
	// Auto-start connection manager if not already started
	if !w.connMgrActive {
		w.StartConnManager(ctx)
	}
	return w.client.Connect(ctx)
}

// StartRegistrationStream establishes bi-directional registration stream with Proxy.
func (w *ProxyClientWrapper) StartRegistrationStream(ctx context.Context) error {
	return w.client.StartRegistrationStream(ctx)
}

// StartMetricsStream establishes bi-directional metrics stream with Proxy.
func (w *ProxyClientWrapper) StartMetricsStream(ctx context.Context) error {
	return w.client.StartMetricsStream(ctx)
}

// StartClusterStateStream establishes bi-directional cluster state stream with Proxy.
func (w *ProxyClientWrapper) StartClusterStateStream(ctx context.Context) error {
	return w.client.StartClusterStateStream(ctx)
}

// Start starts the proxy client with automatic connection and stream setup.
func (w *ProxyClientWrapper) Start(ctx context.Context) error {
	w.ctx = ctx
	return w.client.Start(ctx)
}

// Disconnect closes connection to Proxy.
func (w *ProxyClientWrapper) Disconnect() error {
	w.connMgrActive = false
	return w.client.Disconnect()
}

// SetClusterTopology seeds the wrapped client's cluster collector for testing.
func (w *ProxyClientWrapper) SetClusterTopology(nodes []*databasev1.Node, calls []*fodcv1.Call) error {
	if w == nil || w.collector == nil {
		return fmt.Errorf("cluster collector not available")
	}
	w.collector.SetClusterTopology(agentcluster.TopologyMap{
		Nodes: nodes,
		Calls: calls,
	})
	return nil
}

// StartLifecycleStream establishes bi-directional lifecycle stream with Proxy.
func (w *ProxyClientWrapper) StartLifecycleStream(ctx context.Context) error {
	return w.client.StartLifecycleStream(ctx)
}

// StartCrashStream establishes bi-directional crash diagnostics stream with Proxy.
func (w *ProxyClientWrapper) StartCrashStream(ctx context.Context) error {
	return w.client.StartCrashStream(ctx)
}

// ScanCrashDirectory scans the wrapped client's crash artifact directory.
func (w *ProxyClientWrapper) ScanCrashDirectory() {
	if w == nil || w.crashWatcher == nil {
		return
	}
	w.crashWatcher.Scan()
}

// NewProxyClientWrapper creates a wrapped ProxyClient for testing.
func NewProxyClientWrapper(
	proxyAddr string,
	nodeRole string,
	podName string,
	containerNames []string,
	labels map[string]string,
	heartbeatInterval time.Duration,
	reconnectInterval time.Duration,
	flightRecorder interface{},
	logger *logger.Logger,
	reportDir string,
) *ProxyClientWrapper {
	frTyped, ok := flightRecorder.(*flightrecorder.FlightRecorder)
	if !ok {
		return nil
	}
	clusterCollector := &agentcluster.Collector{}
	lifecycleCollector := agentlifecycle.NewCollector(nil, "", reportDir, 0)
	client := proxy.NewClient(
		proxyAddr,
		nodeRole,
		podName,
		containerNames,
		labels,
		heartbeatInterval,
		reconnectInterval,
		frTyped,
		clusterCollector,
		lifecycleCollector,
		nil,
		nil,
		logger,
	)
	return &ProxyClientWrapper{
		client:             client,
		collector:          clusterCollector,
		lifecycleCollector: lifecycleCollector,
	}
}

// NewProxyClientWrapperWithCrashDir creates a wrapped ProxyClient for testing with crash artifact collection.
func NewProxyClientWrapperWithCrashDir(
	proxyAddr string,
	nodeRole string,
	podName string,
	containerNames []string,
	labels map[string]string,
	heartbeatInterval time.Duration,
	reconnectInterval time.Duration,
	flightRecorder interface{},
	logger *logger.Logger,
	crashDir string,
) *ProxyClientWrapper {
	frTyped, ok := flightRecorder.(*flightrecorder.FlightRecorder)
	if !ok {
		return nil
	}
	clusterCollector := &agentcluster.Collector{}
	crashWatcher := crashcollector.NewDirectoryWatcher(logger, crashDir, crashcollector.Config{})
	client := proxy.NewClient(
		proxyAddr,
		nodeRole,
		podName,
		containerNames,
		labels,
		heartbeatInterval,
		reconnectInterval,
		frTyped,
		clusterCollector,
		nil,
		crashWatcher,
		nil,
		logger,
	)
	return &ProxyClientWrapper{
		client:       client,
		collector:    clusterCollector,
		crashWatcher: crashWatcher,
	}
}
