984 lines
30 KiB
Elixir
984 lines
30 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.Help.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 map_discovery_items(%Scope{user: user}, filters, %DiscoveryViewport{} = viewport) do
|
|
cell_size = DiscoveryViewport.cluster_cell_meters(viewport)
|
|
|
|
points =
|
|
user
|
|
|> open_activities_query(filters, viewport)
|
|
|> where([activity], activity.location_visibility == :approximate_public)
|
|
|> select([activity], %{
|
|
id: activity.id,
|
|
title: activity.title,
|
|
location_label: activity.location_label,
|
|
public_location:
|
|
fragment(
|
|
"ST_SetSRID(ST_MakePoint(round(ST_X(?)::numeric, 2)::double precision, round(ST_Y(?)::numeric, 2)::double precision), 4326)",
|
|
activity.location,
|
|
activity.location
|
|
)
|
|
})
|
|
|
|
points
|
|
|> subquery()
|
|
|> group_by(
|
|
[point],
|
|
fragment(
|
|
"ST_SnapToGrid(ST_Transform(?, 3857), ?)",
|
|
point.public_location,
|
|
constant(^cell_size)
|
|
)
|
|
)
|
|
|> select([point], %{
|
|
count: count(point.id),
|
|
activity_id: fragment("CASE WHEN count(*) = 1 THEN min(?::text) END", point.id),
|
|
title: fragment("CASE WHEN count(*) = 1 THEN min(?) END", point.title),
|
|
location_label: fragment("CASE WHEN count(*) = 1 THEN min(?) END", point.location_label),
|
|
longitude:
|
|
fragment(
|
|
"ST_X(ST_Transform(ST_Centroid(ST_Collect(ST_Transform(?, 3857))), 4326))",
|
|
point.public_location
|
|
),
|
|
latitude:
|
|
fragment(
|
|
"ST_Y(ST_Transform(ST_Centroid(ST_Collect(ST_Transform(?, 3857))), 4326))",
|
|
point.public_location
|
|
),
|
|
cell_x:
|
|
fragment(
|
|
"ST_X(ST_SnapToGrid(ST_Transform(?, 3857), ?))",
|
|
point.public_location,
|
|
constant(^cell_size)
|
|
),
|
|
cell_y:
|
|
fragment(
|
|
"ST_Y(ST_SnapToGrid(ST_Transform(?, 3857), ?))",
|
|
point.public_location,
|
|
constant(^cell_size)
|
|
)
|
|
})
|
|
|> order_by(
|
|
[point],
|
|
asc:
|
|
fragment(
|
|
"ST_X(ST_SnapToGrid(ST_Transform(?, 3857), ?))",
|
|
point.public_location,
|
|
constant(^cell_size)
|
|
),
|
|
asc:
|
|
fragment(
|
|
"ST_Y(ST_SnapToGrid(ST_Transform(?, 3857), ?))",
|
|
point.public_location,
|
|
constant(^cell_size)
|
|
)
|
|
)
|
|
|> Repo.all()
|
|
|> Enum.map(&map_discovery_item(&1, viewport.zoom))
|
|
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
|
|
spatial_condition =
|
|
Enum.reduce(DiscoveryViewport.envelopes(viewport), dynamic(false), fn
|
|
{west, south, east, north}, condition ->
|
|
dynamic(
|
|
[activity],
|
|
^condition or
|
|
fragment(
|
|
"ST_Intersects(?, ST_MakeEnvelope(?, ?, ?, ?, 4326))",
|
|
activity.location,
|
|
^west,
|
|
^south,
|
|
^east,
|
|
^north
|
|
)
|
|
)
|
|
end)
|
|
|
|
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.exact_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, _zoom) do
|
|
%{
|
|
type: "activity",
|
|
id: row.activity_id,
|
|
title: row.title,
|
|
location: row.location_label,
|
|
latitude: row.latitude,
|
|
longitude: row.longitude,
|
|
exact: false
|
|
}
|
|
end
|
|
|
|
defp map_discovery_item(row, zoom) do
|
|
digest =
|
|
:crypto.hash(:sha256, "activity:#{zoom}:#{row.cell_x}:#{row.cell_y}")
|
|
|> Base.url_encode64(padding: false)
|
|
|> binary_part(0, 16)
|
|
|
|
%{
|
|
type: "cluster",
|
|
id: "activity-cluster-#{digest}",
|
|
count: row.count,
|
|
latitude: row.latitude,
|
|
longitude: row.longitude
|
|
}
|
|
end
|
|
|
|
defp escape_like(value) do
|
|
value
|
|
|> String.replace("\\", "\\\\")
|
|
|> String.replace("%", "\\%")
|
|
|> String.replace("_", "\\_")
|
|
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
|