apply method

Future<List<int>> apply()

Returns the versions applied by this invocation, in registration order.

Implementation

Future<List<int>> apply() async {
  if (database.session.engine != history.engine) {
    throw ArgumentError('Migration history and database engines differ.');
  }
  return database.transaction(
    (session) async {
      if (session.engine == Engine.postgresql) {
        // Serialize the first CREATE as well as later history reads. A table
        // lock alone cannot protect two callers before that table exists.
        await session.run(
          "SELECT pg_catalog.pg_advisory_xact_lock(pg_catalog.hashtextextended(pg_catalog.current_database() || ':' || \$1 || ':orm_migrations', 0))",
          parameters: [session.schema],
        );
        // Explicit pg_temp last prevents its implicit precedence for names.
        await session.run(
          'SET LOCAL search_path TO ${quoteIdentifier(session.schema)}, pg_temp',
        );
      }
      final historyTable =
          '${quoteIdentifier(session.schema)}.${quoteIdentifier('_orm_migrations')}';
      final versionType = session.engine == Engine.sqlite
          ? 'INTEGER'
          : 'BIGINT';
      await session.run(
        'CREATE TABLE IF NOT EXISTS $historyTable ("version" $versionType PRIMARY KEY NOT NULL, "name" TEXT NOT NULL, "engine" TEXT NOT NULL, "fingerprint" TEXT NOT NULL)',
      );
      await verifyCatalog(
        session,
        SchemaSnapshot(engine: session.engine, tables: _historyTables),
        const {},
      );
      final readHistory =
          'SELECT "version", "name", "engine", "fingerprint" FROM $historyTable ORDER BY "version"';
      final saved = await session.run(readHistory);
      _verifySaved(saved);
      final versions = <int>[];
      final knownTables = <String>{};
      SchemaSnapshot? previous;
      for (var i = 0; i < saved.rows.length; i++) {
        previous = history.migrations[i].snapshot;
        knownTables.addAll(
          previous.tables.map(
            (t) => physicalIdentity(session.engine, t.name),
          ),
        );
      }
      if (previous != null) {
        await verifyCatalog(
          session,
          previous,
          knownTables.difference(
            previous.tables
                .map((t) => physicalIdentity(session.engine, t.name))
                .toSet(),
          ),
        );
      }
      for (var i = saved.rows.length; i < history.migrations.length; i++) {
        final migration = history.migrations[i];
        for (final step in migration.steps) {
          await session.run(step);
        }
        final names = migration.snapshot.tables
            .map((t) => physicalIdentity(session.engine, t.name))
            .toSet();
        if (i < history.migrations.length - 1) {
          await verifyCatalog(
            session,
            migration.snapshot,
            knownTables.difference(names),
          );
        }
        final placeholders = session.engine == Engine.sqlite
            ? '?, ?, ?, ?'
            : '\$1, \$2, \$3, \$4';
        final inserted = await session.run(
          'INSERT INTO $historyTable ("version", "name", "engine", "fingerprint") VALUES ($placeholders)',
          parameters: [
            migration.version,
            migration.name,
            history.engine.name,
            migration.fingerprint,
          ],
        );
        if (inserted.affectedRows != 1) {
          throw StateError('Migration history marker was not inserted.');
        }
        versions.add(migration.version);
        knownTables.addAll(names);
      }
      if (versions.isNotEmpty) {
        if (session.engine == Engine.postgresql) {
          // Deferred constraint triggers must run before marker verification.
          await session.run('SET CONSTRAINTS ALL IMMEDIATE');
        }
        _verifySaved(await session.run(readHistory), complete: true);
        // Check the final schema after marker triggers have finished.
        final snapshot = history.migrations.last.snapshot;
        await verifyCatalog(
          session,
          SchemaSnapshot(
            engine: session.engine,
            tables: [..._historyTables, ...snapshot.tables],
          ),
          knownTables.difference(
            snapshot.tables
                .map((t) => physicalIdentity(session.engine, t.name))
                .toSet(),
          ),
        );
      }
      return List<int>.unmodifiable(versions);
    },
    isolation: history.engine == Engine.postgresql
        ? Isolation.readCommitted
        : Isolation.serializable,
  );
}