// writeMessage sends a bitcoin Message to the peer with logging.
func (p *peer) writeMessage(msg btcwire.Message) {
	// Don't do anything if we're disconnecting.
	if atomic.LoadInt32(&p.disconnect) != 0 {
		return
	}

	log.Debugf("[PEER] Sending command [%v] to %s", msg.Command(),
		p.addr)

	// Use closures to log expensive operations so they are only run when the
	// logging level requires it.
	log.Tracef("%v", newLogClosure(func() string {
		return "[PEER] msg" + spew.Sdump(msg)
	}))
	log.Tracef("%v", newLogClosure(func() string {
		var buf bytes.Buffer
		err := btcwire.WriteMessage(&buf, msg, p.protocolVersion, p.btcnet)
		if err != nil {
			return err.Error()
		}
		return "[PEER] " + spew.Sdump(buf.Bytes())
	}))

	// Write the message to the peer.
	err := btcwire.WriteMessage(p.conn, msg, p.protocolVersion, p.btcnet)
	if err != nil {
		p.Disconnect()
		p.logError("[PEER] Can't send message: %v", err)
		return
	}
}

// messageToHex serializes a message to the wire protocol encoding using the
// latest protocol version and returns a hex-encoded string of the result.
func messageToHex(msg btcwire.Message) (string, error) {
	var buf bytes.Buffer
	err := msg.BtcEncode(&buf, btcwire.ProtocolVersion)
	if err != nil {
		return "", btcjson.Error{
			Code:    btcjson.ErrInternal.Code,
			Message: err.Error(),
		}
	}
	return hex.EncodeToString(buf.Bytes()), nil
}

// writeMessage sends a bitcoin Message to the peer with logging.
func (p *peer) writeMessage(msg btcwire.Message) {
	// Don't do anything if we're disconnecting.
	if atomic.LoadInt32(&p.disconnect) != 0 {
		return
	}
	if !p.versionKnown {
		switch msg.(type) {
		case *btcwire.MsgVersion:
			// This is OK.
		default:
			// We drop all messages other than version if we
			// haven't done the handshake already.
			return
		}
	}

	// Use closures to log expensive operations so they are only run when
	// the logging level requires it.
	log.Debugf("%v", newLogClosure(func() string {
		// Debug summary of message.
		summary := messageSummary(msg)
		if len(summary) > 0 {
			summary = " (" + summary + ")"
		}
		return fmt.Sprintf("PEER: Sending %v%s to %s", msg.Command(),
			summary, p.addr)
	}))
	log.Tracef("%v", newLogClosure(func() string {
		return "PEER: " + spew.Sdump(msg)
	}))
	log.Tracef("%v", newLogClosure(func() string {
		var buf bytes.Buffer
		err := btcwire.WriteMessage(&buf, msg, p.protocolVersion, p.btcnet)
		if err != nil {
			return err.Error()
		}
		return "PEER: " + spew.Sdump(buf.Bytes())
	}))

	// Write the message to the peer.
	err := btcwire.WriteMessage(p.conn, msg, p.protocolVersion, p.btcnet)
	if err != nil {
		p.Disconnect()
		p.logError("PEER: Can't send message: %v", err)
		return
	}
}

// messageSummary returns a human-readable string which summarizes a message.
// Not all messages have or need a summary.  This is used for debug logging.
func messageSummary(msg btcwire.Message) string {
	switch msg := msg.(type) {
	case *btcwire.MsgVersion:
		return fmt.Sprintf("agent %s, pver %d, block %d",
			msg.UserAgent, msg.ProtocolVersion, msg.LastBlock)

	case *btcwire.MsgVerAck:
		// No summary.

	case *btcwire.MsgGetAddr:
		// No summary.

	case *btcwire.MsgAddr:
		return fmt.Sprintf("%d addr", len(msg.AddrList))

	case *btcwire.MsgPing:
		// No summary - perhaps add nonce.

	case *btcwire.MsgPong:
		// No summary - perhaps add nonce.

	case *btcwire.MsgAlert:
		// No summary.

	case *btcwire.MsgMemPool:
		// No summary.

	case *btcwire.MsgTx:
		hash, _ := msg.TxSha()
		return fmt.Sprintf("hash %s, %d inputs, %d outputs, lock %s",
			hash, len(msg.TxIn), len(msg.TxOut),
			formatLockTime(msg.LockTime))

	case *btcwire.MsgBlock:
		header := &msg.Header
		hash, _ := msg.BlockSha()
		return fmt.Sprintf("hash %s, ver %d, %d tx, %s", hash,
			header.Version, len(msg.Transactions), header.Timestamp)

	case *btcwire.MsgInv:
		return invSummary(msg.InvList)

	case *btcwire.MsgNotFound:
		return invSummary(msg.InvList)

	case *btcwire.MsgGetData:
		return invSummary(msg.InvList)

	case *btcwire.MsgGetBlocks:
		return locatorSummary(msg.BlockLocatorHashes, &msg.HashStop)

	case *btcwire.MsgGetHeaders:
		return locatorSummary(msg.BlockLocatorHashes, &msg.HashStop)

	case *btcwire.MsgHeaders:
		return fmt.Sprintf("num %d", len(msg.Headers))

	case *btcwire.MsgReject:
		// Ensure the variable length strings don't contain any
		// characters which are even remotely dangerous such as HTML
		// control characters, etc.  Also limit them to sane length for
		// logging.
		rejCommand := sanitizeString(msg.Cmd, btcwire.CommandSize)
		rejReason := sanitizeString(msg.Reason, maxRejectReasonLen)
		summary := fmt.Sprintf("cmd %v, code %v, reason %v", rejCommand,
			msg.Code, rejReason)
		if rejCommand == btcwire.CmdBlock || rejCommand == btcwire.CmdTx {
			summary += fmt.Sprintf(", hash %v", msg.Hash)
		}
		return summary
	}

	// No summary for other messages.
	return ""
}

func connHandler(id int, outAddrs chan<- []*btc.NetAddress, outNode chan<- Node, inAddr <-chan *btc.NetAddress) {
	// A worker that deals with the connection to a single bitcoin node.
	// It writes the list of nodes reported by node into out.
	// It also writes a valid node into outNode.
	// It reads from inAddr everytime it closes a connection

	for {
		addr := <-inAddr
		strA := addressFmt(*addr)

		threadLog := func(reported LogLevel, msg string) {
			if reported <= Level {
				logger.Printf("[%d] %s: %s\n", id, strA, msg)
			}
		}
		connProtoVer := btc.ProtocolVersion
		write := composeWrite(threadLog, connProtoVer)

		conn, err := net.DialTimeout("tcp", strA, time.Millisecond*500)
		if err != nil {
			threadLog(Log, err.Error())
			continue
		}
		threadLog(Info, "Connected")

		ver_m, _ := btc.NewMsgVersionFromConn(conn, genNonce(), 0)
		ver_m.AddUserAgent("btcmonitor", "0.0.1")
		write(conn, ver_m)

		// We are looking for successful addr messages
		wins := 0
		// After 10 seconds we just close the conn and handle errors
		time.AfterFunc(time.Second*10, func() { conn.Close() })
	MessageLoop:
		for {
			var resp btc.Message
			resp, _, err := btc.ReadMessage(conn, connProtoVer, btcnet)
			if err != nil {
				threadLog(Log, err.Error())
				break MessageLoop
			}
			threadLog(Info, resp.Command())
			switch resp := resp.(type) {
			case *btc.MsgVersion:
				nodePVer := uint32(resp.ProtocolVersion)
				if nodePVer < connProtoVer {
					connProtoVer = nodePVer
					write = composeWrite(threadLog, connProtoVer)
				}
				node := convNode(*addr, *resp)
				outNode <- node
				verack := btc.NewMsgVerAck()
				write(conn, verack)
				getAddr := btc.NewMsgGetAddr()
				write(conn, getAddr)
			case *btc.MsgAddr:
				wins += 1
				addrs := resp.AddrList
				outAddrs <- addrs
				if wins == 3 {
					break MessageLoop
				}
			case *btc.MsgPing:
				nonce := resp.Nonce
				pong := btc.NewMsgPong(nonce)
				write(conn, pong)
			}
		}
	}
}

// messageSummary returns a human-readable string which summarizes a message.
// Not all messages have or need a summary.  This is used for debug logging.
func messageSummary(msg btcwire.Message) string {
	switch msg := msg.(type) {
	case *btcwire.MsgVersion:
		return fmt.Sprintf("agent %s, pver %d, block %d",
			msg.UserAgent, msg.ProtocolVersion, msg.LastBlock)

	case *btcwire.MsgVerAck:
		// No summary.

	case *btcwire.MsgGetAddr:
		// No summary.

	case *btcwire.MsgAddr:
		return fmt.Sprintf("%d addr", len(msg.AddrList))

	case *btcwire.MsgPing:
		// No summary - perhaps add nonce.

	case *btcwire.MsgPong:
		// No summary - perhaps add nonce.

	case *btcwire.MsgAlert:
		// No summary.

	case *btcwire.MsgMemPool:
		// No summary.

	case *btcwire.MsgTx:
		hash, _ := msg.TxSha()
		return fmt.Sprintf("hash %s, %d inputs, %d outputs, lock %s",
			hash, len(msg.TxIn), len(msg.TxOut),
			formatLockTime(msg.LockTime))

	case *btcwire.MsgBlock:
		header := &msg.Header
		hash, _ := msg.BlockSha()
		return fmt.Sprintf("hash %s, ver %d, %d tx, %s", hash,
			header.Version, header.TxnCount, header.Timestamp)

	case *btcwire.MsgInv:
		return invSummary(msg.InvList)

	case *btcwire.MsgNotFound:
		return invSummary(msg.InvList)

	case *btcwire.MsgGetData:
		return invSummary(msg.InvList)

	case *btcwire.MsgGetBlocks:
		return locatorSummary(msg.BlockLocatorHashes, &msg.HashStop)

	case *btcwire.MsgGetHeaders:
		return locatorSummary(msg.BlockLocatorHashes, &msg.HashStop)

	case *btcwire.MsgHeaders:
		return fmt.Sprintf("num %d", len(msg.Headers))
	}

	// No summary for other messages.
	return ""
}

func nodeHandler(cfg TowerCfg, txStream chan<- *TxMeta, blockStream chan<- *btcwire.MsgBlock) {

	conn := setupConn(cfg.Addr, cfg.Logger)

	connparams := connParams{
		conn:    conn,
		pver:    btcwire.ProtocolVersion,
		btcnet:  cfg.Net,
		logger:  cfg.Logger,
		insetup: true,
	}

	read, write := composeConnOuts(connparams)

	// Initial handshake
	ver_m, _ := btcwire.NewMsgVersionFromConn(conn, genNonce(), int32(cfg.StartHeight))
	ver_m.AddUserAgent("Watchtower", "0.0.0")

	write(ver_m)

	// Wait for responses
	acked, responded := false, false
	for {
		var msg btcwire.Message
		msg = read()
		cfg.Logger.Println(msg.Command())
		switch msg := msg.(type) {
		case *btcwire.MsgVersion:
			responded = true
			ack := btcwire.NewMsgVerAck()
			nodeVer := uint32(msg.ProtocolVersion)
			if nodeVer < connparams.pver {
				connparams.pver = nodeVer
				read, write = composeConnOuts(connparams)
			}
			write(ack)
		case *btcwire.MsgVerAck:
			acked = true
		}
		if responded && acked {
			break
		}
	}

	// We are through the initial handshake, assume functional channel from here
	// on out. If there any errors with the pipe logger.Fatal gets called.
	connparams.insetup = false
	read, write = composeConnOuts(connparams)
	cfg.Logger.Println("Conn Negotiated")

	for {
		// If there are messages to send to peers send them. Otherwise, listen!
		select {
		case msg := <-cfg.MsgChan:
			write(msg)
		default:
			// listen for txs + blocks then push them into the appropriatestreams
			msg := read()
			switch msg := msg.(type) {
			case *btcwire.MsgInv:
				want := btcwire.NewMsgGetData()
				invVec := msg.InvList
				for i := range invVec {
					chunk := invVec[i]
					want.AddInvVect(chunk)
				}
				write(want)
			case *btcwire.MsgTx:
				var empt []byte // evaluates to nil
				meta := TxMeta{MsgTx: msg, BlockSha: empt, Time: time.Now()}
				txStream <- &meta
			case *btcwire.MsgBlock:
				blockStream <- msg
			case *btcwire.MsgPing:
				pong := btcwire.NewMsgPong(msg.Nonce)
				// More fun than anything...
				write(pong)
			}
		}
	}
}