Add support for automatic, idempotent DB migrations

- On boot, the app now checks if the DB version matches its
  expected version and refuses to start if there are pending
  migrations to be run.
- The new `--upgrade` flag runs data migrations from the last
  recorded migration (in the settings table) to the latest one
  in the binary.
- Migrations are DB/arbitrary logic functions in .go files in
  internal/migrations.
- All migration functions are idempotent.

go.mod

@@ -20,6 +20,7 @@ require (
 	github.com/rhnvrm/simples3 v0.5.0
 	github.com/spf13/pflag v1.0.5
 	github.com/ssor/bom v0.0.0-20170718123548-6386211fdfcf // indirect
+	golang.org/x/mod v0.3.0
 	gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f // indirect
 	gopkg.in/volatiletech/null.v6 v6.0.0-20170828023728-0bef4e07ae1b
 	jaytaylor.com/html2text v0.0.0-20200220170450-61d9dc4d7195

go.sum

@@ -114,13 +114,19 @@ github.com/valyala/fasttemplate v1.0.1/go.mod h1:UQGH1tvbgY+Nz5t2n7tXsz52dQxojPU
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2 h1:VklqNMn3ovrHsnt90PveolxSbWFaJdECFbxSq0Mqo2M=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59 h1:3zb4D3T4G8jdExgVU/95+vQXfpEPiMdCaZgmGVxjNHM=
 golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/image v0.0.0-20191009234506-e7c1f5e7dbb8 h1:hVwzHzIUGRjiF7EcUjqNxk3NCfkPxbDKRdnNE1Rpg0U=
 golang.org/x/image v0.0.0-20191009234506-e7c1f5e7dbb8/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
+golang.org/x/mod v0.3.0 h1:RM4zey1++hCTbCVQfnWeKs9/IEsaBLA8vTkd0WVtmH4=
+golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190603091049-60506f45cf65 h1:+rhAzEzT3f4JtomfC371qB+0Ola2caSKcY69NUBZrRQ=
 golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859 h1:R/3boaszxrf1GEUWTVDzSKVwLmSJpwZ1yqXm8j0v2QI=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a h1:1BGLXjeY4akVXGgbC9HugT3Jv3hCI0z56oJR5vAMgBU=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -137,6 +143,9 @@ golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/appengine v1.6.5 h1:tycE03LOZYQNhDpS27tcQdAzLCVMaj7QT2SXxebnpCM=
 google.golang.org/appengine v1.6.5/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc=

init.go

@@ -70,6 +70,7 @@ func initFlags() {
 	f.StringSlice("config", []string{"config.toml"},
 		"path to one or more config files (will be merged in order)")
 	f.Bool("install", false, "run first time installation")
+	f.Bool("upgrade", false, "upgrade database to the current version")
 	f.Bool("version", false, "current version of the build")
 	f.Bool("new-config", false, "generate sample config file")
 	f.String("static-dir", "", "(optional) path to directory with static files")

install.go

@@ -17,29 +17,29 @@ import (
 
 // install runs the first time setup of creating and
 // migrating the database and creating the super user.
-func install(db *sqlx.DB, fs stuffbin.FileSystem, prompt bool) {
+func install(lastVer string, db *sqlx.DB, fs stuffbin.FileSystem, prompt bool) {
 	qMap, _ := initQueries(queryFilePath, db, fs, false)
 
 	fmt.Println("")
-	fmt.Println("** First time installation **")
+	fmt.Println("** first time installation **")
 	fmt.Printf("** IMPORTANT: This will wipe existing listmonk tables and types in the DB '%s' **",
 		ko.String("db.database"))
 	fmt.Println("")
 
 	if prompt {
 		var ok string
-		fmt.Print("Continue (y/n)?  ")
+		fmt.Print("continue (y/n)?  ")
 		if _, err := fmt.Scanf("%s", &ok); err != nil {
-			lo.Fatalf("Error reading value from terminal: %v", err)
+			lo.Fatalf("error reading value from terminal: %v", err)
 		}
 		if strings.ToLower(ok) != "y" {
-			fmt.Println("Installation cancelled.")
+			fmt.Println("install cancelled.")
 			return
 		}
 	}
 
 	// Migrate the tables.
-	err := installMigrate(db, fs)
+	err := installSchema(lastVer, db, fs)
 	if err != nil {
 		lo.Fatalf("Error migrating DB schema: %v", err)
 	}
@@ -133,19 +133,28 @@ func install(db *sqlx.DB, fs stuffbin.FileSystem, prompt bool) {
 	lo.Printf(`Run the program and access the dashboard at %s`, ko.MustString("app.address"))
 }
 
-// installMigrate executes the SQL schema and creates the necessary tables and types.
+// installSchema executes the SQL schema and creates the necessary tables and types.
-func installMigrate(db *sqlx.DB, fs stuffbin.FileSystem) error {
+func installSchema(curVer string, db *sqlx.DB, fs stuffbin.FileSystem) error {
 	q, err := fs.Read("/schema.sql")
 	if err != nil {
 		return err
 	}
 
-	_, err = db.Query(string(q))
+	if _, err := db.Exec(string(q)); err != nil {
-	if err != nil {
 		return err
 	}
 
-	return nil
+	// Insert the current migration version.
+	return recordMigrationVersion(curVer, db)
+}
+
+// recordMigrationVersion inserts the given version (of DB migration) into the
+// `migrations` array in the settings table.
+func recordMigrationVersion(ver string, db *sqlx.DB) error {
+	_, err := db.Exec(fmt.Sprintf(`INSERT INTO settings (key, value)
+	VALUES('migrations', '["%s"]'::JSONB)
+	ON CONFLICT (key) DO UPDATE SET value = settings.value || EXCLUDED.value`, ver))
+	return err
 }
 
 func newConfigFile() error {

internal/migrations/v0.4.0.go

@@ -0,0 +1,26 @@
+package migrations
+
+import (
+	"github.com/jmoiron/sqlx"
+	"github.com/knadh/koanf"
+	"github.com/knadh/stuffbin"
+)
+
+// V0_4_0 performs the DB migrations for v.0.4.0.
+func V0_4_0(db *sqlx.DB, fs stuffbin.FileSystem, ko *koanf.Koanf) error {
+	_, err := db.Exec(`
+	DO $$
+	BEGIN
+		IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'list_optin') THEN
+			CREATE TYPE list_optin AS ENUM ('single', 'double');
+		END IF;
+		IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'campaign_type') THEN
+			CREATE TYPE campaign_type AS ENUM ('regular', 'optin');
+		END IF;
+	END$$;
+
+	ALTER TABLE lists ADD COLUMN IF NOT EXISTS optin list_optin NOT NULL DEFAULT 'single';
+	ALTER TABLE campaigns ADD COLUMN IF NOT EXISTS type campaign_type DEFAULT 'regular';
+	`)
+	return err
+}

internal/migrations/v0.7.0.go

@@ -0,0 +1,112 @@
+package migrations
+
+import (
+	"github.com/jmoiron/sqlx"
+	"github.com/knadh/koanf"
+	"github.com/knadh/stuffbin"
+)
+
+// V0_7_0 performs the DB migrations for v.0.7.0.
+func V0_7_0(db *sqlx.DB, fs stuffbin.FileSystem, ko *koanf.Koanf) error {
+	// Check if the subscriber_status.blocklisted enum value exists. If not,
+	// it has to be created (for the change from blacklisted -> blocklisted).
+	var bl bool
+	if err := db.Get(&bl, `SELECT 'blocklisted' = ANY(ENUM_RANGE(NULL::subscriber_status)::TEXT[])`); err != nil {
+		return err
+	}
+
+	// If `blocklist` doesn't exist, add it to the subscriber_status enum,
+	// and update existing statuses to this value. Unfortunately, it's not possible
+	// to remove the enum value `blacklisted` (until PG10).
+	if !bl {
+		tx, err := db.Begin()
+		if err != nil {
+			return err
+		}
+		defer tx.Rollback()
+		if _, err := tx.Exec(`
+			-- Change the status column to text.
+			ALTER TABLE subscribers ALTER COLUMN status TYPE TEXT;
+
+			-- Change all statuses from 'blacklisted' to 'blocklisted'.
+			UPDATE subscribers SET status='blocklisted' WHERE status='blacklisted';
+	
+			-- Remove the old enum.
+			DROP TYPE subscriber_status CASCADE;
+
+			-- Create new enum with the new values.
+			CREATE TYPE subscriber_status AS ENUM ('enabled', 'disabled', 'blocklisted');
+
+			-- Change the text status column to the new enum.
+			ALTER TABLE subscribers ALTER COLUMN status TYPE subscriber_status
+				USING (status::subscriber_status);
+		`); err != nil {
+			return err
+		}
+		if err := tx.Commit(); err != nil {
+			return err
+		}
+	}
+
+	_, err := db.Exec(`
+	ALTER TABLE media DROP COLUMN IF EXISTS width,
+					  DROP COLUMN IF EXISTS height,
+					  ADD COLUMN IF NOT EXISTS provider TEXT NOT NULL DEFAULT '';
+
+	-- 'blacklisted' to 'blocklisted' ENUM rename is not possible (until pg10),
+	-- so just add the new value and ignore the old one.
+	
+	
+	CREATE TABLE IF NOT EXISTS settings (
+		key             TEXT NOT NULL UNIQUE,
+		value           JSONB NOT NULL DEFAULT '{}',
+		updated_at      TIMESTAMP WITH TIME ZONE DEFAULT NOW()
+	);
+	CREATE INDEX IF NOT EXISTS idx_settings_key ON settings(key);
+	
+	-- Insert default settings if the table is empty.
+	INSERT INTO settings (key, value) SELECT k, v::JSONB FROM (VALUES
+		('app.favicon_url', '""'),
+		('app.from_email', '"listmonk <noreply@listmonk.yoursite.com>"'),
+		('app.logo_url', '"http://localhost:9000/public/static/logo.png"'),
+		('app.concurrency', '10'),
+		('app.message_rate', '10'),
+		('app.batch_size', '1000'),
+		('app.max_send_errors', '1000'),
+		('app.notify_emails', '["admin1@mysite.com", "admin2@mysite.com"]'),
+		('privacy.allow_blocklist', 'true'),
+		('privacy.allow_export', 'true'),
+		('privacy.allow_wipe', 'true'),
+		('privacy.exportable', '["profile", "subscriptions", "campaign_views", "link_clicks"]'),
+		('upload.provider', '"filesystem"'),
+		('upload.filesystem.upload_path', '"uploads"'),
+		('upload.filesystem.upload_uri', '"/uploads"'),
+		('upload.s3.aws_access_key_id', '""'),
+		('upload.s3.aws_secret_access_key', '""'),
+		('upload.s3.aws_default_region', '"ap-south-b"'),
+		('upload.s3.bucket', '""'),
+		('upload.s3.bucket_domain', '""'),
+		('upload.s3.bucket_path', '"/"'),
+		('upload.s3.bucket_type', '"public"'),
+		('upload.s3.expiry', '"14d"'),
+		('smtp',
+			'[{"enabled":true, "host":"smtp.yoursite.com","port":25,"auth_protocol":"cram","username":"username","password":"password","hello_hostname":"","max_conns":10,"idle_timeout":"15s","wait_timeout":"5s","max_msg_retries":2,"tls_enabled":true,"tls_skip_verify":false,"email_headers":[]},
+			  {"enabled":false, "host":"smtp2.yoursite.com","port":587,"auth_protocol":"plain","username":"username","password":"password","hello_hostname":"","max_conns":10,"idle_timeout":"15s","wait_timeout":"5s","max_msg_retries":2,"tls_enabled":false,"tls_skip_verify":false,"email_headers":[]}]'),
+		('messengers', '[]')) vals(k, v) WHERE NOT EXISTS(SELECT * FROM settings LIMIT 1);
+	
+	`)
+	if err != nil {
+		return err
+	}
+
+	// `provider` in the media table is a new field. If there's provider config available
+	// and no provider value exists in the media table, set it.
+	prov := ko.String("upload.provider")
+	if prov != "" {
+		if _, err := db.Exec(`UPDATE media SET provider=$1 WHERE provider=''`, prov); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

main.go

@@ -57,6 +57,7 @@ var (
 )
 
 func init() {
+	lo.Println(buildString)
 	initFlags()
 
 	// Display version.
@@ -85,9 +86,17 @@ func init() {
 	// Installer mode? This runs before the SQL queries are loaded and prepared
 	// as the installer needs to work on an empty DB.
 	if ko.Bool("install") {
-		install(db, fs, !ko.Bool("yes"))
+		// Save the version of the last listed migration.
-		return
+		install(migList[len(migList)-1].version, db, fs, !ko.Bool("yes"))
+		os.Exit(0)
 	}
+	if ko.Bool("upgrade") {
+		upgrade(db, fs, !ko.Bool("yes"))
+		os.Exit(0)
+	}
+
+	// Before the queries are prepared, see if there are pending upgrades.
+	checkUpgrade(db)
 
 	// Load the SQL queries from the filesystem.
 	_, queries := initQueries(queryFilePath, db, fs, true)

schema.sql

@@ -128,7 +128,7 @@ DROP TABLE IF EXISTS media CASCADE;
 CREATE TABLE media (
     id               SERIAL PRIMARY KEY,
     uuid uuid        NOT NULL UNIQUE,
-    provider         TEXT NOT NULL,
+    provider         TEXT NOT NULL DEFAULT '',
     filename         TEXT NOT NULL,
     thumb            TEXT NOT NULL,
     created_at       TIMESTAMP WITH TIME ZONE DEFAULT NOW()