From e6df337ab7c90fa4476a19cb9982c711df626b91 Mon Sep 17 00:00:00 2001
From: saneery <anatolyniky@gmail.com>
Date: Fri, 26 Jul 2019 15:39:27 +0300
Subject: [PATCH 01/16] Send events through postgres notify

---
 apps/explorer/lib/explorer/application.ex     |  3 ++
 .../lib/explorer/chain/events/listener.ex     | 54 +++++++++++++++++++
 .../lib/explorer/chain/events/publisher.ex    | 21 ++++----
 3 files changed, 68 insertions(+), 10 deletions(-)
 create mode 100644 apps/explorer/lib/explorer/chain/events/listener.ex

diff --git a/apps/explorer/lib/explorer/application.ex b/apps/explorer/lib/explorer/application.ex
index c1fb19f600..c509490525 100644
--- a/apps/explorer/lib/explorer/application.ex
+++ b/apps/explorer/lib/explorer/application.ex
@@ -16,6 +16,8 @@ defmodule Explorer.Application do
     Transactions
   }
 
+  alias Explorer.Chain.Events.Listener
+
   alias Explorer.Chain.Supply.RSK
 
   alias Explorer.Market.MarketHistoryCache
@@ -43,6 +45,7 @@ defmodule Explorer.Application do
       {Admin.Recovery, [[], [name: Admin.Recovery]]},
       {TransactionCount, [[], []]},
       {BlockCount, []},
+      {Listener, []},
       con_cache_child_spec(Blocks.cache_name()),
       con_cache_child_spec(NetVersion.cache_name()),
       con_cache_child_spec(MarketHistoryCache.cache_name()),
diff --git a/apps/explorer/lib/explorer/chain/events/listener.ex b/apps/explorer/lib/explorer/chain/events/listener.ex
new file mode 100644
index 0000000000..8ff076a014
--- /dev/null
+++ b/apps/explorer/lib/explorer/chain/events/listener.ex
@@ -0,0 +1,54 @@
+defmodule Explorer.Chain.Events.Listener do
+  @moduledoc """
+  Listens and publishes events from PG
+  """
+
+  use GenServer
+
+  alias Postgrex.Notifications
+
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, "chain_event")
+  end
+
+  def init(channel) do
+    {:ok, pid} =
+      :explorer
+      |> Application.get_env(Explorer.Repo)
+      |> Notifications.start_link()
+
+    ref = Notifications.listen!(pid, channel)
+
+    {:ok, {pid, ref, channel}}
+  end
+
+  def handle_info({:notification, _pid, _ref, _topic, payload}, state) do
+    payload
+    |> decode_payload!()
+    |> broadcast()
+
+    {:noreply, state}
+  end
+
+  defp decode_payload!(payload) do
+    payload
+    |> Base.decode64!()
+    |> :erlang.binary_to_term()
+  end
+
+  defp broadcast({:chain_event, event_type} = event) do
+    Registry.dispatch(Registry.ChainEvents, event_type, fn entries ->
+      for {pid, _registered_val} <- entries do
+        send(pid, event)
+      end
+    end)
+  end
+
+  defp broadcast({:chain_event, event_type, broadcast_type, _data} = event) do
+    Registry.dispatch(Registry.ChainEvents, {event_type, broadcast_type}, fn entries ->
+      for {pid, _registered_val} <- entries do
+        send(pid, event)
+      end
+    end)
+  end
+end
diff --git a/apps/explorer/lib/explorer/chain/events/publisher.ex b/apps/explorer/lib/explorer/chain/events/publisher.ex
index c7e01eef3a..8bb59fac65 100644
--- a/apps/explorer/lib/explorer/chain/events/publisher.ex
+++ b/apps/explorer/lib/explorer/chain/events/publisher.ex
@@ -2,6 +2,7 @@ defmodule Explorer.Chain.Events.Publisher do
   @moduledoc """
   Publishes events related to the Chain context.
   """
+  alias Explorer.Repo
 
   @allowed_events ~w(addresses address_coin_balances blocks block_rewards internal_transactions token_transfers transactions contract_verification_result)a
 
@@ -19,11 +20,8 @@ defmodule Explorer.Chain.Events.Publisher do
   end
 
   defp send_data(event_type) do
-    Registry.dispatch(Registry.ChainEvents, event_type, fn entries ->
-      for {pid, _registered_val} <- entries do
-        send(pid, {:chain_event, event_type})
-      end
-    end)
+    payload = encode_payload({:chain_event, event_type})
+    Repo.query("NOTIFY chain_event, '#{payload}';")
   end
 
   # The :catchup type of event is not being consumed right now.
@@ -32,10 +30,13 @@ defmodule Explorer.Chain.Events.Publisher do
   defp send_data(_event_type, :catchup, _event_data), do: :ok
 
   defp send_data(event_type, broadcast_type, event_data) do
-    Registry.dispatch(Registry.ChainEvents, {event_type, broadcast_type}, fn entries ->
-      for {pid, _registered_val} <- entries do
-        send(pid, {:chain_event, event_type, broadcast_type, event_data})
-      end
-    end)
+    payload = encode_payload({:chain_event, event_type, broadcast_type, event_data})
+    Repo.query("NOTIFY chain_event, '#{payload}';")
+  end
+
+  defp encode_payload(payload) do
+    payload
+    |> :erlang.term_to_binary()
+    |> Base.encode64()
   end
 end

From 19d84abd401c83633fb2cd35afbb17654458f9d2 Mon Sep 17 00:00:00 2001
From: saneery <anatolyniky@gmail.com>
Date: Mon, 29 Jul 2019 15:55:35 +0300
Subject: [PATCH 02/16] Update tests to receive realtime events

---
 .../lib/explorer/chain/events/listener.ex     |  1 +
 .../lib/explorer/chain/events/publisher.ex    | 20 ++++++++++++++++---
 .../explorer/chain/events/publisher_test.exs  |  6 +++---
 .../explorer/chain/events/subscriber_test.exs |  4 ++--
 .../test/explorer/chain/import_test.exs       | 12 +++++------
 5 files changed, 29 insertions(+), 14 deletions(-)

diff --git a/apps/explorer/lib/explorer/chain/events/listener.ex b/apps/explorer/lib/explorer/chain/events/listener.ex
index 8ff076a014..40a4131e84 100644
--- a/apps/explorer/lib/explorer/chain/events/listener.ex
+++ b/apps/explorer/lib/explorer/chain/events/listener.ex
@@ -30,6 +30,7 @@ defmodule Explorer.Chain.Events.Listener do
     {:noreply, state}
   end
 
+  # sobelow_skip ["Misc.BinToTerm"]
   defp decode_payload!(payload) do
     payload
     |> Base.decode64!()
diff --git a/apps/explorer/lib/explorer/chain/events/publisher.ex b/apps/explorer/lib/explorer/chain/events/publisher.ex
index 8bb59fac65..31374e19f0 100644
--- a/apps/explorer/lib/explorer/chain/events/publisher.ex
+++ b/apps/explorer/lib/explorer/chain/events/publisher.ex
@@ -2,6 +2,7 @@ defmodule Explorer.Chain.Events.Publisher do
   @moduledoc """
   Publishes events related to the Chain context.
   """
+  alias Ecto.Adapters.SQL.Sandbox
   alias Explorer.Repo
 
   @allowed_events ~w(addresses address_coin_balances blocks block_rewards internal_transactions token_transfers transactions contract_verification_result)a
@@ -17,11 +18,12 @@ defmodule Explorer.Chain.Events.Publisher do
   @spec broadcast(atom()) :: :ok
   def broadcast(event_type) do
     send_data(event_type)
+    :ok
   end
 
   defp send_data(event_type) do
     payload = encode_payload({:chain_event, event_type})
-    Repo.query("NOTIFY chain_event, '#{payload}';")
+    send_notify(payload)
   end
 
   # The :catchup type of event is not being consumed right now.
@@ -31,12 +33,24 @@ defmodule Explorer.Chain.Events.Publisher do
 
   defp send_data(event_type, broadcast_type, event_data) do
     payload = encode_payload({:chain_event, event_type, broadcast_type, event_data})
-    Repo.query("NOTIFY chain_event, '#{payload}';")
+    send_notify(payload)
   end
 
   defp encode_payload(payload) do
     payload
-    |> :erlang.term_to_binary()
+    |> :erlang.term_to_binary([:compressed])
     |> Base.encode64()
   end
+
+  defp send_notify(payload) do
+    fun = fn ->
+      Repo.query("select pg_notify('chain_event', $1::text);", [payload])
+    end
+
+    if Mix.env() == :test do
+      Sandbox.unboxed_run(Repo, fun)
+    else
+      fun.()
+    end
+  end
 end
diff --git a/apps/explorer/test/explorer/chain/events/publisher_test.exs b/apps/explorer/test/explorer/chain/events/publisher_test.exs
index ef76126086..e0934a53dd 100644
--- a/apps/explorer/test/explorer/chain/events/publisher_test.exs
+++ b/apps/explorer/test/explorer/chain/events/publisher_test.exs
@@ -15,7 +15,7 @@ defmodule Explorer.Chain.Events.PublisherTest do
 
       Publisher.broadcast([{event_type, event_data}], broadcast_type)
 
-      assert_received {:chain_event, ^event_type, ^broadcast_type, []}
+      assert_receive {:chain_event, ^event_type, ^broadcast_type, []}
     end
 
     test "won't send chain_event of catchup type" do
@@ -27,7 +27,7 @@ defmodule Explorer.Chain.Events.PublisherTest do
 
       Publisher.broadcast([{event_type, event_data}], broadcast_type)
 
-      refute_received {:chain_event, ^event_type, ^broadcast_type, []}
+      refute_receive {:chain_event, ^event_type, ^broadcast_type, []}
     end
 
     test "won't send event that is not allowed" do
@@ -59,7 +59,7 @@ defmodule Explorer.Chain.Events.PublisherTest do
 
       Publisher.broadcast(event_type)
 
-      assert_received {:chain_event, ^event_type}
+      assert_receive {:chain_event, ^event_type}
     end
   end
 end
diff --git a/apps/explorer/test/explorer/chain/events/subscriber_test.exs b/apps/explorer/test/explorer/chain/events/subscriber_test.exs
index 531efc24b2..b26841d0ef 100644
--- a/apps/explorer/test/explorer/chain/events/subscriber_test.exs
+++ b/apps/explorer/test/explorer/chain/events/subscriber_test.exs
@@ -15,7 +15,7 @@ defmodule Explorer.Chain.Events.SubscriberTest do
 
       Publisher.broadcast([{event_type, event_data}], broadcast_type)
 
-      assert_received {:chain_event, :blocks, :realtime, []}
+      assert_receive {:chain_event, :blocks, :realtime, []}
     end
   end
 
@@ -27,7 +27,7 @@ defmodule Explorer.Chain.Events.SubscriberTest do
 
       Publisher.broadcast(event_type)
 
-      assert_received {:chain_event, :exchange_rate}
+      assert_receive {:chain_event, :exchange_rate}
     end
   end
 end
diff --git a/apps/explorer/test/explorer/chain/import_test.exs b/apps/explorer/test/explorer/chain/import_test.exs
index dd3bbf1aec..0c559bdfec 100644
--- a/apps/explorer/test/explorer/chain/import_test.exs
+++ b/apps/explorer/test/explorer/chain/import_test.exs
@@ -470,27 +470,27 @@ defmodule Explorer.Chain.ImportTest do
     test "publishes addresses with updated fetched_coin_balance data to subscribers on insert" do
       Subscriber.to(:addresses, :realtime)
       Import.all(@import_data)
-      assert_received {:chain_event, :addresses, :realtime, [%Address{}, %Address{}, %Address{}]}
+      assert_receive {:chain_event, :addresses, :realtime, [%Address{}, %Address{}, %Address{}]}
     end
 
     test "publishes block data to subscribers on insert" do
       Subscriber.to(:blocks, :realtime)
       Import.all(@import_data)
-      assert_received {:chain_event, :blocks, :realtime, [%Block{}]}
+      assert_receive {:chain_event, :blocks, :realtime, [%Block{}]}
     end
 
     test "publishes internal_transaction data to subscribers on insert" do
       Subscriber.to(:internal_transactions, :realtime)
       Import.all(@import_data)
 
-      assert_received {:chain_event, :internal_transactions, :realtime,
-                       [%{transaction_hash: _, index: _}, %{transaction_hash: _, index: _}]}
+      assert_receive {:chain_event, :internal_transactions, :realtime,
+                      [%{transaction_hash: _, index: _}, %{transaction_hash: _, index: _}]}
     end
 
     test "publishes transactions data to subscribers on insert" do
       Subscriber.to(:transactions, :realtime)
       Import.all(@import_data)
-      assert_received {:chain_event, :transactions, :realtime, [%Transaction{}]}
+      assert_receive {:chain_event, :transactions, :realtime, [%Transaction{}]}
     end
 
     test "publishes token_transfers data to subscribers on insert" do
@@ -498,7 +498,7 @@ defmodule Explorer.Chain.ImportTest do
 
       Import.all(@import_data)
 
-      assert_received {:chain_event, :token_transfers, :realtime, [%TokenTransfer{}]}
+      assert_receive {:chain_event, :token_transfers, :realtime, [%TokenTransfer{}]}
     end
 
     test "does not broadcast if broadcast option is false" do

From ea0d47cfa4cbfb485893edb3f244452a1c1a460d Mon Sep 17 00:00:00 2001
From: saneery <anatolyniky@gmail.com>
Date: Thu, 1 Aug 2019 11:47:43 +0300
Subject: [PATCH 03/16] Add mock to send events directly to listener

---
 apps/explorer/config/config.exs               |  3 ++-
 apps/explorer/config/test.exs                 |  3 +++
 .../lib/explorer/chain/events/listener.ex     |  2 +-
 .../lib/explorer/chain/events/publisher.ex    | 19 +++----------------
 .../lib/explorer/chain/events/sender.ex       | 12 ++++++++++++
 .../lib/explorer/chain/events/sender_mock.ex  | 10 ++++++++++
 .../explorer/chain/events/publisher_test.exs  |  2 +-
 7 files changed, 32 insertions(+), 19 deletions(-)
 create mode 100644 apps/explorer/lib/explorer/chain/events/sender.ex
 create mode 100644 apps/explorer/lib/explorer/chain/events/sender_mock.ex

diff --git a/apps/explorer/config/config.exs b/apps/explorer/config/config.exs
index fed5124a2a..7e1914e89c 100644
--- a/apps/explorer/config/config.exs
+++ b/apps/explorer/config/config.exs
@@ -15,7 +15,8 @@ config :explorer,
       "homestead,tangerineWhistle,spuriousDragon,byzantium,constantinople,petersburg,default",
   include_uncles_in_average_block_time:
     if(System.get_env("UNCLES_IN_AVERAGE_BLOCK_TIME") == "false", do: false, else: true),
-  healthy_blocks_period: System.get_env("HEALTHY_BLOCKS_PERIOD") || :timer.minutes(5)
+  healthy_blocks_period: System.get_env("HEALTHY_BLOCKS_PERIOD") || :timer.minutes(5),
+  realtime_events_sender: Explorer.Chain.Events.Sender
 
 average_block_period =
   case Integer.parse(System.get_env("AVERAGE_BLOCK_CACHE_PERIOD", "")) do
diff --git a/apps/explorer/config/test.exs b/apps/explorer/config/test.exs
index d91f4dbdfc..4838fa8e32 100644
--- a/apps/explorer/config/test.exs
+++ b/apps/explorer/config/test.exs
@@ -41,6 +41,9 @@ end
 config :explorer, Explorer.ExchangeRates.Source.TransactionAndLog,
   secondary_source: Explorer.ExchangeRates.Source.OneCoinSource
 
+config :explorer,
+  realtime_events_sender: Explorer.Chain.Events.SenderMock
+
 variant =
   if is_nil(System.get_env("ETHEREUM_JSONRPC_VARIANT")) do
     "parity"
diff --git a/apps/explorer/lib/explorer/chain/events/listener.ex b/apps/explorer/lib/explorer/chain/events/listener.ex
index 40a4131e84..2d93eeada0 100644
--- a/apps/explorer/lib/explorer/chain/events/listener.ex
+++ b/apps/explorer/lib/explorer/chain/events/listener.ex
@@ -8,7 +8,7 @@ defmodule Explorer.Chain.Events.Listener do
   alias Postgrex.Notifications
 
   def start_link(_) do
-    GenServer.start_link(__MODULE__, "chain_event")
+    GenServer.start_link(__MODULE__, "chain_event", name: __MODULE__)
   end
 
   def init(channel) do
diff --git a/apps/explorer/lib/explorer/chain/events/publisher.ex b/apps/explorer/lib/explorer/chain/events/publisher.ex
index 31374e19f0..11a7dc1518 100644
--- a/apps/explorer/lib/explorer/chain/events/publisher.ex
+++ b/apps/explorer/lib/explorer/chain/events/publisher.ex
@@ -2,8 +2,7 @@ defmodule Explorer.Chain.Events.Publisher do
   @moduledoc """
   Publishes events related to the Chain context.
   """
-  alias Ecto.Adapters.SQL.Sandbox
-  alias Explorer.Repo
+  @sender Application.get_env(:explorer, :realtime_events_sender)
 
   @allowed_events ~w(addresses address_coin_balances blocks block_rewards internal_transactions token_transfers transactions contract_verification_result)a
 
@@ -23,7 +22,7 @@ defmodule Explorer.Chain.Events.Publisher do
 
   defp send_data(event_type) do
     payload = encode_payload({:chain_event, event_type})
-    send_notify(payload)
+    @sender.send_notify(payload)
   end
 
   # The :catchup type of event is not being consumed right now.
@@ -33,7 +32,7 @@ defmodule Explorer.Chain.Events.Publisher do
 
   defp send_data(event_type, broadcast_type, event_data) do
     payload = encode_payload({:chain_event, event_type, broadcast_type, event_data})
-    send_notify(payload)
+    @sender.send_notify(payload)
   end
 
   defp encode_payload(payload) do
@@ -41,16 +40,4 @@ defmodule Explorer.Chain.Events.Publisher do
     |> :erlang.term_to_binary([:compressed])
     |> Base.encode64()
   end
-
-  defp send_notify(payload) do
-    fun = fn ->
-      Repo.query("select pg_notify('chain_event', $1::text);", [payload])
-    end
-
-    if Mix.env() == :test do
-      Sandbox.unboxed_run(Repo, fun)
-    else
-      fun.()
-    end
-  end
 end
diff --git a/apps/explorer/lib/explorer/chain/events/sender.ex b/apps/explorer/lib/explorer/chain/events/sender.ex
new file mode 100644
index 0000000000..f64728791f
--- /dev/null
+++ b/apps/explorer/lib/explorer/chain/events/sender.ex
@@ -0,0 +1,12 @@
+defmodule Explorer.Chain.Events.Sender do
+  @moduledoc """
+  Sends events to Postgres.
+  """
+  alias Explorer.Repo
+
+  @callback send_notify(String.t()) :: {:ok, any}
+
+  def send_notify(payload) do
+    Repo.query!("select pg_notify('chain_event', $1::text);", [payload])
+  end
+end
diff --git a/apps/explorer/lib/explorer/chain/events/sender_mock.ex b/apps/explorer/lib/explorer/chain/events/sender_mock.ex
new file mode 100644
index 0000000000..44d55ab70d
--- /dev/null
+++ b/apps/explorer/lib/explorer/chain/events/sender_mock.ex
@@ -0,0 +1,10 @@
+defmodule Explorer.Chain.Events.SenderMock do
+  @moduledoc """
+  Sends events directly to Listener.
+  """
+  alias Explorer.Chain.Events.Listener
+
+  def send_notify(payload) do
+    send(Listener, {:notification, nil, nil, nil, payload})
+  end
+end
diff --git a/apps/explorer/test/explorer/chain/events/publisher_test.exs b/apps/explorer/test/explorer/chain/events/publisher_test.exs
index e0934a53dd..4a51627cc7 100644
--- a/apps/explorer/test/explorer/chain/events/publisher_test.exs
+++ b/apps/explorer/test/explorer/chain/events/publisher_test.exs
@@ -27,7 +27,7 @@ defmodule Explorer.Chain.Events.PublisherTest do
 
       Publisher.broadcast([{event_type, event_data}], broadcast_type)
 
-      refute_receive {:chain_event, ^event_type, ^broadcast_type, []}
+      refute_received {:chain_event, ^event_type, ^broadcast_type, []}
     end
 
     test "won't send event that is not allowed" do

From ad68c42fe06e98b6ebeb9e64cd97499f5b4fc817 Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Wed, 9 Oct 2019 16:50:52 +0300
Subject: [PATCH 04/16] add simple sender

---
 apps/explorer/config/config.exs               |  2 +-
 apps/explorer/config/test.exs                 |  2 +-
 .../lib/explorer/chain/events/db_sender.ex    | 28 +++++++++++++++++++
 .../lib/explorer/chain/events/publisher.ex    | 12 ++------
 .../lib/explorer/chain/events/sender.ex       | 12 --------
 .../lib/explorer/chain/events/sender_mock.ex  | 10 -------
 .../explorer/chain/events/simple_sender.ex    | 23 +++++++++++++++
 7 files changed, 55 insertions(+), 34 deletions(-)
 create mode 100644 apps/explorer/lib/explorer/chain/events/db_sender.ex
 delete mode 100644 apps/explorer/lib/explorer/chain/events/sender.ex
 delete mode 100644 apps/explorer/lib/explorer/chain/events/sender_mock.ex
 create mode 100644 apps/explorer/lib/explorer/chain/events/simple_sender.ex

diff --git a/apps/explorer/config/config.exs b/apps/explorer/config/config.exs
index fcff7aeea5..97f7391241 100644
--- a/apps/explorer/config/config.exs
+++ b/apps/explorer/config/config.exs
@@ -16,7 +16,7 @@ config :explorer,
   include_uncles_in_average_block_time:
     if(System.get_env("UNCLES_IN_AVERAGE_BLOCK_TIME") == "true", do: true, else: false),
   healthy_blocks_period: System.get_env("HEALTHY_BLOCKS_PERIOD") || :timer.minutes(5),
-  realtime_events_sender: Explorer.Chain.Events.Sender
+  realtime_events_sender: Explorer.Chain.Events.DBSender
 
 average_block_period =
   case Integer.parse(System.get_env("AVERAGE_BLOCK_CACHE_PERIOD", "")) do
diff --git a/apps/explorer/config/test.exs b/apps/explorer/config/test.exs
index c7061cd198..58050cba2c 100644
--- a/apps/explorer/config/test.exs
+++ b/apps/explorer/config/test.exs
@@ -44,7 +44,7 @@ config :explorer, Explorer.ExchangeRates.Source.TransactionAndLog,
   secondary_source: Explorer.ExchangeRates.Source.OneCoinSource
 
 config :explorer,
-  realtime_events_sender: Explorer.Chain.Events.SenderMock
+  realtime_events_sender: Explorer.Chain.Events.SimpleSender
 
 variant =
   if is_nil(System.get_env("ETHEREUM_JSONRPC_VARIANT")) do
diff --git a/apps/explorer/lib/explorer/chain/events/db_sender.ex b/apps/explorer/lib/explorer/chain/events/db_sender.ex
new file mode 100644
index 0000000000..0ea1c8eecf
--- /dev/null
+++ b/apps/explorer/lib/explorer/chain/events/db_sender.ex
@@ -0,0 +1,28 @@
+defmodule Explorer.Chain.Events.DBSender do
+  @moduledoc """
+  Sends events to Postgres.
+  """
+  alias Explorer.Repo
+
+  def send_data(event_type) do
+    payload = encode_payload({:chain_event, event_type})
+    send_notify(payload)
+  end
+
+  def send_data(_event_type, :catchup, _event_data), do: :ok
+
+  def send_data(event_type, broadcast_type, event_data) do
+    payload = encode_payload({:chain_event, event_type, broadcast_type, event_data})
+    send_notify(payload)
+  end
+
+  defp encode_payload(payload) do
+    payload
+    |> :erlang.term_to_binary([:compressed])
+    |> Base.encode64()
+  end
+
+  defp send_notify(payload) do
+    Repo.query!("select pg_notify('chain_event', $1::text);", [payload])
+  end
+end
diff --git a/apps/explorer/lib/explorer/chain/events/publisher.ex b/apps/explorer/lib/explorer/chain/events/publisher.ex
index 11a7dc1518..142baee117 100644
--- a/apps/explorer/lib/explorer/chain/events/publisher.ex
+++ b/apps/explorer/lib/explorer/chain/events/publisher.ex
@@ -21,8 +21,7 @@ defmodule Explorer.Chain.Events.Publisher do
   end
 
   defp send_data(event_type) do
-    payload = encode_payload({:chain_event, event_type})
-    @sender.send_notify(payload)
+    @sender.send_data(event_type)
   end
 
   # The :catchup type of event is not being consumed right now.
@@ -31,13 +30,6 @@ defmodule Explorer.Chain.Events.Publisher do
   defp send_data(_event_type, :catchup, _event_data), do: :ok
 
   defp send_data(event_type, broadcast_type, event_data) do
-    payload = encode_payload({:chain_event, event_type, broadcast_type, event_data})
-    @sender.send_notify(payload)
-  end
-
-  defp encode_payload(payload) do
-    payload
-    |> :erlang.term_to_binary([:compressed])
-    |> Base.encode64()
+    @sender.send_data(event_type, broadcast_type, event_data)
   end
 end
diff --git a/apps/explorer/lib/explorer/chain/events/sender.ex b/apps/explorer/lib/explorer/chain/events/sender.ex
deleted file mode 100644
index f64728791f..0000000000
--- a/apps/explorer/lib/explorer/chain/events/sender.ex
+++ /dev/null
@@ -1,12 +0,0 @@
-defmodule Explorer.Chain.Events.Sender do
-  @moduledoc """
-  Sends events to Postgres.
-  """
-  alias Explorer.Repo
-
-  @callback send_notify(String.t()) :: {:ok, any}
-
-  def send_notify(payload) do
-    Repo.query!("select pg_notify('chain_event', $1::text);", [payload])
-  end
-end
diff --git a/apps/explorer/lib/explorer/chain/events/sender_mock.ex b/apps/explorer/lib/explorer/chain/events/sender_mock.ex
deleted file mode 100644
index 44d55ab70d..0000000000
--- a/apps/explorer/lib/explorer/chain/events/sender_mock.ex
+++ /dev/null
@@ -1,10 +0,0 @@
-defmodule Explorer.Chain.Events.SenderMock do
-  @moduledoc """
-  Sends events directly to Listener.
-  """
-  alias Explorer.Chain.Events.Listener
-
-  def send_notify(payload) do
-    send(Listener, {:notification, nil, nil, nil, payload})
-  end
-end
diff --git a/apps/explorer/lib/explorer/chain/events/simple_sender.ex b/apps/explorer/lib/explorer/chain/events/simple_sender.ex
new file mode 100644
index 0000000000..d72fe9b895
--- /dev/null
+++ b/apps/explorer/lib/explorer/chain/events/simple_sender.ex
@@ -0,0 +1,23 @@
+defmodule Explorer.Chain.Events.SimpleSender do
+  @moduledoc """
+  Publishes events through Registry without intermediate levels.
+  """
+
+  def send_data(event_type, broadcast_type, event_data) do
+    Registry.dispatch(Registry.ChainEvents, {event_type, broadcast_type}, fn entries ->
+      for {pid, _registered_val} <- entries do
+        send(pid, {:chain_event, event_type, broadcast_type, event_data})
+      end
+    end)
+  end
+
+  def send_data(_event_type, :catchup, _event_data), do: :ok
+
+  def send_data(event_type) do
+    Registry.dispatch(Registry.ChainEvents, event_type, fn entries ->
+      for {pid, _registered_val} <- entries do
+        send(pid, {:chain_event, event_type})
+      end
+    end)
+  end
+end

From 84b686b05484a411828196b02102af4d5f137524 Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Wed, 9 Oct 2019 16:53:59 +0300
Subject: [PATCH 05/16] fix build

---
 apps/explorer/lib/explorer/chain/events/db_sender.ex     | 2 --
 apps/explorer/lib/explorer/chain/events/simple_sender.ex | 2 --
 2 files changed, 4 deletions(-)

diff --git a/apps/explorer/lib/explorer/chain/events/db_sender.ex b/apps/explorer/lib/explorer/chain/events/db_sender.ex
index 0ea1c8eecf..2891e8bde2 100644
--- a/apps/explorer/lib/explorer/chain/events/db_sender.ex
+++ b/apps/explorer/lib/explorer/chain/events/db_sender.ex
@@ -9,8 +9,6 @@ defmodule Explorer.Chain.Events.DBSender do
     send_notify(payload)
   end
 
-  def send_data(_event_type, :catchup, _event_data), do: :ok
-
   def send_data(event_type, broadcast_type, event_data) do
     payload = encode_payload({:chain_event, event_type, broadcast_type, event_data})
     send_notify(payload)
diff --git a/apps/explorer/lib/explorer/chain/events/simple_sender.ex b/apps/explorer/lib/explorer/chain/events/simple_sender.ex
index d72fe9b895..f2a8da3e06 100644
--- a/apps/explorer/lib/explorer/chain/events/simple_sender.ex
+++ b/apps/explorer/lib/explorer/chain/events/simple_sender.ex
@@ -11,8 +11,6 @@ defmodule Explorer.Chain.Events.SimpleSender do
     end)
   end
 
-  def send_data(_event_type, :catchup, _event_data), do: :ok
-
   def send_data(event_type) do
     Registry.dispatch(Registry.ChainEvents, event_type, fn entries ->
       for {pid, _registered_val} <- entries do

From 5825782ecd566e151ba59d47520c3aab900c5ba6 Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Wed, 9 Oct 2019 17:13:37 +0300
Subject: [PATCH 06/16] make sender configuerable

---
 apps/explorer/config/config.exs | 6 +++++-
 docs/env-variables.md           | 2 +-
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/apps/explorer/config/config.exs b/apps/explorer/config/config.exs
index 97f7391241..39124ba2e0 100644
--- a/apps/explorer/config/config.exs
+++ b/apps/explorer/config/config.exs
@@ -16,7 +16,11 @@ config :explorer,
   include_uncles_in_average_block_time:
     if(System.get_env("UNCLES_IN_AVERAGE_BLOCK_TIME") == "true", do: true, else: false),
   healthy_blocks_period: System.get_env("HEALTHY_BLOCKS_PERIOD") || :timer.minutes(5),
-  realtime_events_sender: Explorer.Chain.Events.DBSender
+  realtime_events_sender:
+    if(System.get_env("DISABLE_WEBAPP") != "true",
+      do: Explorer.Chain.Events.SimpleSender,
+      else: Explorer.Chain.Events.DBSender
+    )
 
 average_block_period =
   case Integer.parse(System.get_env("AVERAGE_BLOCK_CACHE_PERIOD", "")) do
diff --git a/docs/env-variables.md b/docs/env-variables.md
index 88f4cfbc07..0aef06b1ae 100644
--- a/docs/env-variables.md
+++ b/docs/env-variables.md
@@ -59,7 +59,7 @@ $ export NETWORK=POA
 | `UNCLES_IN_AVERAGE_BLOCK_TIME` | Include or exclude nonconsensus blocks in avg block time calculation. Exclude if `false`. | false | v2.0.1+ | | |
 | `AVERAGE_BLOCK_CACHE_PERIOD` | | Update of average block cache, in seconds | 30 minutes | v2.0.2+ | |
 | `MARKET_HISTORY_CACHE_PERIOD` | | Update of market history cache, in seconds | 6 hours | v2.0.2+ | |
-| `DISABLE_WEBAPP` | | If `true`, endpoints to webapp are hidden (compile-time) | `false` | v2.0.3+ | :white_check_mark: | |
+| `DISABLE_WEBAPP` | | If `true`, endpoints to webapp are hidden (compile-time). Also, enabling it makes notifications go through `db_notify` | `false` | v2.0.3+ | :white_check_mark: | |
 | `DISABLE_READ_API` | | If `true`, read-only endpoints to API are hidden (compile-time) | `false` | v2.0.3+ | :white_check_mark: | |
 | `DISABLE_WRITE_API` | | If `true`, write endpoints to API are hidden (compile-time) | `false` | v2.0.3+ | :white_check_mark: | |
 | `DISABLE_INDEXER` | | If `true`, indexer application doesn't run | `false` | v2.0.3+ | :white_check_mark: | |

From 1e570c1346a8b28d24043044ad3d537ad583abbc Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Thu, 10 Oct 2019 09:58:17 +0300
Subject: [PATCH 07/16] configure Explorer.Chain.Events.Listener

---
 apps/explorer/config/config.exs           | 7 +++++++
 apps/explorer/lib/explorer/application.ex | 2 +-
 2 files changed, 8 insertions(+), 1 deletion(-)

diff --git a/apps/explorer/config/config.exs b/apps/explorer/config/config.exs
index 39124ba2e0..1ccc67dad3 100644
--- a/apps/explorer/config/config.exs
+++ b/apps/explorer/config/config.exs
@@ -32,6 +32,13 @@ config :explorer, Explorer.Counters.AverageBlockTime,
   enabled: true,
   period: average_block_period
 
+config :explorer, Explorer.Chain.Events.Listener,
+  enabled:
+    if(System.get_env("DISABLE_WEBAPP") != "true",
+      do: false,
+      else: true
+    )
+
 config :explorer, Explorer.ChainSpec.GenesisData,
   enabled: true,
   chain_spec_path: System.get_env("CHAIN_SPEC_PATH"),
diff --git a/apps/explorer/lib/explorer/application.ex b/apps/explorer/lib/explorer/application.ex
index 5af1477551..0752f60cea 100644
--- a/apps/explorer/lib/explorer/application.ex
+++ b/apps/explorer/lib/explorer/application.ex
@@ -49,7 +49,6 @@ defmodule Explorer.Application do
       {Admin.Recovery, [[], [name: Admin.Recovery]]},
       TransactionCount,
       BlockCount,
-      Listener,
       Blocks,
       NetVersion,
       BlockNumber,
@@ -74,6 +73,7 @@ defmodule Explorer.Application do
       configure(Explorer.ChainSpec.GenesisData),
       configure(Explorer.KnownTokens),
       configure(Explorer.Market.History.Cataloger),
+      configure(Explorer.Chain.Events.Listener),
       configure(Explorer.Counters.AddressesWithBalanceCounter),
       configure(Explorer.Counters.AddressesCounter),
       configure(Explorer.Counters.AverageBlockTime),

From 5fe877e31650aab21995e3ca116dc7c8ea133196 Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Thu, 10 Oct 2019 10:00:39 +0300
Subject: [PATCH 08/16] fix build

---
 apps/explorer/lib/explorer/application.ex | 2 --
 1 file changed, 2 deletions(-)

diff --git a/apps/explorer/lib/explorer/application.ex b/apps/explorer/lib/explorer/application.ex
index 0752f60cea..4f3e5585f9 100644
--- a/apps/explorer/lib/explorer/application.ex
+++ b/apps/explorer/lib/explorer/application.ex
@@ -19,8 +19,6 @@ defmodule Explorer.Application do
     Uncles
   }
 
-  alias Explorer.Chain.Events.Listener
-
   alias Explorer.Chain.Supply.RSK
 
   alias Explorer.Market.MarketHistoryCache

From eb420803cc4c5e553f6ad1de45f174d4cdf904ca Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Thu, 10 Oct 2019 12:31:26 +0300
Subject: [PATCH 09/16] add CHANGELOG entry

---
 CHANGELOG.md | 1 +
 1 file changed, 1 insertion(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index b26ab858ad..93760864db 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,7 @@
 ## Current
 
 ### Features
+- [#2449](https://github.com/poanetwork/blockscout/pull/2449) - add ability to send notification events through postgres notify
 - [#2733](https://github.com/poanetwork/blockscout/pull/2733) - Add cache for first page of uncles
 - [#2735](https://github.com/poanetwork/blockscout/pull/2735) - Add pending transactions cache
 - [#2726](https://github.com/poanetwork/blockscout/pull/2726) - Remove internal_transaction block_number setting from blocks runner

From db5d862c587191679aa7c27f9346860f8e38bd2c Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Tue, 22 Oct 2019 10:38:05 +0300
Subject: [PATCH 10/16] fix listener init

---
 apps/explorer/config/config.exs                      | 2 +-
 apps/explorer/lib/explorer/chain/events/publisher.ex | 9 ++++++---
 2 files changed, 7 insertions(+), 4 deletions(-)

diff --git a/apps/explorer/config/config.exs b/apps/explorer/config/config.exs
index ed88b4f6de..944626ae5b 100644
--- a/apps/explorer/config/config.exs
+++ b/apps/explorer/config/config.exs
@@ -36,7 +36,7 @@ config :explorer, Explorer.Counters.AverageBlockTime,
 
 config :explorer, Explorer.Chain.Events.Listener,
   enabled:
-    if(System.get_env("DISABLE_WEBAPP") != "true",
+    if((System.get_env("DISABLE_WEBAPP") == nil) && (System.get_env("DISABLE_INDEXER") == nil),
       do: false,
       else: true
     )
diff --git a/apps/explorer/lib/explorer/chain/events/publisher.ex b/apps/explorer/lib/explorer/chain/events/publisher.ex
index 142baee117..6bdfe6eda4 100644
--- a/apps/explorer/lib/explorer/chain/events/publisher.ex
+++ b/apps/explorer/lib/explorer/chain/events/publisher.ex
@@ -2,7 +2,6 @@ defmodule Explorer.Chain.Events.Publisher do
   @moduledoc """
   Publishes events related to the Chain context.
   """
-  @sender Application.get_env(:explorer, :realtime_events_sender)
 
   @allowed_events ~w(addresses address_coin_balances blocks block_rewards internal_transactions token_transfers transactions contract_verification_result)a
 
@@ -21,7 +20,11 @@ defmodule Explorer.Chain.Events.Publisher do
   end
 
   defp send_data(event_type) do
-    @sender.send_data(event_type)
+    sender().send_data(event_type)
+  end
+
+  defp sender do
+    Application.get_env(:explorer, :realtime_events_sender) |> IO.inspect
   end
 
   # The :catchup type of event is not being consumed right now.
@@ -30,6 +33,6 @@ defmodule Explorer.Chain.Events.Publisher do
   defp send_data(_event_type, :catchup, _event_data), do: :ok
 
   defp send_data(event_type, broadcast_type, event_data) do
-    @sender.send_data(event_type, broadcast_type, event_data)
+    sender().send_data(event_type, broadcast_type, event_data)
   end
 end

From a9d67c61482ad4ba341cea5204d843fc461fef86 Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Tue, 22 Oct 2019 10:39:05 +0300
Subject: [PATCH 11/16] fix config

---
 apps/explorer/config/config.exs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/apps/explorer/config/config.exs b/apps/explorer/config/config.exs
index 944626ae5b..60b0afdb48 100644
--- a/apps/explorer/config/config.exs
+++ b/apps/explorer/config/config.exs
@@ -36,7 +36,7 @@ config :explorer, Explorer.Counters.AverageBlockTime,
 
 config :explorer, Explorer.Chain.Events.Listener,
   enabled:
-    if((System.get_env("DISABLE_WEBAPP") == nil) && (System.get_env("DISABLE_INDEXER") == nil),
+    if(System.get_env("DISABLE_WEBAPP") == nil && System.get_env("DISABLE_INDEXER") == nil,
       do: false,
       else: true
     )

From df2dda1fdbb053f461b29b269e72992f3f8cc34a Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Tue, 22 Oct 2019 10:42:27 +0300
Subject: [PATCH 12/16] remove IO.inspect

---
 apps/explorer/lib/explorer/chain/events/publisher.ex | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/apps/explorer/lib/explorer/chain/events/publisher.ex b/apps/explorer/lib/explorer/chain/events/publisher.ex
index 6bdfe6eda4..a392d15da8 100644
--- a/apps/explorer/lib/explorer/chain/events/publisher.ex
+++ b/apps/explorer/lib/explorer/chain/events/publisher.ex
@@ -24,7 +24,7 @@ defmodule Explorer.Chain.Events.Publisher do
   end
 
   defp sender do
-    Application.get_env(:explorer, :realtime_events_sender) |> IO.inspect
+    Application.get_env(:explorer, :realtime_events_sender)
   end
 
   # The :catchup type of event is not being consumed right now.

From 247b5fcb9816e675068051cddc45934f3340e1b9 Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Thu, 24 Oct 2019 13:09:50 +0300
Subject: [PATCH 13/16] fix address_to_unique_tokens query

---
 apps/explorer/lib/explorer/chain/token_transfer.ex | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/apps/explorer/lib/explorer/chain/token_transfer.ex b/apps/explorer/lib/explorer/chain/token_transfer.ex
index ef0cb4e876..900a57e759 100644
--- a/apps/explorer/lib/explorer/chain/token_transfer.ex
+++ b/apps/explorer/lib/explorer/chain/token_transfer.ex
@@ -280,7 +280,7 @@ defmodule Explorer.Chain.TokenTransfer do
       tt in TokenTransfer,
       left_join: instance in Instance,
       on: tt.token_contract_address_hash == instance.token_contract_address_hash and tt.token_id == instance.token_id,
-      where: tt.token_contract_address_hash == ^contract_address_hash and tt.token_id == tt.token_id,
+      where: tt.token_contract_address_hash == ^contract_address_hash,
       order_by: [desc: tt.block_number],
       distinct: tt.token_id,
       preload: [:to_address],

From ea101ffdb603ed0c6be8953bb79a1cc5888e7cd1 Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Thu, 24 Oct 2019 13:13:16 +0300
Subject: [PATCH 14/16] add CHANGELOG entry

---
 CHANGELOG.md | 1 +
 1 file changed, 1 insertion(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6ac75430a7..a777d66dc7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,7 @@
 ### Fixes
 
 ### Chore
+- [#2801](https://github.com/poanetwork/blockscout/pull/2801) - remove unused clause in address_to_unique_tokens query
 
 
 ## 2.1.0-beta

From 87c99e090ca2afb1756b83b8cf5ddd2483787913 Mon Sep 17 00:00:00 2001
From: Andrew Gross <andogro@gmail.com>
Date: Mon, 28 Oct 2019 22:12:02 -0600
Subject: [PATCH 15/16] link docker instructions to doc website

---
 docker/README.md | 78 ++----------------------------------------------
 1 file changed, 3 insertions(+), 75 deletions(-)

diff --git a/docker/README.md b/docker/README.md
index ba3458883d..03df272c4a 100644
--- a/docker/README.md
+++ b/docker/README.md
@@ -1,77 +1,5 @@
-# BlockScout Docker integration
+# BlockScout Docker Integration
 
-For now this integration is not production ready. It made only for local usage only !
-
-## How to use ?
-First of all, blockscout requires `PostgreSQL` server for working. 
-It will be provided by starting script (new docker image will be created named `postgres`)
-
-**Starting command**
-`make start` - will set everything up and start blockscout in container.
-To connect it to your local environment you will have to configure it using [env variables](#env-variables)
-
-Example connecting to local `ganache` instance running on port `2000` on Mac/Windows:
-```bash
-COIN=DAI \
-ETHEREUM_JSONRPC_VARIANT=ganache \ 
-ETHEREUM_JSONRPC_HTTP_URL=http://host.docker.internal:2000 \
-ETHEREUM_JSONRPC_WS_URL=ws://host.docker.internal:2000 \
-make start
-```
-
-Blockscout will be available on `localhost:4000`
-
-**Note**
-On mac/Windows Docker provides with a special URL `host.docker.internal` that will be available into container and routed to your local machine.
-On Linux docker is starting using `--network=host` and all services should be available on `localhost`
-
-### Migrations
-
-By default, `Makefile` will do migrations for you on `PostgreSQL` creation. 
-But you could run migrations manually using `make migrate` command.
-
-**WARNING** Migrations will clean up your local database !
-
-## Env variables
-
-BlockScout supports 3 different JSON RPC Variants.
-Variant could be configured using `ETHEREUM_JSONRPC_VARIANT` environment variable.
-
-Example: 
-```bash
-ETHEREUM_JSONRPC_VARIANT=ganache make start
-```
-
-Available options are:
-
- * `parity` - Parity JSON RPC (**Default one**)
- * `geth` - Geth JSON RPC
- * `ganache` - Ganache JSON RPC
- 
-
-| Variable | Description | Default value |
-| -------- | ----------- | ------------- |
-| `ETHEREUM_JSONRPC_VARIANT` | Variant of your JSON RPC service: `parity`, `geth` or `ganache` | `parity` |
-| `ETHEREUM_JSONRPC_HTTP_URL` | HTTP JSON RPC URL Only for `geth` or `ganache` variant | Different per JSONRPC variant |
-| `ETHEREUM_JSONRPC_WS_URL` | WS JSON RPC url | Different per JSONRPC variant |
-| `ETHEREUM_JSONRPC_TRACE_URL` | Trace URL **Only for `parity` variant** | `http://localhost:8545` |
-| `COIN` | Default Coin | `POA` |
-| `LOGO` | Coin logo | Empty | 
-| `NETWORK` | Network | Empty |
-| `SUBNETWORK` | Subnetwork | Empty |
-| `NETWORK_ICON` | Network icon | Empty | 
-| `NETWORK_PATH` | Network path | `/` |
-
-
-`ETHEREUM_JSONRPC_HTTP_URL` default values:
-
- * For `parity` - `http://localhost:8545`
- * For `geth` - `https://mainnet.infura.io/8lTvJTKmHPCHazkneJsY`
- * For `ganache` - `http://localhost:7545`
-
-`ETHEREUM_JSONRPC_WS_URL` default values:
-
- * For `parity` - `ws://localhost:8546`
- * For `geth` - `wss://mainnet.infura.io/8lTvJTKmHPCHazkneJsY/ws`
- * For `ganache` - `ws://localhost:7545`
+This integration is not production ready, and should be used for local BlockScout deployment only.
 
+For usage instructions and ENV variables, see the [docker integration documentation](https://docs.blockscout.com/for-developers/information-and-settings/docker-integration-local-use-only).
\ No newline at end of file

From fd1f7d8dfdc4fe237d620ded72de26d5ce6b65ae Mon Sep 17 00:00:00 2001
From: Andrew Gross <andogro@gmail.com>
Date: Mon, 28 Oct 2019 22:16:13 -0600
Subject: [PATCH 16/16] update changelog

---
 CHANGELOG.md | 1 +
 1 file changed, 1 insertion(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0e75b77e04..18b1ad3994 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,7 @@
 ### Fixes
 
 ### Chore
+- [#2817](https://github.com/poanetwork/blockscout/pull/2817) - move docker integration documentation to blockscout docs
 
 
 ## 2.1.0-beta