♻️ reindex cli: prepare for filtering treated documents (#438)
This commit is contained in:
committed by
ericlinagora
parent
ad46390887
commit
e2c5d3016b
@@ -11,7 +11,11 @@ import { DriveFile, TYPE as DriveFileTYPE } from "../../../services/documents/en
|
|||||||
import Repository, {
|
import Repository, {
|
||||||
FindFilter,
|
FindFilter,
|
||||||
} from "../../../core/platform/services/database/services/orm/repository/repository";
|
} from "../../../core/platform/services/database/services/orm/repository/repository";
|
||||||
import { EntityTarget, SearchServiceAPI } from "../../../core/platform/services/search/api";
|
import {
|
||||||
|
EntityTarget,
|
||||||
|
FindOptions,
|
||||||
|
SearchServiceAPI,
|
||||||
|
} from "../../../core/platform/services/search/api";
|
||||||
import CompanyUser, { TYPE as CompanyUserTYPE } from "../../../services/user/entities/company_user";
|
import CompanyUser, { TYPE as CompanyUserTYPE } from "../../../services/user/entities/company_user";
|
||||||
import runWithPlatform from "../../lib/run_with_platform";
|
import runWithPlatform from "../../lib/run_with_platform";
|
||||||
import SearchRepository from "src/core/platform/services/search/repository";
|
import SearchRepository from "src/core/platform/services/search/repository";
|
||||||
@@ -29,13 +33,15 @@ const defaultPageSize = "100";
|
|||||||
async function iterateOverRepoPages<Entity>(
|
async function iterateOverRepoPages<Entity>(
|
||||||
repository: Repository<Entity>,
|
repository: Repository<Entity>,
|
||||||
forEachPage: (entities: Entity[]) => Promise<void>,
|
forEachPage: (entities: Entity[]) => Promise<void>,
|
||||||
pageSizeAsStringForReasons: string = defaultPageSize,
|
findOptions: FindOptions = {},
|
||||||
filter: FindFilter = {},
|
filter: FindFilter = {},
|
||||||
|
pageSizeAsStringForReasons: string = defaultPageSize,
|
||||||
delayPerPageMS: number = 200,
|
delayPerPageMS: number = 200,
|
||||||
) {
|
) {
|
||||||
let page: Pagination = { limitStr: pageSizeAsStringForReasons };
|
let page: Pagination = { limitStr: pageSizeAsStringForReasons };
|
||||||
do {
|
do {
|
||||||
const list = await repository.find(filter, { pagination: page }, undefined);
|
const options = { ...findOptions, pagination: page };
|
||||||
|
const list = await repository.find(filter, options, undefined);
|
||||||
await forEachPage(list.getEntities());
|
await forEachPage(list.getEntities());
|
||||||
page = list.nextPage as Pagination;
|
page = list.nextPage as Pagination;
|
||||||
await waitTimeoutMS(delayPerPageMS);
|
await waitTimeoutMS(delayPerPageMS);
|
||||||
@@ -87,29 +93,42 @@ abstract class ReindexerCLICommand<Entity> {
|
|||||||
/** Override in sub-classes to translate a page from database to search entities.
|
/** Override in sub-classes to translate a page from database to search entities.
|
||||||
* This is called by `reindexFromDBToSearch` so if that is overriden; this won't be called.
|
* This is called by `reindexFromDBToSearch` so if that is overriden; this won't be called.
|
||||||
*/
|
*/
|
||||||
protected async repairEntities(): Promise<void> {
|
|
||||||
this.statusWarn(`repairEntities > No repair action for ${this.table}`);
|
|
||||||
}
|
|
||||||
|
|
||||||
protected async mapEntitiesToReIndexFromDBToSearch(entities: Entity[]): Promise<Entity[]> {
|
protected async mapEntitiesToReIndexFromDBToSearch(entities: Entity[]): Promise<Entity[]> {
|
||||||
return entities;
|
return entities;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Override in sub-classes that have filters (eg. from arguments) for the entities to re-index/repair */
|
||||||
|
protected async prepareFindOptionsForItemsToReIndex(): Promise<FindOptions | undefined> {
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Override in sub-classes that handle the --repairEntities argument.
|
||||||
|
* It is executed before reindexFromDBToSearch. What repair means depends
|
||||||
|
* on the entity type.
|
||||||
|
*/
|
||||||
|
protected async repairEntities(_findOptions: FindOptions): Promise<void> {
|
||||||
|
this.statusWarn(`repairEntities > No repair action for ${this.table}`);
|
||||||
|
}
|
||||||
|
|
||||||
/** Override in a sub-class to completely replace the re-indexing logic.
|
/** Override in a sub-class to completely replace the re-indexing logic.
|
||||||
* - Iterates over pages of the entity from the database repository
|
* - Iterates over pages of the entity from the database repository
|
||||||
* - calls `mapEntitiesToReIndexFromDBToSearch` to convert each page to a search entity
|
* - calls `mapEntitiesToReIndexFromDBToSearch` to convert each page to a search entity
|
||||||
* - and upserts the result into the search repository
|
* - and upserts the result into the search repository
|
||||||
*/
|
*/
|
||||||
protected async reindexFromDBToSearch(): Promise<void> {
|
protected async reindexFromDBToSearch(findOptions: FindOptions): Promise<void> {
|
||||||
const repository = await this.dbRepository();
|
const repository = await this.dbRepository();
|
||||||
this.statusStart("Start indexing...");
|
this.statusStart("Start indexing...");
|
||||||
let count = 0;
|
let count = 0;
|
||||||
await iterateOverRepoPages(repository, async entities => {
|
await iterateOverRepoPages(
|
||||||
entities = await this.mapEntitiesToReIndexFromDBToSearch(entities);
|
repository,
|
||||||
await this.search.upsert(entities);
|
async entities => {
|
||||||
count += entities.length;
|
entities = await this.mapEntitiesToReIndexFromDBToSearch(entities);
|
||||||
this.statusStart(`Indexed ${count} ${this.table}...`);
|
await this.search.upsert(entities);
|
||||||
});
|
count += entities.length;
|
||||||
|
this.statusStart(`Indexed ${count} ${this.table}...`);
|
||||||
|
},
|
||||||
|
findOptions,
|
||||||
|
);
|
||||||
if (count === 0) this.statusWarn(`Index ${this.table} finished; but 0 items included`);
|
if (count === 0) this.statusWarn(`Index ${this.table} finished; but 0 items included`);
|
||||||
else this.statusSucceed(`${count} ${this.table} indexed`);
|
else this.statusSucceed(`${count} ${this.table} indexed`);
|
||||||
const giveFlushAChanceDurationMS = 10000;
|
const giveFlushAChanceDurationMS = 10000;
|
||||||
@@ -120,8 +139,9 @@ abstract class ReindexerCLICommand<Entity> {
|
|||||||
|
|
||||||
/** Run both operations: repair if requested and re-index */
|
/** Run both operations: repair if requested and re-index */
|
||||||
public async run(): Promise<void> {
|
public async run(): Promise<void> {
|
||||||
if (this.options.repairEntities) await this.repairEntities();
|
const findOptions = await this.prepareFindOptionsForItemsToReIndex();
|
||||||
await this.reindexFromDBToSearch();
|
if (this.options.repairEntities) await this.repairEntities(findOptions);
|
||||||
|
await this.reindexFromDBToSearch(findOptions);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -136,7 +156,7 @@ class UserReindexerCLICommand extends ReindexerCLICommand<User> {
|
|||||||
super(platform, options, UserTYPE, User);
|
super(platform, options, UserTYPE, User);
|
||||||
}
|
}
|
||||||
|
|
||||||
protected override async repairEntities(): Promise<void> {
|
protected override async repairEntities(_findOptions: FindOptions): Promise<void> {
|
||||||
this.statusStart("repairEntities > Adding companies to cache of user");
|
this.statusStart("repairEntities > Adding companies to cache of user");
|
||||||
const companiesUsersRepository = await this.database.getRepository(
|
const companiesUsersRepository = await this.database.getRepository(
|
||||||
CompanyUserTYPE,
|
CompanyUserTYPE,
|
||||||
|
|||||||
Reference in New Issue
Block a user