unknown format "email" ignored in schema at path

[cthorner]

Environment Information

• OS [e.g. Mac, Arch, Windows 10]: Mac
• Node Version [e.g. 8.2.1]: 22.11.0
• NPM Version [e.g. 5.4.2]:10.9.0
• C++ Toolchain [e.g. Visual Studio, llvm, g++]:
• confluent-kafka-javascript version [e.g. 2.3.3]: 1.6.0

Steps to Reproduce
Schema:

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "https://example.com/email-format-example",
  "type": "object",
  "properties": {
     "email": {
         "type": "string",
         "format": "email"
     }
  }
}

const payload = {
    email: "someone@example.com"
}
const topic = 'mytopic'
const id = 1234
const ser = new JsonSerializer(registry, SerdeType.VALUE, { useSchemaId: id, validate: true });
await ser.serialize(topic, payload)

confluent-kafka-javascript Configuration Settings

Additional context
Needs ajv-formats:

import Ajv2020 from "ajv/dist/2020"
import addFormats from "ajv-formats"

const ajv = new Ajv()
addFormats(ajv)

[mlois-efimob]

We have the same problem, some of our JSON Schemas have string formats that are unavaliable in Ajv without using ajv-formats, but there is no option to configure or extend the ajv instance used in the JSON validation.

The only solution that we find, that is not very elegant, and depends on the current implementation, is to extend the JSON Ser/Des to use a Ajv instance with ajv-formats.

This is a POC that we made, our final version is bit refactored but shares the same funcionality, we copy the toValidateFunction and add a line to include ajv-formats:

import {
  Client,
  JsonDeserializer,
  JsonDeserializerConfig,
  JsonSerde,
  JsonSerdeConfig,
  JsonSerializer,
  RefResolver,
  SchemaInfo,
  ValidateFunction
} from '@confluentinc/schemaregistry';
import Ajv2019 from 'ajv/dist/2019';
import Ajv2020 from 'ajv/dist/2020';
import * as draft6MetaSchema from 'ajv/dist/refs/json-schema-draft-06.json';
import * as draft7MetaSchema from 'ajv/dist/refs/json-schema-draft-07.json';
import stringify from 'json-stringify-deterministic';
import addFormats from 'ajv-formats';

export class AjvFormatsJsonSerializer extends JsonSerializer {
  async toValidateFunction(info: SchemaInfo): Promise<ValidateFunction | undefined> {
    return await toValidateFunction(
      this.client,
      this.conf as JsonDeserializerConfig,
      this,
      info,
      async (client, info) => {
        const deps = new Map<string, string>();
        await this.resolveReferences(client, info, deps);
        return deps;
      }
    );
  }
}

export class AjvFormatsJsonDeserializer extends JsonDeserializer {
  async toValidateFunction(info: SchemaInfo): Promise<ValidateFunction | undefined> {
    return await toValidateFunction(
      this.client,
      this.conf as JsonDeserializerConfig,
      this,
      info,
      async (client, info) => {
        const deps = new Map<string, string>();
        await this.resolveReferences(client, info, deps);
        return deps;
      }
    );
  }
}

/**
 * Extended JSON validate function to include ajv-formats to extend JSON Schema capabilities
 */
async function toValidateFunction(
  client: Client,
  conf: JsonSerdeConfig,
  serde: JsonSerde,
  info: SchemaInfo,
  refResolver: RefResolver
): Promise<ValidateFunction | undefined> {
  let fn = serde.schemaToValidateCache.get(stringify(info.schema)) as ValidateFunction | undefined;
  if (fn != null) {
    return fn;
  }

  const deps = await refResolver(client, info);

  const json = JSON.parse(info.schema);
  const spec = json.$schema;
  if (
    spec === 'http://json-schema.org/draft/2020-12/schema' ||
    spec === 'https://json-schema.org/draft/2020-12/schema'
  ) {
    const ajv2020 = new Ajv2020(conf);
    ajv2020.addKeyword('confluent:tags');
    deps.forEach((schema, name) => {
      ajv2020.addSchema(JSON.parse(schema), name);
    });

    // Register additional JSON Schema formats
    addFormats(ajv2020);
    fn = ajv2020.compile(json);
  } else {
    const ajv = new Ajv2019(conf);
    ajv.addKeyword('confluent:tags');
    ajv.addMetaSchema(draft6MetaSchema);
    ajv.addMetaSchema(draft7MetaSchema);
    deps.forEach((schema, name) => {
      ajv.addSchema(JSON.parse(schema), name);
    });

    // Register additional JSON Schema formats
    addFormats(ajv);
    fn = ajv.compile(json) as ValidateFunction;
  }

  serde.schemaToValidateCache.set(stringify(info.schema), fn);
  return fn;
}

Will be great to have some option to configure Ajv without extending the current implementation.