Get notified of user signups and plan changes automatically using Postgres & Phoenix PubSub

mix phx.new pub_sub_demo

cd pub_sub_demo

mix ecto.create

{:httpoison, "~> 1.0"},
{:bamboo, "~> 0.8"}

extra_applications: [:logger, :runtime_tools, :bamboo]

mix phx.gen.context Accounts User users name:string payment_plan:string

mix ecto.migrate
mix deps.get
mix phx.server

mix ecto.gen.migration broadcast_users_table_changes

# broadcast_changes_migration.ex
defmodule PubSubDemo.Repo.Migrations.BroadcastUsersTableChanges do
  use Ecto.Migration

  def up do
    # Create a function that broadcasts row changes
    execute "
      CREATE OR REPLACE FUNCTION broadcast_changes()
      RETURNS trigger AS $$
      DECLARE
        current_row RECORD;
      BEGIN
        IF (TG_OP = 'INSERT' OR TG_OP = 'UPDATE') THEN
          current_row := NEW;
        ELSE
          current_row := OLD;
        END IF;
        IF (TG_OP = 'INSERT') THEN
          OLD := NEW;
        END IF;
      PERFORM pg_notify(
          'users_changes',
          json_build_object(
            'table', TG_TABLE_NAME,
            'type', TG_OP,
            'id', current_row.id,
            'new_row_data', row_to_json(NEW),
            'old_row_data', row_to_json(OLD)
          )::text
        );
      RETURN current_row;
      END;
      $$ LANGUAGE plpgsql;"

    # Create a trigger links the users table to the broadcast function
    execute "
      CREATE TRIGGER notify_user_changes_trigger
      AFTER INSERT OR UPDATE OR DELETE
      ON users
      FOR EACH ROW
      EXECUTE PROCEDURE broadcast_changes();"
  end

  def down do
    execute "DROP TRIGGER notify_user_changes_trigger"
  end
end

# listener.ex
defmodule PubSubDemo.PubSub.Listener do
  use GenServer

  require Logger

  import Poison, only: [decode!: 1]

  @doc """
  Initialize the GenServer
  """
  @spec start_link([String.t], [any])  :: {:ok, pid}
  def start_link(channel, otp_opts \\ []), do: GenServer.start_link(__MODULE__, channel, otp_opts)

  @doc """
  When the GenServer starts subscribe to the given channel
  """
  @spec init([String.t])  :: {:ok, []}
  def init(channel) do
    Logger.debug("Starting #{ __MODULE__ } with channel subscription: #{channel}")
    pg_config = PubSubDemo.Repo.config()
    {:ok, pid} = Postgrex.Notifications.start_link(pg_config)
    {:ok, ref} = Postgrex.Notifications.listen(pid, channel)
    {:ok, {pid, channel, ref}}
  end

  @doc """
  Listen for changes
  """
  def handle_info({:notification, _pid, _ref, "users_changes", payload}, _state) do
    payload
    |> decode!()
    |> IO.inspect()

    {:noreply, :event_handled}
  end

  def handle_info(_, _state), do: {:noreply, :event_received}
end

# application.ex
defmodule PubSubDemo.Application do
  use Application


  def start(_type, _args) do
    # ...
    children = [

      # ...

      # Start the listener
      worker(
        PubSubDemo.PubSub.Listener,
        ["users_changes", [name: PubSubDemo.PubSub.Listener]],
        restart: :permanent
      )
    ]
    # ...
  end
  # ...
end

mix ecto.migrate
mix phx.server

# listener.ex
@doc """
Listen for changes in the users table
"""
def handle_info({:notification, _pid, _ref, "users_changes", payload}, _state) do
  payload
  |> decode!()
  |> handle_user_changes()

  {:noreply, :event_handled}
end

def handle_info(_, _state), do: {:noreply, :event_received}

@doc """
Listen for new users and log when created
"""
def handle_user_changes(%{
  "type" => "INSERT",
  "new_row_data" => %{
    "name" => name
  }
}) do
  IO.puts("New user created with the name: #{name}")
end

@doc """
Listen and log when users change their payment plan
"""
def handle_user_changes(%{
  "type" => "UPDATE",
  "old_row_data" => %{
    "name" => old_name,
    "payment_plan" => old_payment_plan,
  },
  "new_row_data" => %{
    "name" => new_name,
    "payment_plan" => new_payment_plan,
  },
}) when old_payment_plan != new_payment_plan do
  IO.puts("#{new_name} updated their payment plan from #{old_payment_plan} to #{new_payment_plan}")
end

def handle_user_changes(payload), do: nil

# Send emails using Sendgrid
config :pub_sub_demo, PubSubDemo.Mailer,
  adapter: Bamboo.SendgridAdapter,
  api_key: System.get_env("SENDGRID_API_KEY")

# mailer.ex
defmodule PubSubDemo.Mailer do
  use Bamboo.Mailer, otp_app: :pub_sub_demo
end

# listeners.ex
import Bamboo.Email

@doc """
Listen for new users and email when created
"""
def handle_user_changes(%{
  "type" => "INSERT",
  "new_row_data" => %{
    "name" => name
  }
}) do
  subject = "User Created!"
  body = "New user created with the name: #{name}"
  send_email(subject, body)
end

@doc """
Listen and email when users change their payment plan
"""
def handle_user_changes(%{
  "type" => "UPDATE",
  "old_row_data" => %{
    "name" => old_name,
    "payment_plan" => old_payment_plan,
  },
  "new_row_data" => %{
    "name" => new_name,
    "payment_plan" => new_payment_plan,
  },
}) when old_payment_plan != new_payment_plan do
  subject = "Subscription Updated!"
  body = "#{new_name} updated their payment plan from #{old_payment_plan} to #{new_payment_plan}"

  send_email(subject, body)
end

def handle_user_changes(payload), do: nil

def send_email(subject, body) do
  new_email(
    to: "[email protected]",
    from: "[email protected]",
    subject: subject,
    html_body: "<strong>#{body}</strong>",
    text_body: body
  )
  |> PubSubDemo.Mailer.deliver_now()
end