# Send broadcasts inline, and don't run the campaign scheduler.

config :my_app, MyApp.EmailCampaigns, async_delivery: false

config :my_app, MyApp.EmailCampaigns.Scheduler, enabled: false

      {Task.Supervisor, name: MyApp.TaskSupervisor},

      MyApp.EmailCampaigns.Scheduler,

defmodule MyApp.EmailCampaigns do

  @moduledoc """

  Marketing email: broadcasts to every subscriber, and sequences that send a

  series of emails after something happens, like onboarding after sign-up.

  Campaign templates live in code (`MyApp.EmailCampaigns.Templates`), built

  on `MyApp.Emails`. Unlike transactional email, every campaign email checks

  the recipient's subscription and carries an unsubscribe link.

  ## Starting sequences

  Call `trigger/3` when something happens that a sequence should follow up

  on, with any name you like:

      EmailCampaigns.trigger(user, "user.signed_up")

  With the Events & Analytics feature installed, forward every tracked event

  instead, so sequences can start from any event without more code:

      defmodule MyApp.EmailCampaigns.EventHandler do

        @behaviour MyApp.Events.Handler

        @impl true

        def handle_event(%{user: %MyApp.Accounts.User{} = user} = event) do

          MyApp.EmailCampaigns.trigger(user, event.name, event.metadata || %{})

        end

        def handle_event(_event), do: :ok

      end

      config :my_app, MyApp.Events, handlers: [MyApp.EmailCampaigns.EventHandler]

  ## Delivery

  Every email goes through `deliver/3`, which claims a send row and sends in

  one transaction. A second attempt finds the row and skips, so:

    * broadcasts resume after a deploy or crash without resending, and

    * several servers can run `MyApp.EmailCampaigns.Scheduler` at once.

  A crash after the email leaves but before the transaction commits sends it

  again on retry. Email can't be exactly-once, so this errs toward

  delivering.

  """

  require Logger

  alias MyApp.Repo

  alias MyApp.Accounts.User

  alias MyApp.Mailer

  alias MyApp.EmailCampaigns.{

    Broadcast,

    Email,

    Send,

    Sequence,

    SequenceEnrollment,

    SequenceStep,

    Subscription,

    Templates

  }

  @batch_size 100

  ## Subscriptions

  @doc "Whether the user gets campaign emails. Users are subscribed until they opt out."

  def subscribed?(%{id: user_id}) do

    case Repo.get_by(Subscription, user_id: user_id) do

      %Subscription{status: "unsubscribed"} -> false

      _ -> true

    end

  end

  @doc "Opts a user out. Safe to repeat."

  def unsubscribe(user_id), do: put_subscription(user_id, "unsubscribed")

  @doc "Opts a user back in. Safe to repeat."

  def resubscribe(user_id), do: put_subscription(user_id, "subscribed")

  defp put_subscription(user_id, status) do

    %Subscription{}

    |> Subscription.changeset(%{user_id: user_id, status: status})

    |> Repo.insert(

      on_conflict: {:replace, [:status, :updated_at]},

      conflict_target: :user_id,

      returning: true

    )

  end

  def count_subscribed_users, do: Repo.aggregate(Subscription.subscribed_users_query(), :count)

  ## Delivery

  @doc """

  Sends a template to a user once per broadcast or sequence step.

  Takes `:broadcast_id` or `:sequence_step_id`, and `:metadata` for the

  template. Returns `{:ok, :sent}`, `{:ok, :already_sent}`, or

  `{:error, reason}` with nothing recorded, so a retry can try again.

  """

  def deliver(template_key, %User{} = user, opts) do

    now = DateTime.utc_now(:second)

    send = %{

      id: Ecto.UUID.generate(),

      template: template_key,

      user_id: user.id,

      broadcast_id: opts[:broadcast_id],

      sequence_step_id: opts[:sequence_step_id],

      inserted_at: now

    }

    Repo.transaction(fn ->

      case Repo.insert_all(Send, [send], on_conflict: :nothing) do

        {0, _} ->

          :already_sent

        {1, _} ->

          email = Email.build(template_key, user, opts[:metadata] || %{})

          case Mailer.deliver(email) do

            {:ok, _} -> :sent

            {:error, reason} -> Repo.rollback(reason)

          end

      end

    end)

  end

  ## Broadcasts

  @doc """

  Creates a broadcast of a template to every subscribed user and starts

  sending it in the background.

  """

  def create_broadcast(template_key) do

    %Broadcast{}

    |> Broadcast.create_changeset(%{

      template: template_key,

      total_count: count_subscribed_users()

    })

    |> Repo.insert()

    |> tap(fn

      {:ok, broadcast} -> start_delivery(broadcast)

      _error -> :ok

    end)

  end

  defp start_delivery(broadcast) do

    if Application.get_env(:my_app, MyApp.EmailCampaigns, [])[:async_delivery] == false do

      deliver_broadcast(broadcast)

    else

      Task.Supervisor.start_child(MyApp.TaskSupervisor, fn -> deliver_broadcast(broadcast) end)

    end

  end

  @doc """

  Sends a broadcast to every subscribed user who hasn't had it, in batches.

  Resumable: run it again after a crash and it picks up where it stopped.

  Marks the broadcast sent once everyone has it. If some sends fail it stays

  `"sending"`, and the scheduler retries them. If its template was removed,

  it's canceled.

  """

  def deliver_broadcast(%Broadcast{} = broadcast) do

    if Templates.exists?(broadcast.template) do

      deliver_batches(broadcast, [])

    else

      Logger.warning(

        "[EmailCampaigns] canceling broadcast #{broadcast.id}: template #{broadcast.template} was removed"

      )

      finish_broadcast(broadcast, "canceled")

    end

  end

  defp deliver_batches(broadcast, failed_ids) do

    case Repo.all(Broadcast.pending_recipients_query(broadcast, failed_ids, @batch_size)) do

      [] when failed_ids == [] ->

        finish_broadcast(broadcast, "sent")

      [] ->

        Logger.warning(

          "[EmailCampaigns] broadcast #{broadcast.id}: #{length(failed_ids)} sends failed, retrying later"

        )

        {:ok, broadcast}

      users ->

        failed =

          for user <- users,

              match?({:error, _}, deliver(broadcast.template, user, broadcast_id: broadcast.id)),

              do: user.id

        deliver_batches(broadcast, failed_ids ++ failed)

    end

  end

  defp finish_broadcast(broadcast, status) do

    broadcast

    |> Broadcast.finish_changeset(status, DateTime.utc_now(:second))

    |> Repo.update()

  end

  @doc "Picks up broadcasts left sending by a crash, a deploy, or failed sends."

  def resume_broadcasts do

    Broadcast.sending_query()

    |> Repo.all()

    |> Enum.each(&deliver_broadcast/1)

  end

  ## Sequences

  def change_sequence(%Sequence{} = sequence, attrs \\ %{}),

    do: Sequence.changeset(sequence, attrs)

  def create_sequence(attrs) do

    %Sequence{}

    |> Sequence.changeset(attrs)

    |> Repo.insert()

  end

  def update_sequence(%Sequence{} = sequence, attrs) do

    sequence

    |> Sequence.changeset(attrs)

    |> Repo.update()

  end

  def delete_sequence(%Sequence{} = sequence), do: Repo.delete(sequence)

  def change_step(%SequenceStep{} = step, attrs \\ %{}), do: SequenceStep.changeset(step, attrs)

  @doc "Adds a step to the end of a sequence."

  def add_step(%Sequence{} = sequence, attrs) do

    %SequenceStep{

      sequence_id: sequence.id,

      position: Repo.one(SequenceStep.next_position_query(sequence))

    }

    |> SequenceStep.changeset(attrs)

    |> Repo.insert()

  end

  @doc """

  Deletes a step. Users waiting on it move on to the next one.

  """

  def delete_step(%SequenceStep{} = step), do: Repo.delete(step)

  @doc """

  Enrolls a user in a sequence, once. Returns `{:ok, enrollment}` or

  `{:ok, :already_enrolled}`.

  """

  def enroll_user(%Sequence{} = sequence, %{id: user_id}, metadata \\ %{}) do

    if Repo.exists?(SequenceEnrollment.for_user_query(sequence, user_id)) do

      {:ok, :already_enrolled}

    else

      %SequenceEnrollment{}

      |> SequenceEnrollment.create_changeset(%{

        sequence_id: sequence.id,

        user_id: user_id,

        metadata: metadata,

        step_started_at: DateTime.utc_now(:second)

      })

      # A concurrent enrollment wins the race harmlessly.

      |> Repo.insert(on_conflict: :nothing, conflict_target: [:sequence_id, :user_id])

    end

  end

  @doc """

  Enrolls a user in every active sequence with this trigger, if they're

  subscribed. `metadata` reaches the sequence's templates as `@metadata`.

  ## Examples

      trigger(user, "user.signed_up")

      trigger(user, "trial.ending", %{"days_left" => 3})

  """

  def trigger(%User{} = user, trigger, metadata \\ %{}) do

    if subscribed?(user) do

      trigger

      |> Sequence.active_for_trigger_query()

      |> Repo.all()

      |> Enum.each(&enroll_user(&1, user, metadata))

    end

    :ok

  end

  def cancel_enrollment(%SequenceEnrollment{} = enrollment) do

    enrollment

    |> SequenceEnrollment.cancel_changeset()

    |> Repo.update()

  end

  @doc """

  Sends every sequence step that's due. Called by the scheduler.

  Each enrollment is locked while it's handled, and other servers skip locked

  ones, so a step goes out once however many servers run this.

  """

  def process_due_steps do

    now = DateTime.utc_now(:second)

    now

    |> SequenceEnrollment.due_query()

    |> Repo.all()

    |> Enum.each(fn {enrollment, _step} -> process_enrollment(enrollment.id, now) end)

  end

  defp process_enrollment(enrollment_id, now) do

    Repo.transaction(fn ->

      case Repo.one(SequenceEnrollment.lock_due_query(enrollment_id, now)) do

        nil -> :skipped

        {enrollment, step} -> handle_step(enrollment, step, now)

      end

    end)

  end

  defp handle_step(enrollment, step, now) do

    user = Repo.get!(User, enrollment.user_id)

    cond do

      not subscribed?(user) ->

        {:ok, _} = cancel_enrollment(enrollment)

      SequenceStep.template_removed?(step) ->

        Logger.warning(

          "[EmailCampaigns] skipping sequence step #{step.id}: template #{step.template} was removed"

        )

        advance(enrollment, step, now)

      true ->

        case deliver(step.template, user,

               sequence_step_id: step.id,

               metadata: enrollment.metadata

             ) do

          {:ok, _sent} -> advance(enrollment, step, now)

          {:error, reason} -> Repo.rollback(reason)

        end

    end

  end

  defp advance(enrollment, step, now) do

    more_steps? = Repo.exists?(SequenceStep.after_position_query(step.sequence_id, step.position))

    enrollment

    |> SequenceEnrollment.advance_changeset(step, more_steps?, now)

    |> Repo.update!()

  end

end

defmodule MyApp.EmailCampaigns.Broadcast do

  @moduledoc """

  One template sent once to every subscribed user.

  A broadcast is `"sending"` until every subscribed user has a send for it,

  then `"sent"`. It's `"canceled"` if its template was removed mid-send.

  """

  use Ecto.Schema

  import Ecto.Changeset

  import Ecto.Query

  alias MyApp.EmailCampaigns.{Send, Subscription, Templates}

  @primary_key {:id, :binary_id, autogenerate: true}

  @foreign_key_type :binary_id

  schema "email_broadcasts" do

    field :template, :string

    field :status, :string, default: "sending"

    field :total_count, :integer, default: 0

    field :finished_at, :utc_datetime

    timestamps(type: :utc_datetime)

  end

  ## Changesets

  def create_changeset(broadcast, attrs) do

    broadcast

    |> cast(attrs, [:template, :total_count])

    |> validate_required([:template])

    |> validate_inclusion(:template, Templates.keys(), message: "isn't a template")

  end

  def finish_changeset(broadcast, status, now) do

    change(broadcast, status: status, finished_at: now)

  end

  ## Queries

  def sending_query do

    from b in __MODULE__, where: b.status == "sending"

  end

  def latest_query do

    from b in __MODULE__, order_by: [desc: b.inserted_at, desc: b.id]

  end

  @doc """

  Subscribed users who haven't been sent this broadcast yet, minus any that

  already failed this run.

  """

  def pending_recipients_query(broadcast, excluded_user_ids, limit) do

    sent_user_ids = from s in Send, where: s.broadcast_id == ^broadcast.id, select: s.user_id

    from u in Subscription.subscribed_users_query(),

      where: u.id not in subquery(sent_user_ids),

      where: u.id not in ^excluded_user_ids,

      order_by: u.id,

      limit: ^limit

  end

end

defmodule MyApp.EmailCampaigns.Email do

  @moduledoc """

  Builds campaign emails on `MyApp.Emails`, adding what marketing email

  needs: a per-recipient unsubscribe link and the `List-Unsubscribe` and

  `List-Unsubscribe-Post` headers (RFC 8058) that Gmail and Yahoo require

  from bulk senders.

  Campaigns come from the same sender as other email unless configured:

      config :my_app, MyApp.EmailCampaigns, from: {"MyApp", "news@example.com"}

  """

  alias MyApp.EmailCampaigns.Templates

  alias MyApp.Emails

  @doc """

  Builds the email for a template and recipient. `user` needs `:id` and

  `:email`.

  """

  def build(template_key, user, metadata \\ %{}) do

    unsubscribe_url = unsubscribe_url(user)

    template_key

    |> Templates.get!()

    |> Emails.build(user, %{unsubscribe_url: unsubscribe_url, metadata: metadata})

    |> Swoosh.Email.from(from_address())

    |> Swoosh.Email.header("List-Unsubscribe", "<#{unsubscribe_url}>")

    |> Swoosh.Email.header("List-Unsubscribe-Post", "List-Unsubscribe=One-Click")

  end

  @doc """

  A signed link that unsubscribes the user. It never expires, so an email

  from years ago can still opt out.

  """

  def unsubscribe_url(user) do

    token = Phoenix.Token.sign(MyAppWeb.Endpoint, "email unsubscribe", user.id)

    MyAppWeb.Endpoint.url() <> "/emails/unsubscribe/" <> token

  end

  @doc "The user id an unsubscribe token was signed for."

  def verify_unsubscribe_token(token) do

    Phoenix.Token.verify(MyAppWeb.Endpoint, "email unsubscribe", token, max_age: :infinity)

  end

  def from_address do

    :my_app

    |> Application.get_env(MyApp.EmailCampaigns, [])

    |> Keyword.get_lazy(:from, &Emails.from_address/0)

  end

end

defmodule MyApp.EmailCampaigns.Scheduler do

  @moduledoc """

  Every minute, sends sequence steps that are due and resumes broadcasts

  still sending.

  It runs on every server. Delivery claims each send and locks each

  enrollment, so servers never send the same email twice.

  Disable it in tests with:

      config :my_app, MyApp.EmailCampaigns.Scheduler, enabled: false

  """

  use GenServer

  require Logger

  @interval :timer.minutes(1)

  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)

  @doc "A child spec that starts nothing when the scheduler is disabled."

  def child_spec(opts) do

    if Keyword.get(Application.get_env(:my_app, __MODULE__, []), :enabled, true) do

      %{id: __MODULE__, start: {__MODULE__, :start_link, [opts]}}

    else

      %{id: __MODULE__, start: {Function, :identity, [:ignore]}}

    end

  end

  @impl true

  def init(_opts) do

    schedule_tick()

    {:ok, %{}}

  end

  @impl true

  def handle_info(:tick, state) do

    run()

    schedule_tick()

    {:noreply, state}

  end

  @doc "One pass: due sequence steps, then unfinished broadcasts."

  def run do

    MyApp.EmailCampaigns.process_due_steps()

    MyApp.EmailCampaigns.resume_broadcasts()

  rescue

    error -> Logger.error("[EmailCampaigns.Scheduler] #{Exception.message(error)}")

  end

  defp schedule_tick, do: Process.send_after(self(), :tick, @interval)

end

defmodule MyApp.EmailCampaigns.Send do

  @moduledoc """

  A record that a template went to a user, from a broadcast or a sequence

  step.

  Delivery claims the send row first and inserts it in the same transaction

  as the email. Partial unique indexes on `(user_id, broadcast_id)` and

  `(user_id, sequence_step_id)` turn a second attempt into a no-op, so

  retries and several servers sending at once never email anyone twice.

  """

  use Ecto.Schema

  import Ecto.Query

  @primary_key {:id, :binary_id, autogenerate: true}

  @foreign_key_type :binary_id

  schema "email_sends" do

    field :template, :string

    belongs_to :user, MyApp.Accounts.User

    belongs_to :broadcast, MyApp.EmailCampaigns.Broadcast

    belongs_to :sequence_step, MyApp.EmailCampaigns.SequenceStep

    timestamps(updated_at: false, type: :utc_datetime)

  end

  ## Queries

  def for_broadcast_query(broadcast_ids) do

    from s in __MODULE__, where: s.broadcast_id in ^broadcast_ids

  end

  @doc "Send counts per broadcast, as `{broadcast_id, count}`."

  def count_by_broadcast_query(broadcast_ids) do

    from s in for_broadcast_query(broadcast_ids),

      group_by: s.broadcast_id,

      select: {s.broadcast_id, count(s.id)}

  end

  @doc "Send counts per template key, as `{template, count}`."

  def count_by_template_query do

    from s in __MODULE__, group_by: s.template, select: {s.template, count(s.id)}

  end

  def latest_for_template_query(template, limit) do

    from s in __MODULE__,

      where: s.template == ^template,

      order_by: [desc: s.inserted_at, desc: s.id],

      limit: ^limit,

      preload: :user

  end

end

defmodule MyApp.EmailCampaigns.Sequence do

  @moduledoc """

  A series of emails sent to a user after a trigger, like onboarding after

  `user.signed_up`. Sequences start from `MyApp.EmailCampaigns.trigger/3`.

  Only `"active"` sequences enroll users and send. `"paused"` holds everyone

  where they are; `"draft"` is for building one.

  """

  use Ecto.Schema

  import Ecto.Changeset

  import Ecto.Query

  @statuses ~w(draft active paused)

  @primary_key {:id, :binary_id, autogenerate: true}

  @foreign_key_type :binary_id

  schema "email_sequences" do

    field :name, :string

    field :trigger, :string

    field :status, :string, default: "draft"

    has_many :steps, MyApp.EmailCampaigns.SequenceStep, preload_order: [asc: :position]

    timestamps(type: :utc_datetime)

  end

  def statuses, do: @statuses

  ## Changesets

  def changeset(sequence, attrs) do

    sequence

    |> cast(attrs, [:name, :trigger, :status])

    |> validate_required([:name, :trigger, :status])

    |> validate_inclusion(:status, @statuses)

  end

  ## Queries

  def active_for_trigger_query(trigger) do

    from s in __MODULE__, where: s.trigger == ^trigger and s.status == "active"

  end

  def latest_query do

    from s in __MODULE__, order_by: [desc: s.inserted_at, desc: s.id]

  end

end

defmodule MyApp.EmailCampaigns.SequenceEnrollment do

  @moduledoc """

  A user's progress through a sequence.

  `current_step` is the lowest step position not yet sent, so the next step

  is the first one at or after it. `step_started_at` is when that step's

  delay started counting.

  """

  use Ecto.Schema

  import Ecto.Changeset

  import Ecto.Query

  alias MyApp.EmailCampaigns.{Sequence, SequenceStep}

  @primary_key {:id, :binary_id, autogenerate: true}

  @foreign_key_type :binary_id

  schema "email_sequence_enrollments" do

    field :current_step, :integer, default: 0

    field :status, :string, default: "active"

    field :metadata, :map, default: %{}

    field :step_started_at, :utc_datetime

    belongs_to :sequence, Sequence

    belongs_to :user, MyApp.Accounts.User

    timestamps(type: :utc_datetime)

  end

  ## Changesets

  def create_changeset(enrollment, attrs) do

    enrollment

    |> cast(attrs, [:sequence_id, :user_id, :metadata, :step_started_at])

    |> validate_required([:sequence_id, :user_id, :step_started_at])

    |> unique_constraint([:sequence_id, :user_id])

  end

  @doc """

  Moves past a step: sent, or skipped because its template was removed.

  Completes the enrollment when no step follows.

  """

  def advance_changeset(enrollment, step, more_steps?, now) do

    change(enrollment,

      current_step: step.position + 1,

      step_started_at: now,

      status: if(more_steps?, do: "active", else: "completed")

    )

  end

  def cancel_changeset(enrollment), do: change(enrollment, status: "canceled")

  ## Queries

  @doc """

  Active enrollments in active sequences whose next step is due, with that

  step, as `{enrollment, step}`.

  """

  def due_query(now) do

    from e in __MODULE__,

      as: :enrollment,

      join: q in assoc(e, :sequence),

      inner_lateral_join: s in subquery(next_step_query()),

      on: true,

      where: e.status == "active" and q.status == "active",

      where:

        fragment("? + make_interval(mins => ?) <= ?", e.step_started_at, s.delay_minutes, ^now),

      select: {e, s}

  end

  @doc """

  Locks one due enrollment for sending, skipping it if another server holds

  the lock. Re-checks that it's still due, since it may have just been sent.

  """

  def lock_due_query(id, now) do

    from [e, _q, _s] in due_query(now),

      where: e.id == ^id,

      lock: fragment("FOR UPDATE OF ? SKIP LOCKED", e)

  end

  defp next_step_query do

    from s in SequenceStep,

      where: s.sequence_id == parent_as(:enrollment).sequence_id,

      where: s.position >= parent_as(:enrollment).current_step,

      order_by: s.position,

      limit: 1

  end

  def for_user_query(sequence, user_id) do

    from e in __MODULE__, where: e.sequence_id == ^sequence.id and e.user_id == ^user_id

  end

  def for_sequence_query(sequence) do

    from e in __MODULE__, where: e.sequence_id == ^sequence.id

  end

  def latest_query(query, limit) do

    from e in query,

      order_by: [desc: e.inserted_at, desc: e.id],

      limit: ^limit,

      preload: :user

  end

  @doc "Enrollment counts by status, as `{status, count}`."

  def count_by_status_query(query) do

    from e in query, group_by: e.status, select: {e.status, count(e.id)}

  end

end

defmodule MyApp.EmailCampaigns.SequenceStep do

  @moduledoc """

  One email in a sequence, sent `delay_minutes` after the previous step (or

  after enrollment, for the first).

  Positions only order steps. Deleting one leaves a gap, and enrollments

  move on to the next position that exists.

  """

  use Ecto.Schema

  import Ecto.Changeset

  import Ecto.Query

  alias MyApp.EmailCampaigns.Templates

  @primary_key {:id, :binary_id, autogenerate: true}

  @foreign_key_type :binary_id

  schema "email_sequence_steps" do

    field :template, :string

    field :delay_minutes, :integer, default: 0

    field :position, :integer

    belongs_to :sequence, MyApp.EmailCampaigns.Sequence

    timestamps(type: :utc_datetime)

  end

  ## Changesets

  def changeset(step, attrs) do

    step

    |> cast(attrs, [:template, :delay_minutes])

    |> validate_required([:template, :delay_minutes])

    |> validate_inclusion(:template, Templates.keys(), message: "isn't a template")

    |> validate_number(:delay_minutes, greater_than_or_equal_to: 0)

    |> unique_constraint([:sequence_id, :position])

  end

  ## Queries

  def next_position_query(sequence) do

    from s in __MODULE__,

      where: s.sequence_id == ^sequence.id,

      select: coalesce(max(s.position), -1) + 1

  end

  @doc "Whether the sequence has a step after `position`."

  def after_position_query(sequence_id, position) do

    from s in __MODULE__, where: s.sequence_id == ^sequence_id and s.position > ^position

  end

  ## View functions

  def template_removed?(%__MODULE__{template: template}), do: not Templates.exists?(template)

end

defmodule MyApp.EmailCampaigns.Subscription do

  @moduledoc """

  A user's preference for campaign emails. Users are subscribed until they opt out,

  so only users who changed it have a row.

  """

  use Ecto.Schema

  import Ecto.Changeset

  import Ecto.Query

  @primary_key {:id, :binary_id, autogenerate: true}

  @foreign_key_type :binary_id

  schema "email_subscriptions" do

    field :status, :string, default: "subscribed"

    belongs_to :user, MyApp.Accounts.User

    timestamps(type: :utc_datetime)

  end

  ## Changesets

  def changeset(subscription, attrs) do

    subscription

    |> cast(attrs, [:user_id, :status])

    |> validate_required([:user_id, :status])

    |> validate_inclusion(:status, ~w(subscribed unsubscribed))

    |> unique_constraint(:user_id)

  end

  ## Queries

  @doc "Ids of users who opted out."

  def unsubscribed_user_ids_query do

    from s in __MODULE__, where: s.status == "unsubscribed", select: s.user_id

  end

  @doc "Which of the given users opted out."

  def unsubscribed_among_query(user_ids) do

    from s in unsubscribed_user_ids_query(), where: s.user_id in ^user_ids

  end

  @doc "Users who haven't opted out."

  def subscribed_users_query do

    from u in MyApp.Accounts.User, where: u.id not in subquery(unsubscribed_user_ids_query())

  end

end

defmodule MyApp.EmailCampaigns.Template do

  @moduledoc """

  A marketing email: a `MyApp.Emails.Template` sent to subscribers, with an

  unsubscribe link. Besides `:user`, every campaign template gets:

    * `:unsubscribe_url` — the recipient's one-click unsubscribe link, which

      `MyApp.Emails.Layout.layout/1` puts in the footer

    * `:metadata` — whatever was passed to `MyApp.EmailCampaigns.trigger/3`

      for sequence emails, or `%{}` for broadcasts

  Register the module in `MyApp.EmailCampaigns.Templates`.

  ## Example

      defmodule MyApp.EmailCampaigns.Templates.Tips do

        use MyApp.EmailCampaigns.Template

        @impl true

        def name, do: "Tips"

        @impl true

        def subject(_assigns), do: "Three tips"

        @impl true

        def html(assigns) do

          ~H\"\"\"

          <.layout unsubscribe_url={@unsubscribe_url}>

            <.paragraph>Here are three tips.</.paragraph>

          </.layout>

          \"\"\"

        end

        @impl true

        def text(assigns), do: "Here are three tips.\\n\\nUnsubscribe: \#{assigns.unsubscribe_url}"

      end

  """

  defmacro __using__(_opts) do

    quote do

      use MyApp.Emails.Template

      @impl true

      def sample_assigns do

        %{unsubscribe_url: MyAppWeb.Endpoint.url() <> "/emails/unsubscribe/sample", metadata: %{}}

      end

      defoverridable sample_assigns: 0

    end

  end

end

defmodule MyApp.EmailCampaigns.Templates do

  @moduledoc """

  The catalog of marketing templates, the ones broadcasts and sequences can

  send. Each is a module using `MyApp.EmailCampaigns.Template`, registered

  here under a key. Transactional templates live in `MyApp.Emails.Templates`.

  Broadcasts, sequence steps, and sends store that key. Removing a template

  is safe: rows keep the key, the admin panel shows it as removed, a

  broadcast still sending it stops, and a sequence step using it is skipped.

  Renaming a key counts as removing a template and adding another.

  """

  alias MyApp.EmailCampaigns.Templates

  @templates %{

    "welcome" => Templates.Welcome,

    "product_update" => Templates.ProductUpdate

  }

  @doc "Every template, as `{key, module}` sorted by name."

  def all, do: Enum.sort_by(@templates, fn {_key, module} -> module.name() end)

  def keys, do: Map.keys(@templates)

  @doc "The template module for a key, or nil if it's been removed."

  def get(key), do: Map.get(@templates, key)

  def get!(key), do: get(key) || raise(ArgumentError, "no campaign template #{inspect(key)}")

  def exists?(key), do: Map.has_key?(@templates, key)

  @doc """

  A template's display name, or the key marked as removed.

  """

  def name(key) do

    case get(key) do

      nil -> "#{key} (removed)"

      module -> module.name()

    end

  end

end

defmodule MyApp.EmailCampaigns.Templates.ProductUpdate do

  @moduledoc """

  An example broadcast. Copy it for each announcement, under a new key.

  """

  use MyApp.EmailCampaigns.Template

  @impl true

  def name, do: "Product update"

  @impl true

  def description, do: "An example announcement to send as a broadcast."

  @impl true

  def subject(_assigns), do: "What's new this month"

  @impl true

  def preheader(_assigns), do: "Three improvements you asked for."

  @impl true

  def html(assigns) do

    ~H"""

    <.layout unsubscribe_url={@unsubscribe_url} preheader={preheader(assigns)}>

      <.heading>What's new this month</.heading>

      <.paragraph>We shipped three improvements you asked for:</.paragraph>

      <ul style="margin: 0 0 16px; padding-left: 20px;">

        <li>Faster search</li>

        <li>Dark mode</li>

        <li>CSV export</li>

      </ul>

      <.button href={MyAppWeb.Endpoint.url()}>See what's new</.button>

    </.layout>

    """

  end

  @impl true

  def text(assigns) do

    """

    What's new this month

    We shipped three improvements you asked for:

    - Faster search

    - Dark mode

    - CSV export

    See what's new: #{MyAppWeb.Endpoint.url()}

    Unsubscribe: #{assigns.unsubscribe_url}

    """

  end

end

defmodule MyApp.EmailCampaigns.Templates.Welcome do

  @moduledoc """

  The first email of an onboarding sequence, triggered by `user.signed_up`.

  """

  use MyApp.EmailCampaigns.Template

  @impl true

  def name, do: "Welcome"

  @impl true

  def description, do: "The first onboarding email, for a sequence triggered by sign-up."

  @impl true

  def subject(_assigns), do: "Welcome to MyApp"

  @impl true

  def preheader(_assigns), do: "Here's how to get started."

  @impl true

  def html(assigns) do

    ~H"""

    <.layout unsubscribe_url={@unsubscribe_url} preheader={preheader(assigns)}>

      <.heading>Welcome aboard</.heading>

      <.paragraph>Thanks for signing up with {@user.email}. We're glad you're here.</.paragraph>

      <.paragraph>The best way to start is to set up your first project.</.paragraph>

      <.button href={MyAppWeb.Endpoint.url()}>Get started</.button>

    </.layout>

    """

  end

  @impl true

  def text(assigns) do

    """

    Welcome aboard

    Thanks for signing up with #{assigns.user.email}. We're glad you're here.

    The best way to start is to set up your first project: #{MyAppWeb.Endpoint.url()}

    Unsubscribe: #{assigns.unsubscribe_url}

    """

  end

end

defmodule MyAppAdmin.Query.EmailCampaigns do

  @moduledoc """

  Admin reads for email campaigns.

  """

  alias MyApp.Repo

  alias MyApp.EmailCampaigns.{Broadcast, Send, Sequence, SequenceEnrollment, Subscription}

  @latest_limit 50

  ## Templates

  @doc "Send counts per template key, including removed templates."

  def send_counts_by_template do

    Send.count_by_template_query() |> Repo.all() |> Map.new()

  end

  def list_latest_sends(template_key) do

    Repo.all(Send.latest_for_template_query(template_key, @latest_limit))

  end

  ## Broadcasts

  def list_broadcasts, do: Repo.all(Broadcast.latest_query())

  def get_broadcast!(id), do: Repo.get!(Broadcast, id)

  @doc "Sent counts for broadcasts, as a map keyed by broadcast id."

  def sent_counts(broadcasts) do

    broadcasts

    |> Enum.map(& &1.id)

    |> Send.count_by_broadcast_query()

    |> Repo.all()

    |> Map.new()

  end

  ## Sequences

  def list_sequences, do: Repo.all(Sequence.latest_query())

  def get_sequence!(id), do: Sequence |> Repo.get!(id) |> Repo.preload(:steps)

  @doc "Enrollment counts by status for a sequence."

  def enrollment_counts(sequence) do

    sequence

    |> SequenceEnrollment.for_sequence_query()

    |> SequenceEnrollment.count_by_status_query()

    |> Repo.all()

    |> Map.new()

  end

  def list_latest_enrollments(sequence) do

    sequence

    |> SequenceEnrollment.for_sequence_query()

    |> SequenceEnrollment.latest_query(@latest_limit)

    |> Repo.all()

  end

  ## Subscribers

  @doc "Ids of the given users who opted out."

  def unsubscribed_ids(users) do

    users

    |> Enum.map(& &1.id)

    |> Subscription.unsubscribed_among_query()

    |> Repo.all()

    |> MapSet.new()

  end

end

defmodule MyAppAdminWeb.CampaignComponents do

  @moduledoc """

  Components shared by the admin campaign pages.

  """

  use MyAppAdminWeb, :html

  alias MyApp.EmailCampaigns.Templates

  @doc """

  Renders a status as a badge: broadcasts, sequences, enrollments, and

  subscriptions.

  """

  attr :status, :string, required: true

  def campaign_status(assigns) do

    ~H"""

    <span class={["badge badge-sm", status_class(@status)]}>

      {@status |> String.replace("_", " ") |> String.capitalize()}

    </span>

    """

  end

  defp status_class(status) when status in ~w(sent active completed subscribed),

    do: "badge-success"

  defp status_class(status) when status in ~w(sending paused), do: "badge-warning"

  defp status_class(status) when status in ~w(canceled unsubscribed), do: "badge-error"

  defp status_class(_status), do: "badge-ghost"

  @doc """

  Renders a template's name, flagging templates removed from code.

  """

  attr :key, :string, required: true

  def template_name(assigns) do

    ~H"""

    <span :if={Templates.exists?(@key)}>{Templates.name(@key)}</span>

    <span :if={!Templates.exists?(@key)} class="text-error" title="Removed from the template catalog">

      {@key} <span class="badge badge-error badge-sm">Removed template</span>

    </span>

    """

  end

  @doc "Formats a step delay, like \"3 days\" or \"right away\"."

  def format_delay(0), do: "Right away"

  def format_delay(minutes) when rem(minutes, 1440) == 0, do: pluralize(div(minutes, 1440), "day")

  def format_delay(minutes) when rem(minutes, 60) == 0, do: pluralize(div(minutes, 60), "hour")

  def format_delay(minutes), do: pluralize(minutes, "minute")

  defp pluralize(1, unit), do: "1 #{unit} later"

  defp pluralize(n, unit), do: "#{n} #{unit}s later"

end

        <.sidebar_item

          icon="hero-megaphone-mini"

          navigate={~p"/admin/emails/broadcasts"}

          current={@page_selected == :broadcasts}

        >

          Broadcasts

        </.sidebar_item>

        <.sidebar_item

          icon="hero-queue-list-mini"

          navigate={~p"/admin/emails/sequences"}

          current={@page_selected == :sequences}

        >

          Sequences

        </.sidebar_item>

        <.sidebar_item

          icon="hero-users-mini"

          navigate={~p"/admin/emails/subscribers"}

          current={@page_selected == :subscribers}

        >

          Subscribers

        </.sidebar_item>

defmodule MyAppAdminWeb.BroadcastLive.Index do

  use MyAppAdminWeb, :live_view

  import MyAppAdminWeb.CampaignComponents

  alias MyAppAdmin.Query

  def render(assigns) do

    ~H"""

    <Layouts.admin_ui flash={@flash} current_scope={@current_scope} page_selected={:broadcasts}>

      <.header>

        Broadcasts

        <:subtitle>A template sent once to every subscriber.</:subtitle>

        <:actions>

          <.link navigate={~p"/admin/emails/broadcasts/new"}>

            <.button variant="primary">New broadcast</.button>

          </.link>

        </:actions>

      </.header>

      <.table

        :if={@broadcasts != []}

        id="broadcasts"

        rows={@broadcasts}

        row_click={fn broadcast -> JS.navigate(~p"/admin/emails/broadcasts/#{broadcast}") end}

      >

        <:col :let={broadcast} label="Template"><.template_name key={broadcast.template} /></:col>

        <:col :let={broadcast} label="Status"><.campaign_status status={broadcast.status} /></:col>

        <:col :let={broadcast} label="Sent">

          {Map.get(@sent_counts, broadcast.id, 0)} / {broadcast.total_count}

        </:col>

        <:col :let={broadcast} label="Started">{format_datetime(broadcast.inserted_at)}</:col>

      </.table>

      <.table_card :if={@broadcasts == []}>

        <.empty_state id="broadcasts-empty" title="No broadcasts yet" icon="hero-megaphone" />

      </.table_card>

    </Layouts.admin_ui>

    """

  end

  def mount(_params, _session, socket) do

    broadcasts = Query.EmailCampaigns.list_broadcasts()

    {:ok,

     assign(socket,

       broadcasts: broadcasts,

       sent_counts: Query.EmailCampaigns.sent_counts(broadcasts)

     )}

  end

end

defmodule MyAppAdminWeb.BroadcastLive.New do

  use MyAppAdminWeb, :live_view

  alias MyApp.EmailCampaigns

  alias MyApp.EmailCampaigns.Templates

  def render(assigns) do

    ~H"""

    <Layouts.admin_ui flash={@flash} current_scope={@current_scope} page_selected={:broadcasts}>

      <.header>

        New broadcast

        <:subtitle>Sends now to all {@subscriber_count} subscribers.</:subtitle>

      </.header>

      <.form for={@form} id="broadcast-form" phx-submit="send" class="max-w-xl">

        <.input

          field={@form[:template]}

          type="select"

          label="Template"

          options={Enum.map(Templates.all(), fn {key, module} -> {module.name(), key} end)}

        />

        <p class="text-sm text-base-content/70 mb-4">

          Check it first on its <.link navigate={~p"/admin/emails"} class="link">template page</.link>, and send yourself a test.

        </p>

        <div class="flex gap-2">

          <.button

            variant="primary"

            phx-disable-with="Sending..."

            data-confirm={"Send this to all #{@subscriber_count} subscribers? This can't be undone."}

          >

            Send broadcast

          </.button>

          <.link navigate={~p"/admin/emails/broadcasts"}>

            <.button type="button">Cancel</.button>

          </.link>

        </div>

      </.form>

    </Layouts.admin_ui>

    """

  end

  def mount(_params, _session, socket) do

    {:ok,

     assign(socket,

       form: to_form(%{}, as: :broadcast),

       subscriber_count: EmailCampaigns.count_subscribed_users()

     )}

  end

  def handle_event("send", %{"broadcast" => %{"template" => template}}, socket) do

    case EmailCampaigns.create_broadcast(template) do

      {:ok, broadcast} ->

        {:noreply,

         socket

         |> put_flash(:info, "Broadcast started.")

         |> push_navigate(to: ~p"/admin/emails/broadcasts/#{broadcast}")}

      {:error, changeset} ->

        {:noreply, assign(socket, form: to_form(changeset, as: :broadcast))}

    end

  end

end

defmodule MyAppAdminWeb.BroadcastLive.Show do

  use MyAppAdminWeb, :live_view

  import MyAppAdminWeb.CampaignComponents

  alias MyAppAdmin.Query

  @refresh_interval 2_000

  def render(assigns) do

    ~H"""

    <Layouts.admin_ui flash={@flash} current_scope={@current_scope} page_selected={:broadcasts}>

      <.header>

        Broadcast

        <:subtitle><.template_name key={@broadcast.template} /></:subtitle>

        <:actions>

          <.link navigate={~p"/admin/emails/broadcasts"}>

            <.button>Back</.button>

          </.link>

        </:actions>

      </.header>

      <.list>

        <:item title="Status"><.campaign_status status={@broadcast.status} /></:item>

        <:item title="Sent">{@sent_count} of {@broadcast.total_count} subscribers</:item>

        <:item title="Started">{format_datetime(@broadcast.inserted_at)}</:item>

        <:item title="Finished">{format_datetime(@broadcast.finished_at)}</:item>

      </.list>

      <p :if={@broadcast.status == "sending"} class="text-sm text-base-content/70 mt-4">

        Sending in the background. It picks up where it left off after a deploy, without resending.

      </p>

      <p :if={@broadcast.status == "canceled"} class="text-sm text-error mt-4">

        Stopped because its template was removed from code.

      </p>

    </Layouts.admin_ui>

    """

  end

  def mount(%{"id" => id}, _session, socket) do

    {:ok, assign_broadcast(socket, Query.EmailCampaigns.get_broadcast!(id))}

  end

  def handle_info(:refresh, socket) do

    {:noreply,

     assign_broadcast(socket, Query.EmailCampaigns.get_broadcast!(socket.assigns.broadcast.id))}

  end

  defp assign_broadcast(socket, broadcast) do

    if connected?(socket) and broadcast.status == "sending" do

      Process.send_after(self(), :refresh, @refresh_interval)

    end

    assign(socket,

      broadcast: broadcast,

      sent_count: Map.get(Query.EmailCampaigns.sent_counts([broadcast]), broadcast.id, 0)

    )

  end

end

defmodule MyAppAdminWeb.CampaignTemplateLive.Show do

  use MyAppAdminWeb, :live_view

  import MyAppAdminWeb.EmailComponents

  alias MyApp.EmailCampaigns.Templates

  alias MyApp.Emails

  alias MyAppAdmin.Query

  def render(assigns) do

    ~H"""

    <Layouts.admin_ui flash={@flash} current_scope={@current_scope} page_selected={:email_templates}>

      <.header>

        {@template.name()}

        <:subtitle>{@template.description()}</:subtitle>

        <:actions>

          <.link navigate={~p"/admin/emails"}>

            <.button>Back</.button>

          </.link>

        </:actions>

      </.header>

      <.list>

        <:item title="Key"><.copy_button value={@key} /></:item>

        <:item title="Module"><code>{inspect(@template)}</code></:item>

      </.list>

      <.test_send_form form={@test_form} />

      <.template_preview rendered={@rendered} />

      <div class="mt-8">

        <.header>Latest sends</.header>

        <.table id="sends" rows={@sends}>

          <:col :let={send} label="Recipient">{send.user.email}</:col>

          <:col :let={send} label="From">

            {if send.broadcast_id, do: "Broadcast", else: "Sequence"}

          </:col>

          <:col :let={send} label="Sent">{format_datetime(send.inserted_at)}</:col>

        </.table>

      </div>

    </Layouts.admin_ui>

    """

  end

  def mount(%{"key" => key}, _session, socket) do

    case Templates.get(key) do

      nil ->

        {:ok,

         socket

         |> put_flash(:error, "The campaign template #{key} was removed from code.")

         |> push_navigate(to: ~p"/admin/emails")}

      template ->

        {:ok,

         assign(socket,

           key: key,

           template: template,

           rendered: Emails.render_sample(template),

           sends: Query.EmailCampaigns.list_latest_sends(key),

           test_form: test_form(socket.assigns.current_scope)

         )}

    end

  end

  def handle_event("send_test", %{"test" => %{"email" => email}}, socket) do

    case Emails.deliver_test(socket.assigns.template, email) do

      {:ok, _} ->

        {:noreply, put_flash(socket, :info, "Test sent to #{email}.")}

      {:error, _reason} ->

        {:noreply, put_flash(socket, :error, "The test email couldn't be sent.")}

    end

  end

end

  import MyAppAdminWeb.CampaignComponents

  alias MyApp.EmailCampaigns

  alias MyAppAdmin.Query

      <div class="mt-8">

        <.header>

          Campaigns

          <:subtitle>

            Marketing templates for broadcasts and sequences, with an unsubscribe link.

          </:subtitle>

        </.header>

        <.table

          id="campaign-templates"

          rows={@campaign_templates}

          row_click={fn {key, _module} -> JS.navigate(~p"/admin/emails/campaigns/#{key}") end}

        >

          <:col :let={{_key, module}} label="Name">{module.name()}</:col>

          <:col :let={{key, _module}} label="Key"><code>{key}</code></:col>

          <:col :let={{_key, module}} label="Description">{module.description()}</:col>

          <:col :let={{key, _module}} label="Sends">{Map.get(@send_counts, key, 0)}</:col>