Store compiled jobs with GORM in PostgreSQL

cmd/psql-gorm-poc/main.go

@@ -66,7 +66,7 @@ func main() {
 		case float64:
 			theInt = int(v)
 		case int:
-			theInt = int(v)
+			theInt = v
 		default:
 			log.Panic().Interface("value", v).Msg("unexpected type in JSON")
 		}

internal/manager/persistence/db.go

@@ -23,10 +23,11 @@ package persistence
 
 import (
 	"context"
-	"database/sql"
 	"fmt"
 
 	"github.com/rs/zerolog/log"
+	"gorm.io/driver/postgres"
+	"gorm.io/gorm"
 	_ "modernc.org/sqlite"
 
 	"gitlab.com/blender/flamenco-ng-poc/internal/manager/job_compilers"
@@ -34,30 +35,28 @@ import (
 )
 
 // TODO : have this configurable from the CLI.
-const dbURI = "flamenco-manager.sqlite"
+const dbDSN = "host=localhost user=flamenco password=flamenco dbname=flamenco TimeZone=Europe/Amsterdam"
 
 // DB provides the database interface.
 type DB struct {
-	sqldb *sql.DB
+	gormDB *gorm.DB
 }
 
 func OpenDB(ctx context.Context) (*DB, error) {
-	log.Info().Str("uri", dbURI).Msg("opening database")
+	return openDB(ctx, dbDSN)
-	return openDB(ctx, dbURI)
 }
 
 func openDB(ctx context.Context, uri string) (*DB, error) {
-	sqldb, err := sql.Open("sqlite", uri)
+	// TODO: don't log the password.
-	if err != nil {
+	log.Info().Str("dsn", dbDSN).Msg("opening database")
-		return nil, fmt.Errorf("unable to open database: %w", err)
-	}
 
-	if err := sqldb.PingContext(ctx); err != nil {
+	gormDB, err := gorm.Open(postgres.Open(dbDSN), &gorm.Config{})
-		return nil, fmt.Errorf("error accessing database %s: %w", dbURI, err)
+	if err != nil {
+		log.Panic().Err(err).Msg("failed to connect database")
 	}
 
 	db := DB{
-		sqldb: sqldb,
+		gormDB: gormDB,
 	}
 	if err := db.migrate(); err != nil {
 		return nil, err
@@ -67,95 +66,46 @@ func openDB(ctx context.Context, uri string) (*DB, error) {
 }
 
 func (db *DB) StoreJob(ctx context.Context, authoredJob job_compilers.AuthoredJob) error {
-	tx, err := db.sqldb.BeginTx(ctx, nil)
-	if err != nil {
-		return err
-	}
-	defer tx.Rollback()
 
-	_, err = tx.ExecContext(ctx,
+	dbJob := Job{
-		`INSERT INTO jobs (uuid, name, jobType, priority) VALUES (?, ?, ?, ?)`,
+		UUID:     authoredJob.JobID,
-		authoredJob.JobID, authoredJob.Name, authoredJob.JobType, authoredJob.Priority,
+		Name:     authoredJob.Name,
-	)
+		JobType:  authoredJob.JobType,
-	if err != nil {
+		Priority: int8(authoredJob.Priority),
-		return err
+		Settings: JobSettings(authoredJob.Settings),
+		Metadata: JobMetadata(authoredJob.Metadata),
 	}
 
-	for key, value := range authoredJob.Settings {
+	tx := db.gormDB.Create(&dbJob)
-		_, err := tx.ExecContext(ctx,
+	if tx.Error != nil {
-			`INSERT INTO job_settings (job_id, key, value) VALUES (?, ?, ?)`,
+		return fmt.Errorf("error storing job: %v", tx.Error)
-			authoredJob.JobID, key, value,
-		)
-		if err != nil {
-			return err
-		}
 	}
 
-	for key, value := range authoredJob.Metadata {
+	return nil
-		_, err := tx.ExecContext(ctx,
-			`INSERT INTO job_metadata (job_id, key, value) VALUES (?, ?, ?)`,
-			authoredJob.JobID, key, value,
-		)
-		if err != nil {
-			return err
-		}
-	}
-
-	return tx.Commit()
 }
 
 func (db *DB) FetchJob(ctx context.Context, jobID string) (*api.Job, error) {
-	job := api.Job{}
+	dbJob := Job{}
-
+	findResult := db.gormDB.First(&dbJob, "uuid = ?", jobID)
-	err := db.sqldb.
+	if findResult.Error != nil {
-		QueryRowContext(ctx, `SELECT * FROM jobs j where j.uuid=?`, jobID).
+		return nil, findResult.Error
-		Scan(&job.Id, &job.Name, &job.Type, &job.Priority, &job.Created, &job.Updated)
-	if err != nil {
-		return nil, err
 	}
 
-	var settings api.JobSettings
+	apiJob := api.Job{
-	rows, err := db.sqldb.QueryContext(ctx, "SELECT key, value FROM job_settings WHERE job_id=?", jobID)
+		SubmittedJob: api.SubmittedJob{
-	if err != nil {
+			Name:     dbJob.Name,
-		return nil, err
+			Priority: int(dbJob.Priority),
-	}
+			Type:     dbJob.JobType,
-	defer rows.Close()
+		},
-	for rows.Next() {
+
-		var key, value string
+		Id:      dbJob.UUID,
-		if err := rows.Scan(&key, &value); err != nil {
+		Created: dbJob.CreatedAt,
-			return nil, err
+		Updated: dbJob.UpdatedAt,
-		}
+		Status:  api.JobStatus(dbJob.Status),
-		settings.AdditionalProperties[key] = value
-	}
-	if err := rows.Close(); err != nil {
-		return nil, err
-	}
-	if err := rows.Err(); err != nil {
-		return nil, err
 	}
 
-	var metadata api.JobMetadata
+	apiJob.Settings.AdditionalProperties = dbJob.Settings
-	rows, err = db.sqldb.QueryContext(ctx, "SELECT key, value FROM job_metadata WHERE job_id=?", jobID)
+	apiJob.Metadata.AdditionalProperties = dbJob.Metadata
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-	for rows.Next() {
-		var key, value string
-		if err := rows.Scan(&key, &value); err != nil {
-			return nil, err
-		}
-		metadata.AdditionalProperties[key] = value
-	}
-	if err := rows.Close(); err != nil {
-		return nil, err
-	}
-	if err := rows.Err(); err != nil {
-		return nil, err
-	}
 
-	job.Settings = &settings
+	return &apiJob, nil
-	job.Metadata = &metadata
-
-	return &job, nil
 }

internal/manager/persistence/db_migration.go

@@ -21,48 +21,12 @@ package persistence
  * ***** END GPL LICENSE BLOCK ***** */
 
 import (
-	"embed"
 	"fmt"
-
-	"github.com/golang-migrate/migrate/v4"
-	"github.com/golang-migrate/migrate/v4/database/sqlite"
-	"github.com/golang-migrate/migrate/v4/source"
-	"github.com/golang-migrate/migrate/v4/source/iofs"
 )
 
-func init() {
-	source.Register("embedfs", &EmbedFS{})
-}
-
-type EmbedFS struct {
-	iofs.PartialDriver
-}
-
-//go:embed migrations/*.sql
-var embedFS embed.FS
-
 func (db *DB) migrate() error {
-	driver, err := sqlite.WithInstance(db.sqldb, &sqlite.Config{})
+	if err := db.gormDB.AutoMigrate(&Job{}); err != nil {
-	if err != nil {
+		return fmt.Errorf("failed to automigrate database: %v", err)
-		return fmt.Errorf("cannot create migration driver: %w", err)
-	}
-
-	m, err := migrate.NewWithDatabaseInstance("embedfs://", "sqlite", driver)
-	if err != nil {
-		return fmt.Errorf("cannot create migration instance: %w", err)
-	}
-
-	err = m.Up()
-	if err != nil && err != migrate.ErrNoChange {
-		return fmt.Errorf("cannot migrate database: %w", err)
 	}
 	return nil
 }
-
-func (f *EmbedFS) Open(url string) (source.Driver, error) {
-	nf := &EmbedFS{}
-	if err := nf.Init(embedFS, "migrations"); err != nil {
-		return nil, err
-	}
-	return nf, nil
-}

internal/manager/persistence/migrations/000001_create_jobs_table.down.sql

@@ -1 +0,0 @@
-DROP TABLE jobs;

internal/manager/persistence/migrations/000001_create_jobs_table.up.sql

@@ -1,24 +0,0 @@
-CREATE TABLE jobs (
-  uuid CHAR(36) PRIMARY KEY NOT NULL,
-  name VARCHAR(255) NOT NULL,
-  jobType VARCHAR(255) NOT NULL,
-  priority INT NOT NULL,
-  created TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,
-  updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL
-);
-
-CREATE TABLE job_settings (
-  job_id CHAR(36),
-  key VARCHAR(255),
-  value TEXT,
-  FOREIGN KEY(job_id) REFERENCES jobs(uuid)
-);
-CREATE UNIQUE INDEX job_settings_index ON job_settings(job_id, key);
-
-CREATE TABLE job_metadata (
-  job_id CHAR(36),
-  key VARCHAR(255),
-  value TEXT,
-  FOREIGN KEY(job_id) REFERENCES jobs(uuid)
-);
-CREATE UNIQUE INDEX job_metadata_index ON job_metadata(job_id, key);

internal/manager/persistence/models.go

@@ -0,0 +1,68 @@
+package persistence
+
+/* ***** BEGIN GPL LICENSE BLOCK *****
+ *
+ * Original Code Copyright (C) 2022 Blender Foundation.
+ *
+ * This file is part of Flamenco.
+ *
+ * Flamenco is free software: you can redistribute it and/or modify it under
+ * the terms of the GNU General Public License as published by the Free Software
+ * Foundation, either version 3 of the License, or (at your option) any later
+ * version.
+ *
+ * Flamenco is distributed in the hope that it will be useful, but WITHOUT ANY
+ * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+ * A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * Flamenco.  If not, see <https://www.gnu.org/licenses/>.
+ *
+ * ***** END GPL LICENSE BLOCK ***** */
+
+import (
+	"database/sql/driver"
+	"encoding/json"
+	"errors"
+
+	"gorm.io/gorm"
+)
+
+type Job struct {
+	gorm.Model
+	UUID string `gorm:"type:char(36)"`
+
+	Name     string `gorm:"type:varchar(64)"`
+	JobType  string `gorm:"type:varchar(32)"`
+	Priority int8   `gorm:"type:smallint"`
+	Status   string `gorm:"type:varchar(32)"` // See JobStatusXxxx consts in openapi_types.gen.go
+
+	Settings JobSettings `gorm:"type:jsonb"`
+	Metadata JobMetadata `gorm:"type:jsonb"`
+}
+
+type JobSettings map[string]interface{}
+
+func (js JobSettings) Value() (driver.Value, error) {
+	return json.Marshal(js)
+}
+func (js *JobSettings) Scan(value interface{}) error {
+	b, ok := value.([]byte)
+	if !ok {
+		return errors.New("type assertion to []byte failed")
+	}
+	return json.Unmarshal(b, &js)
+}
+
+type JobMetadata map[string]string
+
+func (js JobMetadata) Value() (driver.Value, error) {
+	return json.Marshal(js)
+}
+func (js *JobMetadata) Scan(value interface{}) error {
+	b, ok := value.([]byte)
+	if !ok {
+		return errors.New("type assertion to []byte failed")
+	}
+	return json.Unmarshal(b, &js)
+}

internal/manager/persistence/types.go

@@ -1,44 +0,0 @@
-package persistence
-
-/* ***** BEGIN GPL LICENSE BLOCK *****
- *
- * Original Code Copyright (C) 2022 Blender Foundation.
- *
- * This file is part of Flamenco.
- *
- * Flamenco is free software: you can redistribute it and/or modify it under
- * the terms of the GNU General Public License as published by the Free Software
- * Foundation, either version 3 of the License, or (at your option) any later
- * version.
- *
- * Flamenco is distributed in the hope that it will be useful, but WITHOUT ANY
- * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
- * A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License along with
- * Flamenco.  If not, see <https://www.gnu.org/licenses/>.
- *
- * ***** END GPL LICENSE BLOCK ***** */
-
-import "time"
-
-type Job struct {
-	ID       string // UUID
-	Name     string
-	JobType  string
-	Priority int8
-	Created  time.Time
-	Updated  time.Time
-}
-
-type JobSetting struct {
-	JobID string
-	Key   string
-	Value string
-}
-
-type JobMetadata struct {
-	JobID string
-	Key   string
-	Value string
-}