diff --git a/plugins/catalog-backend/migrations/20220616202842_refresh_keys.js b/plugins/catalog-backend/migrations/20220616202842_refresh_keys.js index dfa0bf5794..8358cbb0cd 100644 --- a/plugins/catalog-backend/migrations/20220616202842_refresh_keys.js +++ b/plugins/catalog-backend/migrations/20220616202842_refresh_keys.js @@ -14,6 +14,9 @@ * limitations under the License. */ +/** + * @param { import("knex").Knex } knex + */ exports.up = async function up(knex) { await knex.schema.createTable('refresh_keys', table => { table.comment( diff --git a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts index 66b87d6b24..3b93e9a222 100644 --- a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts @@ -519,20 +519,28 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { } } - async refreshByRefreshKey( + async refreshByRefreshKeys( txOpaque: Transaction, options: RefreshByKeyOptions, ) { const tx = txOpaque as Knex.Transaction; - const { key } = options; + const { keys } = options; - const rows = await tx('refresh_keys') - .where({ key }) - .select({ - entity_ref: 'refresh_keys.entity_ref', - }); + const query = await tx('refresh_state') + .whereIn('entity_ref', function (tx2) { + tx2 + .whereIn('key', keys) + .select({ + entity_ref: 'refresh_keys.entity_ref', + }) + .from('refresh_keys') + .columns('entity_ref'); + }) + .update({ next_update_at: tx.fn.now() }) + .toSQL() + .toNative(); - await Promise.all(rows.map(r => this.refresh(tx, r.entity_ref))); + console.log(query, '@@@@@@@!!!!!!!@@@@@'); } async setRefreshKeys( @@ -555,16 +563,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { ); } - async deleteRefreshKey( - txOpaque: Transaction, - options: RefreshByKeyOptions, - ): Promise { - const tx = txOpaque as Knex.Transaction; - const { key } = options; - - await tx('refresh_keys').where({ key: key }).delete(); - } - async transaction(fn: (tx: Transaction) => Promise): Promise { try { let result: T | undefined = undefined; diff --git a/plugins/catalog-backend/src/database/types.ts b/plugins/catalog-backend/src/database/types.ts index 23755ad513..4d2b64a721 100644 --- a/plugins/catalog-backend/src/database/types.ts +++ b/plugins/catalog-backend/src/database/types.ts @@ -86,7 +86,7 @@ export type RefreshKeyOptions = { }; export type RefreshByKeyOptions = { - key: string; + keys: string[]; }; export type RefreshOptions = { @@ -165,14 +165,6 @@ export interface ProcessingDatabase { options: RefreshKeyOptions, ): Promise; - /** - * Schedules a refresh for all the entities that have the given refreshKey - */ - setRefreshKeys( - txOpaque: Transaction, - options: RefreshKeyOptions, - ): Promise; - /** * Lists all ancestors of a given entityRef. * diff --git a/plugins/catalog-backend/src/service/DefaultRefreshService.ts b/plugins/catalog-backend/src/service/DefaultRefreshService.ts index d0c2742d97..e5deab1790 100644 --- a/plugins/catalog-backend/src/service/DefaultRefreshService.ts +++ b/plugins/catalog-backend/src/service/DefaultRefreshService.ts @@ -51,7 +51,7 @@ export class DefaultRefreshService implements RefreshService { } async refreshByRefreshKey(options: RefreshByRefreshKeysOptions) { await this.database.transaction(async tx => { - await this.database.refreshByRefreshKey(tx, options); + await this.database.refreshByRefreshKeys(tx, options); }); } } diff --git a/plugins/catalog-backend/src/service/types.ts b/plugins/catalog-backend/src/service/types.ts index cb4debc41d..1e2142fc3a 100644 --- a/plugins/catalog-backend/src/service/types.ts +++ b/plugins/catalog-backend/src/service/types.ts @@ -62,7 +62,7 @@ export type RefreshOptions = { }; export type RefreshByRefreshKeysOptions = { - key: string; + keys: string[]; }; /** * A service that manages refreshes of entities in the catalog.