defmodule PresenceEx do @moduledoc false # credo:disable-for-this-file Credo.Check.Refactor.PipeChainStart import Ecto.Query alias PresenceEx.Repo alias PresenceEx.Schemas.Status alias PresenceEx.Utils # alias PresenceEx.Schemas.Subscribe # alias PresenceEx.Schemas.BeSubscribed alias PresenceEx.Schemas.StatusSubscribe require Logger def insert(account, resource, node, status, features) do %Status{} |> Status.changeset(%{ account: account, resource: resource, node: node, status: status, features: features }) |> Repo.insert(prefix: Utils.prefix(account)) end def update(account, resource, t) do from(Status, where: [account: ^account, resource: ^resource]) |> Repo.update_all([set: t], prefix: Utils.prefix(account)) end def upsert(account, resource, node, status, features) do %Status{} |> Status.changeset(%{ account: account, resource: resource, node: node, status: status, features: features }) |> Repo.insert( prefix: Utils.prefix(account), on_conflict: {:replace, [:node, :status, :features]}, conflict_target: [:account, :resource] ) end @spec get(binary | integer, any) :: any def get(account, resource) do from(Status, where: [account: ^account, resource: ^resource]) |> Repo.replica().all(prefix: Utils.prefix(account)) end def get(account, resource, statuses) do case length(statuses) > 2 do true -> [] false -> case statuses do [] -> [] [s] -> from(Status, where: [account: ^account, resource: ^resource, status: ^s]) |> Repo.replica().all(prefix: Utils.prefix(account)) sl -> [s_exclude] = ["online", "away", "offline"] -- sl from(s in Status, where: s.account == ^account and s.resource == ^resource and s.status != ^s_exclude ) |> Repo.replica().all(prefix: Utils.prefix(account)) end end end def get_by_user(account) do from(Status, where: [account: ^account]) |> Repo.replica().all(prefix: Utils.prefix(account)) end def get_by_user(account, statuses) do case length(statuses) > 2 do true -> [] false -> case statuses do [] -> [] [s] -> from(Status, where: [account: ^account, status: ^s]) |> Repo.replica().all(prefix: Utils.prefix(account)) sl -> [s_exclude] = ["online", "away", "offline"] -- sl from(s in Status, where: s.account == ^account and s.status != ^s_exclude) |> Repo.replica().all(prefix: Utils.prefix(account)) end end end def get_by_node(node) do query = from(Status, where: [node: ^node]) Utils.all_schema_ids() |> Enum.flat_map(&Repo.replica().all(query, prefix: Utils.prefix(&1))) end def delete(account, resource) do from(Status, where: [account: ^account, resource: ^resource]) |> Repo.delete_all(prefix: Utils.prefix(account)) end def delete_by_user(account) do from(Status, where: [account: ^account]) |> Repo.delete_all(prefix: Utils.prefix(account)) end def delete_by_node(node) do query = from(Status, where: [node: ^node]) Utils.all_schema_ids() |> Enum.map(&Repo.delete_all(query, prefix: Utils.prefix(&1))) end def upsert_subscribe(subscriber, resource, subscribed, timestamp) do param = %{ subscriber_token: subscriber, subscriber_resource: resource, timestamp: timestamp, subscribed_token: subscribed } verified_param = StatusSubscribe.verify_input(param) case verified_param.valid? do true -> case Repo.insert(verified_param, prefix: Utils.prefix(subscriber), on_conflict: {:replace, [:timestamp]}, conflict_target: [:subscriber_token, :subscriber_resource, :subscribed_token] ) do {:ok, _e} -> :ok {:error, error} -> Logger.error("error insert_subscribe #{inspect(error)}") :failed end false -> Logger.info("unexpected insert_subscribe error #{inspect(verified_param)}") :failed end end def get_subscribe(subscriber, resource, subscribed) do conditions = [ subscriber_token: subscriber, subscriber_resource: resource, subscribed_token: subscribed ] from(StatusSubscribe, where: ^conditions) |> Repo.replica().all(prefix: Utils.prefix(subscriber)) end def delete_subscribe(subscriber, resource) do from(StatusSubscribe, where: [subscriber_token: ^subscriber, subscriber_resource: ^resource]) |> Repo.delete_all(prefix: Utils.prefix(subscriber)) end def delete_subscribe(subscriber, resource, subscribed) do from(StatusSubscribe, where: [ subscriber_token: ^subscriber, subscriber_resource: ^resource, subscribed_token: ^subscribed ] ) |> Repo.delete_all(prefix: Utils.prefix(subscriber)) end def delete_subscribe(subscribed) do query = from(StatusSubscribe, where: [subscribed_token: ^subscribed]) {Utils.all_schema_ids() |> Enum.reduce( 0, fn id, acc -> {n, nil} = Repo.delete_all(query, prefix: Utils.prefix(id)) acc + n end ), nil} end def get_all_subscribe() do # for debug use, not used in prod!! Utils.all_schema_ids() |> Enum.flat_map(&Repo.replica().all(StatusSubscribe, prefix: Utils.prefix(&1))) end def get_subscribed(subscriber, resource) do conditions = [ subscriber_token: subscriber, subscriber_resource: resource ] from(StatusSubscribe, where: ^conditions) |> Repo.replica().all(prefix: Utils.prefix(subscriber)) end def get_subscriber(subscribed) do query = from(StatusSubscribe, where: [subscribed_token: ^subscribed]) Utils.all_schema_ids() |> Enum.flat_map(&Repo.replica().all(query, prefix: Utils.prefix(&1))) end def start(_, _) do :ok end end