[#IOPID-2343] Introducing Redis cache for PDV Tokenizer calls

AnalyticsProfileStorageQueueInboundProcessorAdapter/index.ts

@@ -2,6 +2,7 @@ import { Context } from "@azure/functions";
 import { NonEmptyString } from "@pagopa/ts-commons/lib/strings";
 import * as t from "io-ts";
 import { RetrievedProfile } from "@pagopa/io-functions-commons/dist/src/models/profile";
+import { Second } from "@pagopa/ts-commons/lib/units";
 import * as KP from "../utils/kafka/KafkaProducerCompact";
 import { ValidableKafkaProducerConfig } from "../utils/kafka/KafkaTypes";
 import { getConfigOrThrow, withTopic } from "../utils/config";
@@ -15,6 +16,7 @@ import { OutboundEnricher } from "../outbound/port/outbound-enricher";
 import { profilesAvroFormatter } from "../utils/formatter/profilesAvroFormatter";
 import { pdvTokenizerClient } from "../utils/pdvTokenizerClient";
 import { httpOrHttpsApiFetch } from "../utils/fetch";
+import { createRedisClientSingleton } from "../utils/redis";
 
 export type RetrievedProfileWithMaybePdvId = t.TypeOf<
   typeof RetrievedProfileWithMaybePdvId
@@ -52,6 +54,8 @@ const pdvTokenizer = pdvTokenizerClient(
   config.PDV_TOKENIZER_BASE_PATH
 );
 
+const redisClientTask = createRedisClientSingleton(config);
+
 const telemetryClient = TA.initTelemetryClient(
   config.APPINSIGHTS_INSTRUMENTATIONKEY
 );
@@ -60,7 +64,13 @@ const telemetryAdapter = TA.create(telemetryClient);
 
 const pdvIdEnricherAdapter: OutboundEnricher<RetrievedProfileWithMaybePdvId> = PDVA.create<
   RetrievedProfileWithMaybePdvId
->(config.ENRICH_PDVID_THROTTLING, pdvTokenizer, telemetryClient);
+>(
+  config.ENRICH_PDVID_THROTTLING,
+  pdvTokenizer,
+  redisClientTask,
+  config.PDV_IDS_TTL as Second,
+  telemetryClient
+);
 
 const run = (_context: Context, document: unknown): Promise<void> =>
   getAnalyticsProcessorForDocuments(

AnalyticsProfilesChangeFeedInboundProcessorAdapter/index.ts

@@ -6,6 +6,7 @@ import * as t from "io-ts";
 import { RetrievedProfile } from "@pagopa/io-functions-commons/dist/src/models/profile";
 import { FiscalCode, NonEmptyString } from "@pagopa/ts-commons/lib/strings";
 
+import { Second } from "@pagopa/ts-commons/lib/units";
 import * as KA from "../outbound/adapter/kafka-outbound-publisher";
 import * as KP from "../utils/kafka/KafkaProducerCompact";
 import * as QA from "../outbound/adapter/queue-outbound-mapper-publisher";
@@ -23,6 +24,7 @@ import { OutboundFilterer } from "../outbound/port/outbound-filterer";
 import { profilesAvroFormatter } from "../utils/formatter/profilesAvroFormatter";
 import { httpOrHttpsApiFetch } from "../utils/fetch";
 import { pdvTokenizerClient } from "../utils/pdvTokenizerClient";
+import { createRedisClientSingleton } from "../utils/redis";
 
 export type RetrievedProfileWithMaybePdvId = t.TypeOf<
   typeof RetrievedProfileWithMaybePdvId
@@ -71,13 +73,21 @@ const pdvTokenizer = pdvTokenizerClient(
   config.PDV_TOKENIZER_BASE_PATH
 );
 
+const redisClientTask = createRedisClientSingleton(config);
+
 const telemetryClient = TA.initTelemetryClient(
   config.APPINSIGHTS_INSTRUMENTATIONKEY
 );
 
 const pdvIdEnricherAdapter: OutboundEnricher<RetrievedProfileWithMaybePdvId> = PDVA.create<
   RetrievedProfileWithMaybePdvId
->(config.ENRICH_PDVID_THROTTLING, pdvTokenizer, telemetryClient);
+>(
+  config.ENRICH_PDVID_THROTTLING,
+  pdvTokenizer,
+  redisClientTask,
+  config.PDV_IDS_TTL as Second,
+  telemetryClient
+);
 
 const telemetryAdapter = TA.create(telemetryClient);
 

AnalyticsServicePreferencesChangeFeedInboundProcessorAdapter/index.ts

@@ -4,6 +4,7 @@ import { Context } from "@azure/functions";
 import { RetrievedServicePreference } from "@pagopa/io-functions-commons/dist/src/models/service_preference";
 import { FiscalCode } from "@pagopa/ts-commons/lib/strings";
 
+import { Second } from "@pagopa/ts-commons/lib/units";
 import * as KA from "../outbound/adapter/kafka-outbound-publisher";
 import * as KP from "../utils/kafka/KafkaProducerCompact";
 import * as QA from "../outbound/adapter/queue-outbound-mapper-publisher";
@@ -22,6 +23,7 @@ import { OutboundFilterer } from "../outbound/port/outbound-filterer";
 import { RetrievedServicePreferenceWithMaybePdvId } from "../utils/types/decoratedTypes";
 import { pdvTokenizerClient } from "../utils/pdvTokenizerClient";
 import { httpOrHttpsApiFetch } from "../utils/fetch";
+import { createRedisClientSingleton } from "../utils/redis";
 
 const config = getConfigOrThrow();
 
@@ -63,13 +65,21 @@ const pdvTokenizer = pdvTokenizerClient(
   config.PDV_TOKENIZER_BASE_PATH
 );
 
+const redisClientTask = createRedisClientSingleton(config);
+
 const telemetryClient = TA.initTelemetryClient(
   config.APPINSIGHTS_INSTRUMENTATIONKEY
 );
 
 const pdvIdEnricherAdapter: OutboundEnricher<RetrievedServicePreferenceWithMaybePdvId> = PDVA.create<
   RetrievedServicePreferenceWithMaybePdvId
->(config.ENRICH_PDVID_THROTTLING, pdvTokenizer, telemetryClient);
+>(
+  config.ENRICH_PDVID_THROTTLING,
+  pdvTokenizer,
+  redisClientTask,
+  config.PDV_IDS_TTL as Second,
+  telemetryClient
+);
 
 const telemetryAdapter = TA.create(telemetryClient);
 

AnalyticsServicePreferencesStorageQueueInboundProcessorAdapter/index.ts

@@ -1,5 +1,6 @@
 import { Context } from "@azure/functions";
 import { RetrievedServicePreference } from "@pagopa/io-functions-commons/dist/src/models/service_preference";
+import { Second } from "@pagopa/ts-commons/lib/units";
 import * as KP from "../utils/kafka/KafkaProducerCompact";
 import { ValidableKafkaProducerConfig } from "../utils/kafka/KafkaTypes";
 import { getConfigOrThrow, withTopic } from "../utils/config";
@@ -14,6 +15,7 @@ import { servicePreferencesAvroFormatter } from "../utils/formatter/servicePrefe
 import { RetrievedServicePreferenceWithMaybePdvId } from "../utils/types/decoratedTypes";
 import { pdvTokenizerClient } from "../utils/pdvTokenizerClient";
 import { httpOrHttpsApiFetch } from "../utils/fetch";
+import { createRedisClientSingleton } from "../utils/redis";
 
 const config = getConfigOrThrow();
 
@@ -44,6 +46,8 @@ const pdvTokenizer = pdvTokenizerClient(
   config.PDV_TOKENIZER_BASE_PATH
 );
 
+const redisClientTask = createRedisClientSingleton(config);
+
 const telemetryClient = TA.initTelemetryClient(
   config.APPINSIGHTS_INSTRUMENTATIONKEY
 );
@@ -52,7 +56,13 @@ const telemetryAdapter = TA.create(telemetryClient);
 
 const pdvIdEnricherAdapter: OutboundEnricher<RetrievedServicePreferenceWithMaybePdvId> = PDVA.create<
   RetrievedServicePreferenceWithMaybePdvId
->(config.ENRICH_PDVID_THROTTLING, pdvTokenizer, telemetryClient);
+>(
+  config.ENRICH_PDVID_THROTTLING,
+  pdvTokenizer,
+  redisClientTask,
+  config.PDV_IDS_TTL as Second,
+  telemetryClient
+);
 
 const run = (_context: Context, document: unknown): Promise<void> =>
   getAnalyticsProcessorForDocuments(

businesslogic/__mocks__/processor.mock.ts

@@ -12,6 +12,9 @@ import { OutboundPublisher } from "../../outbound/port/outbound-publisher";
 import * as pdv from "../../utils/pdv";
 import { Client } from "../../generated/pdv-tokenizer-api/client";
 import { aKafkaResponse, aMockPdvId, aTopic } from "./data.mock";
+import { RedisClientType } from "redis";
+import * as TE from "fp-ts/lib/TaskEither";
+import { Second } from "@pagopa/ts-commons/lib/units";
 
 // Mocks
 export const mockTrackException = jest.fn(_ => void 0);
@@ -39,6 +42,19 @@ export const producerMock = () => ({
   topic: { topic: aTopic }
 });
 
+// Redis mock
+const mockSet = jest.fn().mockResolvedValue("OK");
+// DEFAULT BEHAVIOUR: redis doesn't contain the value in the cache
+const mockGet = jest.fn().mockResolvedValue(undefined);
+const mockRedisClient = ({
+  set: mockSet,
+  setEx: mockSet,
+  get: mockGet
+} as unknown) as RedisClientType;
+
+const mockPDVIdsTTL = 30 as Second;
+//
+
 export const mockGetPdvId = jest
   .spyOn(pdv, "getPdvId")
   .mockReturnValue(RTE.right(aMockPdvId));
@@ -51,6 +67,8 @@ export const getPdvIdEnricherAdapter = <
   PDVA.create<T>(
     10,
     ({} as unknown) as Client /* functionality mocked by mockGetPdvId */,
+    TE.right(mockRedisClient),
+    mockPDVIdsTTL,
     trackerMock
   );
 

docker-compose.yml

@@ -19,6 +19,7 @@ services:
       - io-fn
     depends_on:
       - fnstorage
+      - redis
     volumes:
       - .:/usr/src/app
     labels:
@@ -61,6 +62,19 @@ services:
     networks:
       - io-fn
 
+  redis:
+    image: redis/redis-stack:6.2.6-v17
+    env_file:
+      - .env
+    environment:
+      REDIS_ARGS: "--requirepass ${REDIS_PASSWORD}"
+    ports:
+      - "6379:6379"
+      # redis insights visible in browser
+      - "8001:8001"
+    networks:
+      - io-fn
+
   cosmosdb:
     image: cosmosdb
     env_file:

env.example

@@ -25,3 +25,9 @@ SERVICEID_EXCLUSION_LIST=ServiceId1,ServiceId2
 
 # needed to connect to cosmosdb server
 NODE_TLS_REJECT_UNAUTHORIZED=0
+
+# Redis config
+REDIS_URL="redis"
+REDIS_PASSWORD="foo"
+REDIS_TLS_ENABLED=false
+REDIS_PORT=6379

local.settings.json.example

@@ -37,6 +37,11 @@
     "PROFILES_LEASES_PREFIX" : "profiles-001",
     "PROFILES_FAILURE_QUEUE_NAME" : "<Queue Name>-profiles",
 
-    "SERVICEID_EXCLUSION_LIST": "ServiceId1,ServiceId2"
+    "SERVICEID_EXCLUSION_LIST": "ServiceId1,ServiceId2",
+
+    "REDIS_URL": "localhost",
+    "REDIS_PORT": 6379,
+    "REDIS_PASSWORD": "foo",
+    "REDIS_TLS_ENABLED": false
   }
 }

outbound/adapter/__tests__/pdv-id-outbound-enricher.test.ts

@@ -13,6 +13,10 @@ import {
   aRetrievedServicePreferences,
   aRetrievedServicePreferencesList
 } from "../../../businesslogic/__mocks__/data.mock";
+import { RedisClientType } from "redis";
+import * as TE from "fp-ts/lib/TaskEither";
+import { Second } from "@pagopa/ts-commons/lib/units";
+import { PDVIdPrefix } from "../../../utils/redis";
 
 const mockSave = jest
   .fn()
@@ -27,7 +31,26 @@ const mockTokenizerClient = ({
   saveUsingPUT: mockSave
 } as unknown) as Client;
 
-const enricher = create(2, mockTokenizerClient, mockTelemetryClient);
+// Redis mock
+const mockSet = jest.fn().mockResolvedValue("OK");
+// DEFAULT BEHAVIOUR: redis doesn't contain the value in the cache
+const mockGet = jest.fn().mockResolvedValue(undefined);
+const mockRedisClient = ({
+  set: mockSet,
+  setEx: mockSet,
+  get: mockGet
+} as unknown) as RedisClientType;
+
+const mockPDVIdsTTL = 30 as Second;
+//
+
+const enricher = create(
+  2,
+  mockTokenizerClient,
+  TE.right(mockRedisClient),
+  mockPDVIdsTTL,
+  mockTelemetryClient
+);
 
 describe.each`
   title                        | value                               | isList   | length
@@ -140,3 +163,99 @@ describe.each`
     );
   });
 });
+
+describe("Redis cache introduction", () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  const call = enricher.enrich(aRetrievedProfile);
+
+  it("GIVEN a valid document WHEN the cache already has the token THEN PDV Tokenizer should not be called", async () => {
+    mockGet.mockResolvedValueOnce(aMockPdvId);
+
+    const result = await call();
+    expect(result).toStrictEqual(
+      E.right({
+        ...aRetrievedProfile,
+        userPDVId: aMockPdvId
+      })
+    );
+    expect(mockGet).toHaveBeenCalledTimes(1);
+    expect(mockSave).not.toHaveBeenCalled();
+    expect(mockSet).not.toHaveBeenCalled();
+    expect(mockTrackEvent).not.toHaveBeenCalled();
+  });
+
+  it("GIVEN a valid document WHEN the cache doesn't hold the token THEN PDV Tokenizer should be called along with the cache", async () => {
+    mockGet.mockResolvedValueOnce(undefined);
+
+    const result = await call();
+    expect(result).toStrictEqual(
+      E.right({
+        ...aRetrievedProfile,
+        userPDVId: aMockPdvId
+      })
+    );
+    expect(mockGet).toHaveBeenCalledTimes(1);
+    expect(mockSave).toHaveBeenCalledTimes(1);
+    expect(mockSet).toHaveBeenCalledTimes(1);
+    expect(mockSet).toHaveBeenCalledWith(
+      `${PDVIdPrefix}${aRetrievedProfile.fiscalCode}`,
+      mockPDVIdsTTL,
+      aMockPdvId
+    );
+    expect(mockTrackEvent).not.toHaveBeenCalled();
+  });
+
+  it("GIVEN a valid document WHEN the cache can't be reached THEN an error should be returned", async () => {
+    const faultyEnricher = create(
+      2,
+      mockTokenizerClient,
+      TE.left(Error("error")),
+      mockPDVIdsTTL,
+      mockTelemetryClient
+    );
+    const result = await faultyEnricher.enrich(aRetrievedProfile)();
+    expect(result).toStrictEqual(E.left(Error("error")));
+    expect(mockGet).not.toHaveBeenCalled();
+    expect(mockSave).not.toHaveBeenCalled();
+    expect(mockSet).not.toHaveBeenCalled();
+    expect(mockTrackEvent).toHaveBeenCalledTimes(1);
+  });
+
+  it("GIVEN a valid document WHEN the cache can't reach redis to save the token THEN an error should be returned", async () => {
+    mockSet.mockRejectedValueOnce(Error("error"));
+
+    const result = await call();
+
+    expect(result).toStrictEqual(E.left(Error("error")));
+    expect(mockGet).toHaveBeenCalledTimes(1);
+    expect(mockSave).toHaveBeenCalledTimes(1);
+    expect(mockSet).toHaveBeenCalledTimes(1);
+    expect(mockSet).toHaveBeenCalledWith(
+      `${PDVIdPrefix}${aRetrievedProfile.fiscalCode}`,
+      mockPDVIdsTTL,
+      aMockPdvId
+    );
+    expect(mockTrackEvent).toHaveBeenCalledTimes(1);
+  });
+
+  it("GIVEN a valid document WHEN the cache can't save the token THEN an error should be returned", async () => {
+    mockSet.mockResolvedValueOnce("KO");
+
+    const result = await call();
+
+    expect(result).toStrictEqual(E.left(Error("Error saving the key")));
+    expect(mockGet).toHaveBeenCalledTimes(1);
+    expect(mockSave).toHaveBeenCalledTimes(1);
+    expect(mockSet).toHaveBeenCalledTimes(1);
+
+    expect(mockSet).toHaveBeenCalledWith(
+      `${PDVIdPrefix}${aRetrievedProfile.fiscalCode}`,
+      mockPDVIdsTTL,
+      aMockPdvId
+    );
+    expect(mockTrackEvent).toHaveBeenCalledTimes(1);
+  });
+});