defmodule ConfluentSchemaRegistry do
  @moduledoc """
  Elixir client for [Confluent Schema Registry](https://docs.confluent.io/current/schema-registry/)
  [API](https://docs.confluent.io/current/schema-registry/develop/api.html).
  """

  # Argument types
  @type id :: pos_integer
  @type schema :: binary
  @type subject :: binary
  @type code :: non_neg_integer
  @type reason :: any
  @type version :: pos_integer | binary
  @type level :: binary

  @doc ~S"""
  Create client to talk to Schema Registry.

  Options are:

  * base_url: URL of schema registry (optiona, default "http://localhost:8081")
  * username: username for BasicAuth (optional)
  * password: password for BasicAuth (optional)
  * adapter: Tesla adapter config
  * middleware: List of additional middleware module config (optional)

  ## Examples

      iex> client = ConfluentSchemaRegistry.client()
      %Tesla.Client{
        adapter: nil,
        fun: nil,
        post: [],
        pre: [
          {Tesla.Middleware.BaseUrl, :call, ["http://localhost:8081"]},
          {Tesla.Middleware.Headers, :call,
           [[{"content-type", "application/vnd.schemaregistry.v1+json"}]]},
          {Tesla.Middleware.JSON, :call,
           [[decode_content_types: ["application/vnd.schemaregistry.v1+json"]]]}
        ]
      }
  """
  @spec client(Keyword.t) :: Tesla.Client.t
  def client(opts \\ []) do
    base_url = opts[:base_url] || "http://localhost:8081"
    opts_middleware = opts[:middleware] || []
    adapter = opts[:adapter]

    middleware = opts_middleware ++ [
      {Tesla.Middleware.BaseUrl, base_url},
      {Tesla.Middleware.Headers, [
        {"content-type", "application/vnd.schemaregistry.v1+json"},
        {"accept", "application/vnd.schemaregistry.v1+json, application/vnd.schemaregistry+json, application/json"}
      ]},
      {Tesla.Middleware.JSON, decode_content_types: [
        "application/vnd.schemaregistry.v1+json",
        "application/vnd.schemaregistry+json"
      ]},
    ] ++ basic_auth(opts)

    Tesla.client(middleware, adapter)
  end

  # Configure Tesla.Middleware.BasicAuth
  @spec basic_auth(Keyword.t) :: [{Tesla.Middleware.BasicAuth, map}]
  defp basic_auth(opts) do
    if opts[:username] do
      auth_opts = opts
                  |> Keyword.take([:username, :password])
                  |> Map.new()
      [{Tesla.Middleware.BasicAuth, auth_opts}]
    else
      []
    end
  end


  @doc ~S"""
  Get the schema string identified by the input ID.

  https://docs.confluent.io/current/schema-registry/develop/api.html#get--schemas-ids-int-%20id

  Returns binary schema.

  ## Examples

      iex> ConfluentSchemaRegistry.get_schema(client, 21)
      {:ok, "{\"type\":\"record\",\"name\":\"test\",\"fields\":[{\"name\":\"field1\",\"type\":\"string\"},{\"name\":\"field2\",\"type\":\"int\"}]}"}

  """
  @spec get_schema(Tesla.Client.t, id) :: {:ok, schema} | {:error, code, reason}
  def get_schema(client, id) when is_integer(id) do
    case do_get(client, "/schemas/ids/#{id}") do
      {:ok, %{"schema" => value}} -> {:ok, value}
      {:ok, value} -> {:error, 1, "Unexpected response: " <> value}
      error -> error
    end
  end

  @doc ~S"""
  Get a list of registered subjects.

  https://docs.confluent.io/current/schema-registry/develop/api.html#get--subjects

  Returns list of subject name strings.

  ## Examples

      iex> ConfluentSchemaRegistry.get_subjects(client)
      {:ok, ["test"]}

  """
  @spec get_subjects(Tesla.Client.t) :: {:ok, list(subject)} | {:error, code, reason}
  def get_subjects(client) do
    case do_get(client, "/subjects") do
      {:ok, _} = result -> result
      error -> error
    end
  end

  @doc ~S"""
  Get a list of versions registered under the specified subject.

  https://docs.confluent.io/current/schema-registry/develop/api.html#get--subjects-(string-%20subject)-versions

  ```
  {:ok, [1, 2, 3, 4]} = ConfluentSchemaRegistry.get_versions(client, "test")
  ```

  Returns list of integer ids.
  """
  @spec get_versions(Tesla.Client.t, subject) :: {:ok, list(id)} | {:error, code, reason}
  def get_versions(client, subject) do
    case do_get(client, "/subjects/#{subject}/versions") do
      {:ok, _} = result -> result
      error -> error
    end
  end

  @doc ~S"""
  Deletes the specified subject and its associated compatibility level if registered.
  It is recommended to use this API only when a topic needs to be recycled or
  in development environment.

  https://docs.confluent.io/current/schema-registry/develop/api.html#delete--subjects-(string-%20subject)

  Returns list of integer ids.

  ## Examples

      iex> ConfluentSchemaRegistry.delete_subject(client, "test")
      {:ok, [1]}

  """
  @spec delete_subject(Tesla.Client.t, subject) :: {:ok, list(id)} | {:error, code, reason}
  def delete_subject(client, subject) do
    case do_delete(client, "/subjects/#{subject}") do
      {:ok, _} = result -> result
      error -> error
    end
  end

  @doc ~S"""
  Get a specific version of the schema registered under this subject

  https://docs.confluent.io/current/schema-registry/develop/api.html#delete--subjects-(string-%20subject)

  Returns a map with the following keys:

  * subject (string) -- Name of the subject that this schema is registered under
  * id (int) -- Globally unique identifier of the schema
  * version (int) -- Version of the returned schema
  * schema (string) -- The Avro schema string

  ```
  case ConfluentSchemaRegistry.get_schema(client, "test", "latest") do
    {:ok, reg} ->
      # Already registered
      schema = reg["schema"]
      schema_id = reg["id"]
    {:error, 404, %{"error_code" => 40401}} ->
      # Subject not found
    {:error, 404, %{"error_code" => 40402}} ->
      # Version not found
    {:error, 422, reason} ->
      # Unprocessable Entity, Invalid Avro version
    {:error, code, reason} ->
      # Other error
  end
  ```

  ## Examples

      iex> ConfluentSchemaRegistry.get_schema(client, "test")
      {:ok,
       %{
         "id" => 21,
         "schema" => "{\"type\":\"record\",\"name\":\"test\",\"fields\":[{\"name\":\"field1\",\"type\":\"string\"},{\"name\":\"field2\",\"type\":\"int\"}]}",
         "subject" => "test",
         "version" => 13
       }}

      iex> ConfluentSchemaRegistry.get_schema(client, "test", 13)
      {:ok,
       %{
         "id" => 21,
         "schema" => "{\"type\":\"record\",\"name\":\"test\",\"fields\":[{\"name\":\"field1\",\"type\":\"string\"},{\"name\":\"field2\",\"type\":\"int\"}]}",
         "subject" => "test",
         "version" => 13
       }}

  """
  @spec get_schema(Tesla.Client.t, subject, version) ::
    {:ok, map} | {:error, code, reason}
  def get_schema(client, subject, version \\ "latest") do
    case do_get(client, "/subjects/#{subject}/versions/#{version}") do
      {:ok, _} = result -> result
      error -> error
    end
  end

  # NOTE: /subjects/#{subject}/versions/#{version}/schema not implemented, as
  # it's redundant with get_schema/3

  @doc ~S"""
  Register a new schema under the specified subject. If successfully
  registered, this returns the unique identifier of this schema in the registry.

  https://docs.confluent.io/current/schema-registry/develop/api.html#post--subjects-(string-%20subject)-versions

  Returns the integer id.

  ```
  case ConfluentSchemaRegistry.register_schema(client, "test", schema) do
    {:ok, schema_id} ->
      # Already registered
    {:error, 409, reason} ->
      # Conflict -- Incompatible Avro schema
    {:error, 422, reason} ->
      # Unprocessable Entity, Invalid Avro schema
    {:error, code, reason} ->
      # Other error
  end
  ```

  ## Examples

      iex> ConfluentSchemaRegistry.register_schema(client, "test", schema)
      {:ok, 21}

  """
  @spec register_schema(Tesla.Client.t, subject, schema) :: {:ok, id} | {:error, code, reason}
  def register_schema(client, subject, schema) do
    case do_post(client, "/subjects/#{subject}/versions", %{schema: schema}) do
      {:ok, %{"id" => value}} -> {:ok, value}
      {:ok, value} -> {:error, 1, "Unexpected response: #{inspect value}"}
      error -> error
    end
  end

  @doc ~S"""
  Check if a schema has already been registered under the specified subject. If
  so, this returns the schema string along with its globally unique identifier,
  its version under this subject and the subject name.

  https://docs.confluent.io/current/schema-registry/develop/api.html#post--subjects-(string-%20subject)

  Returns map with the following keys:

  * subject (string) -- Name of the subject that this schema is registered under
  * id (int) -- Globally unique identifier of the schema
  * version (int) -- Version of the returned schema
  * schema (string) -- The Avro schema string

  ```
  schema = "{\"type\":\"record\",\"name\":\"test\",\"fields\":[{\"name\":\"field1\",\"type\":\"string\"},{\"name\":\"field2\",\"type\":\"int\"}]}"
  case ConfluentSchemaRegistry.is_registered(client, "test", schema) do
    {:ok, reg} ->
      # Found
      schema = reg["schema"]
      schema_id = reg["id"]
    {:error, 404, %{"error_code" => 40401}} ->
      # Subject not found
    {:error, 404, %{"error_code" => 40403}} ->
      # Schema not found
    {:error, code, reason} ->
      # Other error
  end
  ```

  ## Examples

      iex> ConfluentSchemaRegistry.is_registered(client, "test", schema)
      {:ok,
       %{
         "id" => 21,
         "schema" => "{\"type\":\"record\",\"name\":\"test\",\"fields\":[{\"name\":\"field1\",\"type\":\"string\"},{\"name\":\"field2\",\"type\":\"int\"}]}",
         "subject" => "test",
         "version" => 1
       }}

      iex> ConfluentSchemaRegistry.is_registered(client, "test2", schema)
      {:error, 404, %{"error_code" => 40401, "message" => "Subject not found. ..."}}

  """
  @spec is_registered(Tesla.Client.t, subject, schema) :: {:ok, map} | {:error, code, reason}
  def is_registered(client, subject, schema) do
    case do_post(client, "/subjects/#{subject}", %{schema: schema}) do
      {:ok, _} = result -> result
      error -> error
    end
  end

  @doc ~S"""
  Deletes a specific version of the schema registered under this subject. This
  only deletes the version and the schema ID remains intact making it still
  possible to decode data using the schema ID.

  https://docs.confluent.io/current/schema-registry/develop/api.html#delete--subjects-(string-%20subject)-versions-(versionId-%20version)

  ```
  {:ok, 1} = ConfluentSchemaRegistry.delete_version(client, "test") # latest
  {:ok, 1} = ConfluentSchemaRegistry.delete_version(client, "test", 1)
  ```

  Returns integer id of deleted version.
  """
  @spec delete_version(Tesla.Client.t, subject, version) :: {:ok, id} | {:error, code, reason}
  def delete_version(client, subject, version \\ "latest") do
    case do_delete(client, "/subjects/#{subject}/versions/#{version}") do
      {:ok, value} when is_integer(value) -> {:ok, value}
      {:ok, value} -> {:error, 1, "Unexpected response: #{inspect value}"}
      error -> error
    end
  end

  @doc ~S"""
  Test input schema against a particular version of a subject's schema for
  compatibility. Note that the compatibility level applied for the check is the
  configured compatibility level for the subject (`get_compatibility/2`).
  If this subject's compatibility level was never changed, then the
  global compatibility level applies (`get_compatibility/1`).

  https://docs.confluent.io/current/schema-registry/develop/api.html#post--compatibility-subjects-(string-%20subject)-versions-(versionId-%20version)

  Returns boolean.

  ## Examples

      iex> ConfluentSchemaRegistry.is_compatible(client, "test", schema)
      {:ok, true}

      iex> ConfluentSchemaRegistry.is_compatible(client, "test", schema, "latest")
      {:ok, true}

      iex> ConfluentSchemaRegistry.is_compatible(client, "test", schema, 1)
      {:ok, true}

  """
  @spec is_compatible(Tesla.Client.t, subject, schema, version) :: {:ok, boolean} | {:error, code, reason}
  def is_compatible(client, subject, schema, version \\ "latest") do
    case do_post(client, "/compatibility/subjects/#{subject}/versions/#{version}", %{schema: schema}) do
      {:ok, %{"is_compatible" => value}} -> {:ok, value}
      {:ok, value} -> {:error, 1, "Unexpected response: " <> value}
      error -> error
    end
  end

  @doc ~S"""
  Update global compatibility level.

  Level is a string which must be one of BACKWARD, BACKWARD_TRANSITIVE, FORWARD,
  FORWARD_TRANSITIVE, FULL, FULL_TRANSITIVE, NONE

  https://docs.confluent.io/current/schema-registry/develop/api.html#put--config

  Returns string.

  ## Examples

      iex> ConfluentSchemaRegistry.update_compatibility(client, "test", "FULL")
      {:ok, "FULL"}

  """
  @spec update_compatibility(Tesla.Client.t, level) :: {:ok, level} | {:error, code, reason}
    when level: binary, code: non_neg_integer, reason: any
  def update_compatibility(client, level) do
    case do_put(client, "/config", %{compatibility: level}) do
      {:ok, %{"compatibility" => value}} -> {:ok, value}
      {:ok, value} -> {:error, 1, "Unexpected response: " <> value}
      error -> error
    end
  end

  @doc ~S"""
  Get global compatibility level.

  Level is a string which will be one of BACKWARD, BACKWARD_TRANSITIVE, FORWARD,
  FORWARD_TRANSITIVE, FULL, FULL_TRANSITIVE, NONE

  https://docs.confluent.io/current/schema-registry/develop/api.html#put--config

  Returns string.

  ## Examples

      iex> ConfluentSchemaRegistry.get_compatibility(client)
      {:ok, "BACKWARD"}

  """
  @spec get_compatibility(Tesla.Client.t) :: {:ok, level} | {:error, code, reason}
    when level: binary, code: non_neg_integer, reason: any
  def get_compatibility(client) do
    case do_get(client, "/config") do
      {:ok, %{"compatibilityLevel" => value}} -> {:ok, value}
      {:ok, value} -> {:error, 1, "Unexpected response: #{inspect value}"}
      error -> error
    end
  end

  @doc ~S"""
  Update compatibility level for the specified subject.

  Leve is a string which must be one of BACKWARD, BACKWARD_TRANSITIVE, FORWARD,
  FORWARD_TRANSITIVE, FULL, FULL_TRANSITIVE, NONE

  https://docs.confluent.io/current/schema-registry/develop/api.html#put--config

  Returns string.

  ## Examples

      iex> ConfluentSchemaRegistry.update_compatibility(client, "test", "FULL")
      {:ok, "FULL"}

  """
  @spec update_compatibility(Tesla.Client.t, subject, level) ::
    {:ok, level} | {:error, code, reason}
  def update_compatibility(client, subject, level) do
    case do_put(client, "/config/#{subject}", %{compatibility: level}) do
      {:ok, %{"compatibility" => value}} -> {:ok, value}
      {:ok, value} -> {:error, 1, "Unexpected response: #{inspect value}"}
      error -> error
    end
  end

  @doc ~S"""
  Get compatibility level for a subject.

  Level is a string which will be one of BACKWARD, BACKWARD_TRANSITIVE, FORWARD,
  FORWARD_TRANSITIVE, FULL, FULL_TRANSITIVE, NONE

  https://docs.confluent.io/current/schema-registry/develop/api.html#put--config

  Returns string.

  ## Examples

      iex> ConfluentSchemaRegistry.get_compatibility(client, "test")
      {:ok, "FULL"}

  """
  @spec get_compatibility(Tesla.Client.t, subject) ::
    {:ok, level} | {:error, code, reason}
  def get_compatibility(client, subject) do
    case do_get(client, "/config/#{subject}") do
      {:ok, %{"compatibilityLevel" => value}} -> {:ok, value}
      {:ok, value} -> {:error, 1, "Unexpected response: " <> value}
      error -> error
    end
  end


  # Internal utility functions

  @spec do_get(Tesla.Client.t, binary) :: {:ok, any} | {:error, code, reason}
  defp do_get(client, url) do
    tesla_response(Tesla.get(client, url))
  end

  @spec do_delete(Tesla.Client.t, binary) :: {:ok, any} | {:error, code, reason}
  defp do_delete(client, url) do
    tesla_response(Tesla.delete(client, url))
  end

  @spec do_post(Tesla.Client.t, binary, any) :: {:ok, any} | {:error, code, reason}
  defp do_post(client, url, data) when is_binary(data) do
    tesla_response(Tesla.post(client, url, data))
  end
  defp do_post(client, url, data) do
    case Jason.encode(data) do
      {:ok, encoded} ->
        do_post(client, url, encoded)
      {:error, reason} ->
        {:error, 0, reason}
    end
  end

  @spec do_put(Tesla.Client.t, binary, any) :: {:ok, any} | {:error, code, reason}
  defp do_put(client, url, data) when is_binary(data) do
    tesla_response(Tesla.put(client, url, data))
  end
  defp do_put(client, url, data) do
    case Jason.encode(data) do
      {:ok, encoded} ->
        do_put(client, url, encoded)
      {:error, reason} ->
        {:error, 0, reason}
    end
  end

  defp tesla_response({:ok, %{status: 200, body: body}}), do: {:ok, body}
  defp tesla_response({:ok, %{status: status, body: body}}), do: {:error, status, body}
  defp tesla_response({:error, reason}), do: {:error, 0, reason}

end