fix(server): run migrations after database checks (#5832)

* run migrations after checks

* optional migrations

* only run checks in server and e2e

* re-add migrations for microservices

* refactor

* move e2e init

* remove assert from migration

* update providers

* update microservices app service

* fixed logging

* refactored version check, added unit tests

* more version tests

* don't use mocks for sut

* refactor tests

* suggest image only if postgres is 14, 15 or 16

* review suggestions

* fixed regexp escape

* fix typing

* update migration

server/src/infra/database.config.ts

@@ -1,4 +1,4 @@
-import { DataSource, QueryRunner } from 'typeorm';
+import { DataSource } from 'typeorm';
 import { PostgresConnectionOptions } from 'typeorm/driver/postgres/PostgresConnectionOptions';
 
 const url = process.env.DB_URL;
@@ -18,58 +18,10 @@ export const databaseConfig: PostgresConnectionOptions = {
   synchronize: false,
   migrations: [__dirname + '/migrations/*.{js,ts}'],
   subscribers: [__dirname + '/subscribers/*.{js,ts}'],
-  migrationsRun: true,
+  migrationsRun: false,
   connectTimeoutMS: 10000, // 10 seconds
   ...urlOrParts,
 };
 
 // this export is used by TypeORM commands in package.json#scripts
 export const dataSource = new DataSource(databaseConfig);
-
-export async function databaseChecks() {
-  if (!dataSource.isInitialized) {
-    await dataSource.initialize();
-  }
-
-  await assertVectors(dataSource);
-  await enablePrefilter(dataSource);
-  await dataSource.runMigrations();
-}
-
-export async function enablePrefilter(runner: DataSource | QueryRunner) {
-  await runner.query(`SET vectors.enable_prefilter = on`);
-}
-
-export async function getExtensionVersion(extName: string, runner: DataSource | QueryRunner): Promise<string | null> {
-  const res = await runner.query(`SELECT extversion FROM pg_extension WHERE extname = $1`, [extName]);
-  return res[0]?.['extversion'] ?? null;
-}
-
-export async function getPostgresVersion(runner: DataSource | QueryRunner): Promise<string> {
-  const res = await runner.query(`SHOW server_version`);
-  return res[0]['server_version'].split('.')[0];
-}
-
-export async function assertVectors(runner: DataSource | QueryRunner) {
-  const postgresVersion = await getPostgresVersion(runner);
-  const expected = ['0.1.1', '0.1.11'];
-  const image = `tensorchord/pgvecto-rs:pg${postgresVersion}-v${expected[expected.length - 1]}`;
-
-  await runner.query('CREATE EXTENSION IF NOT EXISTS vectors').catch((err) => {
-    console.error(
-      'Failed to create pgvecto.rs extension. ' +
-        `If you have not updated your Postgres instance to an image that supports pgvecto.rs (such as '${image}'), please do so. ` +
-        'See the v1.91.0 release notes for more info: https://github.com/immich-app/immich/releases/tag/v1.91.0',
-    );
-    throw err;
-  });
-
-  const version = await getExtensionVersion('vectors', runner);
-  if (version != null && !expected.includes(version)) {
-    throw new Error(
-      `The pgvecto.rs extension version is ${version} instead of the expected version ${
-        expected[expected.length - 1]
-      }.` + `If you're using the 'latest' tag, please switch to '${image}'.`,
-    );
-  }
-}

server/src/infra/infra.module.ts

@@ -6,6 +6,7 @@ import {
   IAuditRepository,
   ICommunicationRepository,
   ICryptoRepository,
+  IDatabaseRepository,
   IJobRepository,
   IKeyRepository,
   ILibraryRepository,
@@ -43,6 +44,7 @@ import {
   AuditRepository,
   CommunicationRepository,
   CryptoRepository,
+  DatabaseRepository,
   FilesystemProvider,
   JobRepository,
   LibraryRepository,
@@ -70,6 +72,7 @@ const providers: Provider[] = [
   { provide: IAuditRepository, useClass: AuditRepository },
   { provide: ICommunicationRepository, useClass: CommunicationRepository },
   { provide: ICryptoRepository, useClass: CryptoRepository },
+  { provide: IDatabaseRepository, useClass: DatabaseRepository },
   { provide: IJobRepository, useClass: JobRepository },
   { provide: ILibraryRepository, useClass: LibraryRepository },
   { provide: IKeyRepository, useClass: ApiKeyRepository },

server/src/infra/logger.ts

@@ -5,7 +5,7 @@ import { LogLevel } from './entities';
 const LOG_LEVELS = [LogLevel.VERBOSE, LogLevel.DEBUG, LogLevel.LOG, LogLevel.WARN, LogLevel.ERROR, LogLevel.FATAL];
 
 export class ImmichLogger extends ConsoleLogger {
-  private static logLevels: LogLevel[] = [];
+  private static logLevels: LogLevel[] = [LogLevel.WARN, LogLevel.ERROR, LogLevel.FATAL];
 
   constructor(context: string) {
     super(context);

server/src/infra/migrations/1700713871511-UsePgVectors.ts

@@ -1,13 +1,11 @@
 import { getCLIPModelInfo } from '@app/domain/smart-info/smart-info.constant';
 import { MigrationInterface, QueryRunner } from 'typeorm';
-import { assertVectors } from '../database.config';
 
 export class UsePgVectors1700713871511 implements MigrationInterface {
   name = 'UsePgVectors1700713871511';
 
   public async up(queryRunner: QueryRunner): Promise<void> {
-    await assertVectors(queryRunner);
-
+    await queryRunner.query(`CREATE EXTENSION IF NOT EXISTS vectors`);
     const faceDimQuery = await queryRunner.query(`
         SELECT CARDINALITY(embedding::real[]) as dimsize
         FROM asset_faces

server/src/infra/repositories/database.repository.ts

@@ -0,0 +1,28 @@
+import { DatabaseExtension, IDatabaseRepository, Version } from '@app/domain';
+import { Injectable } from '@nestjs/common';
+import { InjectDataSource } from '@nestjs/typeorm';
+import { DataSource } from 'typeorm';
+
+@Injectable()
+export class DatabaseRepository implements IDatabaseRepository {
+  constructor(@InjectDataSource() private dataSource: DataSource) {}
+
+  async getExtensionVersion(extension: DatabaseExtension): Promise<Version | null> {
+    const res = await this.dataSource.query(`SELECT extversion FROM pg_extension WHERE extname = $1`, [extension]);
+    const version = res[0]?.['extversion'];
+    return version == null ? null : Version.fromString(version);
+  }
+
+  async getPostgresVersion(): Promise<Version> {
+    const res = await this.dataSource.query(`SHOW server_version`);
+    return Version.fromString(res[0]['server_version']);
+  }
+
+  async createExtension(extension: DatabaseExtension): Promise<void> {
+    await this.dataSource.query(`CREATE EXTENSION IF NOT EXISTS ${extension}`);
+  }
+
+  async runMigrations(options?: { transaction?: 'all' | 'none' | 'each' }): Promise<void> {
+    await this.dataSource.runMigrations(options);
+  }
+}

server/src/infra/repositories/index.ts

@@ -6,6 +6,7 @@ export * from './asset.repository';
 export * from './audit.repository';
 export * from './communication.repository';
 export * from './crypto.repository';
+export * from './database.repository';
 export * from './filesystem.provider';
 export * from './job.repository';
 export * from './library.repository';

server/src/infra/repositories/smart-info.repository.ts

@@ -52,6 +52,7 @@ export class SmartInfoRepository implements ISmartInfoRepository {
     let results: AssetEntity[] = [];
     await this.assetRepository.manager.transaction(async (manager) => {
       await manager.query(`SET LOCAL vectors.k = '${numResults}'`);
+      await manager.query(`SET LOCAL vectors.enable_prefilter = on`);
       results = await manager
         .createQueryBuilder(AssetEntity, 'a')
         .innerJoin('a.smartSearch', 's')

server/src/infra/sql/smart.info.repository.sql

@@ -4,6 +4,8 @@
 START TRANSACTION
 SET
   LOCAL vectors.k = '100'
+SET
+  LOCAL vectors.enable_prefilter = on
 SELECT
   "a"."id" AS "a_id",
   "a"."deviceAssetId" AS "a_deviceAssetId",