Remove message.owner_user_id field and implement db migrations

scnserver/db/impl/logs/database.go

@@ -4,6 +4,7 @@ import (
 	server "blackforestbytes.com/simplecloudnotifier"
 	"blackforestbytes.com/simplecloudnotifier/db/dbtools"
 	"blackforestbytes.com/simplecloudnotifier/db/schema"
+	"blackforestbytes.com/simplecloudnotifier/db/simplectx"
 	"context"
 	"database/sql"
 	"errors"
@@ -63,77 +64,93 @@ func (db *Database) DB() sq.DB {
 	return db.db
 }
 
-func (db *Database) Migrate(ctx context.Context) error {
-	ctx, cancel := context.WithTimeout(context.Background(), 24*time.Second)
-	defer cancel()
+func (db *Database) Migrate(outerctx context.Context) error {
+	innerctx, cancel := context.WithTimeout(outerctx, 24*time.Second)
+	tctx := simplectx.CreateSimpleContext(innerctx, cancel)
 
-	currschema, err := db.ReadSchema(ctx)
+	tx, err := tctx.GetOrCreateTransaction(db)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if tx.Status() == sq.TxStatusInitial || tx.Status() == sq.TxStatusActive {
+			err = tx.Rollback()
+			if err != nil {
+				log.Err(err).Msg("failed to rollback transaction")
+			}
+		}
+	}()
+
+	ppReInit := false
+
+	currschema, err := db.ReadSchema(tctx)
 	if err != nil {
 		return err
 	}
 
 	if currschema == 0 {
+		schemastr := schema.LogsSchema[schema.LogsSchemaVersion].SQL
+		schemahash := schema.LogsSchema[schema.LogsSchemaVersion].Hash
 
-		schemastr := schema.LogsSchema1
-
-		schemahash, err := sq.HashSqliteSchema(ctx, schemastr)
+		_, err = tx.Exec(tctx, schemastr, sq.PP{})
 		if err != nil {
 			return err
 		}
 
-		_, err = db.db.Exec(ctx, schemastr, sq.PP{})
+		err = db.WriteMetaInt(tctx, "schema", int64(schema.LogsSchemaVersion))
 		if err != nil {
 			return err
 		}
 
-		err = db.WriteMetaInt(ctx, "schema", 1)
+		err = db.WriteMetaString(tctx, "schema_hash", schemahash)
 		if err != nil {
 			return err
 		}
 
-		err = db.WriteMetaString(ctx, "schema_hash", schemahash)
+		ppReInit = true
+
+		currschema = schema.LogsSchemaVersion
+	}
+
+	if currschema == 1 {
+		schemHashDB, err := sq.HashSqliteDatabase(tctx, tx)
 		if err != nil {
 			return err
 		}
 
-		err = db.pp.Init(ctx) // Re-Init
+		schemaHashMeta, err := db.ReadMetaString(tctx, "schema_hash")
 		if err != nil {
 			return err
 		}
 
-		return nil
-
-	} else if currschema == 1 {
-
-		schemHashDB, err := sq.HashSqliteDatabase(ctx, db.db)
-		if err != nil {
-			return err
-		}
-
-		schemaHashMeta, err := db.ReadMetaString(ctx, "schema_hash")
-		if err != nil {
-			return err
-		}
-
-		schemHashAsset := schema.LogsHash1
-		if err != nil {
-			return err
-		}
-
-		if schemHashDB != langext.Coalesce(schemaHashMeta, "") || langext.Coalesce(schemaHashMeta, "") != schemHashAsset {
+		if schemHashDB != langext.Coalesce(schemaHashMeta, "") || langext.Coalesce(schemaHashMeta, "") != schema.LogsSchema[currschema].Hash {
 			log.Debug().Str("schemHashDB", schemHashDB).Msg("Schema (logs db)")
 			log.Debug().Str("schemaHashMeta", langext.Coalesce(schemaHashMeta, "")).Msg("Schema (logs db)")
-			log.Debug().Str("schemHashAsset", schemHashAsset).Msg("Schema (logs db)")
+			log.Debug().Str("schemaHashAsset", schema.LogsSchema[currschema].Hash).Msg("Schema (logs db)")
 			return errors.New("database schema does not match (logs db)")
 		} else {
 			log.Debug().Str("schemHash", schemHashDB).Msg("Verified Schema consistency (logs db)")
 		}
+	}
 
-		return nil // current
-	} else {
+	if currschema != schema.LogsSchemaVersion {
 		return errors.New(fmt.Sprintf("Unknown DB schema: %d", currschema))
 	}
 
+	err = tx.Commit()
+	if err != nil {
+		return err
+	}
+
+	if ppReInit {
+		log.Debug().Msg("Re-Init preprocessor")
+		err = db.pp.Init(outerctx) // Re-Init
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
 }
 
 func (db *Database) Ping(ctx context.Context) error {