package repository

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"path/filepath"

	_ "github.com/marcboeker/go-duckdb"
	"github.com/pingcap/log"
	"github.com/tidbcloud/serverless-metering/pkg/config"
	"github.com/tidbcloud/serverless-metering/pkg/util"
	"go.uber.org/zap"
)

const (
	createClusterBranchTableSQL = `CREATE TABLE IF NOT EXISTS cluster_branch (
	primary_id varchar,
	keyspace_id varchar,
	data varchar,
	created_at TIMESTAMP,
	PRIMARY KEY (primary_id, keyspace_id)
)`

	getItemSQL = `SELECT data FROM cluster_branch WHERE %s = '%s' AND created_at > current_localtimestamp() - INTERVAL %d MINUTE`

	createItemSQL = `INSERT INTO cluster_branch (
  primary_id, keyspace_id, data, created_at
)
VALUES
  ('%s', '%d', '%s', current_localtimestamp()) ON CONFLICT DO
UPDATE
SET
  data = EXCLUDED.data,
  created_at = EXCLUDED.created_at`
)

type CacheClusterBranchRepo struct {
	cloudProvider     string
	ttlMinutes        int
	db                *sql.DB
	sources           []string
	dynamoClusterRepo *DynamoClusterRepo
	infraBranchRepo   *InfraBranchRepo
	pdMetaStoreRepo   *PDMetaStoreClient
}

func NewCacheClusterBranchRepo() (*CacheClusterBranchRepo, error) {
	cloudProvider := config.Global().CloudProvider
	cfg := config.Global().Repo
	err := util.EnsureDir(filepath.Dir(cfg.CacheFile))
	if err != nil {
		return nil, err
	}
	db, err := sql.Open("duckdb", cfg.CacheFile)
	if err != nil {
		return nil, fmt.Errorf("failed to open duckdb: %s", err)
	}
	_, err = db.Exec(createClusterBranchTableSQL)
	if err != nil {
		return nil, fmt.Errorf("failed to create table: %s", err)
	}
	infraBranchRepo, err := NewInfraBranchRepo()
	if err != nil {
		return nil, err
	}
	var dynamoClusterRepo *DynamoClusterRepo
	if cloudProvider == "aws" {
		dynamoClusterRepo, err = NewDynamoClusterRepo()
		if err != nil {
			return nil, err
		}
	}
	pdMetaStoreRepo := NewPDMetaStoreClient()
	sources := []string{"local", "pd", "infra"}
	if config.Global().EnableDDBMeta {
		if cloudProvider != "aws" {
			return nil, fmt.Errorf("enable-ddb-meta is true but cloud provider is %s, must be aws", cloudProvider)
		}
		sources = []string{"local", "ddb", "infra"}
	}
	return &CacheClusterBranchRepo{
		cloudProvider:     config.Global().CloudProvider,
		ttlMinutes:        cfg.CacheTTLMinutes,
		db:                db,
		sources:           sources,
		dynamoClusterRepo: dynamoClusterRepo,
		infraBranchRepo:   infraBranchRepo,
		pdMetaStoreRepo:   pdMetaStoreRepo,
	}, nil
}

func (r *CacheClusterBranchRepo) getItemByPrimaryId(ctx context.Context, primaryID string) (*ClusterBranch, error) {
	return r.getItemFromDB(ctx, "primary_id", primaryID)
}

func (r *CacheClusterBranchRepo) getItemByKeyspaceId(ctx context.Context, keyspaceID string) (*ClusterBranch, error) {
	return r.getItemFromDB(ctx, "keyspace_id", keyspaceID)
}

func (r *CacheClusterBranchRepo) getItemFromDB(ctx context.Context, key, value string) (*ClusterBranch, error) {
	row := r.db.QueryRowContext(ctx, fmt.Sprintf(getItemSQL, key, value, r.ttlMinutes))
	var data string
	if err := row.Scan(&data); err != nil {
		return nil, err
	}
	var clusterBranch ClusterBranch
	if err := json.Unmarshal([]byte(data), &clusterBranch); err != nil {
		return nil, err
	}
	return &clusterBranch, nil
}

func (r *CacheClusterBranchRepo) storeItemToDB(ctx context.Context, clusterBranch *ClusterBranch) error {
	data, err := json.Marshal(clusterBranch)
	if err != nil {
		return err
	}
	_, err = r.db.ExecContext(ctx, fmt.Sprintf(
		createItemSQL, clusterBranch.PrimaryID, clusterBranch.KeyspaceID, string(data)))
	if err != nil {
		return err
	}
	return nil
}

func (r *CacheClusterBranchRepo) doQuery(ctx context.Context, logger *zap.Logger, value string, fns map[string]func(context.Context, string) (*ClusterBranch, error)) (*ClusterBranch, error) {
	for _, source := range r.sources {
		fn := fns[source]
		clusterBranch, err := fn(ctx, value)
		if err != nil {
			if !util.IsNotFoundError(err) {
				return nil, err
			}
			logger.Info("cluster branch not found", zap.String("source", source))
			continue
		}
		if source != "local" {
			err = r.storeItemToDB(ctx, clusterBranch)
			if err != nil {
				logger.Warn("failed to store item", zap.Error(err))
			}
		}
		return clusterBranch, nil
	}

	return nil, fmt.Errorf("not found")
}

func (r *CacheClusterBranchRepo) GetClusterBranchByPrimaryID(ctx context.Context, primaryID string) (*ClusterBranch, error) {
	logger := log.L().With(zap.String("primaryID", primaryID))
	fns := map[string]func(context.Context, string) (*ClusterBranch, error){
		"local": r.getItemByPrimaryId,
		"ddb":   r.dynamoClusterRepo.GetClusterByID,
		"infra": r.infraBranchRepo.GetBranchByName,
		"pd":    r.pdMetaStoreRepo.GetClusterByID,
	}
	return r.doQuery(ctx, logger, primaryID, fns)
}

func (r *CacheClusterBranchRepo) GetClusterBranchByKeyspaceID(ctx context.Context, keyspaceID string) (*ClusterBranch, error) {
	logger := log.L().With(zap.String("keyspaceID", keyspaceID))
	fns := map[string]func(context.Context, string) (*ClusterBranch, error){
		"local": r.getItemByKeyspaceId,
		"ddb":   r.dynamoClusterRepo.GetClusterByKeyspaceID,
		"infra": r.infraBranchRepo.GetBranchByKeyspaceID,
		"pd":    r.pdMetaStoreRepo.GetClusterByKeyspaceID,
	}
	return r.doQuery(ctx, logger, keyspaceID, fns)
}