defmodule Pulsar.Schema do @moduledoc """ Schema definition for Pulsar messages. Pulsar schemas enable type-safe messaging by defining the structure of message data. When a producer connects with a schema, the broker registers it and returns a schema version that is included in every message. ## Supported Types - Primitive: `:None`, `:String`, `:Bool`, `:Int8`, `:Int16`, `:Int32`, `:Int64`, `:Float`, `:Double` - Structured: `:Json`, `:Avro`, `:Protobuf`, `:ProtobufNative`, `:KeyValue` - Temporal: `:Date`, `:Time`, `:Timestamp`, `:Instant`, `:LocalDate`, `:LocalTime`, `:LocalDateTime` See `new/1` for usage examples. """ alias Pulsar.Protocol.Binary.Pulsar.Proto, as: Binary @valid_types ~w( None String Json Protobuf Avro Bool Int8 Int16 Int32 Int64 Float Double Date Time Timestamp KeyValue Instant LocalDate LocalTime LocalDateTime ProtobufNative AutoConsume )a @primitive_types ~w(None String Bool Int8 Int16 Int32 Int64 Float Double)a @type schema_type :: unquote(Enum.reduce(@valid_types, &{:|, [], [&1, &2]})) @type t :: %__MODULE__{ type: schema_type(), definition: binary(), name: String.t() | nil, properties: map() | nil } defstruct [:type, :definition, :name, :properties] @doc """ Creates a new schema. ## Options - `:type` - (required) The schema type (e.g., `:Json`, `:String`, `:Avro`) - `:definition` - The schema definition (required for non-primitive types). For `:Json` and `:Avro` schemas, can be a struct, map, or binary string. Structs and maps will be automatically JSON-encoded. - `:name` - Optional schema name - `:properties` - Optional metadata properties as a map ## Returns - `{:ok, schema}` on success - `{:error, reason}` if validation fails ## Examples # JSON schema (Pulsar uses Avro-style record format) {:ok, schema} = Pulsar.Schema.new( type: :Json, definition: %{ type: "record", name: "User", fields: [%{name: "id", type: "int"}] } ) # String schema (primitive, no definition needed) {:ok, schema} = Pulsar.Schema.new(type: :String) # With properties {:ok, schema} = Pulsar.Schema.new( type: :Avro, definition: avro_schema, name: "my-schema", properties: %{"version" => "1.0"} ) """ @spec new(keyword()) :: {:ok, t()} | {:error, term()} def new(opts) when is_list(opts) do with {:ok, type} <- validate_type(opts[:type]), {:ok, definition} <- validate_definition(type, opts[:definition]), {:ok, properties} <- validate_properties(opts[:properties]) do {:ok, %__MODULE__{type: type, definition: definition, name: opts[:name], properties: properties}} end end @doc """ Creates a new schema, raising on error. Same as `new/1` but raises `ArgumentError` if validation fails. """ @spec new!(keyword()) :: t() def new!(opts) when is_list(opts) do case new(opts) do {:ok, schema} -> schema {:error, reason} -> raise ArgumentError, "invalid schema: #{inspect(reason)}" end end @doc """ Converts a schema to the binary protocol format. Returns `nil` if the input is `nil`. """ @spec to_binary(t() | nil) :: Binary.Schema.t() | nil def to_binary(nil), do: nil def to_binary(%__MODULE__{} = schema) do %Binary.Schema{ name: schema.name || "", schema_data: schema.definition || <<>>, type: schema.type, properties: Pulsar.Protocol.to_key_value_list(schema.properties) } end # Private functions defp validate_type(nil), do: {:error, :type_required} defp validate_type(type) when type in @valid_types, do: {:ok, type} defp validate_type(type), do: {:error, {:invalid_type, type}} defp validate_definition(type, def) when type in @primitive_types, do: {:ok, def || <<>>} defp validate_definition(_type, nil), do: {:error, :definition_required} defp validate_definition(_type, def) when is_binary(def), do: {:ok, def} defp validate_definition(type, %{__struct__: _} = def) when type in [:Json, :Avro] do validate_definition(type, Map.from_struct(def)) end defp validate_definition(type, def) when type in [:Json, :Avro] and is_map(def) do case Jason.encode(def) do {:ok, encoded} -> {:ok, encoded} {:error, reason} -> {:error, {:json_encode_failed, reason}} end end defp validate_definition(_type, def), do: {:error, {:invalid_definition, def}} defp validate_properties(nil), do: {:ok, nil} defp validate_properties(props) when is_map(props), do: {:ok, props} defp validate_properties(props), do: {:error, {:invalid_properties, props}} end