make refreshByRefreshKeys accept an array of keys

Signed-off-by: Kiss Miklos <miklos@roadie.io>

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(

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<DbRefreshKeysRow>('refresh_keys')
-      .where({ key })
-      .select({
-        entity_ref: 'refresh_keys.entity_ref',
-      });
+    const query = await tx<DbRefreshStateRow>('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<void> {
-    const tx = txOpaque as Knex.Transaction;
-    const { key } = options;
-
-    await tx<DbRefreshKeysRow>('refresh_keys').where({ key: key }).delete();
-  }
-
   async transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T> {
     try {
       let result: T | undefined = undefined;

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<void>;
 
-  /**
-   * Schedules a refresh for all the entities that have the given refreshKey
-   */
-  setRefreshKeys(
-    txOpaque: Transaction,
-    options: RefreshKeyOptions,
-  ): Promise<void>;
-
   /**
    * Lists all ancestors of a given entityRef.
    *

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);
     });
   }
 }

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.