Initial commit

Basic skeleton and simple op

README.md

@@ -0,0 +1,36 @@
+# pg-roll
+
+:warning: Under development :warning:
+
+PostgreSQL zero-downtime migrations made easy.
+
+## Getting started (development)
+
+* Bring a development PostgreSQL up:
+
+```sh
+docker compose up
+```
+
+* Start a migration:
+
+```sh
+go run . start examples/01_create_tables.json
+```
+
+* Inspect the results:
+
+```sh
+psql postgres://localhost -U postgres
+```
+
+```sql
+\d+ public.*
+\d+ 01_create_tables.*
+```
+
+* Complete the migration:
+
+```sh
+go run . complete examples/01_create_tables.json
+```

cmd/complete.go

@@ -0,0 +1,41 @@
+package cmd
+
+import (
+	"fmt"
+	"path/filepath"
+	"pg-roll/pkg/migrations"
+	"strings"
+
+	"github.com/spf13/cobra"
+)
+
+var completeCmd = &cobra.Command{
+	Use:   "complete <file>",
+	Short: "Complete an ongoing migration with the operations present in the given file",
+	Long:  `TODO: Add long description`,
+	Args:  cobra.ExactArgs(1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		fileName := args[0]
+
+		m, err := migrations.New(cmd.Context(), PGURL)
+		if err != nil {
+			return err
+		}
+		defer m.Close()
+
+		ops, err := migrations.ReadMigrationFile(args[0])
+		if err != nil {
+			return fmt.Errorf("reading migration file: %w", err)
+		}
+
+		version := strings.TrimSuffix(filepath.Base(fileName), filepath.Ext(fileName))
+
+		err = m.Complete(cmd.Context(), version, ops)
+		if err != nil {
+			return err
+		}
+
+		fmt.Println("Migration successful!")
+		return nil
+	},
+}

cmd/root.go

@@ -0,0 +1,31 @@
+package cmd
+
+import (
+	"github.com/spf13/cobra"
+)
+
+var (
+	PGURL string
+)
+
+func init() {
+	rootCmd.PersistentFlags().StringVar(&PGURL, "postgres_url", "postgres://postgres:postgres@localhost?sslmode=disable", "Postgres URL")
+}
+
+var (
+	rootCmd = &cobra.Command{
+		Use:          "pg-roll",
+		Short:        "TODO: Add short description",
+		Long:         `TODO: Add long description`,
+		SilenceUsage: true,
+	}
+)
+
+// Execute executes the root command.
+func Execute() error {
+	// register subcommands
+	rootCmd.AddCommand(startCmd)
+	rootCmd.AddCommand(completeCmd)
+
+	return rootCmd.Execute()
+}

cmd/start.go

@@ -0,0 +1,41 @@
+package cmd
+
+import (
+	"fmt"
+	"path/filepath"
+	"pg-roll/pkg/migrations"
+	"strings"
+
+	"github.com/spf13/cobra"
+)
+
+var startCmd = &cobra.Command{
+	Use:   "start <file>",
+	Short: "Start a migration for the operations present in the given file",
+	Long:  `TODO: Add long description`,
+	Args:  cobra.ExactArgs(1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		fileName := args[0]
+
+		m, err := migrations.New(cmd.Context(), PGURL)
+		if err != nil {
+			return err
+		}
+		defer m.Close()
+
+		ops, err := migrations.ReadMigrationFile(args[0])
+		if err != nil {
+			return fmt.Errorf("reading migration file: %w", err)
+		}
+
+		version := strings.TrimSuffix(filepath.Base(fileName), filepath.Ext(fileName))
+
+		err = m.Start(cmd.Context(), version, ops)
+		if err != nil {
+			return err
+		}
+
+		fmt.Printf("Migration successful!, new version of the schema available under postgres '%s' schema\n", version)
+		return nil
+	},
+}

docker-compose.yml

@@ -0,0 +1,15 @@
+version: '3.8'
+services:
+  db:
+    image: postgres:14-alpine
+    restart: always
+    environment:
+      - POSTGRES_USER=postgres
+      - POSTGRES_PASSWORD=postgres
+    ports:
+      - '5432:5432'
+    volumes: 
+      - data:/var/lib/postgresql/data
+volumes:
+  data:
+    driver: local

examples/01_create_tables.json

@@ -0,0 +1,44 @@
+[
+    {
+        "create_table": {
+            "name": "customers",
+            "columns": [
+                {
+                    "name": "id",
+                    "type": "integer",
+                    "pk": true
+                },
+                {
+                    "name": "name",
+                    "type": "varchar(255)",
+                    "unique": true
+                },
+                {
+                    "name": "credit_card",
+                    "type": "text",
+                    "nullable": true
+                }
+            ]
+        }
+    },
+    {
+        "create_table": {
+            "name": "bills",
+            "columns": [
+                {
+                    "name": "id",
+                    "type": "integer",
+                    "pk": true
+                },
+                {
+                    "name": "date",
+                    "type": "time with time zone"
+                },
+                {
+                    "name": "quantity",
+                    "type": "integer"
+                }
+            ]
+        }
+    }
+]

go.mod

@@ -0,0 +1,13 @@
+module pg-roll
+
+go 1.20
+
+require (
+	github.com/lib/pq v1.10.9
+	github.com/spf13/cobra v1.7.0
+)
+
+require (
+	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/spf13/pflag v1.0.5 // indirect
+)

go.sum

@@ -0,0 +1,12 @@
+github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
+github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
+github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
+github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
+github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/spf13/cobra v1.7.0 h1:hyqWnYt1ZQShIddO5kBpj3vu05/++x6tJ6dg8EC572I=
+github.com/spf13/cobra v1.7.0/go.mod h1:uLxZILRyS/50WlhOIKD7W6V5bgeIt+4sICxh6uRMrb0=
+github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
+github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

main.go

@@ -0,0 +1,9 @@
+package main
+
+import (
+	"pg-roll/cmd"
+)
+
+func main() {
+	cmd.Execute()
+}

pkg/migrations/execute.go

@@ -0,0 +1,73 @@
+package migrations
+
+import (
+	"context"
+	"fmt"
+	"pg-roll/pkg/schema"
+	"strings"
+
+	"github.com/lib/pq"
+)
+
+// Start will apply the required changes to enable supporting the new schema version
+func (m *Migrations) Start(ctx context.Context, version string, ops Operations) error {
+	newSchema := schema.New()
+
+	// execute operations
+	for _, op := range ops {
+		err := op.Start(ctx, m.pgConn, newSchema)
+		if err != nil {
+			return fmt.Errorf("unable to execute start operation: %w", err)
+		}
+	}
+
+	// create schema for the new version
+	_, err := m.pgConn.ExecContext(ctx, fmt.Sprintf("CREATE SCHEMA IF NOT EXISTS %s", pq.QuoteIdentifier(version)))
+	if err != nil {
+		return err
+	}
+
+	// create views in the new schema
+	for name, table := range newSchema.Tables {
+		err = m.createView(ctx, version, name, table)
+		if err != nil {
+			return fmt.Errorf("unable to create view: %w", err)
+		}
+	}
+
+	return nil
+}
+
+// Complete will update the database schema to match the current version
+func (m *Migrations) Complete(ctx context.Context, version string, ops Operations) error {
+	// execute operations
+	for _, op := range ops {
+		err := op.Complete(ctx, m.pgConn)
+		if err != nil {
+			return fmt.Errorf("unable to execute complete operation: %w", err)
+		}
+	}
+
+	// TODO: once we have state, drop views for previous versions
+
+	return nil
+}
+
+// create view creates a view for the new version of the schema
+func (m *Migrations) createView(ctx context.Context, version string, name string, table schema.Table) error {
+	columns := make([]string, 0, len(table.Columns))
+	for k, v := range table.Columns {
+		columns = append(columns, fmt.Sprintf("%s AS %s", pq.QuoteIdentifier(k), pq.QuoteIdentifier(v.Name)))
+	}
+
+	_, err := m.pgConn.ExecContext(ctx,
+		fmt.Sprintf("CREATE OR REPLACE VIEW %s.%s AS SELECT %s FROM %s",
+			pq.QuoteIdentifier(version),
+			pq.QuoteIdentifier(name),
+			strings.Join(columns, ","),
+			pq.QuoteIdentifier(table.Name)))
+	if err != nil {
+		return err
+	}
+	return nil
+}

pkg/migrations/migrations.go

@@ -0,0 +1,43 @@
+package migrations
+
+import (
+	"context"
+	"database/sql"
+	"pg-roll/pkg/schema"
+
+	_ "github.com/lib/pq"
+)
+
+type Migrations struct {
+	pgConn *sql.DB // TODO abstract sql connection
+}
+
+type Operation interface {
+	// Start will apply the required changes to enable supporting the new schema
+	// version in the database (through a view)
+	// update the given views to expose the new schema version
+	Start(ctx context.Context, conn *sql.DB, v *schema.Schema) error
+
+	// Complete will update the database schema to match the current version
+	// after calling Execute.
+	// this method should be called once the previous version is no longer used
+	Complete(ctx context.Context, conn *sql.DB) error
+
+	// TODO
+	// Rollback(ctx context.Context, conn *sql.DB) error
+}
+
+func New(ctx context.Context, pgURL string) (*Migrations, error) {
+	conn, err := sql.Open("postgres", pgURL)
+	if err != nil {
+		return nil, err
+	}
+
+	return &Migrations{
+		pgConn: conn,
+	}, nil
+}
+
+func (m *Migrations) Close() error {
+	return m.pgConn.Close()
+}

pkg/migrations/op_common.go

@@ -0,0 +1,79 @@
+package migrations
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+)
+
+type Operations []Operation
+
+func temporaryName(name string) string {
+	return "_pgroll_new_" + name
+}
+
+func ReadMigrationFile(file string) ([]Operation, error) {
+	// read operations from file
+	jsonFile, err := os.Open(file)
+	if err != nil {
+		return nil, err
+	}
+	defer jsonFile.Close()
+
+	// read our opened xmlFile as a byte array.
+	byteValue, err := io.ReadAll(jsonFile)
+	if err != nil {
+		return nil, err
+	}
+
+	ops := Operations{}
+	err = json.Unmarshal(byteValue, &ops)
+	if err != nil {
+		return nil, err
+	}
+
+	return ops, nil
+}
+
+func (v *Operations) UnmarshalJSON(data []byte) error {
+	var tmp []map[string]json.RawMessage
+	if err := json.Unmarshal(data, &tmp); err != nil {
+		return nil
+	}
+
+	if len(tmp) == 0 {
+		*v = Operations{}
+		return nil
+	}
+
+	ops := make([]Operation, len(tmp))
+	for i, opObj := range tmp {
+		var opName string
+		var logBody json.RawMessage
+		if len(opObj) != 1 {
+			return fmt.Errorf("invalid migration: %v", opObj)
+		}
+		for k, v := range opObj {
+			opName = k
+			logBody = v
+		}
+
+		var item Operation
+		switch opName {
+		case "create_table":
+			item = &OpCreateTable{}
+		default:
+			return fmt.Errorf("unknown migration type: %v", opName)
+		}
+
+		if err := json.Unmarshal(logBody, item); err != nil {
+			return fmt.Errorf("decode migration [%v]: %w", opName, err)
+		}
+
+		ops[i] = item
+	}
+
+	*v = ops
+	return nil
+}

pkg/migrations/op_create_table.go

@@ -0,0 +1,82 @@
+package migrations
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"pg-roll/pkg/schema"
+
+	"github.com/lib/pq"
+)
+
+var _ Operation = (*OpCreateTable)(nil)
+
+type OpCreateTable struct {
+	Name    string   `json:"name"`
+	Columns []column `json:"columns"`
+}
+
+type column struct {
+	Name       string         `json:"name"`
+	Type       string         `json:"type"`
+	Nullable   bool           `json:"nullable"`
+	Unique     bool           `json:"unique"`
+	PrimaryKey bool           `json:"pk"`
+	Default    sql.NullString `json:"default"`
+}
+
+func (o *OpCreateTable) Start(ctx context.Context, conn *sql.DB, s *schema.Schema) error {
+	tempName := temporaryName(o.Name)
+	_, err := conn.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s (%s)",
+		pq.QuoteIdentifier(tempName),
+		columnsToSQL(o.Columns)))
+	if err != nil {
+		return err
+	}
+
+	columns := make(map[string]schema.Column, len(o.Columns))
+	for _, col := range o.Columns {
+		columns[col.Name] = schema.Column{
+			Name: col.Name,
+		}
+	}
+
+	s.Tables[o.Name] = schema.Table{
+		Name:    tempName,
+		Columns: columns,
+	}
+
+	return nil
+}
+
+func columnsToSQL(cols []column) string {
+	var sql string
+	for i, col := range cols {
+		if i > 0 {
+			sql += ", "
+		}
+		sql += fmt.Sprintf("%s %s", pq.QuoteIdentifier(col.Name), col.Type)
+
+		if col.PrimaryKey {
+			sql += " PRIMARY KEY"
+		}
+		if col.Unique {
+			sql += " UNIQUE"
+		}
+		if !col.Nullable {
+			sql += " NOT NULL"
+		}
+		if col.Default.Valid {
+			sql += fmt.Sprintf(" DEFAULT %s", col.Default.String)
+		}
+	}
+	return sql
+}
+
+func (o *OpCreateTable) Complete(ctx context.Context, conn *sql.DB) error {
+	tempName := temporaryName(o.Name)
+	_, err := conn.ExecContext(ctx, fmt.Sprintf("ALTER TABLE %s RENAME TO %s",
+		pq.QuoteIdentifier(tempName),
+		pq.QuoteIdentifier(o.Name)))
+	return err
+}

pkg/schema/schema.go

@@ -0,0 +1,24 @@
+package schema
+
+// XXX we create a view of the schema with the minimum required for us to
+// know how to execute migrations and build views for the new schema version.
+// As of now this is just the table names and column names.
+
+func New() *Schema {
+	return &Schema{
+		Tables: make(map[string]Table),
+	}
+}
+
+type Schema struct {
+	Tables map[string]Table // virtual name -> table mapping
+}
+
+type Table struct {
+	Name    string            // actual name in postgres
+	Columns map[string]Column // virtual name -> column mapping
+}
+
+type Column struct {
+	Name string // actual name in postgres
+}