upgrade method

@override Future<Schema?> upgrade(Schema? fromSchema, List<Migration> withMigrations, { bool temporary = false, })

Implementation

@override
Future<Schema?> upgrade(
  Schema? fromSchema,
  List<Migration> withMigrations, {
  bool temporary = false,
}) async {
  final connection = await getDatabaseConnection();

  Schema? schema = fromSchema;

  await connection.transaction((ctx) async {
    final transactionStore =
        PostgreSQLPersistentStore._transactionProxy(this, ctx);
    await _createVersionTableIfNecessary(ctx, temporary);

    withMigrations.sort((m1, m2) => m1.version!.compareTo(m2.version!));

    for (final migration in withMigrations) {
      migration.database =
          SchemaBuilder(transactionStore, schema, isTemporary: temporary);
      migration.database.store = transactionStore;

      final existingVersionRows = await ctx.query(
        "SELECT versionNumber, dateOfUpgrade FROM $versionTableName WHERE versionNumber >= @v:int4",
        substitutionValues: {"v": migration.version},
      );
      if (existingVersionRows.isNotEmpty) {
        final date = existingVersionRows.first.last;
        throw MigrationException(
          "Trying to upgrade database to version ${migration.version}, but that migration has already been performed on $date.",
        );
      }

      logger.info("Applying migration version ${migration.version}...");
      await migration.upgrade();

      for (final cmd in migration.database.commands) {
        logger.info("\t$cmd");
        await ctx.execute(cmd);
      }

      logger.info(
        "Seeding data from migration version ${migration.version}...",
      );
      await migration.seed();

      await ctx.execute(
        "INSERT INTO $versionTableName (versionNumber, dateOfUpgrade) VALUES (${migration.version}, '${DateTime.now().toUtc().toIso8601String()}')",
      );

      logger
          .info("Applied schema version ${migration.version} successfully.");

      schema = migration.currentSchema;
    }
  });

  return schema;
}