// Must be called while holding no locks and not in a transaction.
func deliver(msg *relay.UserMessage) {

	// If we have a connection to the recipient session,
	// deliver the message.
	sessionsLock.Lock()

	userConn, exists := sessions[msg.Recipient]
	if exists {
		// Try to deliver the message to this connection,
		// without blocking.
		// If its delivery channel is full, just drop it.
		select {
		case userConn.deliver <- msg:
			break
		default:
			break
		}
	}

	sessionsLock.Unlock()

	if exists {
		return
	}

	// Otherwise, decrement TTL. If it's 0, discard the message.
	msg.Ttl--
	if msg.Ttl == 0 {
		return
	}

	// Otherwise, look for another node which is connected to this session.
	store.StartTransaction()

	session := store.GetEntity(msg.Recipient)
	if session == nil || session.Value("kind") != "session" {
		// Unknown session; discard the message.
		store.EndTransaction()
		return
	}

	nodes := session.AllAttached()
	store.EndTransaction()

	if len(nodes) == 0 {
		// Shouldn't happen, sessions are transient,
		// should be deleted before reaching here.
		// Just discard the message.
		return
	}

	// Forward the message to a random one of those nodes.
	r := rand.Intn(len(nodes))
	relay.Forward(uint16(nodes[r]), msg)
}

func Forward(node uint16, userMsg *UserMessage) {
	store.StartTransaction()
	defer store.EndTransaction()

	// While degraded, we drop all messages.
	if store.Degraded() {
		return
	}

	// If we have no connection to that node, we drop the message.
	if len(connections[node]) == 0 {
		return
	}

	// Otherwise, send to the given node.
	var forward rproto.Forward
	forward.Sender = new(uint64)
	forward.Recipient = new(uint64)
	forward.Tag = new(string)
	forward.Content = new(string)
	forward.Ttl = new(uint32)
	*forward.Sender = userMsg.Sender
	*forward.Recipient = userMsg.Recipient
	*forward.Tag = userMsg.Tag
	*forward.Content = userMsg.Content
	*forward.Ttl = uint32(userMsg.Ttl)

	connections[node][0].SendProto(2, &forward)
}

func handleInstructionRequest(f *followConn, content []byte) {
	store.StartTransaction()
	defer store.EndTransaction()
	f.lock.Lock()
	defer f.lock.Unlock()

	if f.closed {
		return
	}

	var msg fproto.InstructionRequest
	if err := proto.Unmarshal(content, &msg); err != nil {
		f.Close()
		return
	}

	relativeSlot := int(*msg.Slot - store.InstructionStart())
	if relativeSlot < 0 {
		f.Close()
		return
	}
	instructions := store.InstructionSlots()
	if relativeSlot >= len(instructions) {
		f.Close()
		return
	}
	slot := instructions[relativeSlot]
	if len(slot) != 1 || !slot[0].IsChosen() {
		f.Close()
		return
	}

	// Convert the change request to our internal format.
	sendInstructionData(f, *msg.Slot, slot[0].ChangeRequest())
}

// Must be called from the processing goroutine.
func processAccept(node uint16, conn *connect.BaseConn, content []byte) {
	var msg coproto.Accept
	if err := proto.Unmarshal(content, &msg); err != nil {
		conn.Close()
		return
	}

	store.StartTransaction()
	defer store.EndTransaction()

	proposal, leader := store.Proposal()
	msgProposal, msgLeader := *msg.Proposal, node
	if proposal != msgProposal || leader != msgLeader {
		// Send a nack message and return,
		// if this accept relates to an earlier proposal.
		if store.CompareProposals(proposal, leader, msgProposal,
			msgLeader) {

			var nack coproto.Nack
			nack.Proposal = new(uint64)
			nack.Leader = new(uint32)
			*nack.Proposal = proposal
			*nack.Leader = uint32(leader)
			conn.SendProto(6, &nack)

			return
		}

		store.SetProposal(msgProposal, msgLeader)
	}

	addAccept(node, &msg)
}

// Must be called from the processing goroutine.
func processNack(node uint16, conn *connect.BaseConn, content []byte) {
	var msg coproto.Nack
	if err := proto.Unmarshal(content, &msg); err != nil {
		conn.Close()
		return
	}

	store.StartTransaction()
	defer store.EndTransaction()

	// If we don't consider ourselves the leader, discard.
	if !amLeader {
		return
	}

	msgProposal, msgLeader := *msg.Proposal, uint16(*msg.Leader)
	proposal, leader := store.Proposal()
	if msgProposal == proposal && msgLeader == leader {
		return
	}
	if store.CompareProposals(msgProposal, msgLeader, proposal, leader) {
		stopBeingLeader()
		store.SetProposal(msgProposal, msgLeader)
	}
}

// Can only be called from the handling goroutine for conn.
func handleStopFollowingUser(conn *userConn, content []byte) {
	var msg cliproto_up.StopFollowingUser
	if err := proto.Unmarshal(content, &msg); err != nil {
		conn.conn.Close()
		return
	}

	// Start transaction.
	store.StartTransaction()
	defer store.EndTransaction()
	sessionsLock.Lock()
	defer sessionsLock.Unlock()

	// Authentication check.
	if conn.session == 0 {
		conn.conn.Close()
		return
	}

	// If the ID exists in our following list, remove it.
	for i, existing := range conn.following {
		if existing == *msg.UserId {
			conn.following = append(conn.following[:i],
				conn.following[i+1:]...)
		}
	}

	// Send "stopped following" message.
	sendStoppedFollowing(conn, *msg.UserId, "By Request")
}

func handleFirstUnapplied(f *followConn, content []byte) {

	// Client nodes ignore this message, and shouldn't send it.
	if !config.IsCore() || f.node > 0x2000 {
		return
	}

	store.StartTransaction()
	defer store.EndTransaction()
	f.lock.Lock()
	defer f.lock.Unlock()

	if f.closed {
		return
	}

	var msg fproto.FirstUnapplied
	if err := proto.Unmarshal(content, &msg); err != nil {
		f.Close()
		return
	}

	// Can't trigger callbacks from the store package to elsewhere,
	// safe to call while holding f's lock.
	store.SetNodeFirstUnapplied(f.node, *msg.FirstUnapplied)
}

// Must be called from the processing goroutine.
func processPromise(node uint16, conn *connect.BaseConn, content []byte) {
	var msg coproto.Promise
	if err := proto.Unmarshal(content, &msg); err != nil {
		conn.Close()
		return
	}

	store.StartTransaction()
	defer store.EndTransaction()

	if receivedPromises == nil {
		// Not attempting to become leader.
		log.Print("core/consensus: discarded promise, not becoming "+
			"leader, from ", node)
		return
	}

	proposal, leader := store.Proposal()
	if proposal != *msg.Proposal || leader != uint16(*msg.Leader) {
		log.Print("core/consensus: rejected promise for wrong "+
			"proposal number from ", node)
		return
	}
	if receivedPromises[node] != nil {
		// Protocol violation; shouldn't get duplicate promises.
		log.Print("core/consensus: PROTOCOL VIOLATION: received "+
			"duplicate promise from node ", node)
		return
	}

	log.Print("core/consensus: received promise from node ", node)
	addPromise(node, &msg)
}

// Must be called from the processing goroutine.
func processAccepted(node uint16, conn *connect.BaseConn, content []byte) {
	var msg coproto.Accepted
	if err := proto.Unmarshal(content, &msg); err != nil {
		conn.Close()
		return
	}

	store.StartTransaction()
	defer store.EndTransaction()

	addAccepted(node, &msg)
}

func incomingConn(node uint16, conn *connect.BaseConn) {

	store.StartTransaction()
	if store.Degraded() {
		conn.Close()
		store.EndTransaction()
		return
	}

	userConn := new(userConn)
	userConn.conn = conn
	userConn.deliver = make(chan *relay.UserMessage, 100)

	// Add to connections set.
	connectionsLock.Lock()
	connections[userConn] = true
	connectionsLock.Unlock()

	store.EndTransaction()

	go handleConn(userConn)
}

func handleBurstDone(f *followConn, content []byte) {
	store.StartTransaction()
	defer store.EndTransaction()
	f.lock.Lock()

	if f.closed {
		f.lock.Unlock()
		return
	}

	var msg fproto.BurstDone
	if err := proto.Unmarshal(content, &msg); err != nil {
		f.Close()
		f.lock.Unlock()
		return
	}

	if !f.receivingBurst {
		f.Close()
		f.lock.Unlock()
		return
	}
	f.receivingBurst = false

	wasWaiting := f.waiting
	f.waiting = nil

	// We need to unlock before we start mutating the store,
	// due to callbacks from the store package to elsewhere.
	f.lock.Unlock()

	chrequests := make([]store.ChangeRequest, len(wasWaiting))
	for i, data := range wasWaiting {
		req := data.Request
		chrequests[i].RequestEntity = *req.RequestEntity
		chrequests[i].RequestNode = uint16(*req.RequestNode)
		chrequests[i].RequestId = *req.RequestId
		chrequests[i].Changeset =
			make([]store.Change, len(req.Changeset))

		chset := chrequests[i].Changeset
		for j, ch := range req.Changeset {
			chset[j].TargetEntity = *ch.TargetEntity
			chset[j].Key = *ch.Key
			chset[j].Value = *ch.Value
		}
	}
	store.EndBurst(*msg.FirstUnapplied, chrequests)
}

func handleBursting(f *followConn, content []byte) {
	store.StartTransaction()
	defer store.EndTransaction()
	f.lock.Lock()

	if f.closed {
		f.lock.Unlock()
		return
	}

	f.receivingBurst = true

	f.lock.Unlock()

	store.Degrade()
}

func Startup() {
	store.StartTransaction()
	defer store.EndTransaction()
	sessionsLock.Lock()
	defer sessionsLock.Unlock()

	// If undegraded, check for attached users with no session,
	// or nameless users in the store.
	if !store.Degraded() {
		checkOrphanAttaches()
		checkNameless()
	}

	// Start accepting client protocol connections.
	go connect.Listen(connect.CLIENT_PROTOCOL, incomingConn)
}

func handleLeader(f *followConn, content []byte) {
	var msg fproto.Leader
	if err := proto.Unmarshal(content, &msg); err != nil {
		f.lock.Lock()
		f.Close()
		f.lock.Unlock()
		return
	}

	store.StartTransaction()
	defer store.EndTransaction()

	proposal, leader := store.Proposal()
	msgProposal, msgLeader := *msg.Proposal, uint16(*msg.Leader)
	if store.CompareProposals(msgProposal, msgLeader, proposal, leader) {
		store.SetProposal(msgProposal, msgLeader)
	}
}

func (f *followConn) firstUnappliedTimeout() {

	store.StartTransaction()
	defer store.EndTransaction()
	f.lock.Lock()
	defer f.lock.Unlock()

	if f.closed {
		return
	}

	msg := new(fproto.FirstUnapplied)
	msg.FirstUnapplied = new(uint64)
	*msg.FirstUnapplied = store.InstructionFirstUnapplied()
	f.conn.SendProto(10, msg)

	f.firstUnappliedTimer = time.AfterFunc(firstUnappliedTimerDuration,
		func() { f.firstUnappliedTimeout() })

}

func handleInstructionChosen(f *followConn, content []byte) {
	store.StartTransaction()
	defer store.EndTransaction()
	f.lock.Lock()
	defer f.lock.Unlock()

	var msg fproto.InstructionChosen
	if err := proto.Unmarshal(content, &msg); err != nil {
		f.Close()
		return
	}

	if f.closed {
		return
	}

	relativeSlot := int(*msg.Slot - store.InstructionStart())
	if relativeSlot < 0 {
		return
	}
	instructions := store.InstructionSlots()
	if relativeSlot < len(instructions) {
		slot := instructions[relativeSlot]
		if len(slot) == 1 && slot[0].IsChosen() {
			return
		}
	}

	if !config.IsCore() {
		// TODO: Should only do this on one follow connection.
		var intReq fproto.InstructionRequest
		intReq.Slot = msg.Slot
		f.conn.SendProto(8, &intReq)
	}

	timeout := config.ROUND_TRIP_TIMEOUT_PERIOD
	f.offerTimers[*msg.Slot] = time.AfterFunc(
		timeout, func() { f.offerTimeout(*msg.Slot, timeout) })
}

func namelessTimeout(id uint64) {
	store.StartTransaction()
	defer store.EndTransaction()

	// If the timeout has been removed, do nothing.
	if namelessRemoveTimeouts[id] == nil {
		return
	}

	// Remove nameless user.
	// TODO: Should make sure we only do this once.
	user := store.GetEntity(id)
	if user != nil {
		chset := make([]store.Change, 1)
		chset[0].TargetEntity = id
		chset[0].Key = "id"
		chset[0].Value = ""

		req := makeRequest(chset)
		go chrequest.Request(req)
	}
}

func handleEntityProperty(f *followConn, content []byte) {
	store.StartTransaction()
	defer store.EndTransaction()
	f.lock.Lock()
	defer f.lock.Unlock()

	if f.closed {
		return
	}

	var msg fproto.EntityProperty
	if err := proto.Unmarshal(content, &msg); err != nil {
		f.Close()
		return
	}

	if !f.receivingBurst {
		f.Close()
		return
	}

	store.BurstEntity(*msg.Entity, *msg.Key, *msg.Value)
}

func orphanTimeout(id uint64) {
	store.StartTransaction()
	defer store.EndTransaction()

	// If the timeout has been removed, do nothing.
	if namelessRemoveTimeouts[id] == nil {
		return
	}

	// Detach ourselves from the session.
	// TODO: Should make sure we only do this once.
	ourAttachStr := "attach " + strconv.FormatUint(uint64(config.Id()), 10)
	session := store.GetEntity(id)
	if session != nil {
		chset := make([]store.Change, 1)
		chset[0].TargetEntity = id
		chset[0].Key = ourAttachStr
		chset[0].Value = ""

		req := makeRequest(chset)
		go chrequest.Request(req)
	}
}

// Can only be called from the handling goroutine for conn.
func handleFollowUser(conn *userConn, content []byte) {
	var msg cliproto_up.FollowUser
	if err := proto.Unmarshal(content, &msg); err != nil {
		conn.conn.Close()
		return
	}

	// Start transaction.
	store.StartTransaction()
	defer store.EndTransaction()
	sessionsLock.Lock()
	defer sessionsLock.Unlock()

	// Authentication check.
	if conn.session == 0 {
		conn.conn.Close()
		return
	}

	// Check we're not already following this user.
	// If we are, discard the message.
	for _, existing := range conn.following {
		if existing == *msg.UserId {
			return
		}
	}

	// Check this ID is actually a user entity.
	otherUser := store.GetEntity(*msg.UserId)
	if otherUser == nil || otherUser.Value("kind") != "user" {
		sendFollowUserIdFail(conn, *msg.UserId, "No Such User")
		return
	}

	// Start following this user.
	followUser(conn, *msg.UserId)
}