fix(reset): ensure pgmq extension is fully rebuilt during remote resets

internal/db/reset/reset.go

@@ -245,15 +245,42 @@ func resetRemote(ctx context.Context, version string, config pgconn.Config, fsys
 	} else if !shouldReset {
 		return errors.New(context.Canceled)
 	}
+
 	fmt.Fprintln(os.Stderr, "Resetting remote database"+toLogMessage(version))
+
 	conn, err := utils.ConnectByConfigStream(ctx, config, io.Discard, options...)
 	if err != nil {
 		return err
 	}
 	defer conn.Close(context.Background())
+	if err := resetPgmqExtension(ctx, conn); err != nil {
+		return err
+	}
 	return down.ResetAll(ctx, version, conn, fsys)
 }
 
+func resetPgmqExtension(ctx context.Context, conn *pgx.Conn) error {
+	var exists bool
+	if err := conn.QueryRow(ctx, "SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'pgmq')").Scan(&exists); err != nil {
+		return errors.Errorf("failed to check pgmq extension: %w", err)
+	}
+	if !exists {
+		return nil
+	}
+	if _, err := conn.Exec(ctx, "DROP EXTENSION IF EXISTS pgmq CASCADE"); err != nil {
+		return errors.Errorf("failed to drop pgmq extension: %w", err)
+	}
+	policy := start.NewBackoffPolicy(ctx, 10*time.Second)
+	createExtension := func() error {
+		_, err := conn.Exec(ctx, "CREATE EXTENSION IF NOT EXISTS pgmq")
+		return err
+	}
+	if err := backoff.Retry(createExtension, policy); err != nil {
+		return errors.Errorf("failed to recreate pgmq extension: %w", err)
+	}
+	return nil
+}
+
 func LikeEscapeSchema(schemas []string) (result []string) {
 	// Treat _ as literal, * as any character
 	replacer := strings.NewReplacer("_", `\_`, "*", "%")