update based on review, add idempotency to db queries

subquery · Feb 13, 2024 · d0b6c36 · d0b6c36
1 parent 7dbb468
commit d0b6c36
Show file tree

Hide file tree

Showing 6 changed files with 279 additions and 262 deletions.
diff --git a/packages/node-core/CHANGELOG.md b/packages/node-core/CHANGELOG.md
@@ -4,8 +4,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-
 ## [Unreleased]
+### Added
+- Schema Migration support for Enums, Relations, Subscription (#2251)
+- 
+### Fixed
+- Fixed non-atomic schema migration execution (#2244)
 
 ## [7.2.1] - 2024-02-07
 ### Added
@@ -14,9 +18,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 - Historical queries not using the correct block height (#2243)
 
-### Fixed
-- Fixed non-autonomous schema migration execution (#2244)
-
 ## [7.2.0] - 2024-01-30
 ### Changed
 - Update `@subql/apollo-links` and add specific logger for it

diff --git a/packages/node-core/src/configure/migration-service/SchemaMigration.service.ts b/packages/node-core/src/configure/migration-service/SchemaMigration.service.ts
@@ -11,6 +11,7 @@ import {sortModels} from '../../utils';
 import {NodeConfig} from '../NodeConfig';
 import {Migration} from './migration';
 import {
+  alignModelOrder,
   compareEnums,
   compareModels,
   compareRelations,
@@ -80,39 +81,6 @@ export class SchemaMigrationService {
     }
   }
 
-  private alignModelOrder(
-    schemaModels: GraphQLModelsType[],
-    models: GraphQLModelsType[] | ModifiedModels
-  ): GraphQLModelsType[] | ModifiedModels {
-    const orderIndex = schemaModels.reduce((acc: Record<string, number>, model, index) => {
-      acc[model.name] = index;
-      return acc;
-    }, {});
-
-    if (Array.isArray(models)) {
-      return models.sort((a, b) => {
-        const indexA = orderIndex[a.name] ?? Number.MAX_VALUE; // Place unknown models at the end
-        const indexB = orderIndex[b.name] ?? Number.MAX_VALUE;
-        return indexA - indexB;
-      });
-    } else {
-      const modelNames = Object.keys(models);
-      const sortedModelNames = modelNames.sort((a, b) => {
-        const indexA = orderIndex[a] ?? Number.MAX_VALUE;
-        const indexB = orderIndex[b] ?? Number.MAX_VALUE;
-        return indexA - indexB;
-      });
-
-      const sortedModifiedModels: ModifiedModels = {};
-      sortedModelNames.forEach((modelName) => {
-        sortedModifiedModels[modelName] = models[modelName];
-      });
-
-      return sortedModifiedModels;
-    }
-  }
-
-  // eslint-disable-next-line complexity
   async run(
     currentSchema: GraphQLSchema | null,
     nextSchema: GraphQLSchema,
@@ -148,61 +116,52 @@ export class SchemaMigrationService {
       getAllEntitiesRelations(nextSchema).relations
     );
 
-    const sortedAddedModels = this.alignModelOrder(sortedSchemaModels, addedModels) as GraphQLModelsType[];
-    const sortedModifiedModels = this.alignModelOrder(sortedSchemaModels, modifiedModels);
+    const sortedAddedModels = alignModelOrder<GraphQLModelsType[]>(sortedSchemaModels, addedModels);
+    const sortedModifiedModels = alignModelOrder<ModifiedModels>(sortedSchemaModels, modifiedModels);
 
     await this.flushCache(true);
     const migrationAction = await Migration.create(
       this.sequelize,
       this.storeService,
       this.dbSchema,
-      nextSchema,
       currentSchema,
       this.config,
-      logger,
       this.dbType
     );
 
+    // TODO this should only be printed if schema migration is enabled and not store.service sync schema
     logger.info(`${schemaChangesLoggerMessage(schemaDifference)}`);
     try {
-      if (addedEnums.length) {
-        for (const enumValue of addedEnums) {
-          await migrationAction.createEnum(enumValue);
-        }
+      for (const enumValue of addedEnums) {
+        await migrationAction.createEnum(enumValue);
       }
 
-      if (removedModels.length) {
-        for (const model of removedModels) {
-          migrationAction.dropTable(model);
-        }
+      for (const model of removedModels) {
+        migrationAction.dropTable(model);
       }
 
-      if (sortedAddedModels.length) {
-        for (const model of sortedAddedModels) {
-          await migrationAction.createTable(model, this.withoutForeignKeys);
-        }
+      for (const model of sortedAddedModels) {
+        await migrationAction.createTable(model, this.withoutForeignKeys);
       }
 
-      if (Object.keys(sortedModifiedModels).length) {
-        const entities = Object.keys(modifiedModels);
-        for (const model of entities) {
-          const modelValue = modifiedModels[model];
+      const entities = Object.keys(sortedModifiedModels);
+      for (const model of entities) {
+        const modelValue = sortedModifiedModels[model];
 
-          for (const index of modelValue.removedIndexes) {
-            migrationAction.dropIndex(modelValue.model, index);
-          }
+        for (const index of modelValue.removedIndexes) {
+          migrationAction.dropIndex(modelValue.model, index);
+        }
 
-          for (const field of modelValue.removedFields) {
-            migrationAction.dropColumn(modelValue.model, field);
-          }
+        for (const field of modelValue.removedFields) {
+          migrationAction.dropColumn(modelValue.model, field);
+        }
 
-          for (const field of modelValue.addedFields) {
-            migrationAction.createColumn(modelValue.model, field);
-          }
+        for (const field of modelValue.addedFields) {
+          migrationAction.createColumn(modelValue.model, field);
+        }
 
-          for (const index of modelValue.addedIndexes) {
-            migrationAction.createIndex(modelValue.model, index);
-          }
+        for (const index of modelValue.addedIndexes) {
+          migrationAction.createIndex(modelValue.model, index);
         }
       }
 
@@ -214,15 +173,11 @@ export class SchemaMigrationService {
         migrationAction.addRelationComments();
       }
 
-      if (removedRelations.length) {
-        for (const relationModel of removedRelations) {
-          migrationAction.dropRelation(relationModel);
-        }
+      for (const relationModel of removedRelations) {
+        migrationAction.dropRelation(relationModel);
       }
-      if (removedEnums.length) {
-        for (const enumValue of removedEnums) {
-          migrationAction.dropEnums(enumValue);
-        }
+      for (const enumValue of removedEnums) {
+        migrationAction.dropEnum(enumValue);
       }
 
       return migrationAction.run(transaction);

diff --git a/packages/node-core/src/configure/migration-service/migration-helpers.ts b/packages/node-core/src/configure/migration-service/migration-helpers.ts
@@ -1,13 +1,15 @@
 // Copyright 2020-2024 SubQuery Pte Ltd authors & contributors
 // SPDX-License-Identifier: GPL-3.0
 
+import {addRelationToMap, enumNameToHash, SmartTags} from '@subql/node-core';
 import {
   GraphQLEntityField,
   GraphQLEntityIndex,
   GraphQLEnumsType,
   GraphQLModelsType,
   GraphQLRelationsType,
 } from '@subql/utils';
+import {QueryTypes, Sequelize} from '@subql/x-sequelize';
 import {isEqual} from 'lodash';
 
 export type ModifiedModels = Record<
@@ -205,3 +207,81 @@ export function schemaChangesLoggerMessage(schemaChanges: SchemaChangesType): st
   }
   return logMessage;
 }
+export function alignModelOrder<T extends GraphQLModelsType[] | ModifiedModels>(
+  schemaModels: GraphQLModelsType[],
+  models: T
+): T {
+  const orderIndex = schemaModels.reduce((acc: Record<string, number>, model, index) => {
+    acc[model.name] = index;
+    return acc;
+  }, {});
+
+  if (Array.isArray(models)) {
+    return models.sort((a, b) => {
+      const indexA = orderIndex[a.name] ?? Number.MAX_VALUE; // Place unknown models at the end
+      const indexB = orderIndex[b.name] ?? Number.MAX_VALUE;
+      return indexA - indexB;
+    }) as T;
+  } else {
+    const modelNames = Object.keys(models);
+    const sortedModelNames = modelNames.sort((a, b) => {
+      const indexA = orderIndex[a] ?? Number.MAX_VALUE;
+      const indexB = orderIndex[b] ?? Number.MAX_VALUE;
+      return indexA - indexB;
+    });
+
+    const sortedModifiedModels: ModifiedModels = {};
+    sortedModelNames.forEach((modelName) => {
+      sortedModifiedModels[modelName] = models[modelName];
+    });
+
+    return sortedModifiedModels as T;
+  }
+}
+
+export function loadExistingForeignKeys(
+  relations: GraphQLRelationsType[],
+  sequelize: Sequelize
+): Map<string, Map<string, SmartTags>> {
+  const foreignKeyMap = new Map<string, Map<string, SmartTags>>();
+  for (const relation of relations) {
+    const model = sequelize.model(relation.from);
+    const relatedModel = sequelize.model(relation.to);
+    Object.values(model.associations).forEach(() => {
+      addRelationToMap(relation, foreignKeyMap, model, relatedModel);
+    });
+  }
+  return foreignKeyMap;
+}
+
+export async function loadExistingEnums(
+  enums: GraphQLEnumsType[],
+  schema: string,
+  sequelize: Sequelize
+): Promise<Map<string, string>> {
+  const enumTypeMap = new Map<string, string>();
+
+  const results = (await sequelize.query(
+    `
+    SELECT t.typname AS enum_type
+FROM pg_type t
+         JOIN pg_catalog.pg_namespace n ON n.oid = t.typnamespace
+WHERE n.nspname = :schema
+  AND t.typtype = 'e'
+ORDER BY t.typname;
+    `,
+    {
+      replacements: {schema: schema},
+      type: QueryTypes.SELECT,
+    }
+  )) as {enum_type: string}[];
+
+  for (const e of enums) {
+    const enumTypeName = enumNameToHash(e.name);
+    if (!results.find((en) => en.enum_type === enumTypeName)) {
+      continue;
+    }
+    enumTypeMap.set(e.name, `"${schema}"."${enumTypeName}"`);
+  }
+  return enumTypeMap;
+}