feat: Add Signet Orders Indexer for cross-chain order tracking

.github/workflows/config.yml

@@ -62,6 +62,7 @@ jobs:
               "rsk",
               "scroll",
               "shibarium",
+              "signet",
               "stability",
               "zetachain",
               "zilliqa",

apps/explorer/config/dev.exs

@@ -32,7 +32,8 @@ for repo <- [
       Explorer.Repo.Suave,
       Explorer.Repo.Zilliqa,
       Explorer.Repo.ZkSync,
-      Explorer.Repo.Neon
+      Explorer.Repo.Neon,
+      Explorer.Repo.Signet
     ] do
   config :explorer, repo, timeout: :timer.seconds(80)
 end

apps/explorer/config/prod.exs

@@ -34,7 +34,8 @@ for repo <- [
       Explorer.Repo.Suave,
       Explorer.Repo.Zilliqa,
       Explorer.Repo.ZkSync,
-      Explorer.Repo.Neon
+      Explorer.Repo.Neon,
+      Explorer.Repo.Signet
     ] do
   config :explorer, repo,
     prepare: :unnamed,

apps/explorer/config/test.exs

@@ -84,7 +84,8 @@ for repo <- [
       Explorer.Repo.Suave,
       Explorer.Repo.Zilliqa,
       Explorer.Repo.ZkSync,
-      Explorer.Repo.Neon
+      Explorer.Repo.Neon,
+      Explorer.Repo.Signet
     ] do
   config :explorer, repo,
     database: database,

apps/explorer/lib/explorer/application.ex

@@ -378,7 +378,8 @@ defmodule Explorer.Application do
         Explorer.Repo.Stability,
         Explorer.Repo.Suave,
         Explorer.Repo.Zilliqa,
-        Explorer.Repo.ZkSync
+        Explorer.Repo.ZkSync,
+        Explorer.Repo.Signet
       ]
     else
       []

apps/explorer/lib/explorer/chain/import/runner/signet/fills.ex

@@ -0,0 +1,106 @@
+defmodule Explorer.Chain.Import.Runner.Signet.Fills do
+  @moduledoc """
+    Bulk imports of Explorer.Chain.Signet.Fill.
+  """
+
+  require Ecto.Query
+
+  import Ecto.Query, only: [from: 2]
+
+  alias Ecto.{Changeset, Multi, Repo}
+  alias Explorer.Chain.Import
+  alias Explorer.Chain.Signet.Fill
+  alias Explorer.Prometheus.Instrumenter
+
+  @behaviour Import.Runner
+
+  # milliseconds
+  @timeout 60_000
+
+  @type imported :: [Fill.t()]
+
+  @impl Import.Runner
+  def ecto_schema_module, do: Fill
+
+  @impl Import.Runner
+  def option_key, do: :signet_fills
+
+  @impl Import.Runner
+  def imported_table_row do
+    %{
+      value_type: "[#{ecto_schema_module()}.t()]",
+      value_description: "List of `t:#{ecto_schema_module()}.t/0`s"
+    }
+  end
+
+  @impl Import.Runner
+  def run(multi, changes_list, %{timestamps: timestamps} = options) do
+    insert_options =
+      options
+      |> Map.get(option_key(), %{})
+      |> Map.take(~w(on_conflict timeout)a)
+      |> Map.put_new(:timeout, @timeout)
+      |> Map.put(:timestamps, timestamps)
+
+    Multi.run(multi, Fill.insert_result_key(), fn repo, _ ->
+      Instrumenter.block_import_stage_runner(
+        fn -> insert(repo, changes_list, insert_options) end,
+        :block_referencing,
+        :signet_fills,
+        :signet_fills
+      )
+    end)
+  end
+
+  @impl Import.Runner
+  def timeout, do: @timeout
+
+  @spec insert(Repo.t(), [map()], %{required(:timeout) => timeout(), required(:timestamps) => Import.timestamps()}) ::
+          {:ok, [Fill.t()]}
+          | {:error, [Changeset.t()]}
+  def insert(repo, changes_list, %{timeout: timeout, timestamps: timestamps} = options) when is_list(changes_list) do
+    on_conflict = Map.get_lazy(options, :on_conflict, &default_on_conflict/0)
+
+    # Enforce Fill ShareLocks order (see docs: sharelock.md)
+    # Sort by composite primary key: chain_type, transaction_hash, log_index
+    ordered_changes_list =
+      Enum.sort_by(changes_list, fn change ->
+        {change.chain_type, change.transaction_hash, change.log_index}
+      end)
+
+    {:ok, inserted} =
+      Import.insert_changes_list(
+        repo,
+        ordered_changes_list,
+        conflict_target: [:chain_type, :transaction_hash, :log_index],
+        on_conflict: on_conflict,
+        for: Fill,
+        returning: true,
+        timeout: timeout,
+        timestamps: timestamps
+      )
+
+    {:ok, inserted}
+  end
+
+  defp default_on_conflict do
+    from(
+      f in Fill,
+      update: [
+        set: [
+          # Don't update primary key fields (chain_type, transaction_hash, log_index)
+          block_number: fragment("COALESCE(EXCLUDED.block_number, ?)", f.block_number),
+          outputs_json: fragment("COALESCE(EXCLUDED.outputs_json, ?)", f.outputs_json),
+          inserted_at: fragment("LEAST(?, EXCLUDED.inserted_at)", f.inserted_at),
+          updated_at: fragment("GREATEST(?, EXCLUDED.updated_at)", f.updated_at)
+        ]
+      ],
+      where:
+        fragment(
+          "(EXCLUDED.block_number, EXCLUDED.outputs_json) IS DISTINCT FROM (?, ?)",
+          f.block_number,
+          f.outputs_json
+        )
+    )
+  end
+end

apps/explorer/lib/explorer/chain/import/runner/signet/orders.ex

@@ -0,0 +1,114 @@
+defmodule Explorer.Chain.Import.Runner.Signet.Orders do
+  @moduledoc """
+    Bulk imports of Explorer.Chain.Signet.Order.
+  """
+
+  require Ecto.Query
+
+  import Ecto.Query, only: [from: 2]
+
+  alias Ecto.{Changeset, Multi, Repo}
+  alias Explorer.Chain.Import
+  alias Explorer.Chain.Signet.Order
+  alias Explorer.Prometheus.Instrumenter
+
+  @behaviour Import.Runner
+
+  # milliseconds
+  @timeout 60_000
+
+  @type imported :: [Order.t()]
+
+  @impl Import.Runner
+  def ecto_schema_module, do: Order
+
+  @impl Import.Runner
+  def option_key, do: :signet_orders
+
+  @impl Import.Runner
+  def imported_table_row do
+    %{
+      value_type: "[#{ecto_schema_module()}.t()]",
+      value_description: "List of `t:#{ecto_schema_module()}.t/0`s"
+    }
+  end
+
+  @impl Import.Runner
+  def run(multi, changes_list, %{timestamps: timestamps} = options) do
+    insert_options =
+      options
+      |> Map.get(option_key(), %{})
+      |> Map.take(~w(on_conflict timeout)a)
+      |> Map.put_new(:timeout, @timeout)
+      |> Map.put(:timestamps, timestamps)
+
+    Multi.run(multi, Order.insert_result_key(), fn repo, _ ->
+      Instrumenter.block_import_stage_runner(
+        fn -> insert(repo, changes_list, insert_options) end,
+        :block_referencing,
+        :signet_orders,
+        :signet_orders
+      )
+    end)
+  end
+
+  @impl Import.Runner
+  def timeout, do: @timeout
+
+  @spec insert(Repo.t(), [map()], %{required(:timeout) => timeout(), required(:timestamps) => Import.timestamps()}) ::
+          {:ok, [Order.t()]}
+          | {:error, [Changeset.t()]}
+  def insert(repo, changes_list, %{timeout: timeout, timestamps: timestamps} = options) when is_list(changes_list) do
+    on_conflict = Map.get_lazy(options, :on_conflict, &default_on_conflict/0)
+
+    # Enforce Order ShareLocks order (see docs: sharelock.md)
+    # Sort by composite primary key: transaction_hash, then log_index
+    ordered_changes_list =
+      Enum.sort_by(changes_list, fn change ->
+        {change.transaction_hash, change.log_index}
+      end)
+
+    {:ok, inserted} =
+      Import.insert_changes_list(
+        repo,
+        ordered_changes_list,
+        conflict_target: [:transaction_hash, :log_index],
+        on_conflict: on_conflict,
+        for: Order,
+        returning: true,
+        timeout: timeout,
+        timestamps: timestamps
+      )
+
+    {:ok, inserted}
+  end
+
+  defp default_on_conflict do
+    from(
+      o in Order,
+      update: [
+        set: [
+          # Don't update primary key fields (transaction_hash, log_index)
+          deadline: fragment("COALESCE(EXCLUDED.deadline, ?)", o.deadline),
+          block_number: fragment("COALESCE(EXCLUDED.block_number, ?)", o.block_number),
+          inputs_json: fragment("COALESCE(EXCLUDED.inputs_json, ?)", o.inputs_json),
+          outputs_json: fragment("COALESCE(EXCLUDED.outputs_json, ?)", o.outputs_json),
+          sweep_recipient: fragment("COALESCE(EXCLUDED.sweep_recipient, ?)", o.sweep_recipient),
+          sweep_token: fragment("COALESCE(EXCLUDED.sweep_token, ?)", o.sweep_token),
+          sweep_amount: fragment("COALESCE(EXCLUDED.sweep_amount, ?)", o.sweep_amount),
+          inserted_at: fragment("LEAST(?, EXCLUDED.inserted_at)", o.inserted_at),
+          updated_at: fragment("GREATEST(?, EXCLUDED.updated_at)", o.updated_at)
+        ]
+      ],
+      where:
+        fragment(
+          "(EXCLUDED.deadline, EXCLUDED.block_number, EXCLUDED.sweep_recipient, EXCLUDED.sweep_token, EXCLUDED.sweep_amount) IS DISTINCT FROM (?, ?, ?, ?, ?)",
+          o.deadline,
+          o.block_number,
+          o.sweep_recipient,
+          o.sweep_token,
+          o.sweep_amount
+        )
+    )
+  end
+end

apps/explorer/lib/explorer/chain/import/stage/chain_type_specific.ex

@@ -68,6 +68,10 @@ defmodule Explorer.Chain.Import.Stage.ChainTypeSpecific do
     ],
     stability: [
       Runner.Stability.Validators
+    ],
+    signet: [
+      Runner.Signet.Orders,
+      Runner.Signet.Fills
     ]
   }
 

apps/explorer/lib/explorer/chain/signet/fill.ex

@@ -0,0 +1,74 @@
+defmodule Explorer.Chain.Signet.Fill do
+  @moduledoc """
+    Models a Signet Filled event from RollupOrders or HostOrders contracts.
+
+    Fills are indexed independently and uniquely identified by their
+    chain_type + transaction_hash + log_index combination.
+
+    Changes in the schema should be reflected in the bulk import module:
+    - Explorer.Chain.Import.Runner.Signet.Fills
+
+    Migrations:
+    - Explorer.Repo.Signet.Migrations.CreateSignetTables
+  """
+
+  use Explorer.Schema
+
+  alias Explorer.Chain.Hash
+
+  @insert_result_key :insert_signet_fills
+
+  @optional_attrs ~w()a
+
+  @required_attrs ~w(chain_type block_number transaction_hash log_index outputs_json)a
+
+  @allowed_attrs @optional_attrs ++ @required_attrs
+
+  @typedoc """
+  Descriptor of a Signet Filled event:
+    * `chain_type` - Whether this fill occurred on :rollup or :host chain (primary key)
+    * `transaction_hash` - The hash of the transaction containing the fill (primary key)
+    * `log_index` - The index of the log within the transaction (primary key)
+    * `block_number` - The block number where the fill was executed
+    * `outputs_json` - List of filled outputs (token, amount, recipient, chainId) stored as JSONB.
+      Note: Uses `:map` Ecto type which accepts both maps and lists - PostgreSQL JSONB
+      stores both natively, and `insert_all` bypasses changeset validation.
+      NOTE: In Filled events, the `chainId` field represents the ORIGIN chain
+      (where the order was created), not the chain where the fill occurred.
+  """
+  @type to_import :: %{
+          chain_type: :rollup | :host,
+          block_number: non_neg_integer(),
+          transaction_hash: binary(),
+          log_index: non_neg_integer(),
+          outputs_json: [map()]
+        }
+
+  @primary_key false
+  typed_schema "signet_fills" do
+    field(:chain_type, Ecto.Enum, values: [:rollup, :host], primary_key: true)
+    field(:transaction_hash, Hash.Full, primary_key: true)
+    field(:log_index, :integer, primary_key: true)
+    field(:block_number, :integer)
+    field(:outputs_json, :map)
+
+    timestamps()
+  end
+
+  @doc """
+    Validates that the `attrs` are valid.
+  """
+  @spec changeset(Ecto.Schema.t(), map()) :: Ecto.Changeset.t()
+  def changeset(%__MODULE__{} = fill, attrs \\ %{}) do
+    fill
+    |> cast(attrs, @allowed_attrs)
+    |> validate_required(@required_attrs)
+    |> unique_constraint([:chain_type, :transaction_hash, :log_index])
+  end
+
+  @doc """
+  Shared result key used by import runners to return inserted Signet fills.
+  """
+  @spec insert_result_key() :: atom()
+  def insert_result_key, do: @insert_result_key
+end