package etcdutil

import (
	"context"
	"path"
	"time"

	"github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/mvcc/mvccpb"
	"github.com/gogo/protobuf/proto"
	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/wormhole/pkg/whpb"
)

var (
	errTxnFailed = errors.New("[etcd] failed to commit transaction")
)

// Client wraps all kv operations, keep it stateless.
type Client struct {
	Client *clientv3.Client
	path   string
	txn    func(cs ...clientv3.Cmp) clientv3.Txn
}

// NewClient retuns a Client with slow log txn
func NewClient(client *clientv3.Client, rootPath string) *Client {
	return &Client{
		Client: client,
		path:   rootPath,
		txn: func(cs ...clientv3.Cmp) clientv3.Txn {
			return NewSlowLogTxn(client)
		},
	}
}

// NewTxnClient returns a client with specify txn function
func NewTxnClient(client *clientv3.Client, path string, txn func(cs ...clientv3.Cmp) clientv3.Txn) *Client {
	return &Client{
		Client: client,
		path:   path,
		txn:    txn,
	}
}

// LoadProducerPartitions loads producer partitions and offsets information.
func (c *Client) LoadProducerPartitions(topic string) ([]*whpb.Partition, error) {
	partitions, err := c.loadPartitions(topic, "producer/latest")
	return partitions, errors.Trace(err)
}

// LoadConsumerPartitions loads consumer partitions and offsets information.
func (c *Client) LoadConsumerPartitions(topic string, jobID string) ([]*whpb.Partition, error) {
	partitions, err := c.loadPartitions(topic, jobID)
	return partitions, errors.Trace(err)
}

func (c *Client) loadPartitions(topic string, jobID string) ([]*whpb.Partition, error) {
	key := PartitionPathPrefix(c.path, topic, jobID)
	data := &whpb.Partitions{}
	ok, err := c.LoadProto(key, data)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if !ok {
		return nil, nil
	}

	return data.Partitions, nil
}

// SaveConsumerPartition saves partition information.
func (c *Client) SaveConsumerPartition(jobID string, topic string, partitions *whpb.Partitions, opts ...clientv3.OpOption) error {
	key := ConsumerPartitionPath(c.path, topic, jobID)
	return errors.Trace(c.SaveProto(key, partitions, opts...))
}

// SaveProducerPartition saves producer partition information.
func (c *Client) SaveProducerPartition(topic string, partitions *whpb.Partitions, opts ...clientv3.OpOption) error {
	key := ProducerPartitionPath(c.path, topic)
	return errors.Trace(c.SaveProto(key, partitions, opts...))
}

// SaveProducerPartitionSavepoint saves producer block partition.
func (c *Client) SaveProducerPartitionSavepoint(topic string, blockID string, partitions *whpb.Partitions, opts ...clientv3.OpOption) error {
	key := ProducerPartitionSavepointPath(c.path, topic, blockID)
	return errors.Trace(c.SaveProto(key, partitions, opts...))
}

// LoadSchemaDetail loads SchemDetail from etcd.
func (c *Client) LoadSchemaDetail(host string, port int32, schema string, table string, detail *whpb.SchemaDetail) (bool, error) {
	key := SchemaDetailPath(c.path, host, port, schema, table)
	gotValue, err := c.LoadProto(key, detail)
	return gotValue, errors.Trace(err)
}

// LoadSchemaDetails loads all schema detail under specific host and port.
func (c *Client) LoadSchemaDetails(host string, port int32, details map[string]map[string]*whpb.SchemaDetail) error {
	if details == nil {
		details = make(map[string]map[string]*whpb.SchemaDetail)
	}

	key := SchemaDetailPathPrefix(c.path, host, port)
	kvPairs, err := c.BatchLoadWithPrefix(key)
	if err != nil {
		return errors.Trace(err)
	}

	for _, kv := range kvPairs {
		detail := &whpb.SchemaDetail{}
		err = detail.Unmarshal(kv.Value)
		if err != nil {
			return errors.Trace(errors.Annotatef(err, "unmarshal error. key %s", string(kv.Key)))
		}
		_, ok := details[detail.Schema]
		if !ok {
			details[detail.Schema] = make(map[string]*whpb.SchemaDetail)
		}
		if detail.Table == "" {
			continue
		}
		details[detail.Schema][detail.Table] = detail
	}
	return nil
}

// SaveSchemaDetail saves SchemaDetail to etcd.
func (c *Client) SaveSchemaDetail(host string, port int32, schema string, table string, detail *whpb.SchemaDetail) error {
	key := SchemaDetailPath(c.path, host, port, schema, table)
	return errors.Trace(c.SaveProto(key, detail))
}

// RenameTable renames table.
func (c *Client) RenameTable(host string, port int32, oldSchema, oldTable, newSchema, newTable string, newDetail *whpb.SchemaDetail) error {
	// create new table
	newKey := SchemaDetailPath(c.path, host, port, newSchema, newTable)
	newValue, err := newDetail.Marshal()
	if err != nil {
		return errors.Trace(err)
	}
	createNewTableOpt := clientv3.OpPut(newKey, string(newValue))

	// delete old table
	oldKey := SchemaDetailPath(c.path, host, port, oldSchema, oldTable)
	deleteOldTableOpt := clientv3.OpDelete(oldKey)

	// put them in one transaction
	resp, err := c.txn().Then(createNewTableOpt, deleteOldTableOpt).Commit()
	if err != nil {
		return errors.Trace(err)
	}
	if !resp.Succeeded {
		return errors.Trace(errTxnFailed)
	}
	return nil
}

// DeleteSchemaDetail deletes schema details. It can delete the whole database or just a single table.
func (c *Client) DeleteSchemaDetail(host string, port int32, schema string, table string) error {
	var key string
	if table != "" {
		key = SchemaDetailPath(c.path, host, port, schema, table)
	} else {
		key = path.Join(SchemaDetailPathPrefix(c.path, host, port), "schema", schema)
	}
	return errors.Trace(c.deletePath(key))
}

func (c *Client) deletePath(key string) (err error) {
	_, err = c.txn().Then(clientv3.OpDelete(key)).Commit()
	if err != nil {
		return errors.Trace(err)
	}

	return nil
}

// SaveSchemaMeta saves schema meta to etcd.
func (c *Client) SaveSchemaMeta(host string, port int32, meta *whpb.SchemaInfo) error {
	key := SchemaMetaPath(c.path, host, port)
	return errors.Trace(c.SaveProto(key, meta))
}

// LoadSchemaMeta loads scheam meta from etcd.
func (c *Client) LoadSchemaMeta(host string, port int32) (bool, error) {
	key := SchemaMetaPath(c.path, host, port)
	meta := &whpb.SchemaInfo{}
	gotValue, err := c.LoadProto(key, meta)
	return gotValue, errors.Trace(err)
}

// BatchLoadWithPrefix loads key/value in one batch
func (c *Client) BatchLoadWithPrefix(key string) ([]*mvccpb.KeyValue, error) {
	resp, err := KVGet(c.Client, key, clientv3.WithPrefix())
	if err != nil {
		return nil, errors.Trace(err)
	}
	return resp.Kvs, nil
}

// BatchLoadWithOpts loads key/value with options
func (c *Client) BatchLoadWithOpts(key string, opts ...clientv3.OpOption) (*clientv3.GetResponse, error) {
	resp, err := KVGet(c.Client, key, opts...)
	return resp, errors.Trace(err)
}

// LoadProto loads proto message from ectd
func (c *Client) LoadProto(key string, msg proto.Message) (bool, error) {
	value, err := c.Load(key)
	if err != nil {
		return false, errors.Trace(err)
	}
	if value == nil {
		log.Warnf("[etcdutil] load proto. value is nil, key %s", key)
		return false, nil
	}
	return true, errors.Trace(proto.Unmarshal(value, msg))
}

// SaveProto saves proto into etcd
func (c *Client) SaveProto(key string, msg proto.Message, opts ...clientv3.OpOption) error {
	value, err := proto.Marshal(msg)
	if err != nil {
		return errors.Trace(err)
	}
	return errors.Trace(c.Save(key, string(value), opts...))
}

// Load loads single value
func (c *Client) Load(key string, opts ...clientv3.OpOption) ([]byte, error) {
	resp, err := KVGet(c.Client, key, opts...)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if n := len(resp.Kvs); n == 0 {
		return nil, nil
	} else if n > 1 {
		return nil, errors.Errorf("[etcd] load more than one kvs: key %v kvs %v", key, n)
	}
	return resp.Kvs[0].Value, nil
}

// Save saves key/value into etcd
func (c *Client) Save(key, value string, opts ...clientv3.OpOption) error {
	resp, err := c.txn().Then(clientv3.OpPut(key, value, opts...)).Commit()
	if err != nil {
		return errors.Trace(err)
	}
	if !resp.Succeeded {
		return errors.Trace(errTxnFailed)
	}
	return nil
}

// Delete deletes key from etcd
func (c *Client) Delete(key string) error {
	resp, err := c.txn().Then(clientv3.OpDelete(key)).Commit()
	if err != nil {
		return errors.Trace(err)
	}
	if !resp.Succeeded {
		return errors.Trace(errTxnFailed)
	}
	return nil
}

// GrantLease grants lease
func (c *Client) GrantLease(lease int64) (*clientv3.LeaseGrantResponse, error) {
	ctx, cancel := context.WithTimeout(c.Client.Ctx(), requestTimeout)
	defer cancel()
	lcr, err := c.Client.Grant(ctx, lease)
	return lcr, errors.Trace(err)
}

// KVGet gets key with options
func KVGet(c *clientv3.Client, key string, opts ...clientv3.OpOption) (*clientv3.GetResponse, error) {
	log.Debugf("[etcd] kv get key:%s", key)
	ctx, cancel := context.WithTimeout(c.Ctx(), requestTimeout)
	defer cancel()

	start := time.Now()
	resp, err := clientv3.NewKV(c).Get(ctx, key, opts...)
	if cost := time.Since(start); cost > slowRequestTime {
		log.Warnf("[etcd] kv gets too slow: key %v cost %v err %v", key, cost, err)
	}

	return resp, errors.Trace(err)
}