refactor: integrate custom schemas with parser and spectral (#579)

package.json

@@ -51,10 +51,10 @@
  "dependencies": {
  "@asyncapi/specs": "^3.1.0",
  "@openapi-contrib/openapi-schema-to-json-schema": "^3.2.0",
- "@stoplight/spectral-core": "^1.10.1",
- "@stoplight/spectral-functions": "^1.5.1",
- "@stoplight/spectral-parsers": "^1.0.1",
- "@stoplight/spectral-rulesets": "^1.4.3",
+ "@stoplight/spectral-core": "^1.13.1",
+ "@stoplight/spectral-functions": "^1.7.1",
+ "@stoplight/spectral-parsers": "^1.0.2",
+ "@stoplight/spectral-rulesets": "^1.12.0",
  "ajv": "^8.11.0",
  "avsc": "^5.7.4",
  "js-yaml": "^3.14.1",

src/constants.ts

@@ -5,6 +5,7 @@ export const xParserMessageName = 'x-parser-message-name';
 export const xParserSchemaId = 'x-parser-schema-id';
 
 export const xParserOriginalSchemaFormat = 'x-parser-original-schema-format';
+export const xParserOriginalPayload = 'x-parser-original-payload';
 export const xParserOriginalTraits = 'x-parser-original-traits';
 
 export const xParserCircular = 'x-parser-circular';

src/custom-operations/parse-schema.ts

@@ -1,8 +1,8 @@
 import { JSONPath } from 'jsonpath-plus';
 import { toPath } from 'lodash';
 
-import { parseSchema, getDefaultSchemaFormat } from '../schema-parser';
-import { xParserOriginalSchemaFormat } from '../constants';
+import { parseSchema, getSchemaFormat, getDefaultSchemaFormat } from '../schema-parser';
+import { xParserOriginalPayload } from '../constants';
 
 import type { Parser } from '../parser';
 import type { ParseSchemaInput } from "../schema-parser";
@@ -14,15 +14,17 @@ interface ToParseItem {
 }
 
 const customSchemasPathsV2 = [
+ // operations
  '$.channels.*.[publish,subscribe].message',
  '$.channels.*.[publish,subscribe].message.oneOf.*',
  '$.components.channels.*.[publish,subscribe].message',
  '$.components.channels.*.[publish,subscribe].message.oneOf.*',
+ // messages
  '$.components.messages.*',
 ];
 
 export async function parseSchemasV2(parser: Parser, detailed: DetailedAsyncAPI) {
- const defaultSchemaFormat = getDefaultSchemaFormat(detailed.parsed.asyncapi as string);
+ const defaultSchemaFormat = getDefaultSchemaFormat(detailed.semver.version);
  const parseItems: Array<ToParseItem> = [];
 
  const visited: Set<unknown> = new Set();
@@ -43,13 +45,16 @@ export async function parseSchemasV2(parser: Parser, detailed: DetailedAsyncAPI)
  return;
  }
 
+ const schemaFormat = getSchemaFormat(value.schemaFormat, detailed.semver.version);
  parseItems.push({
  input: {
  asyncapi: detailed,
  data: payload,
- meta: undefined,
+ meta: {
+ message: value,
+ },
  path: [...toPath(result.path.slice(1)), 'payload'],
- schemaFormat: value.schemaFormat || defaultSchemaFormat,
+ schemaFormat,
  defaultSchemaFormat,
  },
  value,
@@ -62,6 +67,6 @@ export async function parseSchemasV2(parser: Parser, detailed: DetailedAsyncAPI)
 }
 
 async function parseSchemaV2(parser: Parser, item: ToParseItem) {
- item.value[xParserOriginalSchemaFormat] = item.input.schemaFormat;
+ item.value[xParserOriginalPayload] = item.input.data;
  item.value.payload = await parseSchema(parser, item.input);
 }

src/parse.ts

@@ -2,7 +2,7 @@ import { AsyncAPIDocumentInterface, newAsyncAPIDocument } from "./models";
 
 import { customOperations } from './custom-operations';
 import { validate } from "./lint";
-import { createDetailedAsyncAPI, normalizeInput, toAsyncAPIDocument } from "./utils";
+import { createDetailedAsyncAPI, normalizeInput, toAsyncAPIDocument, unfreezeObject } from "./utils";
 
 import { xParserSpecParsed } from './constants';
 
@@ -47,7 +47,7 @@ export async function parse(parser: Parser, asyncapi: ParseInput, options?: Pars
  }
 
  // unfreeze the object - Spectral makes resolved document "freezed" 
- const validatedDoc = JSON.parse(JSON.stringify(validated));
+ const validatedDoc = unfreezeObject(validated);
  validatedDoc[String(xParserSpecParsed)] = true;
 
  const detailed = createDetailedAsyncAPI(asyncapi as string | Record<string, unknown>, validatedDoc);
@@ -67,6 +67,7 @@ export async function parse(parser: Parser, asyncapi: ParseInput, options?: Pars
 
 const defaultOptions: ParseOptions = {
  applyTraits: true,
+ parseSchemas: true,
 };
 function normalizeOptions(options?: ParseOptions): ParseOptions {
  if (!options || typeof options !== 'object') {

src/parser.ts

@@ -1,9 +1,10 @@
 import { Spectral } from "@stoplight/spectral-core";
-import { asyncapi as aasRuleset } from "@stoplight/spectral-rulesets";
 
 import { parse } from "./parse";
 import { lint, validate } from "./lint";
 import { registerSchemaParser } from './schema-parser';
+import { AsyncAPISchemaParser } from "./schema-parser/asyncapi-schema-parser";
+import { configureSpectral } from "./spectral";
 
 import type { IConstructorOpts } from "@stoplight/spectral-core";
 import type { ParseInput, ParseOptions } from "./parse";
@@ -18,16 +19,18 @@ export class Parser {
  public readonly parserRegistry = new Map<string, SchemaParser>();
  public readonly spectral: Spectral;
 
- constructor(options?: ParserOptions) {
- const { spectral } = options || {};
+ constructor(
+ private readonly options?: ParserOptions
+ ) {
+ const { spectral } = this.options || {};
  if (spectral instanceof Spectral) {
  this.spectral = spectral;
  } else {
  this.spectral = new Spectral(spectral);
  }
 
- // TODO: fix type
- this.spectral.setRuleset(aasRuleset as any);
+ this.registerSchemaParser(AsyncAPISchemaParser());
+ configureSpectral(this);
  }
 
  parse(asyncapi: ParseInput, options?: ParseOptions) {

src/schema-parser/asyncapi-schema-parser.ts

@@ -1,15 +1,16 @@
-import { SchemaParser, ParseSchemaInput, ValidateSchemaInput } from "../schema-parser";
-import Ajv, { ErrorObject, ValidateFunction } from "ajv";
-import type { AsyncAPISchema, SchemaValidateResult } from '../types';
+import Ajv from "ajv";
 // @ts-ignore
 import specs from '@asyncapi/specs';
 
+import type { ErrorObject, ValidateFunction } from "ajv";
+import type { AsyncAPISchema, SchemaValidateResult } from '../types';
+import type { SchemaParser, ParseSchemaInput, ValidateSchemaInput } from "../schema-parser";
+
 const ajv = new Ajv({
  allErrors: true,
  strict: false,
  logger: false,
 });
-
 // Only versions compatible with JSON Schema Draf-07 are supported.
 const specVersions = Object.keys(specs).filter((version: string) => !['1.0.0', '1.1.0', '1.2.0', '2.0.0-rc1', '2.0.0-rc2'].includes(version));
 

src/schema-parser/avro-schema-parser.ts

@@ -1,7 +1,9 @@
-import avsc, { type Schema} from "avsc";
-import type { AsyncAPISchema, SchemaValidateResult } from "types";
-import { SchemaParser, type ParseSchemaInput, type ValidateSchemaInput } from "../schema-parser";
-import { type JSONSchema7TypeName } from "json-schema";
+import avsc from "avsc";
+
+import type { JSONSchema7TypeName } from "json-schema";
+import type { Schema } from "avsc";
+import type { SchemaParser, ParseSchemaInput, ValidateSchemaInput } from "../schema-parser";
+import type { AsyncAPISchema, SchemaValidateResult } from '../types';
 
 type AvroSchema = Schema & { [key: string]: any } & any;
 
@@ -32,29 +34,15 @@ async function validate(input: ValidateSchemaInput<unknown, unknown>): Promise<S
 
 async function parse(input: ParseSchemaInput<unknown, unknown>): Promise<AsyncAPISchema> {
  const asyncAPISchema = await avroToJsonSchema(input.data as AvroSchema);
- const message = (input.meta as any).message
 
  // TODO: Should the following modifications to the message object be done in the caller and for all parsers rather than here?
  // remove that function when https:/asyncapi/spec/issues/622 will be introduced in AsyncAPI spec
- async function handleKafkaProtocolKey() {
- if (message.bindings && message.bindings.kafka) {
- const key = message.bindings.kafka.key;
- if (key) {
- const bindingsTransformed = await avroToJsonSchema(key);
- message['x-parser-original-bindings-kafka-key'] = key;
- message.bindings.kafka.key = bindingsTransformed;
- }
- }
- }
-
- if (message !== undefined) {
- // TODO: Should the following modifications to the message object be done in the caller and for all parsers rather than here?
- message['x-parser-original-schema-format'] = input.schemaFormat || input.defaultSchemaFormat;
- message['x-parser-original-payload'] = input.data;
- message.payload = asyncAPISchema;
- delete message.schemaFormat; 
-
- await handleKafkaProtocolKey();
+ const message = (input.meta as any).message
+ const key = message?.bindings?.kafka?.key;
+ if (key) {
+ const bindingsTransformed = await avroToJsonSchema(key);
+ message['x-parser-original-bindings-kafka-key'] = key;
+ message.bindings.kafka.key = bindingsTransformed;
  }
 
  return asyncAPISchema;
@@ -94,7 +82,7 @@ const typeMappings: { [key: string]: JSONSchema7TypeName } = {
  uuid: 'string',
 };
 
-const commonAttributesMapping = (avroDefinition: AvroSchema, jsonSchema: AsyncAPISchema, isTopLevel: boolean) => {
+function commonAttributesMapping(avroDefinition: AvroSchema, jsonSchema: AsyncAPISchema, isTopLevel: boolean): void {
  if (avroDefinition.doc) jsonSchema.description = avroDefinition.doc;
  if (avroDefinition.default !== undefined) jsonSchema.default = avroDefinition.default;
 
@@ -124,7 +112,7 @@ function getFullyQualifiedName(avroDefinition: AvroSchema) {
  * @param parentJsonSchema the parent json schema which contains the required property to enrich
  * @param haveDefaultValue we assure that a required field does not have a default value
  */
-const requiredAttributesMapping = (fieldDefinition: any, parentJsonSchema: AsyncAPISchema, haveDefaultValue: boolean) => {
+function requiredAttributesMapping(fieldDefinition: any, parentJsonSchema: AsyncAPISchema, haveDefaultValue: boolean): void {
  const isUnionWithNull = Array.isArray(fieldDefinition.type) && fieldDefinition.type.includes('null');
 
  // we assume that a union type without null and a field without default value is required
@@ -134,7 +122,7 @@ const requiredAttributesMapping = (fieldDefinition: any, parentJsonSchema: Async
  }
 };
 
-function extractNonNullableTypeIfNeeded(typeInput: any, jsonSchemaInput: AsyncAPISchema) {
+function extractNonNullableTypeIfNeeded(typeInput: any, jsonSchemaInput: AsyncAPISchema): { type: string, jsonSchema: AsyncAPISchema } {
  let type = typeInput;
  let jsonSchema = jsonSchemaInput;
  // Map example to first non-null type
@@ -145,10 +133,10 @@ function extractNonNullableTypeIfNeeded(typeInput: any, jsonSchemaInput: AsyncAP
  jsonSchema = jsonSchema.oneOf[0] as AsyncAPISchema;
  }
  }
- return {type, jsonSchema};
+ return { type, jsonSchema };
 }
 
-const exampleAttributeMapping = (type: any, example: any, jsonSchema: AsyncAPISchema) => {
+function exampleAttributeMapping(type: any, example: any, jsonSchema: AsyncAPISchema): void {
  if (example === undefined || jsonSchema.examples || Array.isArray(type)) return;
 
  switch (type) {
@@ -163,7 +151,7 @@ const exampleAttributeMapping = (type: any, example: any, jsonSchema: AsyncAPISc
  }
 };
 
-const additionalAttributesMapping = (typeInput: any, avroDefinition: AvroSchema, jsonSchemaInput: AsyncAPISchema) => {
+function additionalAttributesMapping(typeInput: any, avroDefinition: AvroSchema, jsonSchemaInput: AsyncAPISchema): void {
  const __ret = extractNonNullableTypeIfNeeded(typeInput, jsonSchemaInput);
  const type = __ret.type;
  const jsonSchema = __ret.jsonSchema;
@@ -201,7 +189,7 @@ const additionalAttributesMapping = (typeInput: any, avroDefinition: AvroSchema,
  }
 };
 
-function validateAvroSchema(avroDefinition: AvroSchema) {
+function validateAvroSchema(avroDefinition: AvroSchema): void | never {
  // don't need to use the output from parsing the
  // avro definition - we're just using this as a
  // validator as this will throw an exception if
@@ -217,13 +205,13 @@ function validateAvroSchema(avroDefinition: AvroSchema) {
  * @param key String | Undefined - the fully qualified name of an avro record
  * @param value JsonSchema - The json schema from the avro record
  */
-function cacheAvroRecordDef(cache: {[key:string]: AsyncAPISchema}, key: string, value: AsyncAPISchema) {
+function cacheAvroRecordDef(cache: {[key:string]: AsyncAPISchema}, key: string, value: AsyncAPISchema): void {
  if (key !== undefined) {
  cache[key] = value;
  }
 }
 
-async function convertAvroToJsonSchema(avroDefinition: AvroSchema , isTopLevel: boolean, recordCache: Map<string, AsyncAPISchema> | any = {}) {
+async function convertAvroToJsonSchema(avroDefinition: AvroSchema , isTopLevel: boolean, recordCache: Map<string, AsyncAPISchema> | any = {}): Promise<AsyncAPISchema> {
  const jsonSchema: AsyncAPISchema = {};
  const isUnion = Array.isArray(avroDefinition);
 
@@ -287,8 +275,8 @@ async function convertAvroToJsonSchema(avroDefinition: AvroSchema , isTopLevel:
  * @param jsonSchema the schema for the record.
  * @returns {Promise<Map<string, any>>}
  */
-async function processRecordSchema(avroDefinition: AvroSchema, recordCache: {[key:string]: AsyncAPISchema}, jsonSchema: AsyncAPISchema) {
- const propsMap = new Map();
+async function processRecordSchema(avroDefinition: AvroSchema, recordCache: {[key:string]: AsyncAPISchema}, jsonSchema: AsyncAPISchema): Promise<Map<string, any>> {
+ const propsMap = new Map<string, any>();
  for (const field of avroDefinition.fields) {
  // If the type is a sub schema it will have been stored in the cache.
  if (recordCache[field.type]) {
@@ -317,9 +305,9 @@ async function processRecordSchema(avroDefinition: AvroSchema, recordCache: {[ke
  * @param avroDefinition the avro schema to be processed
  * @param isTopLevel is this the top level of the schema or is this a sub schema
  * @param recordCache the cache of previously processed record types
- * @returns {Promise<JsonSchema>} the mutated jsonSchema that was provided to the function
+ * @returns {Promise<AsyncAPISchema>} the mutated jsonSchema that was provided to the function
  */
-async function processUnionSchema(jsonSchema: AsyncAPISchema, avroDefinition: AvroSchema, isTopLevel: boolean, recordCache: {[key:string]: AsyncAPISchema}) {
+async function processUnionSchema(jsonSchema: AsyncAPISchema, avroDefinition: AvroSchema, isTopLevel: boolean, recordCache: {[key:string]: AsyncAPISchema}): Promise<AsyncAPISchema> {
  jsonSchema.oneOf = [];
  let nullDef = null;
  for (const avroDef of avroDefinition) {

src/schema-parser/index.ts

@@ -28,16 +28,15 @@ export interface SchemaParser<D = unknown, M = unknown> {
 export async function validateSchema(parser: Parser, input: ValidateSchemaInput) {
  const schemaParser = parser.parserRegistry.get(input.schemaFormat);
  if (schemaParser === undefined) {
- // throw appropriate error
- throw new Error();
+ throw new Error('Unknown schema format');
  }
  return schemaParser.validate(input);
 }
 
 export async function parseSchema(parser: Parser, input: ParseSchemaInput) {
  const schemaParser = parser.parserRegistry.get(input.schemaFormat);
  if (schemaParser === undefined) {
- return;
+ throw new Error('Unknown schema format');
  }
  return schemaParser.parse(input);
 }
@@ -49,14 +48,21 @@ export function registerSchemaParser(parser: Parser, schemaParser: SchemaParser)
  || typeof schemaParser.parse !== 'function' 
  || typeof schemaParser.getMimeTypes !== 'function'
  ) {
- throw new Error('custom parser must have "parse()", "validate()" and "getMimeTypes()" functions.');
+ throw new Error('Custom parser must have "parse()", "validate()" and "getMimeTypes()" functions.');
  }
 
  schemaParser.getMimeTypes().forEach(schemaFormat => {
  parser.parserRegistry.set(schemaFormat, schemaParser);
  });
 }
 
+export function getSchemaFormat(schematFormat: string | undefined, asyncapiVersion: string) {
+ if (typeof schematFormat === 'string') {
+ return schematFormat;
+ }
+ return getDefaultSchemaFormat(asyncapiVersion);
+}
+
 export function getDefaultSchemaFormat(asyncapiVersion: string) {
  return `application/vnd.aai.asyncapi;version=${asyncapiVersion}`;
 }