bfc77d1f388c03d5c0d50ad8cbdd16720753b1f3-PR-1437

Committed 26 Jun 2025 06:13AM UTC coverage: 84.769% (+0.1%) from 84.636%

Build # bfc77d1f388c03d5c0d50ad8cbdd16720753b1f3-PR-1437

Build Type

Pull #1437

github

Committed by

edgurgel

Commit Message

chore: bump version

Pull Request Pull Request #1437: fix: use gen_rpc on MetricsController

Run Details

13 of 13 new or added lines in 1 file covered. (100.0%)

6 existing lines in 3 files now uncovered.

1909 of 2252 relevant lines covered (84.77%)

4739.49 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

100.0

/lib/realtime/api.ex

defmodule Realtime.Api do
  @moduledoc """
  The Api context.
  """
  require Logger

  import Ecto.Query

  alias Realtime.Api.Extensions
  alias Realtime.Api.Tenant
  alias Realtime.GenCounter
  alias Realtime.RateCounter
  alias Realtime.Repo
  alias Realtime.Repo.Replica
  alias Realtime.Tenants
  alias Realtime.Tenants.Connect
  alias RealtimeWeb.SocketDisconnect

  defguard requires_disconnect(changeset)
           when changeset.valid? == true and
                  (is_map_key(changeset.changes, :jwt_secret) or
                     is_map_key(changeset.changes, :jwt_jwks) or
                     is_map_key(changeset.changes, :private_only))

  defguard requires_restarting_db_connection(changeset)
           when changeset.valid? == true and
                  (is_map_key(changeset.changes, :extensions) or
                     is_map_key(changeset.changes, :jwt_secret) or
                     is_map_key(changeset.changes, :jwt_jwks))

  @doc """
  Returns the list of tenants.

  ## Examples

      iex> list_tenants()
      [%Tenant{}, ...]

  """
  def list_tenants do
    repo_replica = Replica.replica()

    Tenant
    |> repo_replica.all()
    |> repo_replica.preload(:extensions)
  end

  @doc """
  Returns list of tenants with filter options:
  * order_by
  * search external id
  * limit
  * ordering (desc / asc)
  """
  def list_tenants(opts) when is_list(opts) do
    repo_replica = Replica.replica()

    field = Keyword.get(opts, :order_by, "inserted_at") |> String.to_atom()
    external_id = Keyword.get(opts, :search)
    limit = Keyword.get(opts, :limit, 50)
    order = Keyword.get(opts, :order, "desc") |> String.to_atom()

    query =
      Tenant
      |> order_by({^order, ^field})
      |> limit(^limit)

    ilike = "#{external_id}%"

    query = if external_id, do: query |> where([t], ilike(t.external_id, ^ilike)), else: query

    query
    |> repo_replica.all()
    |> repo_replica.preload(:extensions)
  end

  @doc """
  Gets a single tenant.

  Raises `Ecto.NoResultsError` if the Tenant does not exist.

  ## Examples

      iex> _by_host!(123) do

      end

      %Tenant{}

      iex> get_tenant!(456)
      ** (Ecto.NoResultsError)

  """
  def get_tenant!(id), do: Replica.replica().get!(Tenant, id)

  @doc """
  Creates a tenant.

  ## Examples

      iex> create_tenant(%{field: value})
      {:ok, %Tenant{}}

      iex> create_tenant(%{field: bad_value})
      {:error, %Ecto.Changeset{}}

  """
  def create_tenant(attrs) do
    Logger.debug("create_tenant #{inspect(attrs, pretty: true)}")

    %Tenant{}
    |> Tenant.changeset(attrs)
    |> Repo.insert()
  end

  @doc """
  Updates a tenant.

  ## Examples

      iex> update_tenant(tenant, %{field: new_value})
      {:ok, %Tenant{}}

      iex> update_tenant(tenant, %{field: bad_value})
      {:error, %Ecto.Changeset{}}

  """
  def update_tenant(%Tenant{} = tenant, attrs) do
    changeset = Tenant.changeset(tenant, attrs)
    updated = Repo.update(changeset)

    case updated do
      {:ok, tenant} ->
        maybe_invalidate_cache(changeset)
        maybe_trigger_disconnect(changeset)
        maybe_restart_db_connection(changeset)
        Logger.debug("Tenant updated: #{inspect(tenant, pretty: true)}")

      {:error, error} ->
        Logger.error("Failed to update tenant: #{inspect(error, pretty: true)}")
    end

    updated
  end

  @doc """
  Deletes a tenant.

  ## Examples

      iex> delete_tenant(tenant)
      {:ok, %Tenant{}}

      iex> delete_tenant(tenant)
      {:error, %Ecto.Changeset{}}

  """
  def delete_tenant(%Tenant{} = tenant) do
    Repo.delete(tenant)
  end

  @spec delete_tenant_by_external_id(String.t()) :: boolean()
  def delete_tenant_by_external_id(id) do
    from(t in Tenant, where: t.external_id == ^id)
    |> Repo.delete_all()
    |> case do
      {num, _} when num > 0 ->
        true

      _ ->
        false
    end
  end

  @doc """
  Returns an `%Ecto.Changeset{}` for tracking tenant changes.

  ## Examples

      iex> change_tenant(tenant)
      %Ecto.Changeset{data: %Tenant{}}

  """
  def change_tenant(%Tenant{} = tenant, attrs \\ %{}) do
    Tenant.changeset(tenant, attrs)
  end

  @spec get_tenant_by_external_id(String.t(), atom()) :: Tenant.t() | nil
  def get_tenant_by_external_id(external_id, repo \\ :replica)
      when repo in [:primary, :replica] do
    repo =
      case repo do
        :primary -> Repo
        :replica -> Replica.replica()
      end

    Tenant
    |> repo.get_by(external_id: external_id)
    |> repo.preload(:extensions)
  end

  def list_extensions(type \\ "postgres_cdc_rls") do
    from(e in Extensions,
      where: e.type == ^type,
      select: e
    )
    |> Replica.replica().all()
  end

  def rename_settings_field(from, to) do
    for extension <- list_extensions("postgres_cdc_rls") do
      {value, settings} = Map.pop(extension.settings, from)
      new_settings = Map.put(settings, to, value)

      extension
      |> Ecto.Changeset.cast(%{settings: new_settings}, [:settings])
      |> Repo.update!()
    end
  end

  def preload_counters(nil) do
    nil
  end

  def preload_counters(%Tenant{} = tenant) do
    id = Tenants.requests_per_second_key(tenant)

    preload_counters(tenant, id)
  end

  def preload_counters(nil, _key) do
    nil
  end

  def preload_counters(%Tenant{} = tenant, counters_key) do
    {:ok, current} = GenCounter.get(counters_key)
    {:ok, %RateCounter{avg: avg}} = RateCounter.get(counters_key)

    tenant
    |> Map.put(:events_per_second_rolling, avg)
    |> Map.put(:events_per_second_now, current)
  end

  defp maybe_invalidate_cache(
         %Ecto.Changeset{changes: changes, valid?: true, data: %{external_id: external_id}} = changeset
       )
       when changes != %{} and requires_restarting_db_connection(changeset) do
    Tenants.Cache.distributed_invalidate_tenant_cache(external_id)
  end

  defp maybe_invalidate_cache(_changeset), do: nil

  defp maybe_trigger_disconnect(%Ecto.Changeset{data: %{external_id: external_id}} = changeset)
       when requires_disconnect(changeset) do
    SocketDisconnect.distributed_disconnect(external_id)
  end

  defp maybe_trigger_disconnect(_changeset), do: nil

  defp maybe_restart_db_connection(%Ecto.Changeset{data: %{external_id: external_id}} = changeset)
       when requires_restarting_db_connection(changeset) do
    Connect.shutdown(external_id)
  end

  defp maybe_restart_db_connection(_changeset), do: nil
end

1	defmodule Realtime.Api do
2	@moduledoc """
3	The Api context.
4	"""
5	require Logger
6
7	import Ecto.Query
8
9	alias Realtime.Api.Extensions
10	alias Realtime.Api.Tenant
11	alias Realtime.GenCounter
12	alias Realtime.RateCounter
13	alias Realtime.Repo
14	alias Realtime.Repo.Replica
15	alias Realtime.Tenants
16	alias Realtime.Tenants.Connect
17	alias RealtimeWeb.SocketDisconnect
18
19	defguard requires_disconnect(changeset)	2✔
20	when changeset.valid? == true and
21	(is_map_key(changeset.changes, :jwt_secret) or
22	is_map_key(changeset.changes, :jwt_jwks) or
23	is_map_key(changeset.changes, :private_only))
24
25	defguard requires_restarting_db_connection(changeset)	2✔
26	when changeset.valid? == true and
27	(is_map_key(changeset.changes, :extensions) or
28	is_map_key(changeset.changes, :jwt_secret) or
29	is_map_key(changeset.changes, :jwt_jwks))
30
31	@doc """
32	Returns the list of tenants.
33
34	## Examples
35
36	iex> list_tenants()
37	[%Tenant{}, ...]
38
39	"""
40	def list_tenants do
41	repo_replica = Replica.replica()	70✔
42
43	Tenant
44	\|> repo_replica.all()
45	\|> repo_replica.preload(:extensions)	70✔
46	end
47
48	@doc """
49	Returns list of tenants with filter options:
50	* order_by
51	* search external id
52	* limit
53	* ordering (desc / asc)
54	"""
55	def list_tenants(opts) when is_list(opts) do
56	repo_replica = Replica.replica()	12✔
57
58	field = Keyword.get(opts, :order_by, "inserted_at") \|> String.to_atom()	12✔
59	external_id = Keyword.get(opts, :search)	12✔
60	limit = Keyword.get(opts, :limit, 50)	12✔
61	order = Keyword.get(opts, :order, "desc") \|> String.to_atom()	12✔
62
63	query =	12✔
64	Tenant
65	\|> order_by({^order, ^field})
66	\|> limit(^limit)	12✔
67
68	ilike = "#{external_id}%"	12✔
69
70	query = if external_id, do: query \|> where([t], ilike(t.external_id, ^ilike)), else: query	12✔
71
72	query
73	\|> repo_replica.all()
74	\|> repo_replica.preload(:extensions)	12✔
75	end
76
77	@doc """
78	Gets a single tenant.
79
80	Raises `Ecto.NoResultsError` if the Tenant does not exist.
81
82	## Examples
83
84	iex> _by_host!(123) do
85
86	end
87
88	%Tenant{}
89
90	iex> get_tenant!(456)
91	** (Ecto.NoResultsError)
92
93	"""
94	def get_tenant!(id), do: Replica.replica().get!(Tenant, id)	6✔
95
96	@doc """
97	Creates a tenant.
98
99	## Examples
100
101	iex> create_tenant(%{field: value})
102	{:ok, %Tenant{}}
103
104	iex> create_tenant(%{field: bad_value})
105	{:error, %Ecto.Changeset{}}
106
107	"""
108	def create_tenant(attrs) do
109	Logger.debug("create_tenant #{inspect(attrs, pretty: true)}")	666✔
110
111	%Tenant{}
112	\|> Tenant.changeset(attrs)
113	\|> Repo.insert()	666✔
114	end
115
116	@doc """
117	Updates a tenant.
118
119	## Examples
120
121	iex> update_tenant(tenant, %{field: new_value})
122	{:ok, %Tenant{}}
123
124	iex> update_tenant(tenant, %{field: bad_value})
125	{:error, %Ecto.Changeset{}}
126
127	"""
128	def update_tenant(%Tenant{} = tenant, attrs) do
129	changeset = Tenant.changeset(tenant, attrs)	774✔
130	updated = Repo.update(changeset)	774✔
131
132	case updated do	774✔
133	{:ok, tenant} ->
134	maybe_invalidate_cache(changeset)	772✔
135	maybe_trigger_disconnect(changeset)	772✔
136	maybe_restart_db_connection(changeset)	772✔
137	Logger.debug("Tenant updated: #{inspect(tenant, pretty: true)}")	772✔
138
139	{:error, error} ->
140	Logger.error("Failed to update tenant: #{inspect(error, pretty: true)}")	2✔
141	end
142
143	updated	774✔
144	end
145
146	@doc """
147	Deletes a tenant.
148
149	## Examples
150
151	iex> delete_tenant(tenant)
152	{:ok, %Tenant{}}
153
154	iex> delete_tenant(tenant)
155	{:error, %Ecto.Changeset{}}
156
157	"""
158	def delete_tenant(%Tenant{} = tenant) do
159	Repo.delete(tenant)	2✔
160	end
161
162	@spec delete_tenant_by_external_id(String.t()) :: boolean()
163	def delete_tenant_by_external_id(id) do
164	from(t in Tenant, where: t.external_id == ^id)
165	\|> Repo.delete_all()
166	\|> case do	8✔
167	{num, _} when num > 0 ->	6✔
168	true
169
170	_ ->	2✔
171	false
172	end
173	end
174
175	@doc """
176	Returns an `%Ecto.Changeset{}` for tracking tenant changes.
177
178	## Examples
179
180	iex> change_tenant(tenant)
181	%Ecto.Changeset{data: %Tenant{}}
182
183	"""
184	def change_tenant(%Tenant{} = tenant, attrs \\ %{}) do
185	Tenant.changeset(tenant, attrs)	2✔
186	end
187
188	@spec get_tenant_by_external_id(String.t(), atom()) :: Tenant.t() \| nil
189	def get_tenant_by_external_id(external_id, repo \\ :replica)	182✔
190	when repo in [:primary, :replica] do
191	repo =	202✔
192	case repo do
193	:primary -> Repo	20✔
194	:replica -> Replica.replica()	182✔
195	end
196
197	Tenant
198	\|> repo.get_by(external_id: external_id)
199	\|> repo.preload(:extensions)	202✔
200	end
201
202	def list_extensions(type \\ "postgres_cdc_rls") do
203	from(e in Extensions,
204	where: e.type == ^type,
205	select: e
206	)
207	\|> Replica.replica().all()	4✔
208	end
209
210	def rename_settings_field(from, to) do
211	for extension <- list_extensions("postgres_cdc_rls") do	2✔
UNCOV 212	{value, settings} = Map.pop(extension.settings, from)	4✔
UNCOV 213	new_settings = Map.put(settings, to, value)	4✔
214
215	extension
216	\|> Ecto.Changeset.cast(%{settings: new_settings}, [:settings])
UNCOV 217	\|> Repo.update!()	4✔
218	end
219	end
220
221	def preload_counters(nil) do	2✔
222	nil
223	end
224
225	def preload_counters(%Tenant{} = tenant) do
226	id = Tenants.requests_per_second_key(tenant)	36✔
227
228	preload_counters(tenant, id)	36✔
229	end
230
231	def preload_counters(nil, _key) do	2✔
232	nil
233	end
234
235	def preload_counters(%Tenant{} = tenant, counters_key) do
236	{:ok, current} = GenCounter.get(counters_key)	36✔
237	{:ok, %RateCounter{avg: avg}} = RateCounter.get(counters_key)	36✔
238
239	tenant
240	\|> Map.put(:events_per_second_rolling, avg)
241	\|> Map.put(:events_per_second_now, current)	36✔
242	end
243
244	defp maybe_invalidate_cache(
245	%Ecto.Changeset{changes: changes, valid?: true, data: %{external_id: external_id}} = changeset
246	)
247	when changes != %{} and requires_restarting_db_connection(changeset) do
248	Tenants.Cache.distributed_invalidate_tenant_cache(external_id)	44✔
249	end
250
251	defp maybe_invalidate_cache(_changeset), do: nil	728✔
252
253	defp maybe_trigger_disconnect(%Ecto.Changeset{data: %{external_id: external_id}} = changeset)
254	when requires_disconnect(changeset) do
255	SocketDisconnect.distributed_disconnect(external_id)	14✔
256	end
257
258	defp maybe_trigger_disconnect(_changeset), do: nil	758✔
259
260	defp maybe_restart_db_connection(%Ecto.Changeset{data: %{external_id: external_id}} = changeset)
261	when requires_restarting_db_connection(changeset) do
262	Connect.shutdown(external_id)	44✔
263	end
264
265	defp maybe_restart_db_connection(_changeset), do: nil	728✔
266	end

supabase / realtime / bfc77d1f388c03d5c0d50ad8cbdd16720753b1f3-PR-1437

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous