From 86ec89cacd36b850a6a29887a86833cc0cc28e8c Mon Sep 17 00:00:00 2001
From: Milosz Korman <milosz.korman@allegro.pl>
Date: Wed, 16 Oct 2024 12:57:12 +0200
Subject: [PATCH] resolve #1907 | handle record references when presenting avro
 schema

---
 hermes-console/json-server/db.json            |   2 +-
 hermes-console/json-server/topics.json        |   2 +-
 hermes-console/src/dummy/topic.ts             |   2 +-
 .../src/views/topic/schema-panel/AvroTypes.ts |   1 +
 .../schema-panel/avro-viewer/AvroNode.vue     |  20 +-
 .../schema-panel/avro-viewer/AvroViewer.vue   |   9 +-
 .../avro-record-types-registry.spec.ts        | 340 ++++++++++++++++++
 .../avro-viewer/avro-record-types-registry.ts | 122 +++++++
 8 files changed, 491 insertions(+), 7 deletions(-)
 create mode 100644 hermes-console/src/views/topic/schema-panel/avro-viewer/avro-record-types-registry.spec.ts
 create mode 100644 hermes-console/src/views/topic/schema-panel/avro-viewer/avro-record-types-registry.ts

diff --git a/hermes-console/json-server/db.json b/hermes-console/json-server/db.json
index 12f97d5dd3..061e982505 100644
--- a/hermes-console/json-server/db.json
+++ b/hermes-console/json-server/db.json
@@ -190,7 +190,7 @@
   "topics": [
     {
       "id": "pl.allegro.public.group.DummyEvent",
-      "schema": "{\"type\":\"record\",\"name\":\"DummyEvent\",\"namespace\":\"pl.allegro.public.group.DummyEvent\",\"doc\":\"Event emitted when notification is sent to a user\",\"fields\":[{\"name\":\"__metadata\",\"type\":[\"null\",{\"type\":\"map\",\"values\":\"string\"}],\"doc\":\"Field internally used by Hermes to propagate metadata.\",\"default\":null},{\"name\":\"waybillId\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"WaybillId\",\"fields\":[{\"name\":\"waybill\",\"type\":\"string\",\"doc\":\"Waybill\"},{\"name\":\"carrierId\",\"type\":\"string\",\"doc\":\"CarrierId\"}]}],\"doc\":\"WaybillId\",\"default\":null},{\"name\":\"notificationId\",\"type\":\"string\",\"doc\":\"Notification Id\"},{\"name\":\"userId\",\"type\":\"string\",\"doc\":\"User Id\"}]}",
+      "schema": "{\"type\":\"record\",\"name\":\"DummyEvent\",\"namespace\":\"pl.allegro.public.group.DummyEvent\",\"doc\":\"Event emitted when notification is sent to a user\",\"fields\":[{\"name\":\"__metadata\",\"type\":[\"null\",{\"type\":\"map\",\"values\":\"string\"}],\"doc\":\"Field internally used by Hermes to propagate metadata.\",\"default\":null},{\"name\":\"waybillId\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"WaybillId\",\"fields\":[{\"name\":\"waybill\",\"type\":\"string\",\"doc\":\"Waybill\"},{\"name\":\"carrierId\",\"type\":\"string\",\"doc\":\"CarrierId\"}]}],\"doc\":\"WaybillId\",\"default\":null},{\"name\":\"otherWaybillId\",\"type\":[\"null\",\"WaybillId\"],\"doc\":\"other WaybillId\",\"default\":null},{\"name\":\"notificationId\",\"type\":\"string\",\"doc\":\"Notification Id\"},{\"name\":\"userId\",\"type\":\"string\",\"doc\":\"User Id\"}]}",
       "name": "pl.allegro.public.group.DummyEvent",
       "description": "Events emitted when notification is sent to a user.",
       "owner": {
diff --git a/hermes-console/json-server/topics.json b/hermes-console/json-server/topics.json
index bea9c1b3e6..ef667b9756 100644
--- a/hermes-console/json-server/topics.json
+++ b/hermes-console/json-server/topics.json
@@ -1,7 +1,7 @@
 [
   {
     "id": "pl.allegro.public.group.DummyEvent",
-    "schema": "{\"type\":\"record\",\"name\":\"DummyEvent\",\"namespace\":\"pl.allegro.public.group.DummyEvent\",\"doc\":\"Event emitted when notification is sent to a user\",\"fields\":[{\"name\":\"__metadata\",\"type\":[\"null\",{\"type\":\"map\",\"values\":\"string\"}],\"doc\":\"Field internally used by Hermes to propagate metadata.\",\"default\":null},{\"name\":\"waybillId\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"WaybillId\",\"fields\":[{\"name\":\"waybill\",\"type\":\"string\",\"doc\":\"Waybill\"},{\"name\":\"carrierId\",\"type\":\"string\",\"doc\":\"CarrierId\"}]}],\"doc\":\"WaybillId\",\"default\":null},{\"name\":\"notificationId\",\"type\":\"string\",\"doc\":\"Notification Id\"},{\"name\":\"userId\",\"type\":\"string\",\"doc\":\"User Id\"}]}",
+    "schema": "{\"type\":\"record\",\"name\":\"DummyEvent\",\"namespace\":\"pl.allegro.public.group.DummyEvent\",\"doc\":\"Event emitted when notification is sent to a user\",\"fields\":[{\"name\":\"__metadata\",\"type\":[\"null\",{\"type\":\"map\",\"values\":\"string\"}],\"doc\":\"Field internally used by Hermes to propagate metadata.\",\"default\":null},{\"name\":\"waybillId\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"WaybillId\",\"fields\":[{\"name\":\"waybill\",\"type\":\"string\",\"doc\":\"Waybill\"},{\"name\":\"carrierId\",\"type\":\"string\",\"doc\":\"CarrierId\"}]}],\"doc\":\"WaybillId\",\"default\":null},{\"name\":\"otherWaybillId\",\"type\":[\"null\",\"WaybillId\"],\"doc\":\"other WaybillId\",\"default\":null},{\"name\":\"notificationId\",\"type\":\"string\",\"doc\":\"Notification Id\"},{\"name\":\"userId\",\"type\":\"string\",\"doc\":\"User Id\"}]}",
     "name": "pl.allegro.public.group.DummyEvent",
     "description": "Events emitted when notification is sent to a user.",
     "owner": {
diff --git a/hermes-console/src/dummy/topic.ts b/hermes-console/src/dummy/topic.ts
index 4697e7ddac..e02fb9fe92 100644
--- a/hermes-console/src/dummy/topic.ts
+++ b/hermes-console/src/dummy/topic.ts
@@ -11,7 +11,7 @@ import type { Owner } from '@/api/owner';
 
 export const dummyTopic: TopicWithSchema = {
   schema:
-    '{"type":"record","name":"DummyEvent","namespace":"pl.allegro.public.group.DummyEvent","doc":"Event emitted when notification is sent to a user","fields":[{"name":"__metadata","type":["null",{"type":"map","values":"string"}],"doc":"Field internally used by Hermes to propagate metadata.","default":null},{"name":"waybillId","type":["null",{"type":"record","name":"WaybillId","fields":[{"name":"waybill","type":"string","doc":"Waybill"},{"name":"carrierId","type":"string","doc":"CarrierId"}]}],"doc":"WaybillId","default":null},{"name":"notificationId","type":"string","doc":"Notification Id"},{"name":"userId","type":"string","doc":"User Id"}]}',
+    '{"type":"record","name":"DummyEvent","namespace":"pl.allegro.public.group.DummyEvent","doc":"Event emitted when notification is sent to a user","fields":[{"name":"__metadata","type":["null",{"type":"map","values":"string"}],"doc":"Field internally used by Hermes to propagate metadata.","default":null},{"name":"waybillId","type":["null",{"type":"record","name":"WaybillId","fields":[{"name":"waybill","type":"string","doc":"Waybill"},{"name":"carrierId","type":"string","doc":"CarrierId"}]}],"doc":"WaybillId","default":null},{"name":"otherWaybillId","type":["null","WaybillId"],"doc":"other WaybillId","default":null},{"name":"notificationId","type":"string","doc":"Notification Id"},{"name":"userId","type":"string","doc":"User Id"}]}',
   name: 'pl.allegro.public.group.DummyEvent',
   description: 'Events emitted when notification is sent to a user.',
   owner: {
diff --git a/hermes-console/src/views/topic/schema-panel/AvroTypes.ts b/hermes-console/src/views/topic/schema-panel/AvroTypes.ts
index 79cee71078..42f9027f32 100644
--- a/hermes-console/src/views/topic/schema-panel/AvroTypes.ts
+++ b/hermes-console/src/views/topic/schema-panel/AvroTypes.ts
@@ -1,6 +1,7 @@
 export interface AvroSchema {
   type: Type;
   name: string;
+  namespace?: string;
   doc: string;
   fields?: Field[];
   default?: DefaultTypes;
diff --git a/hermes-console/src/views/topic/schema-panel/avro-viewer/AvroNode.vue b/hermes-console/src/views/topic/schema-panel/avro-viewer/AvroNode.vue
index 4782218ed0..38adaed0f9 100644
--- a/hermes-console/src/views/topic/schema-panel/avro-viewer/AvroNode.vue
+++ b/hermes-console/src/views/topic/schema-panel/avro-viewer/AvroNode.vue
@@ -1,9 +1,15 @@
 <script lang="ts" setup>
   import { defineProps, ref } from 'vue';
-  import type { AvroSchema, Type } from '@/views/topic/schema-panel/AvroTypes';
+  import { getRecordTypesRegistryEntry } from '@/views/topic/schema-panel/avro-viewer/avro-record-types-registry';
+  import type {
+    AvroSchema,
+    RecordType,
+    Type,
+  } from '@/views/topic/schema-panel/AvroTypes';
 
   const props = defineProps<{
     field: AvroSchema;
+    recordReferenceTypes: Map<string, RecordType>;
     root: boolean;
   }>();
   const expanded = ref(true);
@@ -62,10 +68,17 @@
   };
 
   const types = getTypes(props.field.type);
-  const isRecord = types.some((type: Type) => type.includes('record'));
+  const recordTypeReference = getRecordTypesRegistryEntry(
+    props.recordReferenceTypes,
+    props.field.type,
+  );
+  const isRecord =
+    !!recordTypeReference ||
+    types.some((type: Type) => type.includes('record'));
   const isEnum = types.some((type: Type) => type.includes('enum'));
   const expandable = isRecord || isEnum;
-  const nestedType = isRecord && findNestedType(props.field.type);
+  const nestedType =
+    isRecord && (recordTypeReference || findNestedType(props.field.type));
   const enumSymbols = isEnum && findEnumSymbols(props.field.type);
 
   const toggle = (event: Event) => {
@@ -100,6 +113,7 @@
         v-for="nestedField in nestedType.fields"
         :key="nestedField.name"
         :field="nestedField"
+        :record-reference-types="props.recordReferenceTypes"
         :root="false"
       />
     </div>
diff --git a/hermes-console/src/views/topic/schema-panel/avro-viewer/AvroViewer.vue b/hermes-console/src/views/topic/schema-panel/avro-viewer/AvroViewer.vue
index 3b3d692f9f..784b77ef62 100644
--- a/hermes-console/src/views/topic/schema-panel/avro-viewer/AvroViewer.vue
+++ b/hermes-console/src/views/topic/schema-panel/avro-viewer/AvroViewer.vue
@@ -1,4 +1,5 @@
 <script lang="ts" setup>
+  import { createRecordTypesRegistry } from '@/views/topic/schema-panel/avro-viewer/avro-record-types-registry';
   import AvroNode from '@/views/topic/schema-panel/avro-viewer/AvroNode.vue';
   import type { AvroSchema } from '@/views/topic/schema-panel/AvroTypes';
   const props = defineProps<{
@@ -9,6 +10,7 @@
     const jsonSchema: AvroSchema = JSON.parse(props.schema);
     return {
       name: jsonSchema.name,
+      namespace: jsonSchema.namespace,
       doc: jsonSchema.doc,
       type: {
         type: 'record',
@@ -22,7 +24,12 @@
 
 <template>
   <div class="avro-schema">
-    <AvroNode v-if="rootField()" :field="rootField()" :root="true" />
+    <AvroNode
+      v-if="rootField()"
+      :field="rootField()"
+      :record-reference-types="createRecordTypesRegistry(rootField())"
+      :root="true"
+    />
   </div>
 </template>
 
diff --git a/hermes-console/src/views/topic/schema-panel/avro-viewer/avro-record-types-registry.spec.ts b/hermes-console/src/views/topic/schema-panel/avro-viewer/avro-record-types-registry.spec.ts
new file mode 100644
index 0000000000..6901a1fd78
--- /dev/null
+++ b/hermes-console/src/views/topic/schema-panel/avro-viewer/avro-record-types-registry.spec.ts
@@ -0,0 +1,340 @@
+import {
+  createRecordTypesRegistry,
+  getRecordTypesRegistryEntry,
+} from '@/views/topic/schema-panel/avro-viewer/avro-record-types-registry';
+import { describe, expect } from 'vitest';
+import type {
+  AvroSchema,
+  Field,
+  RecordType,
+} from '@/views/topic/schema-panel/AvroTypes';
+
+describe('avro records registry', () => {
+  const createSchema = (
+    fields: any,
+    name: string = 'Test',
+    namespace?: string,
+  ): AvroSchema => ({
+    name,
+    namespace,
+    fields: fields,
+    type: {
+      type: 'record',
+      fields: fields,
+    },
+    doc: '',
+  });
+
+  const createStringField = (name: string): Field => ({
+    name,
+    type: 'string',
+    fields: [],
+  });
+
+  const createRecordField = (name: string, type: RecordType): Field => ({
+    name,
+    type: ['null', type],
+    fields: [],
+  });
+
+  const createRecordType = (
+    name: string,
+    namespace: string | undefined = undefined,
+    fields: Field[] = [createStringField('x')],
+  ): RecordType => ({
+    type: 'record',
+    name,
+    namespace: namespace as string,
+    fields,
+    doc: '',
+  });
+
+  it('should handle no sub records', () => {
+    // given
+    const schema = createSchema([
+      createStringField('field1'),
+      createStringField('field2'),
+    ]);
+
+    // when
+    const result = createRecordTypesRegistry(schema);
+
+    // then
+    const expectedResult = new Set([]);
+    expect(new Set(result.entries())).toEqual(expectedResult);
+  });
+
+  it('should handle sub record with namespace for root with namespace', () => {
+    // given
+    const subRecord = createRecordType('SubRecord', 'com.example2');
+    const schema = createSchema(
+      [createRecordField('field1', subRecord)],
+      'Test',
+      'com.example',
+    );
+
+    // when
+    const result = createRecordTypesRegistry(schema);
+
+    // then
+    const expectedResult = new Set([['com.example2.SubRecord', subRecord]]);
+    expect(new Set(result.entries())).toEqual(expectedResult);
+  });
+
+  it('should handle sub record with namespace for root without namespace', () => {
+    // given
+    const subRecord = createRecordType('SubRecord', 'com.example2');
+    const schema = createSchema([createRecordField('field1', subRecord)]);
+
+    // when
+    const result = createRecordTypesRegistry(schema);
+
+    // then
+    const expectedResult = new Set([['com.example2.SubRecord', subRecord]]);
+    expect(new Set(result.entries())).toEqual(expectedResult);
+  });
+
+  it('should handle sub record without namespace for root with namespace', () => {
+    // given
+    const subRecord = createRecordType('SubRecord');
+    const schema = createSchema(
+      [createRecordField('field1', subRecord)],
+      'Test',
+      'com.example',
+    );
+
+    // when
+    const result = createRecordTypesRegistry(schema);
+
+    // then
+    const expectedResult = new Set([
+      ['com.example.SubRecord', subRecord],
+      ['SubRecord', subRecord],
+    ]);
+    expect(new Set(result.entries())).toEqual(expectedResult);
+  });
+
+  it('should handle sub record without namespace for root without namespace', () => {
+    // given
+    const subRecord = createRecordType('SubRecord');
+    const schema = createSchema([createRecordField('field1', subRecord)]);
+
+    // when
+    const result = createRecordTypesRegistry(schema);
+
+    // then
+    const expectedResult = new Set([['SubRecord', subRecord]]);
+    expect(new Set(result.entries())).toEqual(expectedResult);
+  });
+
+  it('should handle sub record with namespace in name', () => {
+    // given
+    const subRecord = createRecordType('com.example2.SubRecord');
+    const schema = createSchema(
+      [createRecordField('field1', subRecord)],
+      'Test',
+      'com.example',
+    );
+
+    // when
+    const result = createRecordTypesRegistry(schema);
+
+    // then
+    const expectedResult = new Set([['com.example2.SubRecord', subRecord]]);
+    expect(new Set(result.entries())).toEqual(expectedResult);
+  });
+
+  it('should handle sub record without namespace for root with namespace in name', () => {
+    // given
+    const subRecord = createRecordType('SubRecord');
+    const schema = createSchema(
+      [createRecordField('field1', subRecord)],
+      'com.example.Test',
+    );
+
+    // when
+    const result = createRecordTypesRegistry(schema);
+
+    // then
+    const expectedResult = new Set([
+      ['com.example.SubRecord', subRecord],
+      ['SubRecord', subRecord],
+    ]);
+    expect(new Set(result.entries())).toEqual(expectedResult);
+  });
+
+  it('should handle sub record 2 levels down without namespace inherited from parent', () => {
+    // given
+    const subSubRecord = createRecordType('SubSubRecord');
+    const subRecord = createRecordType('SubRecord', 'com.example2', [
+      createRecordField('field2', subSubRecord),
+    ]);
+    const schema = createSchema([createRecordField('field1', subRecord)]);
+
+    // when
+    const result = createRecordTypesRegistry(schema);
+
+    // then
+    const expectedResult = new Set([
+      ['com.example2.SubRecord', subRecord],
+      ['com.example2.SubSubRecord', subSubRecord],
+    ]);
+    expect(new Set(result.entries())).toEqual(expectedResult);
+  });
+
+  it('should handle sub record 2 levels down without namespace inherited from parent name', () => {
+    // given
+    const subSubRecord = createRecordType('SubSubRecord');
+    const subRecord = createRecordType('com.example2.SubRecord', undefined, [
+      createRecordField('field2', subSubRecord),
+    ]);
+    const schema = createSchema([createRecordField('field1', subRecord)]);
+
+    // when
+    const result = createRecordTypesRegistry(schema);
+
+    // then
+    const expectedResult = new Set([
+      ['com.example2.SubRecord', subRecord],
+      ['com.example2.SubSubRecord', subSubRecord],
+    ]);
+    expect(new Set(result.entries())).toEqual(expectedResult);
+  });
+
+  it('should handle sub record 2 levels down without namespace inherited from root', () => {
+    // given
+    const subSubRecord = createRecordType('SubSubRecord');
+    const subRecord = createRecordType('SubRecord', undefined, [
+      createRecordField('field2', subSubRecord),
+    ]);
+    const schema = createSchema(
+      [createRecordField('field1', subRecord)],
+      'Test',
+      'com.example',
+    );
+
+    // when
+    const result = createRecordTypesRegistry(schema);
+
+    // then
+    const expectedResult = new Set([
+      ['com.example.SubRecord', subRecord],
+      ['SubRecord', subRecord],
+      ['com.example.SubSubRecord', subSubRecord],
+      ['SubSubRecord', subSubRecord],
+    ]);
+    expect(new Set(result.entries())).toEqual(expectedResult);
+  });
+
+  it('should handle two sub records one with namespace one without', () => {
+    // given
+    const subRecord1 = createRecordType('SubRecord1', 'com.example2');
+    const subRecord2 = createRecordType('SubRecord2');
+    const schema = createSchema(
+      [
+        createRecordField('field1', subRecord1),
+        createRecordField('field2', subRecord2),
+      ],
+      'Test',
+      'com.example',
+    );
+
+    // when
+    const result = createRecordTypesRegistry(schema);
+
+    // then
+    const expectedResult = new Set([
+      ['com.example2.SubRecord1', subRecord1],
+      ['com.example.SubRecord2', subRecord2],
+      ['SubRecord2', subRecord2],
+    ]);
+    expect(new Set(result.entries())).toEqual(expectedResult);
+  });
+
+  it('should find record reference when present', () => {
+    // given
+    const record = createRecordType('com.example.SubRecord');
+    const registry = new Map([
+      ['com.example.SubRecord', record],
+      ['SubRecord', record],
+    ]);
+
+    // and
+    const referenceType = 'SubRecord';
+
+    // when
+    const result = getRecordTypesRegistryEntry(registry, referenceType);
+
+    // then
+    expect(result).toEqual(record);
+  });
+
+  it('should find record by fully qualified reference when present', () => {
+    // given
+    const record = createRecordType('com.example2.SubRecord');
+    const registry = new Map([['com.example2.SubRecord', record]]);
+
+    // and
+    const referenceType = 'com.example2.SubRecord';
+
+    // when
+    const result = getRecordTypesRegistryEntry(registry, referenceType);
+
+    // then
+    expect(result).toEqual(record);
+  });
+
+  it('should find record by reference when null or reference type', () => {
+    // given
+    const record = createRecordType('com.example.SubRecord');
+    const registry = new Map([
+      ['com.example.SubRecord', record],
+      ['SubRecord', record],
+    ]);
+
+    // and
+    const referenceType = ['SubRecord', 'null'];
+
+    // when
+    const result = getRecordTypesRegistryEntry(registry, referenceType);
+
+    // then
+    expect(result).toEqual(record);
+  });
+
+  it('should not find record by reference when absent', () => {
+    // given
+    const record = createRecordType('com.example.SubRecord');
+    const registry = new Map([
+      ['com.example.SubRecord', record],
+      ['SubRecord', record],
+    ]);
+
+    // and
+    const referenceType = 'com.example2.SubRecord';
+
+    // when
+    const result = getRecordTypesRegistryEntry(registry, referenceType);
+
+    // then
+    expect(result).toBeUndefined();
+  });
+
+  it('should not find record when not reference type', () => {
+    // given
+    const record = createRecordType('com.example.SubRecord');
+    const registry = new Map([
+      ['com.example.SubRecord', record],
+      ['SubRecord', record],
+    ]);
+
+    // and
+    const referenceType = record;
+
+    // when
+    const result = getRecordTypesRegistryEntry(registry, referenceType);
+
+    // then
+    expect(result).toBeUndefined();
+  });
+});
diff --git a/hermes-console/src/views/topic/schema-panel/avro-viewer/avro-record-types-registry.ts b/hermes-console/src/views/topic/schema-panel/avro-viewer/avro-record-types-registry.ts
new file mode 100644
index 0000000000..a73c52126b
--- /dev/null
+++ b/hermes-console/src/views/topic/schema-panel/avro-viewer/avro-record-types-registry.ts
@@ -0,0 +1,122 @@
+import type {
+  AvroSchema,
+  NameOrType,
+  RecordType,
+  Type,
+} from '@/views/topic/schema-panel/AvroTypes';
+
+interface WorkingRecord {
+  record: RecordType;
+  namespace?: string;
+  justName: string;
+}
+
+const namespace = (
+  { name, namespace }: AvroSchema,
+  parentNamespace?: string,
+): string | undefined => {
+  if (name.includes('.')) {
+    const namespaceNameSeparatorIndex = name.lastIndexOf('.');
+    return name.substring(0, namespaceNameSeparatorIndex);
+  }
+  if (namespace) {
+    return namespace;
+  }
+  return parentNamespace;
+};
+
+const justRecordName = ({ name }: AvroSchema): string => {
+  if (name.includes('.')) {
+    const namespaceNameSeparatorIndex = name.lastIndexOf('.');
+    return name.substring(namespaceNameSeparatorIndex + 1);
+  }
+  return name;
+};
+
+const workingRecord = (
+  record: RecordType,
+  parentNamespace?: string,
+): WorkingRecord => {
+  return {
+    record,
+    namespace: namespace(record, parentNamespace),
+    justName: justRecordName(record),
+  };
+};
+
+const getTypesList = (type: Type): NameOrType[] =>
+  Array.isArray(type) ? type : [type];
+
+const getDirectSubRecords = (parent: WorkingRecord): WorkingRecord[] => {
+  return parent.record.fields
+    .flatMap((field) => getTypesList(field.type))
+    .filter((field) => field.type === 'record')
+    .map((record) => workingRecord(record, parent.namespace));
+};
+
+const getAllSubRecords = (
+  unprocessedRecords: WorkingRecord[],
+  processedRecords: WorkingRecord[],
+): WorkingRecord[] => {
+  if (unprocessedRecords.length === 0) {
+    return processedRecords;
+  }
+  const currentLevelSubRecords =
+    unprocessedRecords.flatMap(getDirectSubRecords);
+  return getAllSubRecords(currentLevelSubRecords, [
+    ...processedRecords,
+    ...currentLevelSubRecords,
+  ]);
+};
+
+const validQualifiers = (
+  { namespace, justName }: WorkingRecord,
+  rootNamespace?: string,
+): string[] => {
+  if (!namespace) {
+    return [justName];
+  }
+  if (namespace === rootNamespace) {
+    return [justName, `${namespace}.${justName}`];
+  }
+  return [`${namespace}.${justName}`];
+};
+
+const associateByValidQualifiers = (
+  records: WorkingRecord[],
+  rootNamespace?: string,
+): Map<string, RecordType> => {
+  return new Map(
+    records.flatMap((record) => {
+      return validQualifiers(record, rootNamespace).map((qualifier) => [
+        qualifier,
+        record.record,
+      ]);
+    }),
+  );
+};
+
+/**
+ * Type resolution based on avro documentation {@link https://avro.apache.org/docs/1.11.1/specification/#names Avro Names}
+ */
+export const createRecordTypesRegistry = (
+  schema: AvroSchema,
+): Map<string, RecordType> => {
+  const workingRootRecord = workingRecord({
+    ...schema.type,
+    name: schema.name,
+    namespace: schema.namespace,
+  });
+  const allRecords = getAllSubRecords([workingRootRecord], []);
+  return associateByValidQualifiers(allRecords, workingRootRecord.namespace);
+};
+
+export const getRecordTypesRegistryEntry = (
+  registry: Map<string, RecordType>,
+  referenceType: Type,
+): RecordType | undefined => {
+  return getTypesList(referenceType)
+    .filter((type) => typeof type == 'string')
+    .map((typeName) => registry.get(typeName))
+    .find((record) => !!record);
+};