[#IOPLT-325] Add Flatten Mapping and refactor enrichment

src/enrichment/__tests__/table.test.ts

@@ -91,4 +91,17 @@ describe("tableEnrich", () => {
       )
     )();
   });
+
+  it("should return flattened enriched input if table document is found and output field name is not provided", async () => {
+    getTableDocumentMock.mockImplementationOnce(() =>
+      TE.right(O.some({ baz: "baz" }))
+    );
+    await pipe(
+      tableEnrich(input, tableClientMock, "partitionKey", "rowKey"),
+      TE.bimap(
+        () => fail("it should not fail"),
+        result => expect(result).toEqual({ ...input, baz: "baz" })
+      )
+    )();
+  });
 });

src/enrichment/blob.ts

@@ -6,6 +6,7 @@ import * as BS from "@azure/storage-blob";
 import { errorsToReadableMessages } from "@pagopa/ts-commons/lib/reporters";
 
 import { NonEmptyString } from "@pagopa/ts-commons/lib/strings";
+import { flattenField } from "../formatter/flatten";
 import { getBlobDocument } from "../utils/blobStorage";
 import { NotInKeys } from "../utils/types";
 import { toJsonObject } from "../utils/data";
@@ -36,10 +37,17 @@ export const blobEnrich = <T, K extends string>(
             outputFieldName,
             O.fromNullable,
             O.map(fieldName => ({ ...input, [fieldName]: blobDocumentObject })),
-            O.getOrElseW(() => ({
-              ...input,
-              ...blobDocumentObject
-            }))
+            O.getOrElseW(() =>
+              pipe(`${String(blobNameField)}_enrich`, flattenFieldName =>
+                flattenField(
+                  {
+                    ...input,
+                    [flattenFieldName]: blobDocumentObject
+                  },
+                  flattenFieldName
+                )
+              )
+            )
           )
         ),
         O.getOrElse(() => input)

src/enrichment/table.ts

@@ -6,6 +6,7 @@ import * as DT from "@azure/data-tables";
 import * as t from "io-ts";
 import { NonEmptyString } from "@pagopa/ts-commons/lib/strings";
 import { errorsToReadableMessages } from "@pagopa/ts-commons/lib/reporters";
+import { flattenField } from "../formatter/flatten";
 import { getTableDocument } from "../utils/tableStorage";
 import { NotInKeys } from "../utils/types";
 
@@ -44,7 +45,19 @@ export const tableEnrich = <T, K extends string>(
             outputFieldName,
             O.fromNullable,
             O.map(fieldName => ({ ...input, [fieldName]: tableDocument })),
-            O.getOrElse(() => ({ ...input, ...tableDocument }))
+            O.getOrElse(() =>
+              pipe(
+                `${String(partitionKeyField)}_${String(rowKeyField)}_enrich`,
+                flattenFieldName =>
+                  flattenField(
+                    {
+                      ...input,
+                      [flattenFieldName]: tableDocument
+                    },
+                    flattenFieldName
+                  )
+              )
+            )
           )
         ),
         O.getOrElse(() => input)

src/formatter/__test__/flatten.test.ts

@@ -0,0 +1,51 @@
+import { withoutUndefinedValues } from "@pagopa/ts-commons/lib/types";
+import { flattenField } from "../flatten";
+
+const aNormalizedField = {
+  bar: 1,
+  baz: false
+};
+
+const aNormalizedFieldWithAlreadyExistingProperty = {
+  ...aNormalizedField,
+  foo: "normalized_foo"
+};
+
+const input = {
+  foo: "foo",
+  normalized: aNormalizedField
+};
+
+const anotherInput = {
+  ...input,
+  normalized: aNormalizedFieldWithAlreadyExistingProperty
+};
+describe("flattenField", () => {
+  it("should return unmodified input if fieldToFlat is not an Object", () => {
+    const result = flattenField(input, "foo");
+    expect(result).toEqual(input);
+  });
+
+  it("should return a flattened input without renaming any field", () => {
+    const result = flattenField(input, "normalized");
+    expect(result).toEqual(
+      withoutUndefinedValues({
+        ...input,
+        normalized: undefined,
+        ...aNormalizedField
+      })
+    );
+  });
+
+  it("should return a flattened input with renaming fields that overlap existing input fields", () => {
+    const result = flattenField(anotherInput, "normalized");
+    expect(result).toEqual(
+      withoutUndefinedValues({
+        ...aNormalizedFieldWithAlreadyExistingProperty,
+        ...anotherInput,
+        normalized: undefined,
+        normalized_foo: aNormalizedFieldWithAlreadyExistingProperty.foo
+      })
+    );
+  });
+});

src/formatter/flatten.ts

@@ -0,0 +1,37 @@
+import * as O from "fp-ts/Option";
+import { withoutUndefinedValues } from "@pagopa/ts-commons/lib/types";
+import { pipe } from "fp-ts/lib/function";
+import { toJsonObject } from "../utils/data";
+
+export const flattenField = <T>(input: T, fieldToFlat: keyof T): T =>
+  pipe(
+    input,
+    O.fromPredicate(stream => stream[fieldToFlat] instanceof Object),
+    O.map(data => pipe(data[fieldToFlat], toJsonObject)),
+    O.map(objToFlat =>
+      pipe(
+        objToFlat,
+        Object.keys,
+        objectFieldNames =>
+          objectFieldNames.reduce(
+            (acc, fieldName) =>
+              Object.keys(input).includes(fieldName)
+                ? {
+                    ...acc,
+                    [`${String(fieldToFlat)}_${fieldName}`]: objToFlat[
+                      fieldName
+                    ]
+                  }
+                : { ...acc, [fieldName]: objToFlat[fieldName] },
+            {}
+          ),
+        flattenedContent => ({ ...input, ...flattenedContent }),
+        result =>
+          withoutUndefinedValues({
+            ...result,
+            [fieldToFlat]: undefined
+          })
+      )
+    ),
+    O.getOrElse(() => input)
+  );

src/utils/data.ts

@@ -1,5 +1,4 @@
-import * as J from "fp-ts/lib/Json";
 import * as R from "fp-ts/Record";
 
-export const toJsonObject = (jsonData: J.Json): Record<string, unknown> =>
+export const toJsonObject = (jsonData: unknown): Record<string, unknown> =>
   R.fromEntries(Object.entries(jsonData));