conn.go

package golangNeo4jBoltDriver

import (
	"bytes"
	"database/sql/driver"
	"io/ioutil"
	"net"
	"time"

	"net/url"
	"strings"

	"crypto/tls"
	"crypto/x509"
	"io"
	"strconv"

	"github.com/tkandal/golang-neo4j-bolt-driver/encoding"
	"github.com/tkandal/golang-neo4j-bolt-driver/errors"
	"github.com/tkandal/golang-neo4j-bolt-driver/log"
	"github.com/tkandal/golang-neo4j-bolt-driver/structures/messages"
)

// Conn represents a connection to Neo4J database.
//
// Implements a neo-friendly interface.
// Some features of this interface implement neo-specific features
// unavailable in the sql/driver compatible interface.
//
// Conn objects, and any prepared statements/transactions within ARE NOT
// THREAD SAFE.  If you want to use multiple go routines with these objects,
// you should use a driver to create a new conn for each routine.
type Conn interface {
	// PrepareNeo prepares a neo4j specific statement
	PrepareNeo(query string) (Stmt, error)
	// PreparePipeline prepares a neo4j specific pipeline statement
	// Useful for running multiple queries at the same time
	PreparePipeline(query ...string) (PipelineStmt, error)
	// QueryNeo queries using the neo4j-specific interface
	QueryNeo(query string, params map[string]interface{}) (Rows, error)
	// QueryNeoAll queries using the neo4j-specific interface and returns all row data and output metadata
	QueryNeoAll(query string, params map[string]interface{}) ([][]interface{}, map[string]interface{}, map[string]interface{}, error)
	// QueryPipeline queries using the neo4j-specific interface
	// pipelining multiple statements
	QueryPipeline(query []string, params ...map[string]interface{}) (PipelineRows, error)
	// ExecNeo executes a query using the neo4j-specific interface
	ExecNeo(query string, params map[string]interface{}) (Result, error)
	// ExecPipeline executes a query using the neo4j-specific interface
	// pipelining multiple statements
	ExecPipeline(query []string, params ...map[string]interface{}) ([]Result, error)
	// Close closes the connection
	Close() error
	// Begin starts a new transaction
	Begin() (driver.Tx, error)
	// SetChunkSize is used to set the max chunk size of the
	// bytes to send to Neo4j at once
	SetChunkSize(uint16) ConnOption
	// SetTimeout sets the read/write timeouts for the
	// connection to Neo4j
	SetTimeout(time.Duration) ConnOption
}

// ConnOption is a type for setting an option on Conn.
type ConnOption func(*boltConn) ConnOption

type boltConn struct {
	connStr       string
	url           *url.URL
	user          string
	password      string
	conn          net.Conn
	connErr       error
	serverVersion []byte
	timeout       time.Duration
	chunkSize     uint16
	keepAlive     bool
	closed        bool
	useTLS        bool
	certFile      string
	caCertFile    string
	keyFile       string
	tlsNoVerify   bool
	transaction   *boltTx
	statement     *boltStmt
	driver        *boltDriver
	poolDriver    DriverPool
	last          time.Time
}

func createBoltConn(connStr string) *boltConn {
	return &boltConn{
		connStr:       connStr,
		timeout:       defaultTimeout,
		chunkSize:     defaultChunkSize,
		serverVersion: make([]byte, 4),
	}
}

// newBoltConn Creates a new bolt connection.
func newBoltConn(connStr string, driver *boltDriver) (*boltConn, error) {
	c := createBoltConn(connStr)
	c.driver = driver
	c.timeout = driver.timeout
	c.chunkSize = driver.chunkSize
	c.keepAlive = driver.keepAlive

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

	return c, nil
}

// newPooledBoltConn Creates a new bolt connection with a pooled driver.
func newPooledBoltConn(connStr string, driver *boltDriverPool) (*boltConn, error) {
	c := createBoltConn(connStr)
	c.poolDriver = driver
	c.timeout = driver.timeout
	c.chunkSize = driver.chunkSize
	c.keepAlive = driver.keepAlive
	return c, nil
}

func (c *boltConn) parseURL() (*url.URL, error) {
	boltURL, err := url.Parse(c.connStr)
	if err != nil {
		return nil, errors.Wrap(err, "an error occurred parsing bolt URL")
	}
	if strings.ToLower(boltURL.Scheme) != "bolt" {
		return nil, errors.New("unsupported connection string scheme %s; driver only supports 'bolt' scheme",
			boltURL.Scheme)
	}

	if boltURL.User != nil {
		c.user = boltURL.User.Username()
		var isSet bool
		c.password, isSet = boltURL.User.Password()
		if !isSet {
			return nil, errors.New("must specify password when passing user")
		}
	}

	timeout := boltURL.Query().Get("timeout")
	if timeout != "" {
		timeoutInt, err := strconv.Atoi(timeout)
		if err != nil {
			return nil, errors.New("invalid format for timeout %s;  must be an integer", timeout)
		}

		c.timeout = time.Duration(timeoutInt) * time.Second
	}

	useTLS := boltURL.Query().Get("tls")
	c.useTLS = strings.HasPrefix(strings.ToLower(useTLS), "t") || useTLS == "1"

	if c.useTLS {
		c.certFile = boltURL.Query().Get("tls_cert_file")
		c.keyFile = boltURL.Query().Get("tls_key_file")
		c.caCertFile = boltURL.Query().Get("tls_ca_cert_file")
		noVerify := boltURL.Query().Get("tls_no_verify")
		c.tlsNoVerify = strings.HasPrefix(strings.ToLower(noVerify), "t") || noVerify == "1"
	}

	log.Trace("bolt host ", boltURL.Host)
	log.Trace("timeout ", c.timeout)
	if len(c.user) > 0 {
		log.Trace("user ", c.user)
	}
	if len(c.password) > 0 {
		log.Trace("password ", strings.Repeat("*", len(c.password)))
	}
	log.Trace("TLS ", c.useTLS)
	log.Trace("TLS No Verify ", c.tlsNoVerify)
	log.Trace("Cert File ", c.certFile)
	log.Trace("Key File ", c.keyFile)
	log.Trace("CA Cert File ", c.caCertFile)

	return boltURL, nil
}

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 connection 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 while dialing to neo4j")
		}
	}
	if c.keepAlive {
		if tc, ok := conn.(*net.TCPConn); ok {
			if err := tc.SetKeepAlive(true); err != nil {
				_ = conn.Close()
				conn = nil
				return nil, err
			}
		}
	}
	c.last = time.Now()
	return conn, nil
}

func (c *boltConn) tlsConfig() (*tls.Config, error) {
	config := &tls.Config{
		MinVersion: tls.VersionTLS12,
		MaxVersion: tls.VersionTLS13,
	}

	if c.caCertFile != "" {
		// Load CA cert - usually for self-signed certificates
		caCert, err := ioutil.ReadFile(c.caCertFile)
		if err != nil {
			return nil, err
		}
		caCertPool := x509.NewCertPool()
		caCertPool.AppendCertsFromPEM(caCert)

		config.RootCAs = caCertPool
	}

	if c.certFile != "" {
		if c.keyFile == "" {
			return nil, errors.New("if you're providing a cert file, you must also provide a key file")
		}

		cert, err := tls.LoadX509KeyPair(c.certFile, c.keyFile)
		if err != nil {
			return nil, err
		}

		config.Certificates = []tls.Certificate{cert}
	}

	if c.tlsNoVerify {
		config.InsecureSkipVerify = true
	}

	return config, nil
}

func (c *boltConn) handShake() error {
	numWritten, err := c.Write(handShake)
	if numWritten != 20 {
		log.Errorf("couldn't write expected bytes for magic preamble + supported versions; written %d; expected 4", numWritten)
		if err != nil {
			err = errors.Wrap(err, "an error occurred writing magic preamble + supported versions")
		}
		return err
	}

	numRead, err := c.Read(c.serverVersion)
	if numRead != 4 {
		log.Errorf("could not read server version response; read %d bytes; expected 4 bytes; output %s", numRead, c.serverVersion)
		if err != nil {
			err = errors.Wrap(err, "an error occurred reading server version")
		}
		return err
	} else if bytes.Equal(c.serverVersion, noVersionSupported) {
		return errors.New("server responded with no supported version")
	}

	return nil
}

func (c *boltConn) initialize() error {
	// Handle recorder. If there is no conn string, assume we're playing back a recording.
	// If there is a recorder and a connection string, assume we're recording the connection
	// Else, just create the connection normally
	var err error
	if c.connStr == "" && c.driver != nil && c.driver.recorder != nil {
		c.conn = c.driver.recorder
	} else if c.driver != nil && c.driver.recorder != nil {
		c.driver.recorder.Conn, err = c.createConn()
		if err != nil {
			// Return the connection back into the pool
			if e := c.Close(); e != nil {
				log.Errorf("an error occurred closing connection; %v", e)
			}
			return err
		}
		c.conn = c.driver.recorder
	} else {
		c.conn, err = c.createConn()
		if err != nil {
			// Return the connection back into the pool
			if e := c.Close(); e != nil {
				log.Errorf("an error occurred closing connection; %v", e)
			}
			return err
		}
	}

	if err := c.handShake(); err != nil {
		if e := c.Close(); e != nil {
			log.Errorf("an error occurred closing connection; %v", e)
		}
		return err
	}

	respInt, err := c.sendInit()
	if err != nil {
		if e := c.Close(); e != nil {
			log.Errorf("an error occurred closing connection; %v", e)
		}
		return err
	}

	switch resp := respInt.(type) {
	case messages.SuccessMessage:
		log.Infof("successfully initiated Bolt connection; %+v", resp)
		return nil
	default:
		log.Errorf("got an unrecognized message when initializing connection; %+v", resp)
		c.connErr = errors.New("unrecognized response from the server; %#v", resp)
		_ = c.Close()
		return driver.ErrBadConn
	}
}

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

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

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

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

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

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

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

	if err != nil {
		c.connErr = errors.Wrap(err, "an error occurred writing to stream")
		err = driver.ErrBadConn
	}
	return n, err
}

// Close closes the connection.
// Driver may allow for pooling in the future, keeping connections alive.
func (c *boltConn) Close() error {

	if c.closed {
		return nil
	}

	if c.statement != nil {
		if err := c.statement.Close(); err != nil {
			return err
		}
	}

	if c.transaction != nil {
		if err := c.transaction.Rollback(); err != nil {
			return errors.Wrap(err, "error rolling back transaction when closing connection")
		}
	}

	if c.poolDriver != nil {
		// If using connection pooling, don't close connection, just reclaim it
		err := c.poolDriver.reclaim(c)
		if err != nil {
			log.Errorf("an error occurred reclaiming connection for pool; %v", err)
			c.connErr = errors.Wrap(err, "an error occurred closing the connection")
			return driver.ErrBadConn
		}
		return nil
	}

	if c.conn == nil {
		c.closed = true
		// c.connErr = fmt.Errorf("no connection to database")
		// return driver.ErrBadConn
		return nil
	}

	if err := c.conn.Close(); err != nil {
		if !connectionNilOrClosed(c) {
			c.connErr = errors.Wrap(err, "an error occurred closing the connection")
			return driver.ErrBadConn
		}
	}
	c.conn = nil
	c.closed = true
	return 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 acknowledging failure; %#v", resp)
			continue
		case messages.SuccessMessage:
			log.Infof("got success message when acknowledging failure; %#v", resp)
			return nil
		case messages.FailureMessage:
			log.Errorf("got failure message when acknowledging failure; %#v", resp)
			return c.reset()
		default:
			log.Errorf("got unrecognized response from acknowledging failure; %#v", resp)
			c.connErr = errors.New("got unrecognized response from acknowledging failure; %#v; closing session", resp)
			_ = c.Close()
			return driver.ErrBadConn
		}
	}
}

func (c *boltConn) reset() error {
	log.Info("resetting session")

	reset := messages.NewResetMessage()
	if err := encoding.NewEncoder(c, c.chunkSize).Encode(reset); err != nil {
		return errors.Wrap(err, "an error occurred encoding reset message")
	}

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

		switch resp := respInt.(type) {
		case messages.IgnoredMessage:
			log.Infof("got ignored message when resetting session; %#v", resp)
			continue
		case messages.SuccessMessage:
			log.Infof("got success message when resetting session; %#v", resp)
			return nil
		case messages.FailureMessage:
			log.Errorf("got failure message when resetting session; %#v", resp)
			err = c.Close()
			if err != nil {
				log.Errorf("an error occurred closing the session; %s", err)
			}
			return errors.Wrap(resp, "error resetting session; closing session")
		default:
			log.Errorf("got unrecognized response from resetting session; %#v", resp)
			c.connErr = errors.New("got unrecognized response from resetting session; %#v; closing session", resp)
			_ = c.Close()
			return driver.ErrBadConn
		}
	}
}

// Prepare prepares a new statement for a query.
func (c *boltConn) Prepare(query string) (driver.Stmt, error) {
	return c.prepare(query)
}

// PrepareNeo prepares a new statement for a query. Implements a Neo-friendly alternative to sql/driver.
func (c *boltConn) PrepareNeo(query string) (Stmt, error) {
	return c.prepare(query)
}

// PreparePipeline prepares a new pipeline statement for a query.
func (c *boltConn) PreparePipeline(queries ...string) (PipelineStmt, error) {
	if c.statement != nil {
		return nil, errors.New("an open statement already exists")
	}
	if c.closed {
		return nil, errors.New("connection already closed")
	}
	c.statement = newPipelineStmt(queries, c)
	return c.statement, nil
}

func (c *boltConn) prepare(query string) (*boltStmt, error) {
	if c.statement != nil {
		return nil, errors.New("an open statement already exists")
	}
	if c.closed {
		return nil, errors.New("connection already closed")
	}
	c.statement = newStmt(query, c)
	return c.statement, nil
}

// 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
}

// SetChunkSize sets the size of the chunks in bytes to write to the stream.
func (c *boltConn) SetChunkSize(chunkSize uint16) ConnOption {
	return func(c *boltConn) ConnOption {
		prev := c.chunkSize
		c.chunkSize = chunkSize
		return c.SetChunkSize(prev)
	}
}

// SetTimeout sets the timeout for reading and writing to the stream.
func (c *boltConn) SetTimeout(timeout time.Duration) ConnOption {
	return func(c *boltConn) ConnOption {
		prev := c.timeout
		c.timeout = timeout
		return c.SetTimeout(prev)
	}
}

func (c *boltConn) consume() (interface{}, error) {
	log.Info("consuming response from bolt stream")
	respInt, err := encoding.NewDecoder(c).Decode()
	if err != nil {
		return respInt, err
	}

	if log.GetLevel() >= log.TraceLevel {
		log.Tracef("consumed response; %#v", respInt)
	}

	if failure, isFail := respInt.(messages.FailureMessage); isFail {
		log.Errorf("got failure message; %#v", failure)
		err := c.ackFailure(failure)
		if err != nil {
			return nil, err
		}
		return failure, errors.Wrap(failure, "neo4J reported a failure for the query")
	}

	return respInt, err
}

func (c *boltConn) consumeAll() ([]interface{}, interface{}, error) {
	log.Info("consuming all responses until success/failure")

	responses := make([]interface{}, 0)
	for {
		respInt, err := c.consume()
		if err != nil {
			return nil, respInt, err
		}

		if success, isSuccess := respInt.(messages.SuccessMessage); isSuccess {
			log.Infof("got success message; %#v", success)
			return responses, success, nil
		}

		responses = append(responses, respInt)
	}
}

func (c *boltConn) consumeAllMultiple(mult int) ([][]interface{}, []interface{}, error) {
	log.Infof("consuming all responses %d times until success/failure", mult)

	responses := make([][]interface{}, mult)
	successes := make([]interface{}, mult)
	for i := 0; i < mult; i++ {

		resp, success, err := c.consumeAll()
		if err != nil {
			return responses, successes, err
		}

		responses[i] = resp
		successes[i] = success
	}

	return responses, successes, nil
}

func (c *boltConn) sendInit() (interface{}, error) {
	log.Infof("sending init message; client-id %s; User %s", ClientID, c.user)

	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) 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
}

func (c *boltConn) sendRunConsume(query string, args map[string]interface{}) (interface{}, error) {
	if err := c.sendRun(query, args); err != nil {
		return nil, err
	}

	return c.consume()
}

func (c *boltConn) sendPullAll() error {
	log.Infof("sending pull all message")

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

	return nil
}

func (c *boltConn) sendPullAllConsume() (interface{}, error) {
	if err := c.sendPullAll(); err != nil {
		return nil, err
	}

	return c.consume()
}

func (c *boltConn) sendRunPullAll(query string, args map[string]interface{}) error {
	if err := c.sendRun(query, args); err != nil {
		return err
	}

	return c.sendPullAll()
}

func (c *boltConn) sendRunPullAllConsumeRun(query string, args map[string]interface{}) (interface{}, error) {
	if err := c.sendRunPullAll(query, args); err != nil {
		return nil, err
	}

	return c.consume()
}

func (c *boltConn) sendRunPullAllConsumeSingle(query string, args map[string]interface{}) (interface{}, interface{}, error) {
	if err := c.sendRunPullAll(query, args); err != nil {
		return nil, nil, err
	}

	runSuccess, err := c.consume()
	if err != nil {
		return runSuccess, nil, err
	}

	pullSuccess, err := c.consume()
	return runSuccess, pullSuccess, err
}

func (c *boltConn) sendRunPullAllConsumeAll(query string, args map[string]interface{}) (interface{}, interface{}, []interface{}, error) {
	if err := c.sendRunPullAll(query, args); err != nil {
		return nil, nil, nil, err
	}

	runSuccess, err := c.consume()
	if err != nil {
		return runSuccess, nil, nil, err
	}

	records, pullSuccess, err := c.consumeAll()
	return runSuccess, pullSuccess, records, err
}

func (c *boltConn) sendDiscardAll() error {
	log.Infof("sending discard all message")

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

	return nil
}

func (c *boltConn) sendDiscardAllConsume() (interface{}, error) {
	if err := c.sendDiscardAll(); err != nil {
		return nil, err
	}

	return c.consume()
}

func (c *boltConn) sendRunDiscardAll(query string, args map[string]interface{}) error {
	if err := c.sendRun(query, args); err != nil {
		return err
	}

	return c.sendDiscardAll()
}

func (c *boltConn) sendRunDiscardAllConsume(query string, args map[string]interface{}) (interface{}, interface{}, error) {
	runResp, err := c.sendRunConsume(query, args)
	if err != nil {
		return runResp, nil, err
	}

	discardResp, err := c.sendDiscardAllConsume()
	return runResp, discardResp, err
}

func (c *boltConn) Query(query string, args []driver.Value) (driver.Rows, error) {
	params, err := driverArgsToMap(args)
	if err != nil {
		return nil, err
	}
	return c.queryNeo(query, params)
}

func (c *boltConn) QueryNeo(query string, params map[string]interface{}) (Rows, error) {
	return c.queryNeo(query, params)
}

func (c *boltConn) QueryNeoAll(query string, params map[string]interface{}) ([][]interface{}, map[string]interface{}, map[string]interface{}, error) {
	rows, err := c.queryNeo(query, params)
	if err != nil {
		return nil, nil, nil, err
	}
	defer func() {
		_ = rows.Close()
	}()

	data, metadata, err := rows.All()
	return data, rows.metadata, metadata, err
}

func (c *boltConn) queryNeo(query string, params map[string]interface{}) (*boltRows, error) {
	if c.statement != nil {
		return nil, errors.New("an open statement already exists")
	}
	if c.closed {
		return nil, errors.New("connection already closed")
	}

	c.statement = newStmt(query, c)

	// Pipeline the run + pull all for this
	successResp, err := c.sendRunPullAllConsumeRun(c.statement.query, params)
	if err != nil {
		_ = c.statement.Close()
		return nil, err
	}
	success, ok := successResp.(messages.SuccessMessage)
	if !ok {
		_ = c.statement.Close()
		return nil, errors.New("unexpected response querying neo from connection; %#v", successResp)
	}

	c.statement.rows = newQueryRows(c.statement, success.Metadata)
	return c.statement.rows, nil
}

func (c *boltConn) QueryPipeline(queries []string, params ...map[string]interface{}) (PipelineRows, error) {
	if c.statement != nil {
		return nil, errors.New("an open statement already exists")
	}
	if c.closed {
		return nil, errors.New("connection already closed")
	}

	c.statement = newPipelineStmt(queries, c)
	rows, err := c.statement.QueryPipeline(params...)
	if err != nil {
		return nil, err
	}

	// Since we're not exposing the statement,
	// tell the rows to close it when they are closed
	rows.(*boltRows).closeStatement = true
	return rows, nil
}

// Exec executes a query that returns no rows. See sql/driver.Stmt.
// You must bolt encode a map to pass as []bytes for the driver value.
func (c *boltConn) Exec(query string, args []driver.Value) (driver.Result, error) {
	if c.statement != nil {
		return nil, errors.New("an open statement already exists")
	}
	if c.closed {
		return nil, errors.New("connection already closed")
	}

	stmt := newStmt(query, c)
	defer func() {
		_ = stmt.Close()
	}()

	return stmt.Exec(args)
}

// ExecNeo executes a query that returns no rows. Implements a Neo-friendly alternative to sql/driver.
func (c *boltConn) ExecNeo(query string, params map[string]interface{}) (Result, error) {
	if c.statement != nil {
		return nil, errors.New("an open statement already exists")
	}
	if c.closed {
		return nil, errors.New("connection already closed")
	}

	stmt := newStmt(query, c)
	defer func() {
		_ = stmt.Close()
	}()

	return stmt.ExecNeo(params)
}

func (c *boltConn) ExecPipeline(queries []string, params ...map[string]interface{}) ([]Result, error) {
	if c.statement != nil {
		return nil, errors.New("an open statement already exists")
	}
	if c.closed {
		return nil, errors.New("connection already closed")
	}

	stmt := newPipelineStmt(queries, c)
	defer func() {
		_ = stmt.Close()
	}()

	return stmt.ExecPipeline(params...)
}