Neo4J Support in Go Migrate [WIP]

.gitignore

@@ -6,3 +6,4 @@ cli/migrate
 .godoc.pid
 vendor/
 .vscode/
+.idea/

Makefile

@@ -1,5 +1,5 @@
 SOURCE ?= file go_bindata github github_ee aws_s3 google_cloud_storage godoc_vfs gitlab
-DATABASE ?= postgres mysql redshift cassandra spanner cockroachdb clickhouse mongodb sqlserver firebird
+DATABASE ?= postgres mysql redshift cassandra spanner cockroachdb clickhouse mongodb sqlserver firebird neo4j
 VERSION ?= $(shell git describe --tags 2>/dev/null | cut -c 2-)
 TEST_FLAGS ?=
 REPO_OWNER ?= $(shell cd .. && basename "$$(pwd)")

database/driver.go

@@ -33,7 +33,7 @@ var drivers = make(map[string]Driver)
 //      All other functions are tested by tests in database/testing.
 //      Saves you some time and makes sure all database drivers behave the same way.
 //   5. Call Register in init().
-//   6. Create a migrate/cli/build_<driver-name>.go file
+//   6. Create a internal/cli/build_<driver-name>.go file
 //   7. Add driver name in 'DATABASE' variable in Makefile
 //
 // Guidelines:
@@ -61,7 +61,7 @@ type Driver interface {
 	// all migrations have been run.
 	Unlock() error
 
-	// Run applies a migration to the database. migration is garantueed to be not nil.
+	// Run applies a migration to the database. migration is guaranteed to be not nil.
 	Run(migration io.Reader) error
 
 	// SetVersion saves version and dirty state.

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

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

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

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

database/neo4j/examples/migrations/1578421725_create_movies.down.cypher

@@ -0,0 +1,2 @@
+MATCH (m:Movie)
+DELETE m

database/neo4j/examples/migrations/1578421725_create_movies.up.cypher

@@ -0,0 +1,2 @@
+CREATE (:Movie {name: "Footloose"})
+CREATE (:Movie {name: "Ghost"})

database/neo4j/neo4j.go

@@ -0,0 +1,202 @@
+package neo4j
+
+import (
+	"fmt"
+	"io"
+	"io/ioutil"
+	neturl "net/url"
+	"sync/atomic"
+
+	"github.com/golang-migrate/migrate/v4/database"
+	"github.com/neo4j/neo4j-go-driver/neo4j"
+)
+
+func init() {
+	db := Neo4j{}
+	database.Register("bolt", &db)
+	database.Register("neo4j", &db)
+}
+
+var DefaultMigrationsLabel = "SchemaMigration"
+
+var (
+	ErrNilConfig = fmt.Errorf("no config")
+)
+
+type Config struct {
+	AuthToken       neo4j.AuthToken
+	URL             string
+	MigrationsLabel string
+}
+
+type Neo4j struct {
+	driver neo4j.Driver
+	lock   uint32
+
+	// Open and WithInstance need to guarantee that config is never nil
+	config *Config
+}
+
+func WithInstance(config *Config) (database.Driver, error) {
+	if config == nil {
+		return nil, ErrNilConfig
+	}
+
+	neoDriver, err := neo4j.NewDriver(config.URL, config.AuthToken)
+	if err != nil {
+		return nil, err
+	}
+
+	driver := &Neo4j{
+		driver: neoDriver,
+		config: config,
+	}
+
+	if err := driver.ensureVersionConstraint(); err != nil {
+		return nil, err
+	}
+
+	return driver, nil
+}
+
+func (n *Neo4j) Open(url string) (database.Driver, error) {
+	uri, err := neturl.Parse(url)
+	if err != nil {
+		return nil, err
+	}
+	password, _ := uri.User.Password()
+	authToken := neo4j.BasicAuth(uri.User.Username(), password, "")
+
+	return WithInstance(&Config{
+		URL:             url,
+		AuthToken:       authToken,
+		MigrationsLabel: DefaultMigrationsLabel,
+	})
+}
+
+func (n *Neo4j) Close() error {
+	return n.driver.Close()
+}
+
+// local locking in order to pass tests, Neo doesn't support database locking
+func (n *Neo4j) Lock() error {
+	if !atomic.CompareAndSwapUint32(&n.lock, 0, 1) {
+		return database.ErrLocked
+	}
+
+	return nil
+}
+
+func (n *Neo4j) Unlock() error {
+	atomic.StoreUint32(&n.lock, 0)
+	return nil
+}
+
+func (n *Neo4j) Run(migration io.Reader) error {
+	body, err := ioutil.ReadAll(migration)
+	if err != nil {
+		return err
+	}
+
+	session, err := n.driver.Session(neo4j.AccessModeWrite)
+	if err != nil {
+		return err
+	}
+	defer session.Close()
+
+	result, err := session.Run(string(body[:]), nil)
+	if err != nil {
+		return err
+	}
+	return result.Err()
+}
+
+func (n *Neo4j) SetVersion(version int, dirty bool) error {
+	session, err := n.driver.Session(neo4j.AccessModeRead)
+	if err != nil {
+		return err
+	}
+	defer session.Close()
+
+	query := fmt.Sprintf("MERGE (sm:%s {version: $version, dirty: $dirty})",
+		n.config.MigrationsLabel)
+	result, err := session.Run(query, map[string]interface{}{"version": version, "dirty": dirty})
+	if err != nil {
+		return err
+	}
+	return result.Err()
+}
+
+type MigrationRecord struct {
+	Version int
+	Dirty bool
+}
+
+func (n *Neo4j) Version() (version int, dirty bool, err error) {
+	session, err := n.driver.Session(neo4j.AccessModeRead)
+	if err != nil {
+		return -1, false, err
+	}
+	defer session.Close()
+
+	query := fmt.Sprintf("MATCH (sm:%s) RETURN sm.version AS version, sm.dirty AS dirty ORDER BY 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 = -1
+			} else {
+				mr.Version = int(versionResult.(int64))
+			}
+
+			dirtyResult, ok := record.Get("dirty")
+			if ok {
+				mr.Dirty = dirtyResult.(bool)
+			}
+
+			return mr, nil
+		}
+		return nil, result.Err()
+	})
+	if err != nil {
+		return -1, false, err
+	}
+	if result == nil {
+		return -1, false, nil
+	}
+	mr := result.(MigrationRecord)
+	return mr.Version, mr.Dirty, nil
+}
+
+func (n *Neo4j) Drop() error {
+	session, err := n.driver.Session(neo4j.AccessModeWrite);
+	if err != nil {
+		return err
+	}
+	defer session.Close()
+
+	_, err = session.Run("MATCH (n) DETACH DELETE n", map[string]interface{}{})
+	return err
+}
+
+func (n *Neo4j) ensureVersionConstraint() (err error) {
+	session, err := n.driver.Session(neo4j.AccessModeWrite);
+	if err != nil {
+		return err
+	}
+	defer session.Close()
+
+	query := fmt.Sprintf("CREATE CONSTRAINT ON (a:%s) ASSERT a.version IS UNIQUE", n.config.MigrationsLabel)
+	result, err := session.Run(query, nil)
+	if err != nil {
+		return err
+	}
+	return result.Err()
+}

database/neo4j/neo4j_test.go

@@ -0,0 +1,106 @@
+package neo4j
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"testing"
+
+	"github.com/dhui/dktest"
+	"github.com/neo4j/neo4j-go-driver/neo4j"
+
+	"github.com/golang-migrate/migrate/v4"
+	dt "github.com/golang-migrate/migrate/v4/database/testing"
+	"github.com/golang-migrate/migrate/v4/dktesting"
+	_ "github.com/golang-migrate/migrate/v4/source/file"
+)
+
+var (
+	opts = dktest.Options{PortRequired: true, ReadyFunc: isReady,
+		Env: map[string]string{"NEO4J_AUTH": "neo4j/migratetest", "NEO4J_ACCEPT_LICENSE_AGREEMENT": "yes"}}
+	specs = []dktesting.ContainerSpec{
+		{ImageName: "neo4j:3.5", Options: opts},
+		{ImageName: "neo4j:3.5-enterprise", Options: opts},
+		{ImageName: "neo4j:3.4", Options: opts},
+		{ImageName: "neo4j:3.4-enterprise", Options: opts},
+	}
+)
+
+func neoConnectionString(host, port string) string {
+	return fmt.Sprintf("bolt://neo4j:migratetest@%s:%s", host, port)
+}
+
+func isReady(ctx context.Context, c dktest.ContainerInfo) bool {
+	ip, port, err := c.Port(7687)
+	if err != nil {
+		return false
+	}
+
+	driver, err := neo4j.NewDriver(neoConnectionString(ip, port), neo4j.BasicAuth("neo4j", "migratetest", ""))
+	if err != nil {
+		return false
+	}
+	defer func() {
+		if err := driver.Close(); err != nil {
+			log.Println("close error:", err)
+		}
+	}()
+	session, err := driver.Session(neo4j.AccessModeRead)
+	if err != nil {
+		return false
+	}
+	result, err := session.Run("RETURN 1", nil)
+	if err != nil {
+		return false
+	} else if result.Err() != nil {
+		return false
+	}
+
+	return true
+}
+
+func Test(t *testing.T) {
+	dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
+		ip, port, err := c.Port(7687)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		n := &Neo4j{}
+		d, err := n.Open(neoConnectionString(ip, port))
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer func() {
+			if err := d.Close(); err != nil {
+				t.Error(err)
+			}
+		}()
+		dt.Test(t, d, []byte("MATCH a RETURN a"))
+	})
+}
+
+func TestMigrate(t *testing.T) {
+	dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
+		ip, port, err := c.Port(7687)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		n := &Neo4j{}
+		d, err := n.Open(neoConnectionString(ip, port))
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer func() {
+			if err := d.Close(); err != nil {
+				t.Error(err)
+			}
+		}()
+		m, err := migrate.NewWithDatabaseInstance("file://./examples/migrations", "neo4j", d)
+		if err != nil {
+			t.Fatal(err)
+		}
+		dt.TestMigrate(t, m, []byte("MATCH a RETURN a"))
+	})
+}

go.mod

@@ -31,6 +31,8 @@ require (
 	github.com/mattn/go-sqlite3 v1.10.0
 	github.com/morikuni/aec v0.0.0-20170113033406-39771216ff4c // indirect
 	github.com/nakagami/firebirdsql v0.0.0-20190310045651-3c02a58cfed8
+	github.com/neo4j-drivers/gobolt v1.7.4 // indirect
+	github.com/neo4j/neo4j-go-driver v1.7.4
 	github.com/pkg/errors v0.8.1 // indirect
 	github.com/satori/go.uuid v1.2.0 // indirect
 	github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24 // indirect