Fix migrations

federationapi/storage/postgres/deltas/2022042812473400_addexpiresat.go

@@ -15,32 +15,28 @@
 package deltas
 
 import (
+	"context"
 	"database/sql"
 	"fmt"
 	"time"
 
-	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/gomatrixserverlib"
 )
 
-func LoadAddExpiresAt(m *sqlutil.Migrations) {
+func UpAddexpiresat(ctx context.Context, tx *sql.Tx) error {
-	m.AddMigration(upAddexpiresat, downAddexpiresat)
+	_, err := tx.ExecContext(ctx, "ALTER TABLE federationsender_queue_edus ADD COLUMN IF NOT EXISTS expires_at BIGINT NOT NULL DEFAULT 0;")
-}
-
-func upAddexpiresat(tx *sql.Tx) error {
-	_, err := tx.Exec("ALTER TABLE federationsender_queue_edus ADD COLUMN IF NOT EXISTS expires_at BIGINT NOT NULL DEFAULT 0;")
 	if err != nil {
 		return fmt.Errorf("failed to execute upgrade: %w", err)
 	}
-	_, err = tx.Exec("UPDATE federationsender_queue_edus SET expires_at = $1", gomatrixserverlib.AsTimestamp(time.Now().Add(time.Hour*24)))
+	_, err = tx.ExecContext(ctx, "UPDATE federationsender_queue_edus SET expires_at = $1", gomatrixserverlib.AsTimestamp(time.Now().Add(time.Hour*24)))
 	if err != nil {
 		return fmt.Errorf("failed to update queue_edus: %w", err)
 	}
 	return nil
 }
 
-func downAddexpiresat(tx *sql.Tx) error {
+func DownAddexpiresat(ctx context.Context, tx *sql.Tx) error {
-	_, err := tx.Exec("ALTER TABLE federationsender_queue_edus DROP COLUMN expires_at;")
+	_, err := tx.ExecContext(ctx, "ALTER TABLE federationsender_queue_edus DROP COLUMN expires_at;")
 	if err != nil {
 		return fmt.Errorf("failed to execute downgrade: %w", err)
 	}

federationapi/storage/postgres/queue_edus_table.go

@@ -19,6 +19,7 @@ import (
 	"database/sql"
 
 	"github.com/lib/pq"
+	"github.com/matrix-org/dendrite/federationapi/storage/postgres/deltas"
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/gomatrixserverlib"
@@ -90,7 +91,22 @@ func NewPostgresQueueEDUsTable(db *sql.DB) (s *queueEDUsStatements, err error) {
 		db: db,
 	}
 	_, err = s.db.Exec(queueEDUsSchema)
-	return s, err
+	if err != nil {
+		return s, err
+	}
+
+	m := sqlutil.NewMigrator(db)
+	m.AddMigrations(
+		sqlutil.Migration{
+			Version: "federationapi: add expiresat column",
+			Up:      deltas.UpAddexpiresat,
+		},
+	)
+	if err := m.Up(context.Background()); err != nil {
+		return s, err
+	}
+
+	return s, nil
 }
 
 func (s *queueEDUsStatements) Prepare() error {

federationapi/storage/sqlite3/deltas/2022042812473400_addexpiresat.go

@@ -15,25 +15,21 @@
 package deltas
 
 import (
+	"context"
 	"database/sql"
 	"fmt"
 	"time"
 
-	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/gomatrixserverlib"
 )
 
-func LoadAddExpiresAt(m *sqlutil.Migrations) {
+func UpAddexpiresat(ctx context.Context, tx *sql.Tx) error {
-	m.AddMigration(upAddexpiresat, downAddexpiresat)
+	_, err := tx.ExecContext(ctx, "ALTER TABLE federationsender_queue_edus RENAME TO federationsender_queue_edus_old;")
-}
-
-func upAddexpiresat(tx *sql.Tx) error {
-	_, err := tx.Exec("ALTER TABLE federationsender_queue_edus RENAME TO federationsender_queue_edus_old;")
 	if err != nil {
 		return fmt.Errorf("failed to rename table: %w", err)
 	}
 
-	_, err = tx.Exec(`
+	_, err = tx.ExecContext(ctx, `
 CREATE TABLE IF NOT EXISTS federationsender_queue_edus (
 	edu_type TEXT NOT NULL,
 	server_name TEXT NOT NULL,
@@ -47,7 +43,7 @@ CREATE UNIQUE INDEX IF NOT EXISTS federationsender_queue_edus_json_nid_idx
 	if err != nil {
 		return fmt.Errorf("failed to create new table: %w", err)
 	}
-	_, err = tx.Exec(`
+	_, err = tx.ExecContext(ctx, `
 INSERT
     INTO federationsender_queue_edus (
         edu_type, server_name, json_nid, expires_at
@@ -56,15 +52,15 @@ INSERT
 	if err != nil {
 		return fmt.Errorf("failed to update queue_edus: %w", err)
 	}
-	_, err = tx.Exec("UPDATE federationsender_queue_edus SET expires_at = $1", gomatrixserverlib.AsTimestamp(time.Now().Add(time.Hour*24)))
+	_, err = tx.ExecContext(ctx, "UPDATE federationsender_queue_edus SET expires_at = $1", gomatrixserverlib.AsTimestamp(time.Now().Add(time.Hour*24)))
 	if err != nil {
 		return fmt.Errorf("failed to update queue_edus: %w", err)
 	}
 	return nil
 }
 
-func downAddexpiresat(tx *sql.Tx) error {
+func DownAddexpiresat(ctx context.Context, tx *sql.Tx) error {
-	_, err := tx.Exec("ALTER TABLE federationsender_queue_edus DROP COLUMN expires_at;")
+	_, err := tx.ExecContext(ctx, "ALTER TABLE federationsender_queue_edus DROP COLUMN expires_at;")
 	if err != nil {
 		return fmt.Errorf("failed to rename table: %w", err)
 	}

federationapi/storage/sqlite3/queue_edus_table.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/matrix-org/dendrite/federationapi/storage/sqlite3/deltas"
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/gomatrixserverlib"
@@ -91,7 +92,22 @@ func NewSQLiteQueueEDUsTable(db *sql.DB) (s *queueEDUsStatements, err error) {
 		db: db,
 	}
 	_, err = db.Exec(queueEDUsSchema)
-	return s, err
+	if err != nil {
+		return s, err
+	}
+
+	m := sqlutil.NewMigrator(db)
+	m.AddMigrations(
+		sqlutil.Migration{
+			Version: "federationapi: add expiresat column",
+			Up:      deltas.UpAddexpiresat,
+		},
+	)
+	if err := m.Up(context.Background()); err != nil {
+		return s, err
+	}
+
+	return s, nil
 }
 
 func (s *queueEDUsStatements) Prepare() error {