func (s *Store) GetPartition(topic string, id int32) (uint64, *structs.Partition, error) {
	sp := s.tracer.StartSpan("store: get partition")
	sp.LogKV("id", id)
	sp.SetTag("node id", s.nodeID)
	defer sp.Finish()
	tx := s.db.Txn(false)
	defer tx.Abort()
	idx := maxIndexTxn(tx, "partitions")
	partition, err := tx.First("partitions", "id", topic, id)
	if err != nil {
		return 0, nil, fmt.Errorf("partition lookup failed: %s", err)
	}
	if partition != nil {
		return idx, partition.(*structs.Partition), nil
	}
	return idx, nil, nil
}
GetPartition is used to get partitions.
package fsm
import (
	"fmt"
	"io"
	"os"
	"strings"
	"sync"
	memdb "github.com/hashicorp/go-memdb"
	"github.com/hashicorp/raft"
	opentracing "github.com/opentracing/opentracing-go"
	"github.com/travisjeffery/jocko/jocko/structs"
	"github.com/travisjeffery/jocko/jocko/util"
	"github.com/travisjeffery/jocko/log"
	"github.com/ugorji/go/codec"
)
var (
	fsmVerboseLogs bool
)
type command func(buf []byte, index uint64) interface{}
type unboundCommand func(c *FSM, buf []byte, index uint64) interface{}
var commands map[structs.MessageType]unboundCommand
func registerCommand(msg structs.MessageType, fn unboundCommand) {
	if commands == nil {
		commands = make(map[structs.MessageType]unboundCommand)
	}
	if commands[msg] != nil {
		panic(fmt.Errorf("Message %d is already registered", msg))
	}
	commands[msg] = fn
}
type NodeID int32
type Tracer opentracing.Tracer
type FSM struct {
	apply     map[structs.MessageType]command
	stateLock sync.RWMutex
	state     *Store
	tracer    opentracing.Tracer
	nodeID    NodeID
}
func New(args ...interface{}) (*FSM, error) {
	var nodeID NodeID
	var tracer Tracer
	for _, arg := range args {
		switch a := arg.(type) {
		case NodeID:
			nodeID = a
		case Tracer:
			tracer = a
		}
	}
	store, err := NewStore(tracer, nodeID)
	if err != nil {
		return nil, err
	}
	fsm := &FSM{
		apply:  make(map[structs.MessageType]command),
		state:  store,
		tracer: tracer,
		nodeID: nodeID,
	}
	for msg, fn := range commands {
		thisFn := fn
		fsm.apply[msg] = func(buf []byte, index uint64) interface{} {
			return thisFn(fsm, buf, index)
		}
	}
	return fsm, nil
}
func (c *FSM) State() *Store {
	c.stateLock.RLock()
	defer c.stateLock.RUnlock()
	return c.state
}
func (c *FSM) Apply(l *raft.Log) interface{} {
	buf := l.Data
	msgType := structs.MessageType(buf[0])
	if fn := c.apply[msgType]; fn != nil {
		return fn(buf[1:], l.Index)
	}
	return nil
}
func (c *FSM) Restore(old io.ReadCloser) error {
	defer old.Close()
	newState, err := NewStore(c.tracer)
	if err != nil {
		return err
	}
	restore := newState.Restore()
	defer restore.Abort()
	dec := codec.NewDecoder(old, msgpackHandle)
	var header snapshotHeader
	if err := dec.Decode(&header); err != nil {
		return err
	}
	msgType := make([]byte, 1)
	for {
		_, err := old.Read(msgType)
		if err == io.EOF {
			break
		} else if err != nil {
			return err
		}
		msg := structs.MessageType(msgType[0])
		if fn := restorers[msg]; fn != nil {
			if err := fn(&header, restore, dec); err != nil {
				return err
			}
		} else {
			return fmt.Errorf("unrecognized msg type %d", msg)
		}
	}
	restore.Commit()
	c.stateLock.Lock()
	oldState := c.state
	c.state = newState
	c.stateLock.Unlock()
	oldState.Abandon()
	return nil
}
func (c *FSM) Snapshot() (raft.FSMSnapshot, error) {
	return &snapshot{c.state.Snapshot()}, nil
}
type schemaFn func() *memdb.TableSchema
var schemas []schemaFn
func registerSchema(fn schemaFn) {
	schemas = append(schemas, fn)
}
type Store struct {
	schema *memdb.DBSchema
	db     *memdb.MemDB
	abandonCh chan struct{}
	tracer    opentracing.Tracer
	nodeID    NodeID
}
func NewStore(args ...interface{}) (*Store, error) {
	dbSchema := &memdb.DBSchema{
		Tables: make(map[string]*memdb.TableSchema),
	}
	for _, fn := range schemas {
		schema := fn()
		if _, ok := dbSchema.Tables[schema.Name]; ok {
			panic(fmt.Sprintf("duplicate table name: %s", schema.Name))
		}
		dbSchema.Tables[schema.Name] = schema
	}
	db, err := memdb.NewMemDB(dbSchema)
	if err != nil {
		return nil, err
	}
	s := &Store{
		schema:    dbSchema,
		db:        db,
		abandonCh: make(chan struct{}),
	}
	for _, arg := range args {
		switch a := arg.(type) {
		case NodeID:
			s.nodeID = a
		case Tracer:
			s.tracer = a
		}
	}
	return s, nil
}
func (s *Store) Abandon() {
	close(s.abandonCh)
}
func (s *Store) AbandonCh() <-chan struct{} {
	return s.abandonCh
}
func (s *Store) GetNode(id int32) (uint64, *structs.Node, error) {
	sp := s.tracer.StartSpan("store: get node")
	sp.LogKV("id", id)
	sp.SetTag("node id", s.nodeID)
	defer sp.Finish()
	tx := s.db.Txn(false)
	defer tx.Abort()
	idx := maxIndexTxn(tx, "nodes")
	node, err := tx.First("nodes", "id", id)
	if err != nil {
		return 0, nil, fmt.Errorf("node lookup failed: %s", err)
	}
	if node != nil {
		return idx, node.(*structs.Node), nil
	}
	return idx, nil, nil
}
func (s *Store) GetNodes() (uint64, []*structs.Node, error) {
	sp := s.tracer.StartSpan("store: get nodes")
	defer sp.Finish()
	tx := s.db.Txn(false)
	defer tx.Abort()
	idx := maxIndexTxn(tx, "nodes")
	it, err := tx.Get("nodes", "id")
	if err != nil {
		return 0, nil, fmt.Errorf("node lookup failed: %s", err)
	}
	var nodes []*structs.Node
	for next := it.Next(); next != nil; next = it.Next() {
		nodes = append(nodes, next.(*structs.Node))
	}
	return idx, nodes, nil
}
func (s *Store) EnsureNode(idx uint64, node *structs.Node) error {
	sp := s.tracer.StartSpan("store: ensure node")
	s.vlog(sp, "node", node)
	sp.SetTag("node id", s.nodeID)
	defer sp.Finish()
	tx := s.db.Txn(true)
	defer tx.Abort()
	if err := s.ensureNodeTxn(tx, idx, node); err != nil {
		return err
	}
	tx.Commit()
	return nil
}
func (s *Store) DeleteNode(idx uint64, id int32) error {
	sp := s.tracer.StartSpan("store: delete node")
	sp.LogKV("id", id)
	sp.SetTag("node id", s.nodeID)
	defer sp.Finish()
	tx := s.db.Txn(true)
	defer tx.Abort()
	if err := s.deleteNodeTxn(tx, idx, id); err != nil {
		return err
	}
	tx.Commit()
	return nil
}
func (s *Store) deleteNodeTxn(tx *memdb.Txn, idx uint64, id int32) error {
	node, err := tx.First("nodes", "id", id)
	if err != nil {
		log.Error.Printf("fsm: node lookup error: %s", err)
		return err
	}
	if node == nil {
		return nil
	}
	if err := tx.Delete("nodes", node); err != nil {
		log.Error.Printf("fsm: deleting node error: %s", err)
		return err
	}
	if err := tx.Insert("index", &IndexEntry{"nodes", idx}); err != nil {
		log.Error.Printf("fsm: updating index error: %s", err)
		return err
	}
	return nil
}
func (s *Store) EnsureRegistration(idx uint64, req *structs.RegisterNodeRequest) error {
	sp := s.tracer.StartSpan("store: ensure registration")
	s.vlog(sp, "req", req)
	sp.SetTag("node id", s.nodeID)
	defer sp.Finish()
	tx := s.db.Txn(true)
	defer tx.Abort()
	if err := s.ensureRegistration(tx, idx, req); err != nil {
		return err
	}
	tx.Commit()
	return nil
}
func (s *Store) ensureRegistration(tx *memdb.Txn, idx uint64, req *structs.RegisterNodeRequest) error {
	existing, err := tx.First("nodes", "id", req.Node.Node)
	if err != nil {
		log.Error.Printf("fsm: node lookup error: %s", err)
		return err
	}
	if existing == nil {
		if err := s.ensureNodeTxn(tx, idx, &req.Node); err != nil {
			return err
		}
	}
	return nil
}
func (s *Store) ensureNodeTxn(tx *memdb.Txn, idx uint64, node *structs.Node) error {
	var n *structs.Node
	existing, err := tx.First("nodes", "id", node.Node)
	if err != nil {
		return fmt.Errorf("node lookup failed: %s", err)
	}
	if existing != nil {
		n = existing.(*structs.Node)
	}
	if n != nil {
		node.CreateIndex = n.CreateIndex
		node.ModifyIndex = idx
	} else {
		node.CreateIndex = idx
		node.ModifyIndex = idx
	}
	if err := tx.Insert("nodes", node); err != nil {
		return fmt.Errorf("failed inserting node: %s", err)
	}
	if err := tx.Insert("index", &IndexEntry{"nodes", idx}); err != nil {
		return fmt.Errorf("failed updating index: %s", err)
	}
	return nil
}
func (s *Store) EnsureTopic(idx uint64, topic *structs.Topic) error {
	sp := s.tracer.StartSpan("store: ensure topic")
	s.vlog(sp, "topic", topic)
	sp.SetTag("node id", s.nodeID)
	if topic.Config == nil {
		topic.Config = structs.NewTopicConfig()
	}
	defer sp.Finish()
	tx := s.db.Txn(true)
	defer tx.Abort()
	if err := s.ensureTopicTxn(tx, idx, topic); err != nil {
		return err
	}
	tx.Commit()
	return nil
}
func (s *Store) ensureTopicTxn(tx *memdb.Txn, idx uint64, topic *structs.Topic) error {
	var t *structs.Topic
	existing, err := tx.First("topics", "id", topic.Topic)
	if err != nil {
		return fmt.Errorf("topic lookup failed: %s", err)
	}
	if existing != nil {
		t = existing.(*structs.Topic)
	}
	if t != nil {
		topic.CreateIndex = t.CreateIndex
		topic.ModifyIndex = idx
	} else {
		topic.CreateIndex = idx
		topic.ModifyIndex = idx
	}
	if err := tx.Insert("topics", topic); err != nil {
		return fmt.Errorf("failed inserting topic: %s", err)
	}
	if err := tx.Insert("index", &IndexEntry{"topics", idx}); err != nil {
		return fmt.Errorf("failed updating index: %s", err)
	}
	return nil
}
func (s *Store) GetTopic(id string) (uint64, *structs.Topic, error) {
	sp := s.tracer.StartSpan("store: get topic")
	sp.LogKV("id", id)
	sp.SetTag("node id", s.nodeID)
	defer sp.Finish()
	tx := s.db.Txn(false)
	defer tx.Abort()
	idx := maxIndexTxn(tx, "topics")
	topic, err := tx.First("topics", "id", id)
	if err != nil {
		return 0, nil, fmt.Errorf("topic lookup failed: %s", err)
	}
	if topic != nil {
		return idx, topic.(*structs.Topic), nil
	}
	return idx, nil, nil
}
func (s *Store) GetTopics() (uint64, []*structs.Topic, error) {
	sp := s.tracer.StartSpan("store: get topics")
	sp.SetTag("node id", s.nodeID)
	defer sp.Finish()
	tx := s.db.Txn(false)
	defer tx.Abort()
	idx := maxIndexTxn(tx, "topics")
	it, err := tx.Get("topics", "id")
	if err != nil {
		return 0, nil, err
	}
	var topics []*structs.Topic
	for next := it.Next(); next != nil; next = it.Next() {
		topics = append(topics, next.(*structs.Topic))
	}
	return idx, topics, nil
}
func (s *Store) DeleteTopic(idx uint64, id string) error {
	tx := s.db.Txn(true)
	defer tx.Abort()
	if err := s.deleteTopicTxn(tx, idx, id); err != nil {
		return err
	}
	tx.Commit()
	return nil
}
func (s *Store) deleteTopicTxn(tx *memdb.Txn, idx uint64, id string) error {
	topic, err := tx.First("topics", "id", id)
	if err != nil {
		log.Error.Printf("fsm: topic lookup error: %s", err)
		return err
	}
	if topic == nil {
		return nil
	}
	if err := tx.Delete("topics", topic); err != nil {
		log.Error.Printf("fsm: deleting topic error: %s", err)
		return err
	}
	if err := tx.Insert("index", &IndexEntry{"topics", idx}); err != nil {
		log.Error.Printf("fsm: updating index error: %s", err)
		return err
	}
	return nil
}
func (s *Store) EnsureGroup(idx uint64, group *structs.Group) error {
	sp := s.tracer.StartSpan("store: ensure group")
	s.vlog(sp, "group", group)
	sp.SetTag("node id", s.nodeID)
	defer sp.Finish()
	tx := s.db.Txn(true)
	defer tx.Abort()
	if err := s.ensureGroupTxn(tx, idx, group); err != nil {
		return err
	}
	tx.Commit()
	return nil
}
func (s *Store) ensureGroupTxn(tx *memdb.Txn, idx uint64, group *structs.Group) error {
	var t *structs.Group
	existing, err := tx.First("groups", "id", group.Group)
	if err != nil {
		return fmt.Errorf("group lookup failed: %s", err)
	}
	if existing != nil {
		t = existing.(*structs.Group)
	}
	if t != nil {
		group.CreateIndex = t.CreateIndex
		group.ModifyIndex = idx
	} else {
		group.CreateIndex = idx
		group.ModifyIndex = idx
	}
	if err := tx.Insert("groups", group); err != nil {
		return fmt.Errorf("failed inserting group: %s", err)
	}
	if err := tx.Insert("index", &IndexEntry{"groups", idx}); err != nil {
		return fmt.Errorf("failed updating index: %s", err)
	}
	return nil
}
func (s *Store) GetGroup(id string) (uint64, *structs.Group, error) {
	sp := s.tracer.StartSpan("store: get group")
	sp.LogKV("id", id)
	sp.SetTag("node id", s.nodeID)
	defer sp.Finish()
	tx := s.db.Txn(false)
	defer tx.Abort()
	idx := maxIndexTxn(tx, "groups")
	group, err := tx.First("groups", "id", id)
	if err != nil {
		return 0, nil, fmt.Errorf("group lookup failed: %s", err)
	}
	if group != nil {
		return idx, group.(*structs.Group), nil
	}
	return idx, nil, nil
}
func (s *Store) GetGroups() (uint64, []*structs.Group, error) {
	sp := s.tracer.StartSpan("store: get groups")
	defer sp.Finish()
	tx := s.db.Txn(false)
	defer tx.Abort()
	idx := maxIndexTxn(tx, "groups")
	it, err := tx.Get("groups", "id")
	if err != nil {
		return 0, nil, fmt.Errorf("group lookup failed: %s", err)
	}
	var groups []*structs.Group
	for next := it.Next(); next != nil; next = it.Next() {
		groups = append(groups, next.(*structs.Group))
	}
	return idx, groups, nil
}
func (s *Store) GetGroupsByCoordinator(coordinator int32) (uint64, []*structs.Group, error) {
	sp := s.tracer.StartSpan("store: get groups by coordinator")
	sp.LogKV("coordinator", coordinator)
	sp.SetTag("node id", s.nodeID)
	defer sp.Finish()
	tx := s.db.Txn(false)
	defer tx.Abort()
	idx := maxIndexTxn(tx, "groups")
	it, err := tx.Get("groups", "coordinator", coordinator)
	if err != nil {
		return 0, nil, fmt.Errorf("group lookup failed: %s", err)
	}
	var groups []*structs.Group
	for next := it.Next(); next != nil; next = it.Next() {
		groups = append(groups, next.(*structs.Group))
	}
	return idx, groups, nil
}
func (s *Store) DeleteGroup(idx uint64, group string) error {
	sp := s.tracer.StartSpan("store: delete group")
	sp.LogKV("group", group, "group")
	sp.SetTag("node id", s.nodeID)
	defer sp.Finish()
	tx := s.db.Txn(true)
	defer tx.Abort()
	if err := s.deleteGroupTxn(tx, idx, group); err != nil {
		return err
	}
	tx.Commit()
	return nil
}
func (s *Store) deleteGroupTxn(tx *memdb.Txn, idx uint64, id string) error {
	group, err := tx.First("groups", "id", id)
	if err != nil {
		log.Error.Printf("fsm: group lookup error: %s", err)
		return err
	}
	if group == nil {
		return nil
	}
	if err := tx.Delete("groups", group); err != nil {
		log.Error.Printf("fsm: deleting group error: %s", err)
		return err
	}
	if err := tx.Insert("index", &IndexEntry{"groups", idx}); err != nil {
		log.Error.Printf("fsm: updating index error: %s", err)
		return err
	}
	return nil
}
func (s *Store) EnsurePartition(idx uint64, partition *structs.Partition) error {
	sp := s.tracer.StartSpan("store: ensure partition")
	s.vlog(sp, "partition", partition)
	sp.SetTag("node id", s.nodeID)
	defer sp.Finish()
	tx := s.db.Txn(true)
	defer tx.Abort()
	if err := s.ensurePartitionTxn(tx, idx, partition); err != nil {
		return err
	}
	tx.Commit()
	return nil
}
func (s *Store) ensurePartitionTxn(tx *memdb.Txn, idx uint64, partition *structs.Partition) error {
	var t *structs.Partition
	existing, err := tx.First("partitions", "id", partition.Topic, partition.Partition)
	if err != nil {
		return fmt.Errorf("partition lookup failed: %s", err)
	}
	if existing != nil {
		t = existing.(*structs.Partition)
	}
	if t != nil {
		partition.CreateIndex = t.CreateIndex
		partition.ModifyIndex = idx
	} else {
		partition.CreateIndex = idx
		partition.ModifyIndex = idx
	}
	if err := tx.Insert("partitions", partition); err != nil {
		return fmt.Errorf("failed inserting partition: %s", err)
	}
	if err := tx.Insert("index", &IndexEntry{"partitions", idx}); err != nil {
		return fmt.Errorf("failed updating index: %s", err)
	}
	return nil
}