1081 lines
33 KiB
Elixir
1081 lines
33 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.Discovery.Rollup
|
|
alias WhoNeedHelp.Help.{DiscoveryCluster, DiscoveryViewport}
|
|
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_user(%Scope{user: user}),
|
|
do: Phoenix.PubSub.subscribe(WhoNeedHelp.PubSub, "activities:user:#{user.id}")
|
|
|
|
def subscribe_activity(id),
|
|
do: Phoenix.PubSub.subscribe(WhoNeedHelp.PubSub, "activity:#{id}")
|
|
|
|
def notify_activity_updated(id) do
|
|
broadcast({:activity_updated, load_activity_summary(id)})
|
|
broadcast_activity(id, {:activity_updated, load_activity(id)})
|
|
: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
|
|
limit = Pagination.limit(options)
|
|
cursor = Pagination.cursor(options)
|
|
viewport = Keyword.get(options, :viewport)
|
|
|
|
user
|
|
|> open_activities_query(filters, viewport)
|
|
|> after_open_activity(cursor)
|
|
|> order_by([activity], asc: activity.starts_at, asc: activity.id)
|
|
|> limit(^(limit + 1))
|
|
|> with_approved_participant_count()
|
|
|> Repo.all()
|
|
|> Repo.preload(category: :parent, creator: Accounts.public_user_query())
|
|
|> Pagination.page(limit, &{&1.starts_at, &1.id})
|
|
end
|
|
|
|
def count_open_activities(%Scope{user: user}, filters \\ %{}, options \\ []) do
|
|
viewport = Keyword.get(options, :viewport)
|
|
|
|
user
|
|
|> open_activities_query(filters, viewport)
|
|
|> Repo.aggregate(:count, :id)
|
|
end
|
|
|
|
def map_discovery_items(%Scope{user: user}, filters, %DiscoveryViewport{} = viewport) do
|
|
level = DiscoveryCluster.level_for_zoom(viewport.zoom)
|
|
|
|
if DiscoveryCluster.rollup_level?(level) and blank_area_filter?(filters) do
|
|
Rollup.activity_items(user, filters, viewport, level)
|
|
else
|
|
raw_map_discovery_items(user, filters, viewport, level)
|
|
end
|
|
end
|
|
|
|
def map_cluster_expansion(%Scope{user: user}, filters, cluster_id) do
|
|
Rollup.activity_expansion(user, filters, cluster_id)
|
|
end
|
|
|
|
defp raw_map_discovery_items(user, filters, viewport, level) do
|
|
cell_count = Integer.pow(2, level) * 1.0
|
|
|
|
points =
|
|
user
|
|
|> open_activities_query(filters, nil)
|
|
|> where([activity], activity.location_visibility == :approximate_public)
|
|
|> filter_discovery_envelopes(
|
|
DiscoveryCluster.viewport_envelopes(viewport, level),
|
|
filters
|
|
)
|
|
|> select([activity], %{
|
|
id: activity.id,
|
|
public_longitude: activity.public_longitude,
|
|
public_latitude: activity.public_latitude
|
|
})
|
|
|
|
points
|
|
|> subquery()
|
|
|> group_by([_point], [selected_as(:cell_x), selected_as(:cell_y)])
|
|
|> select([point], %{
|
|
count: count(point.id),
|
|
activity_id: fragment("min(?::text)", point.id),
|
|
longitude: min(point.public_longitude),
|
|
latitude: min(point.public_latitude),
|
|
west: min(point.public_longitude),
|
|
south: min(point.public_latitude),
|
|
east: max(point.public_longitude),
|
|
north: max(point.public_latitude),
|
|
cell_x:
|
|
selected_as(
|
|
fragment(
|
|
"floor(((? + 180.0) / 360.0) * ?)::bigint",
|
|
point.public_longitude,
|
|
constant(^cell_count)
|
|
),
|
|
:cell_x
|
|
),
|
|
cell_y:
|
|
selected_as(
|
|
fragment(
|
|
"floor(((ln(tan(pi() / 4.0 + radians(least(85.05112878, greatest(-85.05112878, ?))) / 2.0)) / pi() + 1.0) / 2.0) * ?)::bigint",
|
|
point.public_latitude,
|
|
constant(^cell_count)
|
|
),
|
|
:cell_y
|
|
)
|
|
})
|
|
|> order_by([_point], asc: selected_as(:cell_x), asc: selected_as(:cell_y))
|
|
|> Repo.all()
|
|
|> hydrate_activity_map_singletons()
|
|
|> Enum.map(&map_discovery_item(&1, level))
|
|
end
|
|
|
|
def count_map_discovery_items(%Scope{user: user}, filters) do
|
|
user
|
|
|> open_activities_query(filters, nil)
|
|
|> where([activity], activity.location_visibility == :approximate_public)
|
|
|> Repo.aggregate(:count, :id)
|
|
end
|
|
|
|
def list_my_activities(%Scope{user: user}) do
|
|
paginate_my_activities(%Scope{user: user}).entries
|
|
end
|
|
|
|
def visible_open_activity?(
|
|
%Scope{user: user},
|
|
%Activity{} = activity,
|
|
filters \\ %{},
|
|
viewport \\ nil
|
|
) do
|
|
now = DateTime.utc_now(:second)
|
|
category_id = filters["category_id"] || filters[:category_id]
|
|
area = filters["area"] || filters[:area]
|
|
|
|
activity.status == :open and is_nil(activity.hidden_at) and
|
|
DateTime.after?(activity.starts_at, now) and
|
|
DateTime.after?(activity.join_deadline, now) and
|
|
(category_id in [nil, ""] or
|
|
to_string(activity.category_id) == to_string(category_id)) and
|
|
area_matches?(activity, area) and
|
|
viewport_matches?(activity, viewport, area) and
|
|
not Trust.blocked_between?(user.id, activity.creator_id)
|
|
end
|
|
|
|
def member_activity?(%Activity{} = activity, user_id) do
|
|
Enum.any?(
|
|
activity.participants,
|
|
&(&1.user_id == user_id and &1.status in [:requested, :approved])
|
|
)
|
|
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]
|
|
)
|
|
|> 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)
|
|
)
|
|
)
|
|
|> 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
|
|
with {:ok, id} <- Ecto.UUID.cast(id),
|
|
%Activity{} = activity <- get_loaded_activity(id) do
|
|
cond do
|
|
activity.creator_id == user.id ->
|
|
{:ok, activity}
|
|
|
|
not is_nil(activity.hidden_at) ->
|
|
{:error, :not_found}
|
|
|
|
Trust.blocked_between?(activity.creator_id, user.id) ->
|
|
{:error, :not_found}
|
|
|
|
approved_participant?(activity, user.id) ->
|
|
{:ok, activity}
|
|
|
|
activity.status == :open ->
|
|
{:ok, activity_for_viewer(activity, user.id)}
|
|
|
|
true ->
|
|
{:error, :not_found}
|
|
end
|
|
else
|
|
_invalid_or_missing -> {:error, :not_found}
|
|
end
|
|
end
|
|
|
|
def paginate_messages(%Scope{user: user}, %Activity{} = activity, options \\ []) do
|
|
authorized =
|
|
activity.creator_id == user.id or
|
|
(approved_participant?(activity, user.id) and
|
|
not Trust.blocked_between?(activity.creator_id, user.id))
|
|
|
|
if authorized 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
|
|
broadcast({:activity_created, load_activity_summary(activity.id)})
|
|
{:ok, load_activity(activity.id)}
|
|
end
|
|
end
|
|
|
|
def request_to_join(%Scope{user: user} = scope, activity_id) do
|
|
result =
|
|
with {:ok, activity_id} <- cast_id(activity_id),
|
|
{:ok, _limit} <- Trust.authorize_action(scope, :join_activity) do
|
|
Repo.transact(fn ->
|
|
with %Activity{} = activity <- locked_activity(activity_id) do
|
|
now = DateTime.utc_now(:second)
|
|
:ok = Trust.lock_user_pair(activity.creator_id, user.id)
|
|
|
|
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
|
|
else
|
|
nil -> {:error, :not_found}
|
|
end
|
|
end)
|
|
end
|
|
|
|
after_participant_change(result, activity_id, :join_requested)
|
|
end
|
|
|
|
def approve_participant(%Scope{user: organizer}, participant_id) do
|
|
result =
|
|
with {:ok, participant_id} <- cast_id(participant_id) do
|
|
Repo.transact(fn ->
|
|
with %Participant{} = participant <- locked_participant(participant_id),
|
|
%Activity{} = activity <- locked_activity(participant.activity_id) do
|
|
:ok = Trust.lock_user_pair(activity.creator_id, participant.user_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
|
|
else
|
|
nil -> {:error, :not_found}
|
|
end
|
|
end)
|
|
end
|
|
|
|
after_participant_change(result, participant_activity_id(result), :participant_approved)
|
|
end
|
|
|
|
def decline_participant(%Scope{user: organizer}, participant_id) do
|
|
with {:ok, participant_id} <- cast_id(participant_id) do
|
|
transition_participant(organizer.id, participant_id, :declined)
|
|
end
|
|
end
|
|
|
|
def leave_activity(%Scope{user: user}, activity_id) do
|
|
result =
|
|
with {:ok, activity_id} <- cast_id(activity_id) do
|
|
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)
|
|
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
|
|
result =
|
|
with {:ok, activity_id} <- cast_id(activity_id),
|
|
{:ok, _limit} <- Trust.authorize_action(scope, :send_activity_message) do
|
|
Repo.transact(fn ->
|
|
with %Activity{} = activity <- locked_activity(activity_id) do
|
|
:ok = Trust.lock_user_pair(activity.creator_id, user.id)
|
|
|
|
cond do
|
|
activity.status != :open or not is_nil(activity.hidden_at) ->
|
|
{:error, :not_open}
|
|
|
|
Trust.blocked_between?(activity.creator_id, user.id) ->
|
|
{:error, :blocked}
|
|
|
|
not approved_participant_id?(activity.id, user.id) ->
|
|
{:error, :forbidden}
|
|
|
|
true ->
|
|
%Message{}
|
|
|> Message.changeset(%{
|
|
body: attrs["body"] || attrs[:body],
|
|
activity_id: activity.id,
|
|
sender_id: user.id
|
|
})
|
|
|> Repo.insert()
|
|
end
|
|
else
|
|
nil -> {:error, :not_found}
|
|
end
|
|
end)
|
|
end
|
|
|
|
with {:ok, message} <- result do
|
|
message = Repo.preload(message, sender: Accounts.public_user_query())
|
|
|
|
Phoenix.PubSub.broadcast(
|
|
WhoNeedHelp.PubSub,
|
|
"activity:#{message.activity_id}",
|
|
{:activity_message, message}
|
|
)
|
|
|
|
{:ok, message}
|
|
end
|
|
end
|
|
|
|
def coordinates_for(%Scope{user: user}, %Activity{} = activity) do
|
|
authorized =
|
|
activity.creator_id == user.id or
|
|
(approved_participant?(activity, user.id) and
|
|
not Trust.blocked_between?(activity.creator_id, user.id))
|
|
|
|
if authorized 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 approved_participant_id?(activity_id, user_id) do
|
|
Repo.exists?(
|
|
from(participant in Participant,
|
|
where:
|
|
participant.activity_id == ^activity_id and participant.user_id == ^user_id and
|
|
participant.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 ->
|
|
with %Participant{} = participant <- locked_participant(participant_id),
|
|
%Activity{} = activity <- locked_activity(participant.activity_id) do
|
|
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
|
|
else
|
|
nil -> {:error, :not_found}
|
|
end
|
|
end)
|
|
|
|
after_participant_change(result, participant_activity_id(result), :participant_declined)
|
|
end
|
|
|
|
defp transition_activity(user_id, activity_id, target_status) do
|
|
result =
|
|
with {:ok, activity_id} <- cast_id(activity_id) do
|
|
Repo.transact(fn ->
|
|
with %Activity{} = activity <- locked_activity(activity_id) do
|
|
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
|
|
else
|
|
nil -> {:error, :not_found}
|
|
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
|
|
summary = load_activity_summary(activity_id)
|
|
broadcast({:activity_updated, summary})
|
|
broadcast_user(participant.user_id, {:my_activity_updated, summary, member?(participant)})
|
|
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 load_activity_summary(id) do
|
|
public_creator = Accounts.public_user_query()
|
|
|
|
Activity
|
|
|> where([activity], activity.id == ^id)
|
|
|> with_approved_participant_count()
|
|
|> preload([activity], category: :parent, creator: ^public_creator)
|
|
|> Repo.one!()
|
|
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 open_activities_query(user, filters, viewport) do
|
|
now = DateTime.utc_now(:second)
|
|
|
|
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])
|
|
|> maybe_filter_area(filters["area"] || filters[:area])
|
|
|> filter_discovery_viewport(viewport, filters)
|
|
end
|
|
|
|
defp filter_discovery_viewport(query, nil, _filters), do: query
|
|
|
|
defp filter_discovery_viewport(query, %DiscoveryViewport{} = viewport, filters) do
|
|
filter_discovery_envelopes(query, DiscoveryViewport.envelopes(viewport), filters)
|
|
end
|
|
|
|
defp filter_discovery_envelopes(query, envelopes, filters) when is_list(envelopes) do
|
|
candidate_condition =
|
|
Enum.reduce(envelopes, dynamic(false), fn
|
|
{west, south, east, north}, condition ->
|
|
dynamic(
|
|
[activity],
|
|
^condition or
|
|
fragment(
|
|
"""
|
|
(
|
|
? && ST_MakeEnvelope(?, ?, ?, ?, 4326)
|
|
)
|
|
""",
|
|
activity.location,
|
|
^(west - 0.005),
|
|
^(south - 0.005),
|
|
^(east + 0.005),
|
|
^(north + 0.005)
|
|
)
|
|
)
|
|
end)
|
|
|
|
exact_condition =
|
|
Enum.reduce(envelopes, dynamic(false), fn
|
|
{west, south, east, north}, condition ->
|
|
dynamic(
|
|
[activity],
|
|
^condition or
|
|
fragment(
|
|
"""
|
|
? >= ? AND ? <= ? AND ? >= ? AND ? <= ?
|
|
""",
|
|
activity.public_longitude,
|
|
^west,
|
|
activity.public_longitude,
|
|
^east,
|
|
activity.public_latitude,
|
|
^south,
|
|
activity.public_latitude,
|
|
^north
|
|
)
|
|
)
|
|
end)
|
|
|
|
spatial_condition = dynamic([activity], ^candidate_condition and ^exact_condition)
|
|
|
|
area = filters["area"] || filters[:area]
|
|
|
|
condition =
|
|
if is_binary(area) and String.trim(area) != "" do
|
|
dynamic(
|
|
[activity],
|
|
(activity.location_visibility == :approximate_public and ^spatial_condition) or
|
|
activity.location_visibility == :hidden
|
|
)
|
|
else
|
|
dynamic(
|
|
[activity],
|
|
activity.location_visibility == :approximate_public and ^spatial_condition
|
|
)
|
|
end
|
|
|
|
where(query, ^condition)
|
|
end
|
|
|
|
defp maybe_filter_area(query, value) when value in [nil, ""], do: query
|
|
|
|
defp maybe_filter_area(query, value) when is_binary(value) do
|
|
value = value |> String.trim() |> String.slice(0, 100)
|
|
|
|
if value == "" do
|
|
query
|
|
else
|
|
pattern = "%#{escape_like(value)}%"
|
|
|
|
where(
|
|
query,
|
|
[activity],
|
|
fragment("? ILIKE ? ESCAPE E'\\\\'", activity.location_label, ^pattern)
|
|
)
|
|
end
|
|
end
|
|
|
|
defp maybe_filter_area(query, _value), do: where(query, [activity], false)
|
|
|
|
defp area_matches?(_activity, value) when value in [nil, ""], do: true
|
|
|
|
defp area_matches?(activity, value) when is_binary(value) do
|
|
normalized = value |> String.trim() |> String.downcase()
|
|
|
|
normalized == "" or
|
|
String.contains?(String.downcase(activity.location_label || ""), normalized)
|
|
end
|
|
|
|
defp area_matches?(_activity, _value), do: false
|
|
|
|
defp viewport_matches?(_activity, nil, _area), do: true
|
|
|
|
defp viewport_matches?(
|
|
%Activity{location_visibility: :hidden},
|
|
%DiscoveryViewport{},
|
|
area
|
|
),
|
|
do: is_binary(area) and String.trim(area) != ""
|
|
|
|
defp viewport_matches?(activity, %DiscoveryViewport{} = viewport, _area) do
|
|
%{latitude: latitude, longitude: longitude} = Activity.public_coordinates(activity)
|
|
|
|
latitude >= viewport.south and latitude <= viewport.north and
|
|
longitude_in_viewport?(longitude, viewport)
|
|
end
|
|
|
|
defp longitude_in_viewport?(longitude, %DiscoveryViewport{west: west, east: east})
|
|
when west < east,
|
|
do: longitude >= west and longitude <= east
|
|
|
|
defp longitude_in_viewport?(longitude, %DiscoveryViewport{west: west, east: east}),
|
|
do: longitude >= west or longitude <= east
|
|
|
|
defp map_discovery_item(%{count: 1} = row, level) do
|
|
%{
|
|
type: "activity",
|
|
id: row.activity_id,
|
|
parent_cluster_id: DiscoveryCluster.parent_id(:activity, level, row.cell_x, row.cell_y),
|
|
hierarchy_level: level,
|
|
title: row.title,
|
|
location: row.location_label,
|
|
latitude: row.latitude,
|
|
longitude: row.longitude,
|
|
exact: false
|
|
}
|
|
end
|
|
|
|
defp map_discovery_item(row, level) do
|
|
cluster_id = DiscoveryCluster.cluster_id(:activity, level, row.cell_x, row.cell_y)
|
|
{longitude, latitude} = DiscoveryCluster.cell_center(level, row.cell_x, row.cell_y)
|
|
|
|
%{
|
|
type: "cluster",
|
|
id: cluster_id,
|
|
cluster_id: cluster_id,
|
|
parent_cluster_id: DiscoveryCluster.parent_id(:activity, level, row.cell_x, row.cell_y),
|
|
hierarchy_level: level,
|
|
point_count: row.count,
|
|
expansion_zoom:
|
|
DiscoveryCluster.expansion_zoom_for_bounds(
|
|
level,
|
|
row.west,
|
|
row.south,
|
|
row.east,
|
|
row.north
|
|
),
|
|
count: row.count,
|
|
latitude: latitude,
|
|
longitude: longitude,
|
|
bounds: %{
|
|
west: row.west,
|
|
south: row.south,
|
|
east: row.east,
|
|
north: row.north
|
|
}
|
|
}
|
|
end
|
|
|
|
defp hydrate_activity_map_singletons(rows) do
|
|
singleton_ids =
|
|
rows
|
|
|> Enum.filter(&(&1.count == 1))
|
|
|> Enum.map(& &1.activity_id)
|
|
|
|
details = activity_map_singleton_details(singleton_ids)
|
|
|
|
Enum.map(rows, fn
|
|
%{count: 1, activity_id: activity_id} = row ->
|
|
Map.merge(row, Map.fetch!(details, activity_id))
|
|
|
|
row ->
|
|
row
|
|
end)
|
|
end
|
|
|
|
defp activity_map_singleton_details([]), do: %{}
|
|
|
|
defp activity_map_singleton_details(singleton_ids) do
|
|
Activity
|
|
|> where([activity], activity.id in ^singleton_ids)
|
|
|> select([activity], {
|
|
activity.id,
|
|
%{title: activity.title, location_label: activity.location_label}
|
|
})
|
|
|> Repo.all()
|
|
|> Map.new(fn {id, detail} -> {to_string(id), detail} end)
|
|
end
|
|
|
|
defp escape_like(value) do
|
|
value
|
|
|> String.replace("\\", "\\\\")
|
|
|> String.replace("%", "\\%")
|
|
|> String.replace("_", "\\_")
|
|
end
|
|
|
|
defp blank_area_filter?(filters) do
|
|
case filters["area"] || filters[:area] do
|
|
nil -> true
|
|
area when is_binary(area) -> String.trim(area) == ""
|
|
_invalid -> false
|
|
end
|
|
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
|
|
case Ecto.UUID.cast(value) do
|
|
{:ok, category_id} -> where(query, [activity], activity.category_id == ^category_id)
|
|
:error -> where(query, [activity], false)
|
|
end
|
|
end
|
|
|
|
defp cast_id(value) do
|
|
case Ecto.UUID.cast(value) do
|
|
{:ok, id} -> {:ok, id}
|
|
:error -> {:error, :not_found}
|
|
end
|
|
end
|
|
|
|
defp broadcast(message), do: Phoenix.PubSub.broadcast(WhoNeedHelp.PubSub, @topic, message)
|
|
|
|
defp broadcast_user(user_id, message),
|
|
do: Phoenix.PubSub.broadcast(WhoNeedHelp.PubSub, "activities:user:#{user_id}", message)
|
|
|
|
defp broadcast_activity(id, message),
|
|
do: Phoenix.PubSub.broadcast(WhoNeedHelp.PubSub, "activity:#{id}", message)
|
|
|
|
defp member?(%Participant{status: status}), do: status in [:requested, :approved]
|
|
end
|