Disable cache

packages/node-core/src/configure/ProjectUpgrade.service.ts

@@ -121,13 +121,20 @@ export class ProjectUpgradeService<P extends ISubqueryProject = ISubqueryProject
   private onProjectUpgrade?: OnProjectUpgradeCallback<P>;
   private migrationService?: SchemaMigrationService;
 
-  private constructor(private _projects: BlockHeightMap<P>, currentHeight: number, private _isRewindable = true) {
+  private constructor(
+    private _projects: BlockHeightMap<P>,
+    currentHeight: number,
+    private _isRewindable = true
+  ) {
     logger.info(
       `Projects: ${JSON.stringify(
-        [..._projects.getAll().entries()].reduce((acc, curr) => {
-          acc[curr[0]] = curr[1].id;
-          return acc;
-        }, {} as Record<number, string>),
+        [..._projects.getAll().entries()].reduce(
+          (acc, curr) => {
+            acc[curr[0]] = curr[1].id;
+            return acc;
+          },
+          {} as Record<number, string>
+        ),
         undefined,
         2
       )}`
@@ -150,16 +157,10 @@ export class ProjectUpgradeService<P extends ISubqueryProject = ISubqueryProject
       return;
     }
     this.#initialized = true;
-    this.#storeCache = storeService.storeCache;
+    this.#storeCache = storeService.storeModel;
     this.config = config;
 
-    this.migrationService = new SchemaMigrationService(
-      sequelize,
-      storeService,
-      storeService.storeCache._flushCache.bind(storeService.storeCache),
-      schema,
-      config
-    );
+    this.migrationService = new SchemaMigrationService(sequelize, storeService, schema, config);
 
     const indexedDeployments = await this.getDeploymentsMetadata();
 

packages/node-core/src/db/migration-service/SchemaMigration.service.test.ts

@@ -46,13 +46,7 @@ async function setup(
   await storeService.initCoreTables(schemaName);
   await initDbSchema(schemaName, storeService);
 
-  return new SchemaMigrationService(
-    sequelize,
-    storeService,
-    storeCache._flushCache.bind(storeCache),
-    schemaName,
-    config
-  );
+  return new SchemaMigrationService(sequelize, storeService, schemaName, config);
 }
 
 function loadGqlSchema(fileName: string): GraphQLSchema {

packages/node-core/src/db/migration-service/SchemaMigration.service.ts

@@ -29,7 +29,6 @@ export class SchemaMigrationService {
   constructor(
     private sequelize: Sequelize,
     private storeService: StoreService,
-    private flushCache: (flushAll?: boolean) => Promise<void>,
     private dbSchema: string,
     private config: NodeConfig,
     private dbType: SUPPORT_DB = SUPPORT_DB.postgres
@@ -115,7 +114,7 @@ export class SchemaMigrationService {
     const sortedAddedModels = alignModelOrder<GraphQLModelsType[]>(sortedSchemaModels, addedModels);
     const sortedModifiedModels = alignModelOrder<ModifiedModels>(sortedSchemaModels, modifiedModels);
 
-    await this.storeService.storeCache._flushCache(true);
+    await this.storeService.storeModel.flushData?.(true);
     const migrationAction = await Migration.create(
       this.sequelize,
       this.storeService,
@@ -184,10 +183,10 @@ export class SchemaMigrationService {
       const modelChanges = await migrationAction.run(transaction);
 
       // Update any relevant application state so the right models are used
-      this.storeService.storeCache.updateModels(modelChanges);
+      this.storeService.storeModel.updateModels(modelChanges);
       await this.storeService.updateModels(this.dbSchema, getAllEntitiesRelations(nextSchema));
 
-      await this.flushCache();
+      await this.storeService.storeModel.flushData?.(true);
     } catch (e: any) {
       logger.error(e, 'Failed to execute Schema Migration');
       throw e;

packages/node-core/src/indexer/blockDispatcher/base-block-dispatcher.ts

@@ -218,7 +218,7 @@ export abstract class BaseBlockDispatcher<Q extends IQueue, DS, B> implements IB
 
     await this.storeModelService.applyPendingChanges(height, !this.projectService.hasDataSourcesAfterHeight(height));
 
-    // if (this.storeModelService instanceof StoreCacheService) {
+    // if (this.storeModelService instanceof StorCeacheService) {
     //   if (this.nodeConfig.storeCacheAsync) {
     //     // Flush all completed block data and don't wait
     //     await this.storeModelService.flushAndWaitForCapacity(false)?.catch((e) => {

packages/node-core/src/indexer/poi/poi.service.ts

@@ -22,9 +22,12 @@ const logger = getLogger('PoiService');
 @Injectable()
 export class PoiService implements OnApplicationShutdown {
   private isShutdown = false;
-  private _poiRepo?: CachePoiModel;
+  private _poiRepo?: IPoi;
 
-  constructor(protected readonly nodeConfig: NodeConfig, private storeCache: IStoreModelService) {}
+  constructor(
+    protected readonly nodeConfig: NodeConfig,
+    private storeCache: IStoreModelService
+  ) {}
 
   onApplicationShutdown(): void {
     this.isShutdown = true;
@@ -40,15 +43,20 @@ export class PoiService implements OnApplicationShutdown {
     };
   }
 
-  get poiRepo(): CachePoiModel {
+  get poiRepo(): IPoi {
     if (!this._poiRepo) {
       throw new Error(`No poi repo inited`);
     }
     return this._poiRepo;
   }
 
   get plainPoiRepo(): PlainPoiModel {
-    return this.poiRepo.plainPoiModel;
+    if (this.poiRepo instanceof CachePoiModel) {
+      return this.poiRepo.plainPoiModel;
+    } else if (this.poiRepo instanceof PlainPoiModel) {
+      return this.poiRepo;
+    }
+    throw new Error(`No plainPoiRepo repo inited`);
   }
 
   /**

packages/node-core/src/indexer/project.service.ts

@@ -106,7 +106,7 @@ export abstract class BaseProjectService<
       await this.storeService.initCoreTables(this._schema);
       await this.ensureMetadata();
       // DynamicDsService is dependent on metadata so we need to ensure it exists first
-      await this.dynamicDsService.init(this.storeService.storeCache.metadata);
+      await this.dynamicDsService.init(this.storeService.storeModel.metadata);
 
       /**
        * WARNING: The order of the following steps is very important.
@@ -146,7 +146,7 @@ export abstract class BaseProjectService<
       }
 
       // Flush any pending operations to set up DB
-      await this.storeService.storeCache.flushCache(true);
+      await this.storeService.storeModel.flushData?.(true);
     } else {
       assert(startHeight, 'ProjectService must be initalized with a start height in workers');
       this.projectUpgradeService.initWorker(startHeight, this.handleProjectChange.bind(this));
@@ -195,7 +195,7 @@ export abstract class BaseProjectService<
   }
 
   private async ensureMetadata(): Promise<void> {
-    const metadata = this.storeService.storeCache.metadata;
+    const metadata = this.storeService.storeModel.metadata;
 
     this.eventEmitter.emit(IndexerEvent.NetworkMetadata, this.apiService.networkMeta);
 
@@ -269,7 +269,7 @@ export abstract class BaseProjectService<
   }
 
   protected async getLastProcessedHeight(): Promise<number | undefined> {
-    return this.storeService.storeCache.metadata.find('lastProcessedHeight');
+    return this.storeService.storeModel.metadata.find('lastProcessedHeight');
   }
 
   private async nextProcessHeight(): Promise<number | undefined> {

packages/node-core/src/indexer/store.service.ts

@@ -79,7 +79,7 @@ export class StoreService {
   constructor(
     private sequelize: Sequelize,
     private config: NodeConfig,
-    readonly storeCache: IStoreModelService,
+    readonly storeModel: IStoreModelService,
     @Inject('ISubqueryProject') private subqueryProject: ISubqueryProject<IProjectNetworkConfig>
   ) {}
 
@@ -125,7 +125,7 @@ export class StoreService {
       this._lastTimeDbSizeChecked = Date.now();
       return getDbSizeAndUpdateMetadata(this.sequelize, this.schema);
     } else {
-      return this.storeCache.metadata.find('dbSize').then((cachedDbSize) => {
+      return this.storeModel.metadata.find('dbSize').then((cachedDbSize) => {
         if (cachedDbSize !== undefined) {
           return cachedDbSize;
         } else {
@@ -177,9 +177,9 @@ export class StoreService {
     }
     logger.info(`Historical state is ${this.historical ? 'enabled' : 'disabled'}`);
 
-    this.storeCache.init(this.historical, this.dbType === SUPPORT_DB.cockRoach, this.metaDataRepo, this.poiRepo);
+    this.storeModel.init(this.historical, this.dbType === SUPPORT_DB.cockRoach, this.metaDataRepo, this.poiRepo);
 
-    this._metadataModel = this.storeCache.metadata;
+    this._metadataModel = this.storeModel.metadata;
 
     await this.initHotSchemaReloadQueries(schema);
 
@@ -199,13 +199,7 @@ export class StoreService {
       On SyncSchema, if no schema migration is introduced, it would consider current schema to be null, and go all db operations again
       every start up is a migration
        */
-      const schemaMigrationService = new SchemaMigrationService(
-        this.sequelize,
-        this,
-        this.storeCache._flushCache.bind(this.storeCache),
-        schema,
-        this.config
-      );
+      const schemaMigrationService = new SchemaMigrationService(this.sequelize, this, schema, this.config);
 
       await schemaMigrationService.run(null, this.subqueryProject.schema, tx);
 
@@ -330,10 +324,13 @@ export class StoreService {
           {type: QueryTypes.SELECT}
         );
 
-        const store = res.reduce(function (total, current) {
-          total[current.key] = current.value;
-          return total;
-        }, {} as {[key: string]: string | boolean});
+        const store = res.reduce(
+          function (total, current) {
+            total[current.key] = current.value;
+            return total;
+          },
+          {} as {[key: string]: string | boolean}
+        );
 
         const useHistorical =
           store.historicalStateEnabled === undefined ? !disableHistorical : (store.historicalStateEnabled as boolean);
@@ -476,7 +473,7 @@ group by
   }
 
   getStore(): Store {
-    return new Store(this.config, this.storeCache, this);
+    return new Store(this.config, this.storeModel, this);
   }
 }
 

packages/node-core/src/indexer/store/store.ts

@@ -27,12 +27,12 @@ type Context = {
 export class Store implements IStore {
   /* These need to explicily be private using JS style private properties in order to not leak these in the sandbox */
   #config: NodeConfig;
-  #storeCache: IStoreModelService;
+  #storeModel: IStoreModelService;
   #context: Context;
 
-  constructor(config: NodeConfig, storeCache: IStoreModelService, context: Context) {
+  constructor(config: NodeConfig, storeModel: IStoreModelService, context: Context) {
     this.#config = config;
-    this.#storeCache = storeCache;
+    this.#storeModel = storeModel;
     this.#context = context;
   }
 
@@ -52,7 +52,7 @@ export class Store implements IStore {
 
   async get<T extends Entity>(entity: string, id: string): Promise<T | undefined> {
     try {
-      const raw = await this.#storeCache.getModel<T>(entity).get(id);
+      const raw = await this.#storeModel.getModel<T>(entity).get(id);
       monitorWrite(`-- [Store][get] Entity ${entity} ID ${id}, data: ${handledStringify(raw)}`);
       return EntityClass.create<T>(entity, raw, this);
     } catch (e) {
@@ -72,7 +72,7 @@ export class Store implements IStore {
 
       this.#queryLimitCheck('getByField', entity, options);
 
-      const raw = await this.#storeCache
+      const raw = await this.#storeModel
         .getModel<T>(entity)
         .getByFields([Array.isArray(value) ? [field, 'in', value] : [field, '=', value]], options);
       monitorWrite(`-- [Store][getByField] Entity ${entity}, data: ${handledStringify(raw)}`);
@@ -98,7 +98,7 @@ export class Store implements IStore {
 
       this.#queryLimitCheck('getByFields', entity, options);
 
-      const raw = await this.#storeCache.getModel<T>(entity).getByFields(filter, options);
+      const raw = await this.#storeModel.getModel<T>(entity).getByFields(filter, options);
       monitorWrite(`-- [Store][getByFields] Entity ${entity}, data: ${handledStringify(raw)}`);
       return raw.map((v) => EntityClass.create<T>(entity, v, this)) as T[];
     } catch (e) {
@@ -110,7 +110,7 @@ export class Store implements IStore {
     try {
       const indexed = this.#context.isIndexedHistorical(entity, field as string);
       assert(indexed, `to query by field ${String(field)}, a unique index must be created on model ${entity}`);
-      const [raw] = await this.#storeCache
+      const [raw] = await this.#storeModel
         .getModel<T>(entity)
         .getByFields([Array.isArray(value) ? [field, 'in', value] : [field, '=', value]], {limit: 1});
       monitorWrite(`-- [Store][getOneByField] Entity ${entity}, data: ${handledStringify(raw)}`);
@@ -122,7 +122,7 @@ export class Store implements IStore {
 
   async set(entity: string, _id: string, data: Entity): Promise<void> {
     try {
-      await this.#storeCache.getModel(entity).set(_id, data, this.#context.blockHeight, this.#context.transaction);
+      await this.#storeModel.getModel(entity).set(_id, data, this.#context.blockHeight, this.#context.transaction);
       monitorWrite(
         `-- [Store][set] Entity ${entity}, height: ${this.#context.blockHeight}, data: ${handledStringify(data)}`
       );
@@ -134,7 +134,7 @@ export class Store implements IStore {
 
   async bulkCreate(entity: string, data: Entity[]): Promise<void> {
     try {
-      await this.#storeCache.getModel(entity).bulkCreate(data, this.#context.blockHeight, this.#context.transaction);
+      await this.#storeModel.getModel(entity).bulkCreate(data, this.#context.blockHeight, this.#context.transaction);
       for (const item of data) {
         this.#context.operationStack?.put(OperationType.Set, entity, item);
       }
@@ -148,7 +148,7 @@ export class Store implements IStore {
 
   async bulkUpdate(entity: string, data: Entity[], fields?: string[]): Promise<void> {
     try {
-      await this.#storeCache
+      await this.#storeModel
         .getModel(entity)
         .bulkUpdate(data, this.#context.blockHeight, fields, this.#context.transaction);
       for (const item of data) {
@@ -164,7 +164,7 @@ export class Store implements IStore {
 
   async remove(entity: string, id: string): Promise<void> {
     try {
-      await this.#storeCache.getModel(entity).bulkRemove([id], this.#context.blockHeight, this.#context.transaction);
+      await this.#storeModel.getModel(entity).bulkRemove([id], this.#context.blockHeight, this.#context.transaction);
       this.#context.operationStack?.put(OperationType.Remove, entity, id);
       monitorWrite(`-- [Store][remove] Entity ${entity}, height: ${this.#context.blockHeight}, id: ${id}`);
     } catch (e) {
@@ -174,7 +174,7 @@ export class Store implements IStore {
 
   async bulkRemove(entity: string, ids: string[]): Promise<void> {
     try {
-      await this.#storeCache.getModel(entity).bulkRemove(ids, this.#context.blockHeight, this.#context.transaction);
+      await this.#storeModel.getModel(entity).bulkRemove(ids, this.#context.blockHeight, this.#context.transaction);
 
       for (const id of ids) {
         this.#context.operationStack?.put(OperationType.Remove, entity, id);

packages/node-core/src/indexer/storeCache/baseCache.service.ts

@@ -30,7 +30,7 @@ export abstract class BaseCacheService
   }
 
   @profiler()
-  async flushCache(forceFlush?: boolean): Promise<void> {
+  async flushData(forceFlush?: boolean): Promise<void> {
     const flushCacheGuarded = async (forceFlush?: boolean): Promise<void> => {
       // When we force flush, this will ensure not interrupt current block flushing,
       // Force flush will continue after last block flush tx committed.
@@ -55,12 +55,12 @@ export abstract class BaseCacheService
     return this.queuedFlush;
   }
 
-  async resetCache(): Promise<void> {
+  async resetData(): Promise<void> {
     await this._resetCache();
   }
 
   async beforeApplicationShutdown(): Promise<void> {
-    await timeout(this.flushCache(true), 60, 'Before shutdown flush cache timeout');
+    await timeout(this.flushData(true), 60, 'Before shutdown flush cache timeout');
     this.logger.info(`Force flush cache successful!`);
     await this.flushExportStores();
     this.logger.info(`Force flush exports successful!`);

packages/node-core/src/indexer/storeCache/metadata/metadata.ts

@@ -22,6 +22,8 @@ export interface IMetadata {
   setNewDynamicDatasource(item: DatasourceParams, tx?: Transaction): Promise<void>;
 
   bulkRemove<K extends MetadataKey>(keys: K[], tx?: Transaction): Promise<void>;
+
+  flush?(tx: Transaction, blockHeight: number): Promise<void>;
 }
 
 export class MetadataModel implements IMetadata {