chore: upgrade Elixir to 1.20.2 and Erlang to 27.3.4.13

.tool-versions

@@ -1,5 +1,5 @@
-elixir 1.19.5-otp-27
-erlang 27.3.4.6
+elixir 1.20.2-otp-27
+erlang 27.3.4.13
 nodejs 22.9.0
 protoc 28.0
 rust 1.94.1

Dockerfile

@@ -1,6 +1,6 @@
-ARG ELIXIR_VERSION=1.19.5
-ARG OTP_VERSION=27.3.4.6
-ARG DEBIAN_VERSION=trixie-20260112-slim
+ARG ELIXIR_VERSION=1.20.2
+ARG OTP_VERSION=27.3.4.13
+ARG DEBIAN_VERSION=trixie-20260623-slim
 
 ARG RUST_VERSION=1.94.1
 ARG BUILDER_IMAGE="hexpm/elixir:${ELIXIR_VERSION}-erlang-${OTP_VERSION}-debian-${DEBIAN_VERSION}"

Dockerfile.base

@@ -1,6 +1,6 @@
-ARG ELIXIR_VERSION=1.19.5
-ARG OTP_VERSION=27.3.4.6
-ARG DEBIAN_VERSION=trixie-20260112-slim
+ARG ELIXIR_VERSION=1.20.2
+ARG OTP_VERSION=27.3.4.13
+ARG DEBIAN_VERSION=trixie-20260623-slim
 
 ARG RUST_VERSION=1.94.1
 ARG BUILDER_IMAGE="hexpm/elixir:${ELIXIR_VERSION}-erlang-${OTP_VERSION}-debian-${DEBIAN_VERSION}"

lib/logflare/backends/adaptor/clickhouse_adaptor/ingester.ex

@@ -315,10 +315,6 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.Ingester do
      ]}
   end
 
-  defp build_connection_opts(_backend) do
-    {:error, "Unable to build connection options"}
-  end
-
   @spec build_request_url(
           connection_opts :: Keyword.t(),
           table :: String.t(),

lib/logflare/backends/adaptor/otlp_adaptor/protobuf_formatter.ex

@@ -116,14 +116,14 @@ defmodule Logflare.Backends.Adaptor.OtlpAdaptor.ProtobufFormatter do
   defp build_log_record_fields({"span_id", id}) when is_binary(id), do: [span_id: id]
   defp build_log_record_fields(_unmatched), do: []
 
+  # TODO: distinguish from string
   defp make_value(v) when is_binary(v), do: %AnyValue{value: {:string_value, v}}
+  # defp make_value(v) when is_binary(v), do: %AnyValue{value: {:bytes_value, v}}
   defp make_value(v) when is_boolean(v), do: %AnyValue{value: {:bool_value, v}}
   defp make_value(v) when is_integer(v), do: %AnyValue{value: {:int_value, v}}
   defp make_value(v) when is_float(v), do: %AnyValue{value: {:double_value, v}}
   defp make_value(v) when is_list(v), do: %AnyValue{value: {:array_value, make_array(v)}}
   defp make_value(v) when is_map(v), do: %AnyValue{value: {:kvlist_value, make_key_value_list(v)}}
-  # TODO: distinguish from string
-  defp make_value(v) when is_binary(v), do: %AnyValue{value: {:bytes_value, v}}
 
   defp make_key_value({k, v}), do: %KeyValue{key: to_string(k), value: make_value(v)}
 

lib/logflare/backends/adaptor/postgres_adaptor.ex

@@ -320,7 +320,7 @@ defmodule Logflare.Backends.Adaptor.PostgresAdaptor do
     state = %__MODULE__{
       config: backend.config,
       backend: backend,
-      backend_token: if(backend, do: backend.token, else: nil),
+      backend_token: backend.token,
       source_token: source.token,
       source: source,
       pipeline_name: Backends.via_source(source, Pipeline, backend.id)

lib/logflare/backends/ingest_event_queue/queue_janitor.ex

@@ -193,19 +193,22 @@ defmodule Logflare.Backends.IngestEventQueue.QueueJanitor do
   defp process_stale_events(_table_key, []), do: {0, 0}
 
   defp process_stale_events(table_key, stale_ids) do
-    with tid when tid != nil <- IngestEventQueue.get_tid(table_key) do
-      Enum.reduce(stale_ids, {0, 0}, fn id, {resets, drops} ->
-        case act_on_stale_event(tid, id) do
-          :reset -> {resets + 1, drops}
-          :drop -> {resets, drops + 1}
-          :skip -> {resets, drops}
-        end
-      end)
-    else
-      _ -> {0, 0}
+    case IngestEventQueue.get_tid(table_key) do
+      nil -> {0, 0}
+      tid -> tally_stale_events(tid, stale_ids)
     end
   end
 
+  defp tally_stale_events(tid, stale_ids) do
+    Enum.reduce(stale_ids, {0, 0}, fn id, {resets, drops} ->
+      case act_on_stale_event(tid, id) do
+        :reset -> {resets + 1, drops}
+        :drop -> {resets, drops + 1}
+        :skip -> {resets, drops}
+      end
+    end)
+  end
+
   # Pass the exact row observed by the lookup to IngestEventQueue's CAS helpers, so an event
   # acked, deleted, or re-claimed between lookup and write is neither dropped nor reset (a
   # re-claimed row carries a newer claimed_at and will not match). The :reset/:drop/:skip

lib/logflare/endpoints.ex

@@ -286,17 +286,20 @@ defmodule Logflare.Endpoints do
   @spec maybe_kill_endpoint_caches(EndpointQuery.t(), map()) :: :ok
   defp maybe_kill_endpoint_caches(endpoint, changes) do
     if should_kill_caches?(changes) do
-      for pid <- Resolver.list_caches(endpoint) do
-        Utils.Tasks.async(fn ->
-          ResultsCache.invalidate(pid)
-        end)
-      end
+      endpoint
+      |> Resolver.list_caches()
+      |> Enum.map(&invalidate_cache_async/1)
       |> Task.await_many(30_000)
     end
 
     :ok
   end
 
+  @spec invalidate_cache_async(pid()) :: Task.t()
+  defp invalidate_cache_async(pid) do
+    Utils.Tasks.async(fn -> ResultsCache.invalidate(pid) end)
+  end
+
   @spec get_endpoint_query_version(integer(), integer()) :: Version.t() | nil
   def get_endpoint_query_version(endpoint_id, version_number)
       when is_integer(endpoint_id) and is_integer(version_number) do

lib/logflare/google/bigquery/bigquery.ex

@@ -67,15 +67,11 @@ defmodule Logflare.Google.BigQuery do
     conn = GenUtils.get_conn()
     table_name = GenUtils.format_table_name(source_id)
 
-    %{user_id: user_id, bigquery_project_id: project_id, bigquery_dataset_id: dataset_id} =
+    %{user_id: _user_id, bigquery_project_id: project_id, bigquery_dataset_id: dataset_id} =
       GenUtils.get_bq_user_info(source_id)
 
     conn
-    |> Api.Tables.bigquery_tables_delete(
-      project_id,
-      dataset_id || Integer.to_string(user_id) <> GCPConfig.dataset_id_append(),
-      table_name
-    )
+    |> Api.Tables.bigquery_tables_delete(project_id, dataset_id, table_name)
     |> GenUtils.maybe_parse_google_api_result()
   end
 
@@ -186,14 +182,8 @@ defmodule Logflare.Google.BigQuery do
       bigquery_dataset_id: dataset_id
     } = GenUtils.get_bq_user_info(source_id)
 
-    dataset_id = dataset_id || GenUtils.get_account_id(source_id) <> GCPConfig.dataset_id_append()
-
     conn
-    |> Api.Tables.bigquery_tables_get(
-      project_id,
-      dataset_id,
-      table_name
-    )
+    |> Api.Tables.bigquery_tables_get(project_id, dataset_id, table_name)
     |> GenUtils.maybe_parse_google_api_result()
   end
 
@@ -307,14 +297,14 @@ defmodule Logflare.Google.BigQuery do
     emails =
       for x <- [user | team_users], x.provider == "google", do: x.email
 
-    if Enum.count(user.sources) > 0 do
+    if Enum.empty?(user.sources) do
+      {:ok, :nothing_patched}
+    else
       Tasks.start_child(fn ->
         patch(dataset_id, emails, project_id, user.id)
       end)
 
       {:ok, :patch_attempted}
-    else
-      {:ok, :nothing_patched}
     end
   end
 

lib/logflare/logs/rejected_log_events.ex

@@ -53,11 +53,9 @@ defmodule Logflare.Logs.RejectedLogEvents do
 
   @spec get_by_source(Source.t()) :: list(LE.t())
   def get_by_source(%Source{token: token}) do
-    get!(token)
-    |> case do
-      %{log_events: events} -> events
-      other -> other
-    end
+    token
+    |> get!()
+    |> Map.fetch!(:log_events)
   end
 
   def count(%Source{} = s) do

lib/logflare/lql/backend_transformer/bigquery.ex

@@ -65,7 +65,6 @@ defmodule Logflare.Lql.BackendTransformer.BigQuery do
       end
 
     case normalized_rules do
-      [] -> query
       [%{wildcard: true}] -> query
       rules -> build_combined_select(query, rules)
     end

lib/logflare/lql/backend_transformer/clickhouse.ex

@@ -66,7 +66,6 @@ defmodule Logflare.Lql.BackendTransformer.ClickHouse do
       end
 
     case normalized_rules do
-      [] -> query
       [%{wildcard: true}] -> query
       rules -> build_combined_select(query, rules)
     end

lib/logflare/lql/backend_transformer/postgres.ex

@@ -687,7 +687,7 @@ defmodule Logflare.Lql.BackendTransformer.Postgres do
   end
 
   @spec build_jsonb_accessor([String.t()]) :: String.t()
-  defp build_jsonb_accessor(path) when is_list(path) and length(path) > 0 do
+  defp build_jsonb_accessor(path) when is_non_empty_list(path) do
     # For ["metadata", "status"]: body->'metadata'->>'status'
     # For ["metadata", "user", "email"]: body->'metadata'->'user'->>'email'
     {last, rest} = List.pop_at(path, -1)

lib/logflare/sources/source/bigquery/pipeline.ex

@@ -85,7 +85,7 @@ defmodule Logflare.Sources.Source.BigQuery.Pipeline do
             source_token: source.token,
             bq_storage_write_api: source.bq_storage_write_api,
             source_id: source.id,
-            backend_id: Map.get(backend || %{}, :id),
+            backend_id: Map.get(backend, :id),
             user_id: source.user_id,
             system_source: source.system_source
           }
@@ -138,14 +138,7 @@ defmodule Logflare.Sources.Source.BigQuery.Pipeline do
 
       source ->
         metrics = Sources.get_source_metrics_for_ingest(source.token)
-
-        for %{data: {id, tid, _size}} <- successful do
-          if metrics.avg > 100 do
-            IngestEventQueue.delete_id(tid, id)
-          else
-            IngestEventQueue.update_status(tid, id, :ingested)
-          end
-        end
+        finalize_acked_events(successful, metrics)
     end
 
     :telemetry.execute(
@@ -157,6 +150,23 @@ defmodule Logflare.Sources.Source.BigQuery.Pipeline do
     :ok
   end
 
+  # avg ingest rate above threshold: drop from the queue rather than mark :ingested
+  # to shed load; below threshold: keep the event and finalize its status.
+  @spec finalize_acked_events([Message.t()], map()) :: :ok
+  defp finalize_acked_events(successful, %{avg: avg}) when avg > 100 do
+    for %{data: {id, tid, _size}} <- successful,
+        do: IngestEventQueue.delete_id(tid, id)
+
+    :ok
+  end
+
+  defp finalize_acked_events(successful, _metrics) do
+    for %{data: {id, tid, _size}} <- successful,
+        do: IngestEventQueue.update_status(tid, id, :ingested)
+
+    :ok
+  end
+
   @impl Broadway
   def handle_message(_processor_name, message, context) do
     Logger.metadata(

lib/logflare/sources/source/slack_hook_server/slack_hook_server.ex

@@ -50,7 +50,7 @@ defmodule Logflare.Sources.Source.SlackHookServer do
            true <- new_count > 0,
            true <- source.slack_hook_url != nil,
            events when is_list(events) <- fetch_events(source, new_count),
-           true <- Enum.count(events) > 0 do
+           true <- not Enum.empty?(events) do
         SlackAdaptor.send_message(source, events, new_count)
         |> handle_response(source)
 

lib/logflare/sources/source/supervisor.ex

@@ -40,11 +40,6 @@
       {:error, :already_started} ->
         :noop
 
-      {:error, _} = err ->
-        Logger.error(
-          "Source.Supervisor -  Failed to start SourceSup: #{source_token}, #{inspect(err)}"
-        )
-
       _ ->
         :noop
     end
@@ -79,13 +74,6 @@
 
       {:error, :already_started} ->
         :noop
-
-      {:error, _reason} = err ->
-        Logger.error(
-          "Failed to start SourceSup when attempting restart: #{source_token} , #{inspect(err)} "
-        )
-
-        :noop
     end
 
     {:noreply, state}
@@ -107,7 +95,7 @@
 
   def delete_source(source_token) do
     GenServer.abcast(__MODULE__, {:stop, source_token})
     # TODO: move to adaptor callback
     unless do_pg_ops?() do
       BigQuery.delete_table(source_token)
     end
@@ -129,13 +117,13 @@
   end
 
   def delete_all_user_sources(user) do
     # TODO: use context func
     Repo.all(Ecto.assoc(user, :sources))
     |> Enum.each(fn s -> delete_source(s.token) end)
   end
 
   def reset_all_user_sources(user) do
     # TODO: use context func
     Repo.all(Ecto.assoc(user, :sources))
     |> Enum.each(fn s -> reset_source(s.token) end)
   end

lib/logflare/sql/dialect_translation.ex

@@ -80,10 +80,8 @@
   @spec do_parameter_positions_mapping(query :: String.t(), params :: [String.t()]) :: %{
           pos_integer() => String.t()
         }
-  defp do_parameter_positions_mapping(_query, []), do: %{}
-
   defp do_parameter_positions_mapping(query, params)
-       when is_non_empty_binary(query) and is_list(params) do
+       when is_non_empty_binary(query) and is_non_empty_list(params) do
     str =
       params
       |> Enum.uniq()
@@ -982,7 +980,7 @@
          {"Query" = k, %{"body" => %{"Select" => %{"from" => [_ | _] = from_list}}} = v},
          %{in_cte_tables_tree: false} = data
        ) do
     # TODO: refactor
     aliases =
       for from <- from_list,
           value = get_in(from, ["relation", "Table", "alias", "name", "value"]),
@@ -1020,7 +1018,7 @@
           } = v},
          %{in_cte_tables_tree: true} = data
        ) do
     # TODO: refactor
     aliases =
       for from <- from_list,
           value = get_in(from, ["relation", "Table", "alias", "name", "value"]),

lib/logflare/utils/guards.ex

@@ -25,6 +25,11 @@ defmodule Logflare.Utils.Guards do
   """
   defguard is_non_empty_binary(value) when is_binary(value) and value != ""
 
+  @doc """
+  Guard that indicates if the value provided is a list with at least one element.
+  """
+  defguard is_non_empty_list(value) when is_list(value) and value != []
+
   @doc """
   Guard that indicates if the value provided is a map with at least one key.
   """

lib/logflare_web/controllers/api/endpoint_controller.ex

@@ -63,7 +63,6 @@ defmodule LogflareWeb.Api.EndpointController do
       |> put_status(201)
       |> json(query)
     else
-      nil -> {:error, :not_found}
       err -> err
     end
   end

lib/logflare_web/controllers/billing_controller.ex

@@ -342,7 +342,7 @@ defmodule LogflareWeb.BillingController do
 
   defp billing_account_has_subscription?(billing_account) do
     if subscriptions = billing_account.stripe_subscriptions["data"] do
-      Enum.count(subscriptions) > 0
+      not Enum.empty?(subscriptions)
     else
       false
     end

lib/logflare_web/live/alerts/alerts_live.ex

@@ -1,8 +1,6 @@
 defmodule LogflareWeb.AlertsLive do
   @moduledoc false
-
   use LogflareWeb, :live_view
-  use Phoenix.Component
 
   import Ecto.Query
   import LogflareWeb.Utils, only: [stringify_changeset_errors: 2, time_ago: 1]

lib/logflare_web/live/backends/actions/show.html.heex

@@ -22,7 +22,7 @@
       </li>
       <li :if={@backend.metadata} class="list-group-item tw-flex-col tw-gap-2 ">
         <strong>metadata:</strong>
-        <span :for={{k, v} when is_binary(v) <- @backend.metadata || %{}} :if={@backend.metadata} class="tw-block tw-ml-4">{k}: {v}</span>
+        <span :for={{k, v} when is_binary(v) <- @backend.metadata} :if={@backend.metadata} class="tw-block tw-ml-4">{k}: {v}</span>
       </li>
     </ul>
   </div>

lib/logflare_web/live/endpoints/endpoints_live.ex

@@ -1,8 +1,6 @@
 defmodule LogflareWeb.EndpointsLive do
   @moduledoc false
-
   use LogflareWeb, :live_view
-  use Phoenix.Component
 
   import Ecto.Query, only: [from: 2]
 

lib/logflare_web/live/endpoints/endpoints_versions_live.ex

@@ -1,7 +1,6 @@
 defmodule LogflareWeb.EndpointsVersionsLive do
   @moduledoc false
   use LogflareWeb, :live_view
-  use Phoenix.Component
 
   import Ecto.Query
   import LogflareWeb.Utils, only: [time_ago: 1]

lib/logflare_web/live/query_live.ex

@@ -1,8 +1,6 @@
 defmodule LogflareWeb.QueryLive do
   @moduledoc false
-
   use LogflareWeb, :live_view
-  use Phoenix.Component
 
   alias Logflare.Alerting
   alias Logflare.Backends
@@ -220,8 +218,7 @@ defmodule LogflareWeb.QueryLive do
 
     next_backend_id = backend_id_from_form(socket)
 
-    if query_string != nil and
-         (query_string != prev_query_string or next_backend_id != prev_backend_id) do
+    if query_string != prev_query_string or next_backend_id != prev_backend_id do
       send(self(), :parse_query)
     end