package store

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"os"
	"strings"
	"time"

	duckdb "github.com/marcboeker/go-duckdb"
	"github.com/pingcap/log"
	dto "github.com/prometheus/client_model/go"
	"github.com/stretchr/testify/mock"
	"go.uber.org/zap"
)

type MockRemoteScanner struct {
	data      [][]*Metric
	fileNames []string
	cursor    int
	startTime time.Time
}

func NewMockRemoteScanner(startTime time.Time) RemoteScanner {
	scanner := &MockRemoteScanner{data: make([][]*Metric, 0), startTime: startTime}
	scanner.generateRuMetrics()
	scanner.generateStorageMetrics()
	scanner.write()
	return scanner
}

// for debug
func (m *MockRemoteScanner) write() {
	m.writeJson()
	m.writeParquet()
}

// TODO: clean files
func (m *MockRemoteScanner) writeParquet() {
	ctx := context.TODO()
	fileName := fmt.Sprintf("/Users/clark/coding/serverless-metering/tmp/data/debug-%d.db", m.startTime.Unix())
	connector, err := duckdb.NewConnector(fileName, nil)
	if err != nil {
		log.Fatal("failed to create duckdb connector", zap.Error(err))
	}

	conn, err := connector.Connect(ctx)
	if err != nil {
		log.Fatal("failed to connect to duckdb", zap.Error(err))
	}
	db := sql.OpenDB(connector)
	tableName := "test"
	_, err = db.ExecContext(ctx, fmt.Sprintf(createMetricTableSQL, tableName))
	if err != nil {
		log.Fatal("failed to create usage table", zap.Error(err))
	}

	appender, err := duckdb.NewAppenderFromConn(conn, "", tableName)
	if err != nil {
		log.Fatal("failed to create metric appender", zap.Error(err))
	}
	defer appender.Close()

	for _, data := range m.data {
		for _, metric := range data {
			labels, err := json.Marshal(metric.Labels)
			if err != nil {
				log.Fatal("marshal labels failed", zap.Error(err))
			}
			err = appender.AppendRow(metric.Name, metric.Type, labels, metric.Timestamp, metric.Value)
			if err != nil {
				log.Fatal("append row failed", zap.Error(err))
			}
		}
	}
	err = appender.Flush()
	if err != nil {
		log.Fatal("append row failed", zap.Error(err))
	}

	parquetFileName := fmt.Sprintf("/Users/clark/coding/serverless-metering/tmp/data/debug-%d.parquet", m.startTime.Unix())
	_, err = db.Exec(fmt.Sprintf("COPY %s TO '%s' (FORMAT PARQUET)", tableName, parquetFileName))
	if err != nil {
		log.Fatal("failed to export table to parquet", zap.Error(err))
	}

	m.fileNames = append(m.fileNames, parquetFileName)
}

func (m *MockRemoteScanner) writeJson() {
	f, err := os.Create("/Users/clark/coding/serverless-metering/tmp/data/debug.json")
	if err != nil {
		panic(err)
	}
	for _, metrics := range m.data {
		for _, metric := range metrics {
			b, _ := json.Marshal(metric)
			_, _ = f.Write(b)
			_, _ = f.WriteString("\n")
		}
	}
}

func (m *MockRemoteScanner) ScanMetrics(ctx context.Context) (metrics []*Metric, more bool, err error) {
	if m.cursor >= len(m.data) {
		return nil, false, nil
	}
	metrics = m.data[m.cursor]
	more = true
	err = nil
	m.cursor++
	return
}

func (m *MockRemoteScanner) ScanRemoteFiles(ctx context.Context) (rf *RemoteFile, more bool, err error) {
	if m.cursor >= len(m.fileNames) {
		return nil, false, nil
	}
	fileName := m.fileNames[m.cursor]
	file, err := os.Open(fileName)
	if err != nil {
		return nil, false, err
	}
	m.cursor++

	return &RemoteFile{Name: strings.ReplaceAll(fileName, "/", "-"), Body: file}, true, nil
}

func (m *MockRemoteScanner) appendCounter(reset bool, name string, labels map[string]string) {
	metrics := make([]*Metric, 0, 100)
	for i := 0; i < 100; i++ {
		ts := m.startTime.Add(time.Minute * time.Duration(i))
		value := int64(i * 100)
		// if reset, the counter maybe start from 0 in the middle.
		if reset && i >= 50 {
			value = int64((i - 50) * 100)
		}
		metrics = append(metrics, &Metric{
			Timestamp: ts.Unix(),
			Name:      name,
			Type:      int32(dto.MetricType_COUNTER),
			Value:     float64(value),
			Labels:    labels,
		})
	}
	m.data = append(m.data, metrics)
}

func (m *MockRemoteScanner) generateRuMetrics() {
	// load data ru
	m.appendCounter(false, "tikv_worker_load_data_wru_cost_counter", map[string]string{"keyspace_id": "1", "pod": "ins-1", "task_id": "1-ins-1"})
	m.appendCounter(false, "tikv_worker_load_data_wru_cost_counter", map[string]string{"keyspace_id": "1", "pod": "ins-2", "task_id": "1-ins-2"})
	m.appendCounter(false, "tikv_worker_load_data_wru_cost_counter", map[string]string{"keyspace_id": "2", "task_id": "2-ins-1"})
	m.appendCounter(true, "tikv_worker_load_data_wru_cost_counter", map[string]string{"keyspace_id": "3", "task_id": "3-ins-1"})
	// traffic
	m.appendCounter(false, "gateway_connection_flow_bytes_total", map[string]string{"direction": "out", "cluster_id": "1001", "pod": "ins-1", "mode": "public"})
	m.appendCounter(false, "gateway_connection_flow_bytes_total", map[string]string{"direction": "out", "cluster_id": "1001", "pod": "ins-2", "mode": "public"})
	m.appendCounter(false, "gateway_connection_flow_bytes_total", map[string]string{"direction": "out", "cluster_id": "1001", "pod": "ins-3", "mode": "public"})
	m.appendCounter(true, "gateway_connection_flow_bytes_total", map[string]string{"direction": "out", "cluster_id": "1002", "pod": "ins-3", "mode": "public"})
	m.appendCounter(false, "gateway_connection_flow_bytes_total", map[string]string{"direction": "out", "cluster_id": "1001", "pod": "ins-1", "mode": "privatelink"})
	m.appendCounter(false, "gateway_connection_flow_bytes_total", map[string]string{"direction": "out", "cluster_id": "1001", "pod": "ins-2", "mode": "privatelink"})
	m.appendCounter(false, "gateway_connection_flow_bytes_total", map[string]string{"direction": "out", "cluster_id": "1001", "pod": "ins-3", "mode": "privatelink"})
	m.appendCounter(true, "gateway_connection_flow_bytes_total", map[string]string{"direction": "out", "cluster_id": "1002", "pod": "ins-3", "mode": "privatelink"})
	// pd
	m.appendCounter(false, "resource_manager_resource_unit_read_request_unit_sum", map[string]string{"name": "1"})
	m.appendCounter(false, "resource_manager_resource_unit_read_request_unit_sum", map[string]string{"type": "tp", "name": "1"})
	m.appendCounter(false, "resource_manager_resource_unit_write_request_unit_sum", map[string]string{"name": "1"})
	m.appendCounter(false, "resource_manager_resource_unit_write_request_unit_sum", map[string]string{"type": "tp", "name": "1"})
	m.appendCounter(false, "resource_manager_resource_unit_sql_layer_request_unit", map[string]string{"name": "1"})
	m.appendCounter(false, "resource_manager_resource_unit_sql_layer_request_unit", map[string]string{"type": "tp", "name": "1"})
	m.appendCounter(false, "resource_manager_resource_unit_sql_layer_request_unit_sum", map[string]string{"name": "1"})
	m.appendCounter(false, "resource_manager_resource_unit_sql_layer_request_unit_sum", map[string]string{"type": "tp", "name": "1"})
	m.appendCounter(false, "resource_manager_resource_unit_read_request_unit_sum", map[string]string{"type": "ap", "name": "1"})
	m.appendCounter(false, "resource_manager_resource_unit_write_request_unit_sum", map[string]string{"type": "ap", "name": "1", "pod": "ins-1"})
	m.appendCounter(false, "resource_manager_resource_unit_write_request_unit_sum", map[string]string{"type": "ap", "name": "1", "pod": "ins-2"})
	m.appendCounter(false, "resource_manager_resource_unit_request_unit_v2_sum", map[string]string{"type": "tp", "name": "1"})
	m.appendCounter(false, "resource_manager_resource_unit_request_unit_v2_sum", map[string]string{"type": "ap", "name": "1"})
	// executor statement total
	m.appendCounter(false, "tidb_executor_statement_total", map[string]string{"db": "test_db", "cluster_id": "1001"})
	m.appendCounter(false, "tidb_executor_statement_total", map[string]string{"db": "another_db", "cluster_id": "1001"})
	m.appendCounter(true, "tidb_executor_statement_total", map[string]string{"db": "test_db", "cluster_id": "1002"})
	m.appendCounter(false, "tidb_executor_statement_total", map[string]string{"db": "prod_db", "cluster_id": "1003"})
}

func (m *MockRemoteScanner) generateStorageMetrics() {
	start := m.startTime.Truncate(time.Hour)
	fn := func(step int, labels map[string]string) {
		metrics := make([]*Metric, 0, 10)
		for i := 0; i < 60; i++ {
			value := float64((i % 10) * step * 1024 * 1024)
			ts := start.Add(time.Minute * time.Duration(i))
			metrics = append(metrics, &Metric{
				Timestamp: ts.Unix(),
				Name:      "tikv_store_size_bytes",
				Type:      int32(dto.MetricType_GAUGE),
				Value:     value,
				Labels:    labels,
			})
		}
		m.data = append(m.data, metrics)
	}
	fn(10, map[string]string{"keyspace_id": "1", "region_id": "1", "type": "used"})
	fn(20, map[string]string{"keyspace_id": "1", "region_id": "2", "type": "used"})
	fn(30, map[string]string{"keyspace_id": "2", "region_id": "3", "type": "used", "pod": "ins1"})
	fn(40, map[string]string{"keyspace_id": "2", "region_id": "3", "type": "used", "pod": "ins2"})
	fn(10, map[string]string{"keyspace_id": "1", "region_id": "1", "type": "tiflash_used"})
	fn(20, map[string]string{"keyspace_id": "1", "region_id": "2", "type": "tiflash_used"})
	fn(1000, map[string]string{"keyspace_id": "3", "region_id": "2", "type": "used"})
	fn(1000, map[string]string{"keyspace_id": "3", "region_id": "2", "type": "tiflash_used"})
}

type MockStorage struct{}

func NewMockStorage() Storage {
	return &MockStorage{}
}

func (m *MockStorage) Log(labels map[string]string, metricFamilies map[string]*dto.MetricFamily) error {
	return nil
}

func (m *MockStorage) BuildRemoteScanner(ctx context.Context, ts int64) (scanner RemoteScanner, err error) {
	startTime := time.Unix(ts, 0)
	return NewMockRemoteScanner(startTime), nil
}

func (m *MockStorage) PutCheckPoint(ctx context.Context, name string, checkpoint *CheckPoint) error {
	return nil
}

func (m *MockStorage) GetCheckPoint(ctx context.Context, name string) (*CheckPoint, error) {
	return &CheckPoint{}, nil
}

func (m *MockStorage) PutIntermediateResult(ctx context.Context, ts int64, digest string, data []byte) error {
	return nil
}

func (m *MockStorage) GetIntermediateResult(ctx context.Context, ts int64, digest string) ([]byte, error) {
	return nil, nil
}

type MockRcuStore struct {
	mock.Mock
}

func NewMockRcuStore() *MockRcuStore {
	return &MockRcuStore{}
}

func (m *MockRcuStore) WriteMetrics(labels map[string]string, metricFamilies *dto.MetricFamily) error {
	m.Called(labels, metricFamilies)
	return nil
}

func (m *MockRcuStore) CalRcuUsages(keyspaceIds []uint32, startTs int64, endTs int64, useRUV2 bool) ([]*RcuUsage, error) {
	args := m.Called(keyspaceIds, startTs, endTs, useRUV2)
	return args.Get(0).([]*RcuUsage), args.Error(1)
}

func (m *MockRcuStore) WriteRcuCapacity(capacity *RcuCapacity) error {
	// m.Called(capacity)
	return nil
}

func (m *MockRcuStore) ListRcuCapacities() ([]*RcuCapacity, error) {
	args := m.Called()
	return args.Get(0).([]*RcuCapacity), args.Error(1)
}

func (m *MockRcuStore) DeleteRcuCapacity(keyspaceID uint32) error {
	// m.Called(keyspaceID)
	return nil
}