feat: upgraded Neo4j driver to v5

Author: joschi

database/neo4j/README.md

@@ -1,20 +1,33 @@
 # neo4j
-The Neo4j driver (bolt) does not natively support executing multiple statements in a single query. To allow for multiple statements in a single migration, you can use the `x-multi-statement` param. 
-This mode splits the migration text into separately-executed statements by a semi-colon `;`. Thus `x-multi-statement` cannot be used when a statement in the migration contains a string with a semi-colon.
-The queries **should** run in a single transaction, so partial migrations should not be a concern, but this is untested.
+The Neo4j driver (bolt) does not natively support executing multiple statements
+in a single query. To allow for multiple statements in a single migration, you
+can use the `x-multi-statement` param. This mode splits the migration text into
+separately-executed statements by a semicolon `;`. Thus `x-multi-statement`
+cannot be used when a statement in the migration contains a string with a
+semicolon. The queries **should** run in a single transaction, so partial
+migrations should not be a concern, but this is untested.
 
+Here are possible connection URLs:
 
-`neo4j://user:password@host:port/`
+- `neo4j://user:password@host:port/`
+- `neo4j+s://user:password@host:port/`
+- `neo4j+ssc://user:password@host:port/`
+- `bolt://user:password@host:port/`
+- `bolt+s://user:password@host:port/`
+- `bolt+ssc://user:password@host:port/`
+
+| URL Query           | WithInstance Config           | Description                                                                                          |
+|---------------------|-------------------------------|------------------------------------------------------------------------------------------------------|
+| `x-multi-statement` | `MultiStatement`              | Enable multiple statements to be ran in a single migration (See note above)                          |
+| `user`              | Contained within `AuthConfig` | The user to sign in as                                                                               |
+| `password`          | Contained within `AuthConfig` | The user's password                                                                                  |
+| `host`              |                               | The host to connect to. Values that start with / are for unix domain sockets. (default is localhost) |
+| `port`              |                               | The port to bind to. (default is 7687)                                                               |
+|                     | `MigrationsLabel`             | Name of the migrations node label                                                                    |
 
-| URL Query  | WithInstance Config | Description |
-|------------|---------------------|-------------|
-| `x-multi-statement` | `MultiStatement` | Enable multiple statements to be ran in a single migration (See note above) |
-| `user` | Contained within `AuthConfig` | The user to sign in as |
-| `password` | Contained within `AuthConfig` | The user's password |
-| `host` | | The host to connect to. Values that start with / are for unix domain sockets. (default is localhost) |
-| `port` | | The port to bind to. (default is 7687) |
-|  | `MigrationsLabel` | Name of the migrations node label |
 
 ## Supported versions
 
-Only Neo4j v3.5+ is [supported](https://github.com/neo4j/neo4j-go-driver/issues/64#issuecomment-625133600)
+Neo4j v4.4 LTS and v5+ is supported.
+
+Make sure to check [End Of Life dates](https://neo4j.com/developer/kb/neo4j-supported-versions/) of Neo4j versions.

database/neo4j/examples/migrations/1578421040_create_movies_constraint.down.cypher

@@ -1 +1 @@
-DROP CONSTRAINT ON (m:Movie) ASSERT m.Name IS UNIQUE
+DROP CONSTRAINT FOR (m:Movie) REQUIRE m.Name IS UNIQUE

database/neo4j/examples/migrations/1578421040_create_movies_constraint.up.cypher

@@ -1 +1 @@
-CREATE CONSTRAINT ON (m:Movie) ASSERT m.Name IS UNIQUE
+CREATE CONSTRAINT FOR (m:Movie) REQUIRE m.Name IS UNIQUE

database/neo4j/neo4j.go

@@ -2,17 +2,19 @@ package neo4j
 
 import (
 	"bytes"
+	"context"
 	"fmt"
-	"golang.org/x/mod/semver"
 	"io"
 	neturl "net/url"
 	"strconv"
 	"sync/atomic"
 
+	"golang.org/x/mod/semver"
+
 	"github.com/golang-migrate/migrate/v4/database"
 	"github.com/golang-migrate/migrate/v4/database/multistmt"
 	"github.com/hashicorp/go-multierror"
-	"github.com/neo4j/neo4j-go-driver/v4/neo4j"
+	"github.com/neo4j/neo4j-go-driver/v5/neo4j"
 )
 
 func init() {
@@ -38,14 +40,14 @@ type Config struct {
 }
 
 type Neo4j struct {
-	driver neo4j.Driver
+	driver neo4j.DriverWithContext
 	lock   uint32
 
 	// Open and WithInstance need to guarantee that config is never nil
 	config *Config
 }
 
-func WithInstance(driver neo4j.Driver, config *Config) (database.Driver, error) {
+func WithInstance(driver neo4j.DriverWithContext, config *Config) (database.Driver, error) {
 	if config == nil {
 		return nil, ErrNilConfig
 	}
@@ -70,31 +72,16 @@ func (n *Neo4j) Open(url string) (database.Driver, error) {
 	password, _ := uri.User.Password()
 	authToken := neo4j.BasicAuth(uri.User.Username(), password, "")
 	uri.User = nil
-	uri.Scheme = "bolt"
 	msQuery := uri.Query().Get("x-multi-statement")
 
-	// Whether to turn on/off TLS encryption.
-	tlsEncrypted := uri.Query().Get("x-tls-encrypted")
 	multi := false
-	encrypted := false
 	if msQuery != "" {
 		multi, err = strconv.ParseBool(uri.Query().Get("x-multi-statement"))
 		if err != nil {
 			return nil, err
 		}
 	}
 
-	if tlsEncrypted != "" {
-		encrypted, err = strconv.ParseBool(tlsEncrypted)
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	if encrypted {
-		uri.Scheme += "+s"
-	}
-
 	multiStatementMaxSize := DefaultMultiStatementMaxSize
 	if s := uri.Query().Get("x-multi-statement-max-size"); s != "" {
 		multiStatementMaxSize, err = strconv.Atoi(s)
@@ -105,11 +92,15 @@ func (n *Neo4j) Open(url string) (database.Driver, error) {
 
 	uri.RawQuery = ""
 
-	driver, err := neo4j.NewDriver(uri.String(), authToken, func(config *neo4j.Config) {})
+	driver, err := neo4j.NewDriverWithContext(uri.String(), authToken)
 	if err != nil {
 		return nil, err
 	}
 
+	if err = driver.VerifyConnectivity(context.Background()); err != nil {
+		return nil, err
+	}
+
 	return WithInstance(driver, &Config{
 		MigrationsLabel:       DefaultMigrationsLabel,
 		MultiStatement:        multi,
@@ -118,7 +109,7 @@ func (n *Neo4j) Open(url string) (database.Driver, error) {
 }
 
 func (n *Neo4j) Close() error {
-	return n.driver.Close()
+	return n.driver.Close(context.Background())
 }
 
 // local locking in order to pass tests, Neo doesn't support database locking
@@ -138,60 +129,71 @@ func (n *Neo4j) Unlock() error {
 }
 
 func (n *Neo4j) Run(migration io.Reader) (err error) {
-	session := n.driver.NewSession(neo4j.SessionConfig{AccessMode: neo4j.AccessModeWrite})
+	ctx := context.Background()
+	session := n.driver.NewSession(ctx, neo4j.SessionConfig{AccessMode: neo4j.AccessModeWrite})
 	defer func() {
-		if cerr := session.Close(); cerr != nil {
+		if cerr := session.Close(ctx); cerr != nil {
 			err = multierror.Append(err, cerr)
 		}
 	}()
 
 	if n.config.MultiStatement {
-		_, err = session.WriteTransaction(func(transaction neo4j.Transaction) (interface{}, error) {
-			var stmtRunErr error
-			if err := multistmt.Parse(migration, StatementSeparator, n.config.MultiStatementMaxSize, func(stmt []byte) bool {
-				trimStmt := bytes.TrimSpace(stmt)
-				if len(trimStmt) == 0 {
-					return true
-				}
-				trimStmt = bytes.TrimSuffix(trimStmt, StatementSeparator)
-				if len(trimStmt) == 0 {
-					return true
-				}
-
-				result, err := transaction.Run(string(trimStmt), nil)
-				if _, err := neo4j.Collect(result, err); err != nil {
-					stmtRunErr = err
-					return false
-				}
+		tx, err := session.BeginTransaction(ctx)
+		if err != nil {
+			return err
+		}
+		defer func() {
+			if cerr := tx.Close(ctx); cerr != nil {
+				err = multierror.Append(err, cerr)
+			}
+		}()
+
+		var stmtRunErr error
+		if err := multistmt.Parse(migration, StatementSeparator, n.config.MultiStatementMaxSize, func(stmt []byte) bool {
+			trimStmt := bytes.TrimSpace(stmt)
+			if len(trimStmt) == 0 {
 				return true
-			}); err != nil {
-				return nil, err
 			}
-			return nil, stmtRunErr
-		})
-		return err
+			trimStmt = bytes.TrimSuffix(trimStmt, StatementSeparator)
+			if len(trimStmt) == 0 {
+				return true
+			}
+
+			result, err := tx.Run(ctx, string(trimStmt), nil)
+			if _, err := neo4j.CollectWithContext(ctx, result, err); err != nil {
+				stmtRunErr = err
+				return false
+			}
+			return true
+		}); err != nil {
+			return err
+		}
+		return stmtRunErr
 	}
 
 	body, err := io.ReadAll(migration)
 	if err != nil {
 		return err
 	}
 
-	_, err = neo4j.Collect(session.Run(string(body[:]), nil))
+	res, err := session.Run(ctx, string(body[:]), nil)
+	_, err = neo4j.CollectWithContext(ctx, res, err)
 	return err
 }
 
 func (n *Neo4j) SetVersion(version int, dirty bool) (err error) {
-	session := n.driver.NewSession(neo4j.SessionConfig{AccessMode: neo4j.AccessModeWrite})
+	ctx := context.Background()
+	session := n.driver.NewSession(ctx, neo4j.SessionConfig{AccessMode: neo4j.AccessModeWrite})
 	defer func() {
-		if cerr := session.Close(); cerr != nil {
+		if cerr := session.Close(ctx); cerr != nil {
 			err = multierror.Append(err, cerr)
 		}
 	}()
 
 	query := fmt.Sprintf("MERGE (sm:%s {version: $version}) SET sm.dirty = $dirty, sm.ts = datetime()",
 		n.config.MigrationsLabel)
-	_, err = neo4j.Collect(session.Run(query, map[string]interface{}{"version": version, "dirty": dirty}))
+	res, err := session.Run(ctx, query, map[string]interface{}{"version": version, "dirty": dirty})
+	_, err = neo4j.CollectWithContext(ctx, res, err)
 	if err != nil {
 		return err
 	}
@@ -204,75 +206,73 @@ type MigrationRecord struct {
 }
 
 func (n *Neo4j) Version() (version int, dirty bool, err error) {
-	session := n.driver.NewSession(neo4j.SessionConfig{AccessMode: neo4j.AccessModeRead})
+	ctx := context.Background()
+	session := n.driver.NewSession(ctx, neo4j.SessionConfig{AccessMode: neo4j.AccessModeRead})
 	defer func() {
-		if cerr := session.Close(); cerr != nil {
+		if cerr := session.Close(ctx); cerr != nil {
 			err = multierror.Append(err, cerr)
 		}
 	}()
 
 	query := fmt.Sprintf(`MATCH (sm:%s) RETURN sm.version AS version, sm.dirty AS dirty
 ORDER BY COALESCE(sm.ts, datetime({year: 0})) DESC, sm.version DESC LIMIT 1`,
 		n.config.MigrationsLabel)
-	result, err := session.ReadTransaction(func(transaction neo4j.Transaction) (interface{}, error) {
-		result, err := transaction.Run(query, nil)
-		if err != nil {
-			return nil, err
-		}
-		if result.Next() {
-			record := result.Record()
-			mr := MigrationRecord{}
-			versionResult, ok := record.Get("version")
-			if !ok {
-				mr.Version = database.NilVersion
-			} else {
-				mr.Version = int(versionResult.(int64))
-			}
 
-			dirtyResult, ok := record.Get("dirty")
-			if ok {
-				mr.Dirty = dirtyResult.(bool)
-			}
+	tx, err := session.BeginTransaction(ctx)
 
-			return mr, nil
-		}
-		return nil, result.Err()
-	})
+	result, err := tx.Run(ctx, query, nil)
 	if err != nil {
 		return database.NilVersion, false, err
 	}
-	if result == nil {
-		return database.NilVersion, false, err
+	if result.Next(ctx) {
+		record := result.Record()
+		mr := MigrationRecord{}
+		versionResult, ok := record.Get("version")
+		if !ok {
+			mr.Version = database.NilVersion
+		} else {
+			mr.Version = int(versionResult.(int64))
+		}
+
+		dirtyResult, ok := record.Get("dirty")
+		if ok {
+			mr.Dirty = dirtyResult.(bool)
+		}
+
+		return mr.Version, mr.Dirty, nil
 	}
-	mr := result.(MigrationRecord)
-	return mr.Version, mr.Dirty, err
+
+	return database.NilVersion, false, err
 }
 
 func (n *Neo4j) Drop() (err error) {
-	session := n.driver.NewSession(neo4j.SessionConfig{AccessMode: neo4j.AccessModeWrite})
+	ctx := context.Background()
+	session := n.driver.NewSession(ctx, neo4j.SessionConfig{AccessMode: neo4j.AccessModeWrite})
 	defer func() {
-		if cerr := session.Close(); cerr != nil {
+		if cerr := session.Close(ctx); cerr != nil {
 			err = multierror.Append(err, cerr)
 		}
 	}()
 
-	if _, err := neo4j.Collect(session.Run("MATCH (n) DETACH DELETE n", nil)); err != nil {
+	res, err := session.Run(ctx, "MATCH (n) DETACH DELETE n", nil)
+	if _, err := neo4j.CollectWithContext(ctx, res, err); err != nil {
 		return err
 	}
 	return nil
 }
 
 func (n *Neo4j) ensureVersionConstraint() (err error) {
-	session := n.driver.NewSession(neo4j.SessionConfig{AccessMode: neo4j.AccessModeWrite})
+	ctx := context.Background()
+	session := n.driver.NewSession(ctx, neo4j.SessionConfig{AccessMode: neo4j.AccessModeWrite})
 	defer func() {
-		if cerr := session.Close(); cerr != nil {
+		if cerr := session.Close(ctx); cerr != nil {
 			err = multierror.Append(err, cerr)
 		}
 	}()
 
 	var neo4jVersion string
-
-	res, err := neo4j.Collect(session.Run("call dbms.components() yield versions unwind versions as version return version", nil))
+	result, err := session.Run(ctx, "call dbms.components() yield versions unwind versions as version return version", nil)
+	res, err := neo4j.CollectWithContext(ctx, result, err)
 	if err != nil {
 		return err
 	}
@@ -287,7 +287,8 @@ func (n *Neo4j) ensureVersionConstraint() (err error) {
 	using db.labels() to support Neo4j 3 and 4.
 	Neo4J 3 doesn't support db.constraints() YIELD name
 	*/
-	res, err = neo4j.Collect(session.Run(fmt.Sprintf("CALL db.labels() YIELD label WHERE label=\"%s\" RETURN label", n.config.MigrationsLabel), nil))
+	result, err = session.Run(ctx, fmt.Sprintf("CALL db.labels() YIELD label WHERE label=\"%s\" RETURN label", n.config.MigrationsLabel), nil)
+	res, err = neo4j.CollectWithContext(ctx, result, err)
 	if err != nil {
 		return err
 	}
@@ -299,13 +300,14 @@ func (n *Neo4j) ensureVersionConstraint() (err error) {
 	switch neo4jVersion {
 	case "v5":
 		query = fmt.Sprintf("CREATE CONSTRAINT FOR (a:%s) REQUIRE a.version IS UNIQUE", n.config.MigrationsLabel)
-	case "v3", "v4":
+	case "v4":
 		query = fmt.Sprintf("CREATE CONSTRAINT ON (a:%s) ASSERT a.version IS UNIQUE", n.config.MigrationsLabel)
 	default:
 		return fmt.Errorf("unsupported neo4j version %v", neo4jVersion)
 	}
 
-	if _, err := neo4j.Collect(session.Run(query, nil)); err != nil {
+	result, err = session.Run(ctx, query, nil)
+	if _, err := neo4j.CollectWithContext(ctx, result, err); err != nil {
 		return err
 	}
 	return nil