652 lines
19 KiB
Elixir
652 lines
19 KiB
Elixir
defmodule WhoNeedHelp.Activities do
|
|
@moduledoc """
|
|
Social plans are separate from urgent-help assignments and reputation.
|
|
"""
|
|
|
|
import Ecto.Query
|
|
|
|
alias WhoNeedHelp.Accounts
|
|
alias WhoNeedHelp.Accounts.Scope
|
|
alias WhoNeedHelp.Activities.{Activity, Message, Participant}
|
|
alias WhoNeedHelp.Catalog
|
|
alias WhoNeedHelp.Catalog.Category
|
|
alias WhoNeedHelp.Pagination
|
|
alias WhoNeedHelp.Repo
|
|
alias WhoNeedHelp.Trust
|
|
alias WhoNeedHelp.Trust.Block
|
|
|
|
@topic "activities"
|
|
|
|
def subscribe, do: Phoenix.PubSub.subscribe(WhoNeedHelp.PubSub, @topic)
|
|
|
|
def subscribe_activity(id),
|
|
do: Phoenix.PubSub.subscribe(WhoNeedHelp.PubSub, "activity:#{id}")
|
|
|
|
def notify_activity_updated(id) do
|
|
activity = load_activity(id)
|
|
broadcast({:activity_updated, activity})
|
|
broadcast_activity(id, {:activity_updated, activity})
|
|
:ok
|
|
end
|
|
|
|
def list_open_activities(%Scope{user: user}, filters \\ %{}) do
|
|
paginate_open_activities(%Scope{user: user}, filters).entries
|
|
end
|
|
|
|
def paginate_open_activities(%Scope{user: user}, filters \\ %{}, options \\ []) do
|
|
now = DateTime.utc_now(:second)
|
|
limit = Pagination.limit(options)
|
|
cursor = Pagination.cursor(options)
|
|
public_creator = Accounts.public_user_query(social_identities: true)
|
|
|
|
Activity
|
|
|> where(
|
|
[activity],
|
|
activity.status == :open and activity.starts_at > ^now and
|
|
activity.join_deadline > ^now and is_nil(activity.hidden_at)
|
|
)
|
|
|> where(
|
|
[activity],
|
|
activity.creator_id not in subquery(
|
|
from block in Block, where: block.blocker_id == ^user.id, select: block.blocked_id
|
|
)
|
|
)
|
|
|> where(
|
|
[activity],
|
|
activity.creator_id not in subquery(
|
|
from block in Block, where: block.blocked_id == ^user.id, select: block.blocker_id
|
|
)
|
|
)
|
|
|> maybe_filter_category(filters["category_id"] || filters[:category_id])
|
|
|> after_open_activity(cursor)
|
|
|> order_by([activity], asc: activity.starts_at, asc: activity.id)
|
|
|> limit(^(limit + 1))
|
|
|> with_approved_participant_count()
|
|
|> preload([activity], category: :parent, creator: ^public_creator)
|
|
|> Repo.all()
|
|
|> Pagination.page(limit, &{&1.starts_at, &1.id})
|
|
end
|
|
|
|
def list_my_activities(%Scope{user: user}) do
|
|
paginate_my_activities(%Scope{user: user}).entries
|
|
end
|
|
|
|
def paginate_my_activities(%Scope{user: user}, options \\ []) do
|
|
limit = Pagination.limit(options)
|
|
cursor = Pagination.cursor(options)
|
|
|
|
Activity
|
|
|> join(:inner, [activity], participant in Participant,
|
|
on:
|
|
participant.activity_id == activity.id and participant.user_id == ^user.id and
|
|
participant.status in [:requested, :approved]
|
|
)
|
|
|> before_my_activity(cursor)
|
|
|> order_by([activity], desc: activity.starts_at, desc: activity.id)
|
|
|> limit(^(limit + 1))
|
|
|> with_approved_participant_count()
|
|
|> preload([activity], category: :parent)
|
|
|> Repo.all()
|
|
|> Pagination.page(limit, &{&1.starts_at, &1.id})
|
|
end
|
|
|
|
defp after_open_activity(query, nil), do: query
|
|
|
|
defp after_open_activity(query, {starts_at, id}) do
|
|
where(
|
|
query,
|
|
[activity],
|
|
activity.starts_at > ^starts_at or
|
|
(activity.starts_at == ^starts_at and activity.id > ^id)
|
|
)
|
|
end
|
|
|
|
defp before_my_activity(query, nil), do: query
|
|
|
|
defp before_my_activity(query, {starts_at, id}) do
|
|
where(
|
|
query,
|
|
[activity],
|
|
activity.starts_at < ^starts_at or
|
|
(activity.starts_at == ^starts_at and activity.id < ^id)
|
|
)
|
|
end
|
|
|
|
defp before_message(query, nil), do: query
|
|
|
|
defp before_message(query, {inserted_at, id}) do
|
|
where(
|
|
query,
|
|
[message],
|
|
message.inserted_at < ^inserted_at or
|
|
(message.inserted_at == ^inserted_at and message.id < ^id)
|
|
)
|
|
end
|
|
|
|
def get_activity(%Scope{user: user}, id) do
|
|
case get_loaded_activity(id) do
|
|
nil ->
|
|
{:error, :not_found}
|
|
|
|
activity ->
|
|
cond do
|
|
activity.creator_id == user.id ->
|
|
{:ok, activity}
|
|
|
|
approved_participant?(activity, user.id) ->
|
|
{:ok, activity}
|
|
|
|
not is_nil(activity.hidden_at) ->
|
|
{:error, :not_found}
|
|
|
|
Trust.blocked_between?(activity.creator_id, user.id) ->
|
|
{:error, :not_found}
|
|
|
|
activity.status == :open ->
|
|
{:ok, activity_for_viewer(activity, user.id)}
|
|
|
|
true ->
|
|
{:error, :not_found}
|
|
end
|
|
end
|
|
end
|
|
|
|
def paginate_messages(%Scope{user: user}, %Activity{} = activity, options \\ []) do
|
|
if activity.creator_id == user.id or approved_participant?(activity, user.id) do
|
|
limit = Pagination.limit(options, 50)
|
|
cursor = Pagination.cursor(options)
|
|
public_user = Accounts.public_user_query()
|
|
|
|
page =
|
|
Message
|
|
|> where([message], message.activity_id == ^activity.id)
|
|
|> before_message(cursor)
|
|
|> order_by([message], desc: message.inserted_at, desc: message.id)
|
|
|> limit(^(limit + 1))
|
|
|> preload([message], sender: ^public_user)
|
|
|> Repo.all()
|
|
|> Pagination.page(limit, &{&1.inserted_at, &1.id})
|
|
|
|
%{page | entries: Enum.reverse(page.entries)}
|
|
else
|
|
%Pagination.Page{}
|
|
end
|
|
end
|
|
|
|
def change_activity(%Activity{} = activity, attrs \\ %{}) do
|
|
activity
|
|
|> Activity.create_changeset(attrs)
|
|
|> validate_category_and_data()
|
|
end
|
|
|
|
def create_activity(%Scope{user: user} = scope, attrs) do
|
|
result =
|
|
with {:ok, _limit} <- Trust.authorize_action(scope, :create_activity) do
|
|
Repo.transact(fn ->
|
|
with {:ok, activity} <-
|
|
%Activity{creator_id: user.id}
|
|
|> Activity.create_changeset(attrs)
|
|
|> validate_category_and_data()
|
|
|> Repo.insert(),
|
|
{:ok, _participant} <-
|
|
%Participant{}
|
|
|> Participant.changeset(%{
|
|
activity_id: activity.id,
|
|
user_id: user.id,
|
|
role: :organizer,
|
|
status: :approved,
|
|
reviewed_at: DateTime.utc_now(:second)
|
|
})
|
|
|> Repo.insert(),
|
|
{:ok, _audit} <-
|
|
Trust.audit(user.id, "activity.created", "activity", activity.id, %{
|
|
"category_id" => activity.category_id
|
|
}) do
|
|
{:ok, activity}
|
|
end
|
|
end)
|
|
end
|
|
|
|
with {:ok, activity} <- result do
|
|
activity = load_activity(activity.id)
|
|
broadcast({:activity_created, activity})
|
|
{:ok, activity}
|
|
end
|
|
end
|
|
|
|
def request_to_join(%Scope{user: user} = scope, activity_id) do
|
|
result =
|
|
with {:ok, _limit} <- Trust.authorize_action(scope, :join_activity) do
|
|
Repo.transact(fn ->
|
|
activity = locked_activity(activity_id)
|
|
now = DateTime.utc_now(:second)
|
|
|
|
cond do
|
|
activity.creator_id == user.id ->
|
|
{:error, :organizer_already_joined}
|
|
|
|
activity.status != :open or not is_nil(activity.hidden_at) ->
|
|
{:error, :not_open}
|
|
|
|
DateTime.compare(activity.join_deadline, now) != :gt ->
|
|
{:error, :join_closed}
|
|
|
|
Trust.blocked_between?(activity.creator_id, user.id) ->
|
|
{:error, :blocked}
|
|
|
|
approved_count(activity.id) >= activity.capacity ->
|
|
{:error, :capacity_reached}
|
|
|
|
true ->
|
|
upsert_join_request(activity.id, user.id, now)
|
|
end
|
|
end)
|
|
end
|
|
|
|
after_participant_change(result, activity_id, :join_requested)
|
|
end
|
|
|
|
def approve_participant(%Scope{user: organizer}, participant_id) do
|
|
result =
|
|
Repo.transact(fn ->
|
|
participant = locked_participant(participant_id)
|
|
activity = locked_activity(participant.activity_id)
|
|
|
|
cond do
|
|
activity.creator_id != organizer.id ->
|
|
{:error, :forbidden}
|
|
|
|
participant.status != :requested ->
|
|
{:error, :invalid_transition}
|
|
|
|
activity.status != :open ->
|
|
{:error, :not_open}
|
|
|
|
Trust.blocked_between?(activity.creator_id, participant.user_id) ->
|
|
{:error, :blocked}
|
|
|
|
approved_count(activity.id) >= activity.capacity ->
|
|
{:error, :capacity_reached}
|
|
|
|
true ->
|
|
with {:ok, participant} <-
|
|
participant
|
|
|> Participant.changeset(%{
|
|
status: :approved,
|
|
reviewed_at: DateTime.utc_now(:second)
|
|
})
|
|
|> Repo.update(),
|
|
{:ok, _audit} <-
|
|
Trust.audit(
|
|
organizer.id,
|
|
"activity.participant_approved",
|
|
"activity_participant",
|
|
participant.id,
|
|
%{"activity_id" => activity.id, "user_id" => participant.user_id}
|
|
) do
|
|
{:ok, participant}
|
|
end
|
|
end
|
|
end)
|
|
|
|
after_participant_change(result, participant_activity_id(result), :participant_approved)
|
|
end
|
|
|
|
def decline_participant(%Scope{user: organizer}, participant_id) do
|
|
transition_participant(organizer.id, participant_id, :declined)
|
|
end
|
|
|
|
def leave_activity(%Scope{user: user}, activity_id) do
|
|
result =
|
|
Repo.transact(fn ->
|
|
participant =
|
|
Participant
|
|
|> where(
|
|
[participant],
|
|
participant.activity_id == ^activity_id and participant.user_id == ^user.id
|
|
)
|
|
|> lock("FOR UPDATE")
|
|
|> Repo.one()
|
|
|
|
cond do
|
|
is_nil(participant) ->
|
|
{:error, :not_found}
|
|
|
|
participant.role == :organizer ->
|
|
{:error, :organizer_cannot_leave}
|
|
|
|
participant.status not in [:requested, :approved] ->
|
|
{:error, :invalid_transition}
|
|
|
|
true ->
|
|
with {:ok, participant} <-
|
|
participant
|
|
|> Participant.changeset(%{
|
|
status: :left,
|
|
left_at: DateTime.utc_now(:second)
|
|
})
|
|
|> Repo.update(),
|
|
{:ok, _audit} <-
|
|
Trust.audit(user.id, "activity.left", "activity", activity_id) do
|
|
{:ok, participant}
|
|
end
|
|
end
|
|
end)
|
|
|
|
after_participant_change(result, activity_id, :participant_left)
|
|
end
|
|
|
|
def cancel_activity(%Scope{user: user}, activity_id) do
|
|
transition_activity(user.id, activity_id, :cancelled)
|
|
end
|
|
|
|
def complete_activity(%Scope{user: user}, activity_id) do
|
|
transition_activity(user.id, activity_id, :completed)
|
|
end
|
|
|
|
def send_message(%Scope{user: user} = scope, activity_id, attrs) do
|
|
with {:ok, _limit} <- Trust.authorize_action(scope, :send_activity_message),
|
|
{:ok, activity} <- get_activity(scope, activity_id),
|
|
true <- activity.status == :open,
|
|
true <- approved_participant?(activity, user.id) do
|
|
%Message{}
|
|
|> Message.changeset(%{
|
|
body: attrs["body"] || attrs[:body],
|
|
activity_id: activity.id,
|
|
sender_id: user.id
|
|
})
|
|
|> Repo.insert()
|
|
|> case do
|
|
{:ok, message} ->
|
|
message = Repo.preload(message, sender: Accounts.public_user_query())
|
|
|
|
Phoenix.PubSub.broadcast(
|
|
WhoNeedHelp.PubSub,
|
|
"activity:#{activity.id}",
|
|
{:activity_message, message}
|
|
)
|
|
|
|
{:ok, message}
|
|
|
|
other ->
|
|
other
|
|
end
|
|
else
|
|
false -> {:error, :forbidden}
|
|
other -> other
|
|
end
|
|
end
|
|
|
|
def coordinates_for(%Scope{user: user}, %Activity{} = activity) do
|
|
if activity.creator_id == user.id or approved_participant?(activity, user.id) do
|
|
Activity.exact_coordinates(activity)
|
|
else
|
|
Activity.public_coordinates(activity)
|
|
end
|
|
end
|
|
|
|
def participant_for(%Activity{} = activity, user_id) do
|
|
Enum.find(activity.participants, &(&1.user_id == user_id))
|
|
end
|
|
|
|
def approved_participant?(%Activity{} = activity, user_id) do
|
|
Enum.any?(
|
|
activity.participants,
|
|
&(&1.user_id == user_id and &1.status == :approved)
|
|
)
|
|
end
|
|
|
|
defp validate_category_and_data(changeset) do
|
|
category_id = Ecto.Changeset.get_field(changeset, :category_id)
|
|
|
|
case category_id && Repo.get(Category, category_id) do
|
|
%Category{mode: :activity, active: true} = category ->
|
|
data = Ecto.Changeset.get_field(changeset, :structured_data)
|
|
|
|
case Catalog.validate_structured_data(category, data) do
|
|
{:ok, clean} ->
|
|
Ecto.Changeset.put_change(changeset, :structured_data, clean)
|
|
|
|
{:error, errors} ->
|
|
Enum.reduce(
|
|
errors,
|
|
changeset,
|
|
&Ecto.Changeset.add_error(&2, :structured_data, &1)
|
|
)
|
|
end
|
|
|
|
_ ->
|
|
Ecto.Changeset.add_error(changeset, :category_id, "select an activity category")
|
|
end
|
|
end
|
|
|
|
defp upsert_join_request(activity_id, user_id, now) do
|
|
case Repo.get_by(Participant, activity_id: activity_id, user_id: user_id) do
|
|
nil ->
|
|
%Participant{}
|
|
|> Participant.changeset(%{
|
|
activity_id: activity_id,
|
|
user_id: user_id,
|
|
role: :participant,
|
|
status: :requested
|
|
})
|
|
|> Repo.insert()
|
|
|
|
%Participant{status: status} = participant when status in [:declined, :left] ->
|
|
participant
|
|
|> Participant.changeset(%{
|
|
status: :requested,
|
|
reviewed_at: nil,
|
|
left_at: nil
|
|
})
|
|
|> Repo.update()
|
|
|
|
%Participant{} ->
|
|
{:error, :already_joined}
|
|
end
|
|
|> case do
|
|
{:ok, participant} ->
|
|
with {:ok, _audit} <-
|
|
Trust.audit(user_id, "activity.join_requested", "activity", activity_id, %{
|
|
"requested_at" => DateTime.to_iso8601(now)
|
|
}) do
|
|
{:ok, participant}
|
|
end
|
|
|
|
other ->
|
|
other
|
|
end
|
|
end
|
|
|
|
defp transition_participant(organizer_id, participant_id, status) do
|
|
result =
|
|
Repo.transact(fn ->
|
|
participant = locked_participant(participant_id)
|
|
activity = locked_activity(participant.activity_id)
|
|
|
|
cond do
|
|
activity.creator_id != organizer_id ->
|
|
{:error, :forbidden}
|
|
|
|
participant.status != :requested ->
|
|
{:error, :invalid_transition}
|
|
|
|
true ->
|
|
with {:ok, participant} <-
|
|
participant
|
|
|> Participant.changeset(%{
|
|
status: status,
|
|
reviewed_at: DateTime.utc_now(:second)
|
|
})
|
|
|> Repo.update(),
|
|
{:ok, _audit} <-
|
|
Trust.audit(
|
|
organizer_id,
|
|
"activity.participant_#{status}",
|
|
"activity_participant",
|
|
participant.id,
|
|
%{"activity_id" => activity.id, "user_id" => participant.user_id}
|
|
) do
|
|
{:ok, participant}
|
|
end
|
|
end
|
|
end)
|
|
|
|
after_participant_change(result, participant_activity_id(result), :participant_declined)
|
|
end
|
|
|
|
defp transition_activity(user_id, activity_id, target_status) do
|
|
result =
|
|
Repo.transact(fn ->
|
|
activity = locked_activity(activity_id)
|
|
|
|
cond do
|
|
activity.creator_id != user_id ->
|
|
{:error, :forbidden}
|
|
|
|
activity.status != :open ->
|
|
{:error, :invalid_transition}
|
|
|
|
true ->
|
|
timestamp = DateTime.utc_now(:second)
|
|
|
|
attrs =
|
|
case target_status do
|
|
:cancelled -> %{status: :cancelled, cancelled_at: timestamp}
|
|
:completed -> %{status: :completed, completed_at: timestamp}
|
|
end
|
|
|
|
with {:ok, activity} <- activity |> Ecto.Changeset.change(attrs) |> Repo.update(),
|
|
{:ok, _audit} <-
|
|
Trust.audit(
|
|
user_id,
|
|
"activity.#{target_status}",
|
|
"activity",
|
|
activity.id
|
|
) do
|
|
{:ok, activity}
|
|
end
|
|
end
|
|
end)
|
|
|
|
with {:ok, activity} <- result do
|
|
activity = load_activity(activity.id)
|
|
broadcast({:activity_updated, activity})
|
|
broadcast_activity(activity.id, {:activity_updated, activity})
|
|
{:ok, activity}
|
|
end
|
|
end
|
|
|
|
defp after_participant_change({:ok, participant}, activity_id, event)
|
|
when is_binary(activity_id) do
|
|
activity = load_activity(activity_id)
|
|
broadcast({:activity_updated, activity})
|
|
broadcast_activity(activity.id, {event, participant})
|
|
{:ok, participant}
|
|
end
|
|
|
|
defp after_participant_change(result, _activity_id, _event), do: result
|
|
|
|
defp participant_activity_id({:ok, %Participant{activity_id: activity_id}}), do: activity_id
|
|
defp participant_activity_id(_result), do: nil
|
|
|
|
defp approved_count(activity_id) do
|
|
Repo.aggregate(
|
|
from(
|
|
participant in Participant,
|
|
where: participant.activity_id == ^activity_id and participant.status == :approved
|
|
),
|
|
:count
|
|
)
|
|
end
|
|
|
|
defp load_activity(id) do
|
|
Activity
|
|
|> Repo.get!(id)
|
|
|> preload_activity()
|
|
end
|
|
|
|
defp get_loaded_activity(id) do
|
|
Activity
|
|
|> Repo.get(id)
|
|
|> preload_activity()
|
|
end
|
|
|
|
defp preload_activity(nil), do: nil
|
|
|
|
defp preload_activity(activity) do
|
|
public_user = Accounts.public_user_query(social_identities: true)
|
|
public_sender = Accounts.public_user_query()
|
|
|
|
recent_messages =
|
|
Message
|
|
|> order_by([message], desc: message.inserted_at, desc: message.id)
|
|
|> limit(50)
|
|
|> preload([message], sender: ^public_sender)
|
|
|
|
Repo.preload(
|
|
activity,
|
|
category: :parent,
|
|
creator: public_user,
|
|
participants: [user: public_user],
|
|
messages: recent_messages
|
|
)
|
|
end
|
|
|
|
defp activity_for_viewer(activity, user_id) do
|
|
%{
|
|
activity
|
|
| messages: [],
|
|
participants:
|
|
Enum.filter(
|
|
activity.participants,
|
|
&(&1.status == :approved or &1.user_id == user_id)
|
|
)
|
|
}
|
|
end
|
|
|
|
defp with_approved_participant_count(query) do
|
|
counts =
|
|
Participant
|
|
|> where([participant], participant.status == :approved)
|
|
|> group_by([participant], participant.activity_id)
|
|
|> select([participant], %{
|
|
activity_id: participant.activity_id,
|
|
approved_count: count(participant.id)
|
|
})
|
|
|
|
query
|
|
|> join(:left, [activity], count in subquery(counts),
|
|
on: count.activity_id == activity.id,
|
|
as: :approved_counts
|
|
)
|
|
|> select_merge([approved_counts: count], %{
|
|
approved_participant_count: fragment("coalesce(?, 0)", count.approved_count)
|
|
})
|
|
end
|
|
|
|
defp locked_activity(id) do
|
|
Activity
|
|
|> where([activity], activity.id == ^id)
|
|
|> lock("FOR UPDATE")
|
|
|> Repo.one!()
|
|
end
|
|
|
|
defp locked_participant(id) do
|
|
Participant
|
|
|> where([participant], participant.id == ^id)
|
|
|> lock("FOR UPDATE")
|
|
|> Repo.one!()
|
|
end
|
|
|
|
defp maybe_filter_category(query, value) when value in [nil, ""], do: query
|
|
|
|
defp maybe_filter_category(query, value),
|
|
do: where(query, [activity], activity.category_id == ^value)
|
|
|
|
defp broadcast(message), do: Phoenix.PubSub.broadcast(WhoNeedHelp.PubSub, @topic, message)
|
|
|
|
defp broadcast_activity(id, message),
|
|
do: Phoenix.PubSub.broadcast(WhoNeedHelp.PubSub, "activity:#{id}", message)
|
|
end
|