From b0e1aa76205fc86cb642460046921f8e74a57a7f Mon Sep 17 00:00:00 2001
From: Patrik Oldsberg <poldsberg@gmail.com>
Date: Wed, 22 Sep 2021 17:35:52 +0200
Subject: [PATCH] catalog-backend: add scoped readLocation cache
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-authored-by: Fredrik Adelöw <freben@gmail.com>
Co-authored-by: blam <ben@blam.sh>
Co-authored-by: Johan Haals <johan.haals@gmail.com>
Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>
---
 .../DefaultCatalogProcessingOrchestrator.ts   |  2 +-
 .../processing/ProcessorCacheManager.test.ts  | 41 +++++++++++++
 .../next/processing/ProcessorCacheManager.ts  | 59 +++++++++++++++++--
 3 files changed, 97 insertions(+), 5 deletions(-)

diff --git a/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts
index 01640f34b1..6261c0e2ff 100644
--- a/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts
+++ b/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts
@@ -318,7 +318,7 @@ export class DefaultCatalogProcessingOrchestrator
               false,
               context.collector.onEmit,
               this.options.parser,
-              context.cache.forProcessor(processor),
+              context.cache.forProcessor(processor, target),
             );
             if (read) {
               didRead = true;
diff --git a/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts b/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts
index e7ee27b667..44c96da1ca 100644
--- a/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts
+++ b/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts
@@ -75,3 +75,44 @@ describe('ProcessorCacheManager', () => {
     ).resolves.toBe('my-new-value');
   });
 });
+
+describe('ScopedProcessorCache', () => {
+  const myProcessor = new MyProcessor();
+
+  it('should forward existing state and collect new state', async () => {
+    const cache = new ProcessorCacheManager({
+      'my-processor': { 'scope-1': { 'my-key': 'my-value' } },
+    });
+
+    const scopedCache1 = cache.forProcessor(myProcessor, 'scope-1');
+    const scopedCache2 = cache.forProcessor(myProcessor, 'scope-2');
+
+    // Should be empty to begin with
+    expect(cache.collect()).toEqual({
+      'my-processor': { 'scope-1': { 'my-key': 'my-value' } },
+    });
+
+    await scopedCache2.set('my-new-key-2', 'my-new-value-2');
+
+    expect(cache.collect()).toEqual({
+      'my-processor': {
+        'scope-1': { 'my-key': 'my-value' },
+        'scope-2': { 'my-new-key-2': 'my-new-value-2' },
+      },
+    });
+
+    await scopedCache1.set('my-new-key', 'my-new-value');
+
+    await expect(scopedCache1.get<string>('my-key')).resolves.toBe('my-value');
+    await expect(
+      scopedCache1.get<string>('my-new-key'),
+    ).resolves.toBeUndefined();
+
+    expect(cache.collect()).toEqual({
+      'my-processor': {
+        'scope-1': { 'my-new-key': 'my-new-value' },
+        'scope-2': { 'my-new-key-2': 'my-new-value-2' },
+      },
+    });
+  });
+});
diff --git a/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts b/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts
index 5da466fe86..16945b7cc9 100644
--- a/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts
+++ b/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts
@@ -19,7 +19,7 @@ import { CatalogProcessor } from '../../ingestion/processors';
 import { CatalogProcessorCache } from '../../ingestion/processors/types';
 import { isObject } from './util';
 
-class SingleProcessorCache implements CatalogProcessorCache {
+class SingleProcessorSubCache implements CatalogProcessorCache {
   private newState?: JsonObject;
 
   constructor(private readonly existingState?: JsonObject) {}
@@ -46,17 +46,68 @@ class SingleProcessorCache implements CatalogProcessorCache {
   }
 }
 
+class SingleProcessorCache implements CatalogProcessorCache {
+  private newState?: JsonObject;
+  private subCaches: Map<string, SingleProcessorSubCache> = new Map();
+
+  constructor(private readonly existingState?: JsonObject) {}
+
+  async get<ItemType extends JsonValue>(
+    key: string,
+  ): Promise<ItemType | undefined> {
+    return this.existingState?.[key] as ItemType | undefined;
+  }
+
+  async set<ItemType extends JsonValue>(
+    key: string,
+    value: ItemType,
+  ): Promise<void> {
+    if (!this.newState) {
+      this.newState = {};
+    }
+
+    this.newState[key] = value;
+  }
+
+  withKey(key: string) {
+    const existingSubCache = this.subCaches.get(key);
+    if (existingSubCache) {
+      return existingSubCache;
+    }
+    const existing = this.existingState?.[key];
+    const subCache = new SingleProcessorSubCache(
+      isObject(existing) ? existing : undefined,
+    );
+    this.subCaches.set(key, subCache);
+    return subCache;
+  }
+
+  collect(): JsonObject | undefined {
+    let obj = this.newState ?? this.existingState;
+    for (const [key, subCache] of this.subCaches) {
+      const subCacheValue = subCache.collect();
+      if (subCacheValue) {
+        obj = { ...obj, [key]: subCacheValue };
+      }
+    }
+    return obj;
+  }
+}
+
 export class ProcessorCacheManager {
   private caches = new Map<string, SingleProcessorCache>();
 
   constructor(private readonly existingState: JsonObject) {}
 
-  forProcessor(processor: CatalogProcessor): CatalogProcessorCache {
+  forProcessor(
+    processor: CatalogProcessor,
+    key?: string,
+  ): CatalogProcessorCache {
     // constructor name will be deprecated in the future when we make `getProcessorName` required in the implementation
     const name = processor.getProcessorName?.() ?? processor.constructor.name;
     const cache = this.caches.get(name);
     if (cache) {
-      return cache;
+      return key ? cache.withKey(key) : cache;
     }
 
     const existing = this.existingState[name];
@@ -65,7 +116,7 @@ export class ProcessorCacheManager {
       isObject(existing) ? existing : undefined,
     );
     this.caches.set(name, newCache);
-    return newCache;
+    return key ? newCache.withKey(key) : newCache;
   }
 
   collect(): JsonObject {