Initial migrations

actions/context.go

@@ -0,0 +1,14 @@
+package actions
+
+import (
+	"github.com/scoiatael/archai/persistence"
+)
+
+type Context interface {
+	Persistence() persistence.Provider
+	Migrations() map[string]persistence.Migration
+}
+
+type Action interface {
+	Run(Context) error
+}

actions/migrate.go

@@ -0,0 +1,35 @@
+package actions
+
+import (
+	"log"
+)
+
+// Migrate implements Action interface
+type Migrate struct{}
+
+// Run all migrations
+func (a Migrate) Run(c Context) error {
+	persistenceProvider := c.Persistence()
+	migrationSession, err := persistenceProvider.MigrationSession()
+	if err != nil {
+		return err
+	}
+	for name, m := range c.Migrations() {
+		shouldRun, err := migrationSession.ShouldRunMigration(name)
+		if err != nil {
+			return err
+		}
+		if shouldRun {
+			log.Println("Executing ", name)
+			err = m.Run(migrationSession)
+			if err != nil {
+				return err
+			}
+			err = migrationSession.DidRunMigration(name)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}

config.go

@@ -0,0 +1,26 @@
+package main
+
+import (
+	"github.com/scoiatael/archai/actions"
+	"github.com/scoiatael/archai/persistence"
+)
+
+type Config struct {
+	keyspace string
+	hosts    []string
+	actions  []actions.Action
+}
+
+func (c Config) Migrations() map[string]persistence.Migration {
+	m := make(map[string]persistence.Migration)
+	m["create_events_table"] = persistence.CreateEventsTable
+	return m
+}
+
+func (c Config) Persistence() persistence.Provider {
+	hosts := make([]string, 1)
+	hosts[0] = "127.0.0.1"
+	provider := persistence.CassandraProvider{Hosts: hosts,
+		Keyspace: c.keyspace}
+	return &provider
+}

main.go

@@ -0,0 +1,21 @@
+package main
+
+import (
+	"os"
+
+	"github.com/scoiatael/archai/actions"
+	"github.com/urfave/cli"
+)
+
+func main() {
+	app := cli.NewApp()
+	app.Name = "archai"
+	app.Usage = "eventstore replacement"
+	app.Version = Version
+	app.Action = func(c *cli.Context) error {
+		config := Config{keyspace: "archai_test"}
+		return actions.Migrate{}.Run(config)
+	}
+
+	app.Run(os.Args)
+}

persistence/event.go

@@ -0,0 +1,4 @@
+package persistence
+
+type Event struct {
+}

persistence/migration_session.go

@@ -0,0 +1,51 @@
+package persistence
+
+import (
+	"fmt"
+
+	"log"
+
+	"github.com/gocql/gocql"
+)
+
+type MigrationSession interface {
+	ShouldRunMigration(string) (bool, error)
+	DidRunMigration(string) error
+	Query(string) error
+}
+
+type CassandraMigrationSession struct {
+	session *gocql.Session
+}
+
+func (sess *CassandraMigrationSession) Query(query string) error {
+	return sess.session.Query(query).Exec()
+}
+
+const migrationTable = "archai_migrations"
+
+var createMigrationTable = fmt.Sprintf(`
+    CREATE TABLE IF NOT EXISTS %s (
+        name VARCHAR,
+        PRIMARY KEY (name)
+    )
+`, migrationTable)
+
+var findMigration = fmt.Sprintf(`SELECT name FROM %s WHERE name = ? LIMIT 1`, migrationTable)
+
+var insertMigration = fmt.Sprintf(`INSERT INTO %s (name) VALUES (?)`, migrationTable)
+
+func (sess *CassandraMigrationSession) ShouldRunMigration(name string) (bool, error) {
+	if err := sess.session.Query(createMigrationTable).Exec(); err != nil {
+		return false, err
+	}
+	log.Println("Looking for migration ", name)
+	iter := sess.session.Query(findMigration, name).Iter()
+	found := iter.Scan(nil)
+	err := iter.Close()
+	return !found, err
+}
+
+func (sess *CassandraMigrationSession) DidRunMigration(name string) error {
+	return sess.session.Query(insertMigration, name).Exec()
+}

persistence/migrations.go

@@ -0,0 +1,24 @@
+package persistence
+
+type Migration interface {
+	Run(MigrationSession) error
+}
+
+type SimpleMigration struct {
+	Query string
+}
+
+func (simpleMigration SimpleMigration) Run(session MigrationSession) error {
+	err := session.Query(simpleMigration.Query)
+	return err
+}
+
+var CreateEventsTable = SimpleMigration{Query: `
+	CREATE TABLE IF NOT EXISTS events (
+		id TIMEUUID,
+		blob BLOB,
+		stream VARCHAR,
+		meta MAP<TEXT, TEXT>,
+		PRIMARY KEY (stream, id)
+	)
+`}

persistence/provider.go

@@ -0,0 +1,46 @@
+package persistence
+
+import (
+	"fmt"
+
+	"github.com/gocql/gocql"
+)
+
+type Provider interface {
+	Session() (Session, error)
+	MigrationSession() (MigrationSession, error)
+}
+
+type CassandraProvider struct {
+	Hosts    []string
+	Keyspace string
+}
+
+func (cp *CassandraProvider) newCluster() *gocql.ClusterConfig {
+	return gocql.NewCluster(cp.Hosts...)
+}
+
+func (cp *CassandraProvider) Session() (Session, error) {
+	cluster := cp.newCluster()
+	cluster.Keyspace = cp.Keyspace
+	cluster.Consistency = gocql.Quorum
+	sess, err := cluster.CreateSession()
+	return &CassandraSession{session: sess}, err
+}
+
+func (cp *CassandraProvider) MigrationSession() (MigrationSession, error) {
+	cluster := cp.newCluster()
+	cluster.Consistency = gocql.All
+	sess, err := cluster.CreateSession()
+	if err != nil {
+		return &CassandraMigrationSession{}, err
+	}
+	err = sess.Query(fmt.Sprintf(`CREATE KEYSPACE IF NOT EXISTS %s WITH replication = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 };`, cp.Keyspace)).Exec()
+	if err != nil {
+		return &CassandraMigrationSession{}, err
+	}
+	cluster.Keyspace = cp.Keyspace
+	sess, err = cluster.CreateSession()
+
+	return &CassandraMigrationSession{session: sess}, err
+}

persistence/session.go

@@ -0,0 +1,25 @@
+package persistence
+
+import (
+	"github.com/gocql/gocql"
+)
+
+type Session interface {
+	WriteEvent(string, []byte, map[string]string) error
+	ReadEvents(string, string, int) []Event
+}
+
+type CassandraSession struct {
+	session *gocql.Session
+}
+
+const insertEvent = `INSERT INTO events (id, stream, blob, meta) VALUES (now(), ?, ?, ?)`
+
+func (sess *CassandraSession) WriteEvent(stream string, blob []byte, meta map[string]string) error {
+	return sess.session.Query(insertEvent, stream, blob, meta).Exec()
+}
+
+func (sess *CassandraSession) ReadEvents(stream string, cursor string, amount int) []Event {
+	events := make([]Event, 0)
+	return events
+}

version.go

@@ -0,0 +1,3 @@
+package main
+
+const Version = "0.1.0"