// Open loads an existing store at `brigPath/$ID/index.bolt`, if it does not
// exist, it is created.  For full function, Connect() should be called
// afterwards.
func Open(brigPath string, owner id.Peer, IPFS *ipfsutil.Node) (*Store, error) {
	dbDir := filepath.Join(
		brigPath,
		"bolt."+strings.Replace(string(owner.ID()), "/", "-", -1),
	)

	kv, err := NewBoltKV(dbDir)
	if err != nil {
		return nil, err
	}

	fs := NewFilesystem(kv)

	st := &Store{
		repoPath: brigPath,
		IPFS:     IPFS,
		fs:       fs,
		kv:       kv,
	}

	// This version does not attempt any version checking:
	if err := fs.MetadataPut("version", []byte("1")); err != nil {
		return nil, err
	}

	if err := st.setStoreOwner(owner); err != nil {
		return nil, err
	}

	return st, err
}

// LastSeen returns the time when we've last seen `peer`
func (cp *conversationPool) LastSeen(peer id.Peer) time.Time {
	cp.mu.Lock()
	defer cp.mu.Unlock()

	if pinger := cp.heartbeat[peer.ID()]; pinger != nil {
		return pinger.LastSeen()
	}

	return time.Unix(0, 0)
}

func (st *Store) setStoreOwner(owner id.Peer) error {
	if err := st.fs.MetadataPut("id", []byte(owner.ID())); err != nil {
		return err
	}

	if err := st.fs.MetadataPut("hash", []byte(owner.Hash())); err != nil {
		return err
	}

	return nil
}

// NewConversation returns a conversation that exchanges data over `conn`.
func NewConversation(conn net.Conn, node *ipfsutil.Node, peer id.Peer) (*Conversation, error) {
	proto, err := wrapConnAsProto(conn, node, peer.Hash())
	if err != nil {
		return nil, err
	}

	cnv := &Conversation{
		conn:     conn,
		node:     node,
		peer:     peer,
		proto:    proto,
		notifees: make(map[int64]transfer.AsyncFunc),
	}

	// Cater responses:
	go func() {
		for {
			resp := wire.Response{}
			err := cnv.proto.Recv(&resp)

			if isEOFError(err) {
				break
			}

			if err != nil {
				log.Warningf("Error while receiving data: %v", err)
				continue
			}

			respID := resp.ID

			cnv.Lock()
			fn, ok := cnv.notifees[respID]
			if !ok {
				log.Warningf("No such id: %v", respID)
				cnv.Unlock()
				continue
			}

			// Remove the callback
			delete(cnv.notifees, respID)
			cnv.Unlock()

			fn(&resp)
		}
	}()

	return cnv, nil
}

// Forget removes a peer from the pool and cleans up after it.
func (cp *conversationPool) Forget(peer id.Peer) error {
	cp.mu.Lock()
	defer cp.mu.Unlock()

	peerID := peer.ID()
	cnv, ok := cp.open[peerID]
	if !ok {
		return repo.ErrNoSuchRemote(peerID)
	}

	if pinger, ok := cp.heartbeat[peerID]; ok {
		pinger.Close()
	}

	delete(cp.open, peerID)
	delete(cp.heartbeat, peerID)
	return cnv.Close()
}

// Dial returns an APIClient that is connected to `peer`.
// If there's already a conversation to that peer no new
// connection will be created.
//
// It will return ErrOffline if the connector is not connected.
//
// Note: All connections will be multiplexed over ipfs' swarm.
func (cn *Connector) Dial(peer id.Peer) (*APIClient, error) {
	if !cn.IsInOnlineMode() {
		return nil, ErrOffline
	}

	// Check if we're allowed to connect to that node:
	if _, err := cn.rp.Remotes.Get(peer.ID()); err != nil {
		return nil, err
	}

	cnv, err := cn.layer.Dial(peer)
	if err != nil {
		return nil, err
	}

	if err := cn.cp.Remember(peer, cnv); err != nil {
		return nil, err
	}

	return newAPIClient(cnv, cn.rp.IPFS)
}

func (cp *conversationPool) rememberUnlocked(peer id.Peer, cnv Conversation) error {
	cp.open[peer.ID()] = cnv

	// Create a new pinger if not already done:
	if _, ok := cp.heartbeat[peer.ID()]; !ok {
		pinger, err := cp.rp.IPFS.Ping(peer.Hash())
		if err != nil {
			return err
		}
		cp.heartbeat[peer.ID()] = pinger
	}

	return nil
}

func NewRemoteFromPeer(peer id.Peer) Remote {
	return NewRemote(peer.ID(), peer.Hash())
}