// Rollback rolls back and closes the transaction
func (t *boltTx) Rollback() error {
	if t.closed {
		return errors.New("Transaction already closed")
	}
	if t.conn.statement != nil {
		if err := t.conn.statement.Close(); err != nil {
			return errors.Wrap(err, "An error occurred closing open rows in transaction Rollback")
		}
	}

	successInt, pullInt, err := t.conn.sendRunPullAllConsumeSingle("ROLLBACK", nil)
	if err != nil {
		return errors.Wrap(err, "An error occurred rolling back transaction")
	}

	success, ok := successInt.(messages.SuccessMessage)
	if !ok {
		return errors.New("Unrecognized response type rolling back transaction: %#v", success)
	}

	log.Infof("Got success message rolling back transaction: %#v", success)

	pull, ok := pullInt.(messages.SuccessMessage)
	if !ok {
		return errors.New("Unrecognized response type pulling transaction: %#v", pull)
	}

	log.Infof("Got success message pulling transaction: %#v", pull)

	t.conn.transaction = nil
	t.closed = true
	return err
}

func (s *boltStmt) QueryPipeline(params ...map[string]interface{}) (PipelineRows, error) {
	if s.closed {
		return nil, errors.New("Neo4j Bolt statement already closed")
	}
	if s.rows != nil {
		return nil, errors.New("Another query is already open")
	}

	if len(params) != len(s.queries) {
		return nil, errors.New("Must pass same number of params as there are queries")
	}

	for i, query := range s.queries {
		err := s.conn.sendRunPullAll(query, params[i])
		if err != nil {
			return nil, errors.Wrap(err, "Error running query:\n\n%s\n\nWith Params:\n%#v", query, params[i])
		}
	}

	log.Info("Successfully ran all pipeline queries")

	resp, err := s.conn.consume()
	if err != nil {
		return nil, errors.Wrap(err, "An error occurred consuming initial pipeline command")
	}

	success, ok := resp.(messages.SuccessMessage)
	if !ok {
		return nil, errors.New("Got unexpected return message when consuming initial pipeline command: %#v", resp)
	}

	s.rows = newPipelineRows(s, success.Metadata, 0)
	return s.rows, nil
}

// Read out the object bytes to decode
func (d Decoder) read() (*bytes.Buffer, error) {
	output := &bytes.Buffer{}
	for {
		lengthBytes := make([]byte, 2)
		if numRead, err := d.r.Read(lengthBytes); numRead != 2 {
			return nil, errors.Wrap(err, "Couldn't read expected bytes for message length. Read: %d Expected: 2.", numRead)
		}

		// Chunk header contains length of current message
		messageLen := binary.BigEndian.Uint16(lengthBytes)
		if messageLen == 0 {
			// If the length is 0, the chunk is done.
			return output, nil
		}

		data, err := d.readData(messageLen)
		if err != nil {
			return output, errors.Wrap(err, "An error occurred reading message data")
		}

		numWritten, err := output.Write(data)
		if numWritten < len(data) {
			return output, errors.New("Didn't write full data on output. Expected: %d Wrote: %d", len(data), numWritten)
		} else if err != nil {
			return output, errors.Wrap(err, "Error writing data to output")
		}
	}
}

func (c *boltConn) createConn() (net.Conn, error) {

	var err error
	c.url, err = c.parseURL()
	if err != nil {
		return nil, errors.Wrap(err, "An error occurred parsing the conn URL")
	}

	var conn net.Conn
	if c.useTLS {
		config, err := c.tlsConfig()
		if err != nil {
			return nil, errors.Wrap(err, "An error occurred setting up TLS configuration")
		}
		conn, err = tls.Dial("tcp", c.url.Host, config)
		if err != nil {
			return nil, errors.Wrap(err, "An error occurred dialing to neo4j")
		}
	} else {
		conn, err = net.DialTimeout("tcp", c.url.Host, c.timeout)
		if err != nil {
			return nil, errors.Wrap(err, "An error occurred dialing to neo4j")
		}
	}

	return conn, nil
}

func (c *boltConn) ackFailure(failure messages.FailureMessage) error {
	log.Infof("Acknowledging Failure: %#v", failure)

	ack := messages.NewAckFailureMessage()
	err := encoding.NewEncoder(c, c.chunkSize).Encode(ack)
	if err != nil {
		return errors.Wrap(err, "An error occurred encoding ack failure message")
	}

	for {
		respInt, err := encoding.NewDecoder(c).Decode()
		if err != nil {
			return errors.Wrap(err, "An error occurred decoding ack failure message response")
		}

		switch resp := respInt.(type) {
		case messages.IgnoredMessage:
			log.Infof("Got ignored message when acking failure: %#v", resp)
			continue
		case messages.SuccessMessage:
			log.Infof("Got success message when acking failure: %#v", resp)
			return nil
		case messages.FailureMessage:
			log.Errorf("Got failure message when acking failure: %#v", resp)
			return c.reset()
		default:
			log.Errorf("Got unrecognized response from acking failure: %#v", resp)
			err := c.Close()
			if err != nil {
				log.Errorf("An error occurred closing the session: %s", err)
			}
			return errors.New("Got unrecognized response from acking failure: %#v. CLOSING SESSION!", resp)
		}
	}
}

// Close closes the rows
func (r *boltRows) Close() error {
	if r.closed {
		return nil
	}

	if !r.consumed {
		// Discard all messages if not consumed
		respInt, err := r.statement.conn.sendDiscardAllConsume()
		if err != nil {
			return errors.Wrap(err, "An error occurred discarding messages on row close")
		}

		switch resp := respInt.(type) {
		case messages.SuccessMessage:
			log.Infof("Got success message: %#v", resp)
		default:
			return errors.New("Unrecognized response type discarding all rows: Value: %#v", resp)
		}

	} else if !r.finishedConsume {
		// If this is a pipeline statement, we need to "consume all" multiple times
		numConsume := 1
		if r.statement.queries != nil {
			numQueries := len(r.statement.queries)
			if numQueries > 0 {
				// So, every pipeline statement has two successes
				// but by the time you get to the row object, one has
				// been consumed. Hence we need to clear out the
				// rest of the messages on close by taking the current
				// index * 2 but removing the first success
				numConsume = ((numQueries - r.pipelineIndex) * 2) - 1
			}
		}

		// Clear out all unconsumed messages if we
		// never finished consuming them.
		_, _, err := r.statement.conn.consumeAllMultiple(numConsume)
		if err != nil {
			return errors.Wrap(err, "An error occurred clearing out unconsumed stream")
		}
	}

	r.closed = true
	r.statement.rows = nil

	if r.closeStatement {
		return r.statement.Close()
	}
	return nil
}

func (s *boltStmt) ExecPipeline(params ...map[string]interface{}) ([]Result, error) {
	if s.closed {
		return nil, errors.New("Neo4j Bolt statement already closed")
	}
	if s.rows != nil {
		return nil, errors.New("Another query is already open")
	}

	if len(params) != len(s.queries) {
		return nil, errors.New("Must pass same number of params as there are queries")
	}

	for i, query := range s.queries {
		err := s.conn.sendRunPullAll(query, params[i])
		if err != nil {
			return nil, errors.Wrap(err, "Error running exec query:\n\n%s\n\nWith Params:\n%#v", query, params[i])
		}
	}

	log.Info("Successfully ran all pipeline queries")

	results := make([]Result, len(s.queries))
	for i := range s.queries {
		runResp, err := s.conn.consume()
		if err != nil {
			return nil, errors.Wrap(err, "An error occurred getting result of exec command: %#v", runResp)
		}

		success, ok := runResp.(messages.SuccessMessage)
		if !ok {
			return nil, errors.New("Unexpected response when getting exec query result: %#v", runResp)
		}

		_, pullResp, err := s.conn.consumeAll()
		if err != nil {
			return nil, errors.Wrap(err, "An error occurred getting result of exec discard command: %#v", pullResp)
		}

		success, ok = pullResp.(messages.SuccessMessage)
		if !ok {
			return nil, errors.New("Unexpected response when getting exec query discard result: %#v", pullResp)
		}

		results[i] = newResult(success.Metadata)

	}

	return results, nil
}

// Read reads the data from the underlying connection
func (c *boltConn) Read(b []byte) (n int, err error) {
	if err := c.conn.SetReadDeadline(time.Now().Add(c.timeout)); err != nil {
		return 0, errors.Wrap(err, "An error occurred setting read deadline")
	}

	n, err = c.conn.Read(b)

	if log.GetLevel() >= log.TraceLevel {
		log.Tracef("Read %d bytes from stream:\n\n%s\n", n, sprintByteHex(b))
	}

	if err != nil && err != io.EOF {
		err = errors.Wrap(err, "An error occurred reading from stream")
	}
	return n, err
}

// Write writes the data to the underlying connection
func (c *boltConn) Write(b []byte) (n int, err error) {
	if err := c.conn.SetWriteDeadline(time.Now().Add(c.timeout)); err != nil {
		return 0, errors.Wrap(err, "An error occurred setting write deadline")
	}

	n, err = c.conn.Write(b)

	if log.GetLevel() >= log.TraceLevel {
		log.Tracef("Wrote %d of %d bytes to stream:\n\n%s\n", len(b), n, sprintByteHex(b[:n]))
	}

	if err != nil {
		err = errors.Wrap(err, "An error occurred writing to stream")
	}
	return n, err
}

// Begin begins a new transaction with the Neo4J Database
func (c *boltConn) Begin() (driver.Tx, error) {
	if c.transaction != nil {
		return nil, errors.New("An open transaction already exists")
	}
	if c.statement != nil {
		return nil, errors.New("Cannot open a transaction when you already have an open statement")
	}
	if c.closed {
		return nil, errors.New("Connection already closed")
	}

	successInt, pullInt, err := c.sendRunPullAllConsumeSingle("BEGIN", nil)
	if err != nil {
		return nil, errors.Wrap(err, "An error occurred beginning transaction")
	}

	success, ok := successInt.(messages.SuccessMessage)
	if !ok {
		return nil, errors.New("Unrecognized response type beginning transaction: %#v", success)
	}

	log.Infof("Got success message beginning transaction: %#v", success)

	success, ok = pullInt.(messages.SuccessMessage)
	if !ok {
		return nil, errors.New("Unrecognized response type pulling transaction:  %#v", success)
	}

	log.Infof("Got success message pulling transaction: %#v", success)

	return newTx(c), nil
}

func (c *boltConn) sendRun(query string, args map[string]interface{}) error {
	log.Infof("Sending RUN message: query %s (args: %#v)", query, args)
	runMessage := messages.NewRunMessage(query, args)
	if err := encoding.NewEncoder(c, c.chunkSize).Encode(runMessage); err != nil {
		return errors.Wrap(err, "An error occurred running query")
	}

	return nil
}

// encodeInt encodes a nil object to the stream
func (e Encoder) encodeInt(val int64) error {
	var err error
	switch {
	case val >= math.MinInt64 && val < math.MinInt32:
		// Write as INT_64
		if _, err = e.Write([]byte{Int64Marker}); err != nil {
			return err
		}
		err = binary.Write(e, binary.BigEndian, val)
	case val >= math.MinInt32 && val < math.MinInt16:
		// Write as INT_32
		if _, err = e.Write([]byte{Int32Marker}); err != nil {
			return err
		}
		err = binary.Write(e, binary.BigEndian, int32(val))
	case val >= math.MinInt16 && val < math.MinInt8:
		// Write as INT_16
		if _, err = e.Write([]byte{Int16Marker}); err != nil {
			return err
		}
		err = binary.Write(e, binary.BigEndian, int16(val))
	case val >= math.MinInt8 && val < -16:
		// Write as INT_8
		if _, err = e.Write([]byte{Int8Marker}); err != nil {
			return err
		}
		err = binary.Write(e, binary.BigEndian, int8(val))
	case val >= -16 && val <= math.MaxInt8:
		// Write as TINY_INT
		err = binary.Write(e, binary.BigEndian, int8(val))
	case val > math.MaxInt8 && val <= math.MaxInt16:
		// Write as INT_16
		if _, err = e.Write([]byte{Int16Marker}); err != nil {
			return err
		}
		err = binary.Write(e, binary.BigEndian, int16(val))
	case val > math.MaxInt16 && val <= math.MaxInt32:
		// Write as INT_32
		if _, err = e.Write([]byte{Int32Marker}); err != nil {
			return err
		}
		err = binary.Write(e, binary.BigEndian, int32(val))
	case val > math.MaxInt32 && val <= math.MaxInt64:
		// Write as INT_64
		if _, err = e.Write([]byte{Int64Marker}); err != nil {
			return err
		}
		err = binary.Write(e, binary.BigEndian, val)
	default:
		return errors.New("Int too long to write: %d", val)
	}
	if err != nil {
		return errors.Wrap(err, "An error occured writing an int to bolt")
	}
	return err
}

func (c *boltConn) sendInit() (interface{}, error) {
	log.Infof("Sending INIT Message. ClientID: %s User: %s Password: %s", ClientID, c.user, c.password)

	initMessage := messages.NewInitMessage(ClientID, c.user, c.password)
	if err := encoding.NewEncoder(c, c.chunkSize).Encode(initMessage); err != nil {
		return nil, errors.Wrap(err, "An error occurred sending init message")
	}

	return c.consume()
}

func (c *boltConn) sendDiscardAll() error {
	log.Infof("Sending DISCARD_ALL message")

	discardAllMessage := messages.NewDiscardAllMessage()
	err := encoding.NewEncoder(c, c.chunkSize).Encode(discardAllMessage)
	if err != nil {
		return errors.Wrap(err, "An error occurred encoding discard all query")
	}

	return nil
}

func (c *boltConn) sendPullAll() error {
	log.Infof("Sending PULL_ALL message")

	pullAllMessage := messages.NewPullAllMessage()
	err := encoding.NewEncoder(c, c.chunkSize).Encode(pullAllMessage)
	if err != nil {
		return errors.Wrap(err, "An error occurred encoding pull all query")
	}

	return nil
}

// flush finishes the encoding stream by flushing it to the writer
func (e Encoder) flush() error {
	length := e.buf.Len()
	if length > 0 {
		if err := binary.Write(e.w, binary.BigEndian, uint16(length)); err != nil {
			return errors.Wrap(err, "An error occured writing length bytes during flush")
		}

		if _, err := e.buf.WriteTo(e.w); err != nil {
			return errors.Wrap(err, "An error occured writing message bytes during flush")
		}
	}

	_, err := e.w.Write(EndMessage)
	if err != nil {
		return errors.Wrap(err, "An error occurred ending encoding message")
	}
	e.buf.Reset()

	return nil
}

// encodeFloat encodes a nil object to the stream
func (e Encoder) encodeFloat(val float64) error {
	if _, err := e.Write([]byte{FloatMarker}); err != nil {
		return err
	}

	err := binary.Write(e, binary.BigEndian, val)
	if err != nil {
		return errors.Wrap(err, "An error occured writing a float to bolt")
	}

	return err
}

func (d Decoder) readData(messageLen uint16) ([]byte, error) {
	output := make([]byte, messageLen)
	var totalRead uint16
	for totalRead < messageLen {
		data := make([]byte, messageLen-totalRead)
		numRead, err := d.r.Read(data)
		if err != nil {
			return nil, errors.Wrap(err, "An error occurred reading from stream")
		} else if numRead == 0 {
			return nil, errors.Wrap(err, "Couldn't read expected bytes for message. Read: %d Expected: %d.", totalRead, messageLen)
		}

		for idx, b := range data {
			output[uint16(idx)+totalRead] = b
		}

		totalRead += uint16(numRead)
	}

	return output, nil
}

// newBoltConn Creates a new bolt connection
func newBoltConn(connStr string, driver *boltDriver) (*boltConn, error) {

	c := createBoltConn(connStr)
	c.driver = driver

	err := c.initialize()
	if err != nil {
		return nil, errors.Wrap(err, "An error occurred initializing connection")
	}

	return c, nil
}

// encodeMessageStructure encodes a nil object to the stream
func (e Encoder) encodeStructure(val structures.Structure) error {

	fields := val.AllFields()
	length := len(fields)
	switch {
	case length <= 15:
		if _, err := e.Write([]byte{byte(TinyStructMarker + length)}); err != nil {
			return err
		}
	case length > 15 && length <= math.MaxUint8:
		if _, err := e.Write([]byte{Struct8Marker}); err != nil {
			return err
		}
		if err := binary.Write(e, binary.BigEndian, int8(length)); err != nil {
			return err
		}
	case length > math.MaxUint8 && length <= math.MaxUint16:
		if _, err := e.Write([]byte{Struct16Marker}); err != nil {
			return err
		}
		if err := binary.Write(e, binary.BigEndian, int16(length)); err != nil {
			return err
		}
	default:
		return errors.New("Structure too long to write: %+v", val)
	}

	_, err := e.Write([]byte{byte(val.Signature())})
	if err != nil {
		return errors.Wrap(err, "An error occurred writing to encoder a struct field")
	}

	for _, field := range fields {
		if err := e.encode(field); err != nil {
			return errors.Wrap(err, "An error occurred encoding a struct field")
		}
	}

	return nil
}