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