package store

import (
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"os"
	"path/filepath"
	"time"

	"github.com/hashicorp/raft"
	raftboltdb "github.com/hashicorp/raft-boltdb"
	"github.com/sjy-dv/scdb/scdb/pkg/log"
)

type Store struct {
	RaftDir  string
	RaftBind string
	raft     *raft.Raft
	fsm      *fsm
}

var (
	ErrNotLeader error = errors.New("not leader")
)

func NewStore(raftdir, raftbind string) (*Store, error) {
	fsm, err := NewFSM(raftdir)
	if err != nil {
		return nil, err
	}
	return &Store{
		fsm:      fsm,
		RaftDir:  raftdir,
		RaftBind: raftbind,
	}, nil
}

func (s *Store) Open(bootstrap bool, localID string) error {
	config := raft.DefaultConfig()
	config.LocalID = raft.ServerID(localID)
	config.SnapshotThreshold = 1024
	log.Info(s.RaftBind)

	fullAddress := fmt.Sprintf("127.0.0.1%s", s.RaftBind)
	addr, err := net.ResolveTCPAddr("tcp", fullAddress)
	if err != nil {
		fmt.Println(1)
		return err
	}
	log.Info(addr)

	transport, err := raft.NewTCPTransport(fullAddress, addr, 3, 10*time.Second, os.Stderr)
	if err != nil {
		fmt.Println(2)
		return err
	}

	ss, err := raft.NewFileSnapshotStore(s.RaftDir, 2, os.Stderr)
	if err != nil {
		fmt.Println(3)
		return err
	}

	// boltDB implement log store and stable store interface
	boltDB, err := raftboltdb.NewBoltStore(filepath.Join(s.RaftDir, "raft.db"))
	if err != nil {
		return err
	}

	// raft system
	r, err := raft.NewRaft(config, s.fsm, boltDB, boltDB, ss, transport)
	if err != nil {
		return err
	}
	s.raft = r

	if bootstrap {
		configuration := raft.Configuration{
			Servers: []raft.Server{
				{
					ID:      config.LocalID,
					Address: transport.LocalAddr(),
				},
			},
		}
		s.raft.BootstrapCluster(configuration)
	}
	return nil
}

func (s *Store) Get(key string) (string, error) {
	return s.fsm.Get(key)
}

func (s *Store) Set(key, value string) error {
	if s.raft.State() != raft.Leader {
		return ErrNotLeader
	}

	c := NewSetCommand(key, value)

	msg, err := json.Marshal(c)
	if err != nil {
		return err
	}

	f := s.raft.Apply(msg, 10*time.Second)

	return f.Error()
}

func (s *Store) Delete(key string) error {
	if s.raft.State() != raft.Leader {
		return ErrNotLeader
	}

	c := NewDeleteCommand(key)

	msg, err := json.Marshal(c)
	if err != nil {
		return err
	}

	f := s.raft.Apply(msg, 10*time.Second)

	return f.Error()
}

func (s *Store) Join(nodeID, addr string) error {
	log.Infof("received join request for remote node %s, addr %s", nodeID, addr)

	cf := s.raft.GetConfiguration()

	if err := cf.Error(); err != nil {
		log.Infof("failed to get raft configuration")
		return err
	}

	for _, server := range cf.Configuration().Servers {
		if server.ID == raft.ServerID(nodeID) {
			log.Infof("node %s already joined raft cluster", nodeID)
			return nil
		}
	}

	f := s.raft.AddVoter(raft.ServerID(nodeID), raft.ServerAddress(addr), 0, 0)
	if err := f.Error(); err != nil {
		return err
	}

	log.Infof("node %s at %s joined successfully", nodeID, addr)

	return nil
}

func (s *Store) Leave(nodeID string) error {
	log.Infof("received leave request for remote node %s", nodeID)

	cf := s.raft.GetConfiguration()

	if err := cf.Error(); err != nil {
		log.Infof("failed to get raft configuration")
		return err
	}

	for _, server := range cf.Configuration().Servers {
		if server.ID == raft.ServerID(nodeID) {
			f := s.raft.RemoveServer(server.ID, 0, 0)
			if err := f.Error(); err != nil {
				log.Infof("failed to remove server %s", nodeID)
				return err
			}

			log.Infof("node %s leaved successfully", nodeID)
			return nil
		}
	}

	log.Infof("node %s not exists in raft group", nodeID)

	return nil
}

func (s *Store) Snapshot() error {
	log.Info("doing snapshot mannually")
	f := s.raft.Snapshot()
	return f.Error()
}
