defmodule Electric.Plug.Utils do @moduledoc """ Utility functions for Electric endpoints, e.g. for parsing and validating path and query parameters. """ @doc """ Parse columns parameter from a string consisting of a comma separated list of potentially quoted column names into a sorted list of strings. ## Examples iex> Electric.Plug.Utils.parse_columns_param("") {:error, "Invalid zero-length delimited identifier"} iex> Electric.Plug.Utils.parse_columns_param("foo,") {:error, "Invalid zero-length delimited identifier"} iex> Electric.Plug.Utils.parse_columns_param("id") {:ok, ["id"]} iex> Electric.Plug.Utils.parse_columns_param("id,name") {:ok, ["id", "name"]} iex> Electric.Plug.Utils.parse_columns_param(~S|"PoT@To",PoTaTo|) {:ok, ["PoT@To", "potato"]} iex> Electric.Plug.Utils.parse_columns_param(~S|"PoTaTo,sunday",foo|) {:ok, ["PoTaTo,sunday", "foo"]} iex> Electric.Plug.Utils.parse_columns_param(~S|"fo""o",bar|) {:ok, [~S|fo"o|, "bar"]} iex> Electric.Plug.Utils.parse_columns_param(~S|"id,"name"|) {:error, ~S|Invalid unquoted identifier contains special characters: "id|} """ @spec parse_columns_param(binary()) :: {:ok, [String.t(), ...]} | {:error, term()} def parse_columns_param(columns) when is_binary(columns) do columns # Split by commas that are not inside quotes |> String.split(~r/,(?=(?:[^"]*"[^"]*")*[^"]*$)/) |> Enum.reduce_while([], fn column, acc -> case Electric.Postgres.Identifiers.parse(column) do {:ok, casted_column} -> {:cont, [casted_column | acc]} {:error, reason} -> {:halt, {:error, reason}} end end) |> then(fn result -> case result do # TODO: convert output to MapSet? parsed_cols when is_list(parsed_cols) -> {:ok, Enum.reverse(parsed_cols)} {:error, reason} -> {:error, reason} end end) end def parse_columns_param([col | _] = columns) when is_binary(col) do {:ok, columns} end @doc """ Calculate the next interval that should be used for long polling based on the current time and previous interval used. Timestamp returned is in seconds and uses a custom epoch of 9th of October 2024, UTC. """ @oct9th2024 ~U[2024-10-09 00:00:00Z] @spec get_next_interval_timestamp(integer(), binary() | nil) :: integer() def get_next_interval_timestamp(long_poll_timeout_ms, prev_interval \\ nil) def get_next_interval_timestamp(long_poll_timeout_ms, _) when div(long_poll_timeout_ms, 1000) == 0, do: 0 def get_next_interval_timestamp(long_poll_timeout_ms, prev_interval) do long_poll_timeout_sec = div(long_poll_timeout_ms, 1000) diff_in_seconds = DateTime.diff(DateTime.utc_now(), @oct9th2024, :second) next_interval = ceil(diff_in_seconds / long_poll_timeout_sec) * long_poll_timeout_sec if "#{next_interval}" == prev_interval do next_interval + Enum.random(1..3_600) else next_interval end end alias OpenTelemetry.SemConv, as: SC def common_open_telemetry_attrs(%Plug.Conn{assigns: assigns} = conn) do query_params_map = if is_struct(conn.query_params, Plug.Conn.Unfetched) do %{} else Map.new(conn.query_params, fn {k, v} -> {"http.query_param.#{k}", v} end) end %{ "error.type" => assigns[:error_str], "http.request_id" => assigns[:plug_request_id], "http.query_string" => conn.query_string, SC.ClientAttributes.client_address() => conn.remote_ip, SC.ServerAttributes.server_address() => conn.host, SC.ServerAttributes.server_port() => conn.port, SC.HTTPAttributes.http_request_method() => conn.method, SC.HTTPAttributes.http_response_status_code() => conn.status, SC.Incubating.HTTPAttributes.http_response_size() => assigns[:streaming_bytes_sent], SC.NetworkAttributes.network_transport() => :tcp, SC.NetworkAttributes.network_local_port() => conn.port, SC.UserAgentAttributes.user_agent_original() => user_agent(conn), SC.Incubating.URLAttributes.url_path() => conn.request_path, SC.URLAttributes.url_scheme() => conn.scheme, SC.URLAttributes.url_full() => %URI{ scheme: to_string(conn.scheme), host: conn.host, port: conn.port, path: conn.request_path, query: conn.query_string } |> to_string() } |> Map.filter(fn {_k, v} -> not is_nil(v) end) |> Map.merge(Map.get(conn.private, :telemetry_span_attrs, %{})) |> Map.merge(query_params_map) |> Map.merge(Map.new(conn.req_headers, fn {k, v} -> {"http.request.header.#{k}", v} end)) |> Map.merge(Map.new(conn.resp_headers, fn {k, v} -> {"http.response.header.#{k}", v} end)) end defp user_agent(%Plug.Conn{} = conn) do case Plug.Conn.get_req_header(conn, "user-agent") do [] -> "" [head | _] -> head end end defmodule CORSHeaderPlug do @behaviour Plug import Plug.Conn def init(opts), do: opts def call(conn, opts), do: conn |> put_resp_header("access-control-allow-origin", get_allowed_origin(conn, opts)) |> put_resp_header("access-control-expose-headers", headers_to_expose()) |> put_resp_header("access-control-allow-methods", get_allowed_methods(conn, opts)) defp get_allowed_methods(_conn, opts), do: Access.get(opts, :methods, []) |> Enum.join(", ") defp get_allowed_origin(conn, opts) do Access.get( opts, :origin, case Plug.Conn.get_req_header(conn, "origin") do [origin] -> origin [] -> "*" end ) end defp headers_to_expose do Enum.join(Electric.Shapes.Api.Response.electric_headers(), ",") end end defmodule PassAssignToOptsPlug do @behaviour Plug def init(plug: plug, assign_key: key) when is_atom(plug), do: {plug, key} def call(conn, {plug, key}), do: plug.call(conn, plug.init(conn.assigns[key])) end end