package cassandra2

import (
	"context"
	"errors"
	"fmt"
	"os"
	"strings"
	"time"

	gocql "github.com/apache/cassandra-gocql-driver/v2"

	"github.com/seaweedfs/seaweedfs/weed/filer"
	"github.com/seaweedfs/seaweedfs/weed/glog"
	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
	"github.com/seaweedfs/seaweedfs/weed/util"
)

func init() {
	filer.Stores = append(filer.Stores, &Cassandra2Store{})
}

type Cassandra2Store struct {
	cluster                 *gocql.ClusterConfig
	session                 *gocql.Session
	superLargeDirectoryHash map[string]string
}

func (store *Cassandra2Store) GetName() string {
	return "cassandra2"
}

func (store *Cassandra2Store) Initialize(configuration util.Configuration, prefix string) (err error) {
	enableHostVerification := true
	if val := configuration.GetString(prefix + "ssl_enable_host_verification"); val != "" {
		enableHostVerification = configuration.GetBool(prefix + "ssl_enable_host_verification")
	}

	return store.initialize(
		configuration.GetString(prefix+"keyspace"),
		configuration.GetStringSlice(prefix+"hosts"),
		configuration.GetString(prefix+"username"),
		configuration.GetString(prefix+"password"),
		configuration.GetString(prefix+"ssl_ca_path"),
		configuration.GetString(prefix+"ssl_cert_path"),
		configuration.GetString(prefix+"ssl_key_path"),
		configuration.GetStringSlice(prefix+"superLargeDirectories"),
		configuration.GetString(prefix+"localDC"),
		configuration.GetInt(prefix+"connection_timeout_millisecond"),
		enableHostVerification,
	)
}

func (store *Cassandra2Store) isSuperLargeDirectory(dir string) (dirHash string, isSuperLargeDirectory bool) {
	dirHash, isSuperLargeDirectory = store.superLargeDirectoryHash[dir]
	return
}

func (store *Cassandra2Store) initialize(keyspace string, hosts []string, username string, password string, sslCaPath string, sslCertPath string, sslKeyPath string, superLargeDirectories []string, localDC string, timeout int, enableHostVerification bool) (err error) {
	store.cluster = gocql.NewCluster(hosts...)
	if username != "" && password != "" {
		store.cluster.Authenticator = gocql.PasswordAuthenticator{Username: username, Password: password}
	}
	if sslCaPath != "" || sslCertPath != "" || sslKeyPath != "" {
		if (sslCertPath != "" && sslKeyPath == "") || (sslCertPath == "" && sslKeyPath != "") {
			return fmt.Errorf("both ssl_cert_path and ssl_key_path must be provided for mTLS, or neither")
		}

		for _, path := range []string{sslCaPath, sslCertPath, sslKeyPath} {
			if path != "" {
				if _, err := os.Stat(path); err != nil {
					return fmt.Errorf("ssl file %s not found: %v", path, err)
				}
			}
		}

		store.cluster.SslOpts = &gocql.SslOptions{
			CaPath:                 sslCaPath,
			CertPath:               sslCertPath,
			KeyPath:                sslKeyPath,
			EnableHostVerification: enableHostVerification,
		}

		// check if port is already specified in hosts
		hasPort := false
		for _, host := range hosts {
			if strings.Contains(host, ":") {
				hasPort = true
				break
			}
		}
		if !hasPort {
			// standard cassandra port is 9042, but AWS keyspaces uses 9142
			store.cluster.Port = 9142
		}

		if sslCertPath != "" {
			glog.V(0).Infof("TLS enabled: mTLS with cert %s", sslCertPath)
		} else {
			glog.V(0).Infof("TLS enabled: server-verification with ca %s", sslCaPath)
		}
	}
	store.cluster.Keyspace = keyspace
	store.cluster.Timeout = time.Duration(timeout) * time.Millisecond
	glog.V(0).Infof("timeout = %d", timeout)
	fallback := gocql.RoundRobinHostPolicy()
	if localDC != "" {
		fallback = gocql.DCAwareRoundRobinPolicy(localDC)
	}
	store.cluster.PoolConfig.HostSelectionPolicy = gocql.TokenAwareHostPolicy(fallback)
	store.cluster.Consistency = gocql.LocalQuorum

	store.session, err = store.cluster.CreateSession()
	if err != nil {
		glog.V(0).Infof("Failed to open cassandra2 store, hosts %v, keyspace %s", hosts, keyspace)
	}

	// set directory hash
	store.superLargeDirectoryHash = make(map[string]string)
	existingHash := make(map[string]string)
	for _, dir := range superLargeDirectories {
		// adding dir hash to avoid duplicated names
		dirHash := util.Md5String([]byte(dir))[:4]
		store.superLargeDirectoryHash[dir] = dirHash
		if existingDir, found := existingHash[dirHash]; found {
			glog.Fatalf("directory %s has the same hash as %s", dir, existingDir)
		}
		existingHash[dirHash] = dir
	}
	return
}

func (store *Cassandra2Store) BeginTransaction(ctx context.Context) (context.Context, error) {
	return ctx, nil
}
func (store *Cassandra2Store) CommitTransaction(ctx context.Context) error {
	return nil
}
func (store *Cassandra2Store) RollbackTransaction(ctx context.Context) error {
	return nil
}

func (store *Cassandra2Store) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) {

	dir, name := entry.FullPath.DirAndName()
	if dirHash, ok := store.isSuperLargeDirectory(dir); ok {
		dir, name = dirHash+name, ""
	}

	meta, err := entry.EncodeAttributesAndChunks()
	if err != nil {
		return fmt.Errorf("encode %s: %s", entry.FullPath, err)
	}

	if len(entry.GetChunks()) > filer.CountEntryChunksForGzip {
		meta = util.MaybeGzipData(meta)
	}

	if err := store.session.Query(
		"INSERT INTO filemeta (dirhash,directory,name,meta) VALUES(?,?,?,?) USING TTL ? ",
		util.HashStringToLong(dir), dir, name, meta, entry.TtlSec).Exec(); err != nil {
		return fmt.Errorf("insert %s: %s", entry.FullPath, err)
	}

	return nil
}

func (store *Cassandra2Store) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) {

	return store.InsertEntry(ctx, entry)
}

func (store *Cassandra2Store) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer.Entry, err error) {

	dir, name := fullpath.DirAndName()
	if dirHash, ok := store.isSuperLargeDirectory(dir); ok {
		dir, name = dirHash+name, ""
	}

	var data []byte
	if err := store.session.Query(
		"SELECT meta FROM filemeta WHERE dirhash=? AND directory=? AND name=?",
		util.HashStringToLong(dir), dir, name).Scan(&data); err != nil {
		if errors.Is(err, gocql.ErrNotFound) {
			return nil, filer_pb.ErrNotFound
		}
		return nil, err
	}

	entry = &filer.Entry{
		FullPath: fullpath,
	}
	err = entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data))
	if err != nil {
		return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
	}

	return entry, nil
}

func (store *Cassandra2Store) DeleteEntry(ctx context.Context, fullpath util.FullPath) error {

	dir, name := fullpath.DirAndName()
	if dirHash, ok := store.isSuperLargeDirectory(dir); ok {
		dir, name = dirHash+name, ""
	}

	if err := store.session.Query(
		"DELETE FROM filemeta WHERE dirhash=? AND directory=? AND name=?",
		util.HashStringToLong(dir), dir, name).Exec(); err != nil {
		return fmt.Errorf("delete %s : %v", fullpath, err)
	}

	return nil
}

func (store *Cassandra2Store) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) error {
	if _, ok := store.isSuperLargeDirectory(string(fullpath)); ok {
		return nil // filer.ErrUnsupportedSuperLargeDirectoryListing
	}

	if err := store.session.Query(
		"DELETE FROM filemeta WHERE dirhash=? AND directory=?",
		util.HashStringToLong(string(fullpath)), fullpath).Exec(); err != nil {
		return fmt.Errorf("delete %s : %v", fullpath, err)
	}

	return nil
}

func (store *Cassandra2Store) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
	return lastFileName, filer.ErrUnsupportedListDirectoryPrefixed
}

func (store *Cassandra2Store) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {

	if _, ok := store.isSuperLargeDirectory(string(dirPath)); ok {
		return // nil, filer.ErrUnsupportedSuperLargeDirectoryListing
	}

	cqlStr := "SELECT NAME, meta FROM filemeta WHERE dirhash=? AND directory=? AND name>? ORDER BY NAME ASC LIMIT ?"
	if includeStartFile {
		cqlStr = "SELECT NAME, meta FROM filemeta WHERE dirhash=? AND directory=? AND name>=? ORDER BY NAME ASC LIMIT ?"
	}

	var data []byte
	var name string
	iter := store.session.Query(cqlStr, util.HashStringToLong(string(dirPath)), string(dirPath), startFileName, limit+1).Iter()
	for iter.Scan(&name, &data) {
		entry := &filer.Entry{
			FullPath: util.NewFullPath(string(dirPath), name),
		}
		lastFileName = name
		if decodeErr := entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data)); decodeErr != nil {
			err = decodeErr
			glog.V(0).InfofCtx(ctx, "list %s : %v", entry.FullPath, err)
			break
		}

		resEachEntryFunc, resEachEntryFuncErr := eachEntryFunc(entry)
		if resEachEntryFuncErr != nil {
			err = fmt.Errorf("failed to process eachEntryFunc for entry %q: %w", entry.FullPath, resEachEntryFuncErr)
			glog.V(0).InfofCtx(ctx, "failed to process eachEntryFunc for entry %q: %v", entry.FullPath, resEachEntryFuncErr)
			break
		}

		if !resEachEntryFunc {
			break
		}
	}

	if errClose := iter.Close(); errClose != nil {
		glog.V(0).InfofCtx(ctx, "list iterator close: %v", errClose)
		if err == nil {
			return lastFileName, errClose
		}
	}

	return lastFileName, err
}

func (store *Cassandra2Store) Shutdown() {
	store.session.Close()
}