442 lines
14 KiB
Elixir
442 lines
14 KiB
Elixir
defmodule WhoNeedHelp.Tracking do
|
|
@moduledoc """
|
|
Consent-driven active-match tracking.
|
|
|
|
Browsers share only while the LiveView remains open. The Android client can
|
|
continue through a user-started foreground service with a persistent
|
|
notification. Only the current position is retained while a session is
|
|
active.
|
|
"""
|
|
|
|
import Ecto.Query
|
|
alias WhoNeedHelp.Accounts
|
|
alias WhoNeedHelp.Accounts.Scope
|
|
alias WhoNeedHelp.Help
|
|
alias WhoNeedHelp.Help.Assignment
|
|
alias WhoNeedHelp.Repo
|
|
alias WhoNeedHelp.Tracking.{Position, TrackingSession}
|
|
alias WhoNeedHelp.Trust
|
|
|
|
def subscribe(assignment_id) do
|
|
Phoenix.PubSub.subscribe(WhoNeedHelp.PubSub, "tracking:#{assignment_id}")
|
|
end
|
|
|
|
def active_session?(%Scope{user: user} = scope, %Assignment{} = assignment),
|
|
do: not is_nil(active_session(scope, assignment, user.id))
|
|
|
|
def active_session(%Scope{user: user} = scope, %Assignment{} = assignment),
|
|
do: active_session(scope, assignment, user.id)
|
|
|
|
def start_session(
|
|
%Scope{user: user} = scope,
|
|
%Assignment{} = assignment,
|
|
visibility \\ :active_match
|
|
) do
|
|
with {:ok, _limit} <- Trust.authorize_action(scope, :start_tracking) do
|
|
Repo.transact(fn ->
|
|
with {:ok, current} <- active_assignment(scope, assignment.id, lock: true) do
|
|
now = DateTime.utc_now(:second)
|
|
|
|
%TrackingSession{}
|
|
|> TrackingSession.changeset(%{
|
|
assignment_id: current.id,
|
|
user_id: user.id,
|
|
visibility: visibility,
|
|
started_at: now
|
|
})
|
|
|> Repo.insert(
|
|
on_conflict: [
|
|
set: [active: true, ended_at: nil, started_at: now, visibility: visibility]
|
|
],
|
|
conflict_target: {:unsafe_fragment, "(assignment_id, user_id) WHERE active"}
|
|
)
|
|
end
|
|
end)
|
|
end
|
|
end
|
|
|
|
def update_position(%Scope{user: user} = scope, %Assignment{} = assignment, attrs) do
|
|
with {:ok, _limit} <- Trust.authorize_action(scope, :tracking_position) do
|
|
result =
|
|
Repo.transact(fn ->
|
|
with {:ok, current} <- active_assignment(scope, assignment.id, lock: true) do
|
|
session =
|
|
TrackingSession
|
|
|> where(
|
|
[session],
|
|
session.assignment_id == ^current.id and session.user_id == ^user.id and
|
|
session.active
|
|
)
|
|
|> lock("FOR UPDATE")
|
|
|> Repo.one()
|
|
|
|
if session do
|
|
persist_position(session, current, user, attrs)
|
|
else
|
|
{:error, :tracking_not_active}
|
|
end
|
|
end
|
|
end)
|
|
|
|
with {:ok, {position, evidence}} <- result do
|
|
Phoenix.PubSub.broadcast(
|
|
WhoNeedHelp.PubSub,
|
|
"tracking:#{assignment.id}",
|
|
{:position_updated, user.id, public_position(position), evidence}
|
|
)
|
|
|
|
{:ok, position}
|
|
end
|
|
end
|
|
end
|
|
|
|
def stop_session(%Scope{user: user} = scope, %Assignment{} = assignment) do
|
|
if Help.participant?(scope, assignment) do
|
|
stop_locked_session(assignment.id, user.id)
|
|
else
|
|
{:error, :forbidden}
|
|
end
|
|
end
|
|
|
|
def stop_browser_session(assignment_id, user_id, tracking_session_id) do
|
|
with {:ok, assignment_id} <- Ecto.UUID.cast(assignment_id),
|
|
{:ok, user_id} <- Ecto.UUID.cast(user_id),
|
|
{:ok, tracking_session_id} <- Ecto.UUID.cast(tracking_session_id),
|
|
%Assignment{} = assignment <-
|
|
Assignment
|
|
|> Repo.get(assignment_id)
|
|
|> Repo.preload(:request),
|
|
%Accounts.User{} = user <- Repo.get(Accounts.User, user_id) do
|
|
if Help.participant?(Scope.for_user(user), assignment) do
|
|
stop_locked_session(assignment.id, user.id, tracking_session_id)
|
|
else
|
|
{:error, :forbidden}
|
|
end
|
|
else
|
|
_missing_or_invalid -> {:ok, :already_stopped}
|
|
end
|
|
end
|
|
|
|
def list_current_positions(%Scope{} = scope, %Assignment{} = assignment) do
|
|
with true <- Trust.eligible?(scope),
|
|
{:ok, current} <- active_assignment(scope, assignment.id) do
|
|
TrackingSession
|
|
|> where([s], s.assignment_id == ^current.id and s.active)
|
|
|> join(:inner, [s], p in assoc(s, :position))
|
|
|> select([s, p], {s.user_id, p})
|
|
|> Repo.all()
|
|
|> Map.new(fn {user_id, position} -> {user_id, public_position(position)} end)
|
|
else
|
|
_ -> %{}
|
|
end
|
|
end
|
|
|
|
def cleanup_finished_sessions do
|
|
now = DateTime.utc_now(:second)
|
|
|
|
session_ids =
|
|
TrackingSession
|
|
|> join(:inner, [session], assignment in Assignment,
|
|
on: assignment.id == session.assignment_id
|
|
)
|
|
|> where(
|
|
[session, assignment],
|
|
session.active and assignment.status in [:completed, :cancelled]
|
|
)
|
|
|> select([session], session.id)
|
|
|
|
Repo.transaction(fn ->
|
|
{deleted_positions, _} =
|
|
Position
|
|
|> where([position], position.tracking_session_id in subquery(session_ids))
|
|
|> Repo.delete_all()
|
|
|
|
{ended_sessions, _} =
|
|
TrackingSession
|
|
|> where([session], session.id in subquery(session_ids))
|
|
|> Repo.update_all(set: [active: false, ended_at: now, updated_at: now])
|
|
|
|
%{positions_deleted: deleted_positions, sessions_ended: ended_sessions}
|
|
end)
|
|
end
|
|
|
|
def stop_all_sessions(user_id) do
|
|
result =
|
|
Repo.transact(fn ->
|
|
active_sessions =
|
|
TrackingSession
|
|
|> where([session], session.active and session.user_id == ^user_id)
|
|
|> select([session], {session.id, session.assignment_id})
|
|
|> lock("FOR UPDATE")
|
|
|> Repo.all()
|
|
|
|
session_ids = Enum.map(active_sessions, &elem(&1, 0))
|
|
|
|
if session_ids != [] do
|
|
now = DateTime.utc_now(:second)
|
|
|
|
Position
|
|
|> where([position], position.tracking_session_id in ^session_ids)
|
|
|> Repo.delete_all()
|
|
|
|
TrackingSession
|
|
|> where([session], session.id in ^session_ids)
|
|
|> Repo.update_all(set: [active: false, ended_at: now, updated_at: now])
|
|
end
|
|
|
|
{:ok, Enum.map(active_sessions, &elem(&1, 1))}
|
|
end)
|
|
|
|
with {:ok, assignment_ids} <- result do
|
|
Enum.each(assignment_ids, fn assignment_id ->
|
|
Phoenix.PubSub.broadcast(
|
|
WhoNeedHelp.PubSub,
|
|
"tracking:#{assignment_id}",
|
|
{:tracking_stopped, user_id}
|
|
)
|
|
end)
|
|
|
|
{:ok, %{sessions_ended: length(assignment_ids)}}
|
|
end
|
|
end
|
|
|
|
defp active_session(scope, assignment, user_id) do
|
|
with {:ok, current} <- active_assignment(scope, assignment.id) do
|
|
Repo.get_by(TrackingSession,
|
|
assignment_id: current.id,
|
|
user_id: user_id,
|
|
active: true
|
|
)
|
|
else
|
|
_ -> nil
|
|
end
|
|
end
|
|
|
|
defp active_assignment(%Scope{user: user} = scope, assignment_id, options \\ []) do
|
|
query =
|
|
Assignment
|
|
|> where([assignment], assignment.id == ^assignment_id)
|
|
|> preload(:request)
|
|
|
|
query = if Keyword.get(options, :lock, false), do: lock(query, "FOR UPDATE"), else: query
|
|
|
|
case Repo.one(query) do
|
|
nil ->
|
|
{:error, :not_found}
|
|
|
|
%Assignment{} = assignment ->
|
|
request = assignment.request
|
|
|
|
counterpart_id =
|
|
if user.id == assignment.helper_id, do: request.requester_id, else: assignment.helper_id
|
|
|
|
if Keyword.get(options, :lock, false) do
|
|
:ok = Trust.lock_user_pair(user.id, counterpart_id)
|
|
end
|
|
|
|
cond do
|
|
not Help.participant?(scope, assignment) ->
|
|
{:error, :forbidden}
|
|
|
|
assignment.status not in [:accepted, :in_progress] ->
|
|
{:error, :assignment_inactive}
|
|
|
|
Trust.blocked_between?(user.id, counterpart_id) ->
|
|
{:error, :blocked}
|
|
|
|
true ->
|
|
{:ok, assignment}
|
|
end
|
|
end
|
|
end
|
|
|
|
defp stop_locked_session(assignment_id, user_id, tracking_session_id \\ nil) do
|
|
result =
|
|
Repo.transact(fn ->
|
|
query =
|
|
from session in TrackingSession,
|
|
where:
|
|
session.assignment_id == ^assignment_id and session.user_id == ^user_id and
|
|
session.active,
|
|
lock: "FOR UPDATE"
|
|
|
|
query =
|
|
if tracking_session_id,
|
|
do: where(query, [session], session.id == ^tracking_session_id),
|
|
else: query
|
|
|
|
case Repo.one(query) do
|
|
nil ->
|
|
{:ok, :already_stopped}
|
|
|
|
session ->
|
|
Position
|
|
|> where([position], position.tracking_session_id == ^session.id)
|
|
|> Repo.delete_all()
|
|
|
|
now = DateTime.utc_now(:second)
|
|
|
|
case Repo.update(TrackingSession.changeset(session, %{active: false, ended_at: now})) do
|
|
{:ok, _session} -> {:ok, :stopped}
|
|
{:error, changeset} -> {:error, changeset}
|
|
end
|
|
end
|
|
end)
|
|
|
|
case result do
|
|
{:ok, :stopped} ->
|
|
Phoenix.PubSub.broadcast(
|
|
WhoNeedHelp.PubSub,
|
|
"tracking:#{assignment_id}",
|
|
{:tracking_stopped, user_id}
|
|
)
|
|
|
|
{:ok, :stopped}
|
|
|
|
{:ok, :already_stopped} ->
|
|
{:ok, :already_stopped}
|
|
|
|
{:error, reason} ->
|
|
{:error, reason}
|
|
end
|
|
end
|
|
|
|
defp public_position(%Position{
|
|
position: %Geo.Point{coordinates: {lng, lat}},
|
|
accuracy_meters: accuracy,
|
|
captured_at: captured_at
|
|
}) do
|
|
%{latitude: lat, longitude: lng, accuracy: accuracy, captured_at: captured_at}
|
|
end
|
|
|
|
defp persist_position(session, assignment, user, attrs) do
|
|
previous =
|
|
Position
|
|
|> where([position], position.tracking_session_id == ^session.id)
|
|
|> Repo.one()
|
|
|
|
now = DateTime.utc_now(:second)
|
|
attrs = Map.put(attrs, "captured_at", now)
|
|
|
|
with {:ok, position} <-
|
|
%Position{tracking_session_id: session.id}
|
|
|> Position.changeset(attrs)
|
|
|> Repo.insert(
|
|
on_conflict: {:replace, [:position, :accuracy_meters, :captured_at, :updated_at]},
|
|
conflict_target: :tracking_session_id,
|
|
returning: true
|
|
) do
|
|
movement = movement_evidence(previous, position)
|
|
movement_at = session.movement_observed_at || if(movement > 0, do: now)
|
|
|
|
{:ok, session} =
|
|
session
|
|
|> TrackingSession.changeset(%{
|
|
sample_count: session.sample_count + 1,
|
|
distance_meters: session.distance_meters + movement,
|
|
movement_observed_at: movement_at
|
|
})
|
|
|> Repo.update()
|
|
|
|
maybe_mark_helper_movement(assignment, user.id, session.movement_observed_at, now)
|
|
maybe_mark_proximity(assignment, session, position, now)
|
|
evidence = assignment_evidence(assignment.id)
|
|
|
|
{:ok, {position, evidence}}
|
|
end
|
|
end
|
|
|
|
defp maybe_mark_helper_movement(assignment, user_id, movement_observed_at, now) do
|
|
if assignment.helper_id == user_id and not is_nil(movement_observed_at) and
|
|
is_nil(assignment.helper_movement_observed_at) do
|
|
Assignment
|
|
|> where(
|
|
[candidate],
|
|
candidate.id == ^assignment.id and
|
|
is_nil(candidate.helper_movement_observed_at)
|
|
)
|
|
|> Repo.update_all(set: [helper_movement_observed_at: now, updated_at: now])
|
|
end
|
|
|
|
:ok
|
|
end
|
|
|
|
defp maybe_mark_proximity(%Assignment{proximity_observed_at: observed}, _, _, _)
|
|
when not is_nil(observed),
|
|
do: :ok
|
|
|
|
defp maybe_mark_proximity(assignment, session, position, now) do
|
|
counterpart_positions =
|
|
TrackingSession
|
|
|> where(
|
|
[candidate],
|
|
candidate.assignment_id == ^assignment.id and candidate.active and
|
|
candidate.user_id != ^session.user_id
|
|
)
|
|
|> join(:inner, [candidate], point in assoc(candidate, :position))
|
|
|> select([_candidate, point], point)
|
|
|> Repo.all()
|
|
|
|
if Enum.any?(counterpart_positions, &accuracy_envelopes_overlap?(&1, position)) do
|
|
Assignment
|
|
|> where(
|
|
[candidate],
|
|
candidate.id == ^assignment.id and is_nil(candidate.proximity_observed_at)
|
|
)
|
|
|> Repo.update_all(set: [proximity_observed_at: now, updated_at: now])
|
|
end
|
|
|
|
:ok
|
|
end
|
|
|
|
defp assignment_evidence(assignment_id) do
|
|
Assignment
|
|
|> where([assignment], assignment.id == ^assignment_id)
|
|
|> select([assignment], %{
|
|
helper_movement_observed_at: assignment.helper_movement_observed_at,
|
|
proximity_observed_at: assignment.proximity_observed_at
|
|
})
|
|
|> Repo.one!()
|
|
end
|
|
|
|
defp movement_evidence(nil, _current), do: 0.0
|
|
|
|
defp movement_evidence(previous, current) do
|
|
with previous_accuracy when is_number(previous_accuracy) <- previous.accuracy_meters,
|
|
current_accuracy when is_number(current_accuracy) <- current.accuracy_meters do
|
|
max(distance(previous, current) - previous_accuracy - current_accuracy, 0.0)
|
|
else
|
|
_ -> 0.0
|
|
end
|
|
end
|
|
|
|
defp accuracy_envelopes_overlap?(first, second) do
|
|
with first_accuracy when is_number(first_accuracy) <- first.accuracy_meters,
|
|
second_accuracy when is_number(second_accuracy) <- second.accuracy_meters do
|
|
distance(first, second) <= first_accuracy + second_accuracy
|
|
else
|
|
_ -> false
|
|
end
|
|
end
|
|
|
|
defp distance(
|
|
%Position{position: %Geo.Point{coordinates: {first_lng, first_lat}}},
|
|
%Position{position: %Geo.Point{coordinates: {second_lng, second_lat}}}
|
|
) do
|
|
earth_radius_meters = 6_371_008.8
|
|
latitude_delta = radians(second_lat - first_lat)
|
|
longitude_delta = radians(second_lng - first_lng)
|
|
|
|
a =
|
|
:math.sin(latitude_delta / 2) ** 2 +
|
|
:math.cos(radians(first_lat)) * :math.cos(radians(second_lat)) *
|
|
:math.sin(longitude_delta / 2) ** 2
|
|
|
|
a = a |> max(0.0) |> min(1.0)
|
|
|
|
earth_radius_meters * 2 * :math.atan2(:math.sqrt(a), :math.sqrt(1 - a))
|
|
end
|
|
|
|
defp radians(degrees), do: degrees * :math.pi() / 180
|
|
end
|