defmodule Avrora.Storage.Registry do @moduledoc """ A small wrapper for [Confluent Schema Registry](https://docs.confluent.io/current/schema-registry/develop/api.html), with as less as possible functionality. Inspired by [Schemex](https://github.com/bencebalogh/schemex). """ require Logger alias Avrora.{HttpClient, Name, Schema} @behaviour Avrora.Storage @content_type "application/vnd.schemaregistry.v1+json" @doc """ Fetch the latest version of the schema registered under a subject name. ## Examples iex> {:ok, avro} = Avrora.Storage.Registry.get("io.confluent.Payment") iex> avro.schema.schema.qualified_names ["io.confluent.Payment"] """ def get(key) when is_binary(key) do with {:ok, schema_name} <- Name.parse(key), {name, version} <- {schema_name.name, schema_name.version || "latest"}, {:ok, response} <- http_client_get("subjects/#{name}/versions/#{version}"), {:ok, version} <- Map.fetch(response, "version"), {:ok, schema} <- Map.fetch(response, "schema"), {:ok, schema} <- Schema.parse(schema) do {:ok, %{schema | version: version}} end end @doc """ Fetch a schema by a globally unique ID. FIXME: Can't use a real docs with iex> because don't know how to make it work ## Examples ...> {:ok, avro} = Avrora.Storage.Registry.get(1) ...> avro.schema.schema.qualified_names ...> ["io.confluent.examples.Payment"] """ def get(key) when is_integer(key) do with {:ok, response} <- http_client_get("schemas/ids/#{key}"), {:ok, schema} <- Map.fetch(response, "schema"), {:ok, schema} <- Schema.parse(schema) do {:ok, %{schema | id: key}} end end @doc """ Register a new version of a schema under the subject name. ## Examples iex> schema = %{"namespace" => "io.confluent", "type" => "record", "name" => "Payment", "fields" => [%{"name" => "id", "type" => "string"}, %{"name" => "amount", "type" => "double"}]} iex> {:ok, avro} = Avrora.Storage.Registry.put("io.confluent.examples.Payment", schema) iex> avro.schema.schema.qualified_names ["io.confluent.Payment"] """ def put(key, value) when is_binary(key) and (is_map(value) or is_binary(value)) do with {:ok, schema_name} <- Name.parse(key), {:ok, response} <- http_client_post("subjects/#{schema_name.name}/versions", value), {:ok, id} <- Map.fetch(response, "id"), {:ok, schema} <- Schema.parse(value) do unless is_nil(schema_name.version) do Logger.warn( "storing schema with version is not allowed, `#{schema_name.name}` used instead" ) end {:ok, %{schema | id: id}} end end @doc false def put(_key, _value), do: {:error, :unsupported} @doc false @spec configured?() :: true | false def configured?, do: !is_nil(Application.get_env(:avrora, :registry_url)) defp http_client_get(path) do if configured?(), do: path |> to_url() |> http_client().get() |> handle(), else: {:error, :unconfigured_registry_url} end defp http_client_post(path, payload) do if configured?() do path |> to_url() |> http_client().post(payload, content_type: @content_type) |> handle() else {:error, :unconfigured_registry_url} end end defp to_url(path), do: "#{Application.get_env(:avrora, :registry_url)}/#{path}" defp handle({:error, payload} = _response) when is_map(payload) do reason = case Map.get(payload, "error_code") do 40401 -> :unknown_subject 40402 -> :unknown_version 40403 -> :unknown_schema 409 -> :conflict 422 -> :unprocessable _ -> payload end {:error, reason} end defp handle(response), do: response defp http_client, do: Application.get_env(:avrora, :http_client, HttpClient) end