defmodule Microwaveprop.Valkey do @moduledoc """ Thin wrapper around a single Redix connection to the cluster's Valkey instance. Used by `Microwaveprop.Weather.GridCache` to back the cross-pod weather-grid cache so the per-pod ETS replica (~768 MiB worst case × N pods) collapses to a single shared copy. Connection URL is read from `:microwaveprop, :valkey_url` at boot (set in `runtime.exs` from the `VALKEY_URL` env var). When unset the child returns `:ignore` and `configured?/0` is `false`, so callers can fall back to ETS / disk in dev and test without Valkey running. Public read/write helpers term-encode their values with `:erlang.term_to_binary/1` so atom-keyed maps and DateTimes round-trip cleanly. Decode uses the `:safe` flag — only existing atoms are resolvable, so a malicious Valkey can't blow the atom table. """ require Logger @conn __MODULE__.Conn @spec child_spec(keyword()) :: Supervisor.child_spec() def child_spec(_opts) do %{ id: __MODULE__, start: {__MODULE__, :start_link, []}, type: :worker, restart: :permanent } end @spec start_link() :: GenServer.on_start() | :ignore def start_link do case url() do nil -> Logger.info("Microwaveprop.Valkey not configured (VALKEY_URL unset) — GridCache will use ETS fallback.") :ignore url -> # sync_connect: false lets the app boot even when Valkey is # briefly unreachable; reads/writes during the gap return # {:error, _} and the GridCache caller falls back to disk. Redix.start_link(url, name: @conn, sync_connect: false, exit_on_disconnection: false ) end end @doc "True when VALKEY_URL is configured AND the Redix process is alive." @spec configured?() :: boolean() def configured? do case Process.whereis(@conn) do nil -> false pid when is_pid(pid) -> Process.alive?(pid) end end @doc """ GET a single key. Returns `{:ok, term}` on hit, `:miss` when the key doesn't exist or is expired, `{:error, reason}` on connection / decode failure. The error tuple lets callers fall back to disk without crashing the request. """ @spec get(String.t()) :: {:ok, term()} | :miss | {:error, term()} def get(key) when is_binary(key) do case command(["GET", key]) do {:ok, nil} -> :miss {:ok, bin} when is_binary(bin) -> safe_decode(bin) {:error, _} = err -> err end end @doc """ MGET a list of keys. Returns `{:ok, [term_or_nil]}` preserving the input order, where each element is the decoded term or `nil` on miss. Errors propagate as `{:error, _}` so the caller falls back to disk for the entire bounds query. """ @spec mget([String.t()]) :: {:ok, [term() | nil]} | {:error, term()} def mget([]), do: {:ok, []} def mget(keys) when is_list(keys) do case command(["MGET" | keys]) do {:ok, vals} -> decoded = Enum.map(vals, fn nil -> nil bin -> bin |> safe_decode() |> elem_or_nil() end) {:ok, decoded} {:error, _} = err -> err end end @doc """ SET key with the term-encoded value and an EX TTL in seconds. Used by the cache fill path; on connection failure returns `{:error, _}` and the caller logs + continues (cache miss next time, refilled then). """ @spec set(String.t(), term(), pos_integer()) :: :ok | {:error, term()} def set(key, term, ttl_seconds) when is_binary(key) and is_integer(ttl_seconds) and ttl_seconds > 0 do bin = :erlang.term_to_binary(term) case command(["SET", key, bin, "EX", Integer.to_string(ttl_seconds)]) do {:ok, "OK"} -> :ok {:error, _} = err -> err other -> {:error, other} end end @doc """ Pipeline N SETs with TTL in a single round-trip. Returns `:ok` if every command succeeded, `{:error, _}` otherwise. """ @spec mset_with_ttl([{String.t(), term()}], pos_integer()) :: :ok | {:error, term()} def mset_with_ttl([], _ttl), do: :ok def mset_with_ttl(pairs, ttl_seconds) when is_list(pairs) and is_integer(ttl_seconds) do cmds = Enum.map(pairs, fn {key, term} -> ["SET", key, :erlang.term_to_binary(term), "EX", Integer.to_string(ttl_seconds)] end) case pipeline(cmds) do {:ok, results} -> if Enum.all?(results, &(&1 == "OK")), do: :ok, else: {:error, {:partial, results}} {:error, _} = err -> err end end @doc """ Sorted-set add. Used by `GridCache` to maintain a `prop:wg:vts` index of cached valid_times so `latest_valid_time/0` can `ZREVRANGE` instead of scanning all keys. """ @spec zadd(String.t(), number(), String.t()) :: :ok | {:error, term()} def zadd(key, score, member) do case command(["ZADD", key, to_string(score), member]) do {:ok, _} -> :ok {:error, _} = err -> err end end @spec zrevrange(String.t(), integer(), integer()) :: {:ok, [String.t()]} | {:error, term()} def zrevrange(key, start, stop) do case command(["ZREVRANGE", key, to_string(start), to_string(stop)]) do {:ok, members} -> {:ok, members} {:error, _} = err -> err end end @spec zrangebyscore(String.t(), String.t(), String.t()) :: {:ok, [String.t()]} | {:error, term()} def zrangebyscore(key, min, max) do case command(["ZRANGEBYSCORE", key, min, max]) do {:ok, members} -> {:ok, members} {:error, _} = err -> err end end @spec zrem(String.t(), [String.t()]) :: :ok | {:error, term()} def zrem(_key, []), do: :ok def zrem(key, members) when is_list(members) do case command(["ZREM", key | members]) do {:ok, _} -> :ok {:error, _} = err -> err end end @spec del([String.t()]) :: :ok | {:error, term()} def del([]), do: :ok def del(keys) when is_list(keys) do case command(["DEL" | keys]) do {:ok, _} -> :ok {:error, _} = err -> err end end @doc """ Run a single SCAN MATCH pass and return all matching keys. Used only by maintenance / test paths — the read path uses ZSET indexes instead. """ @spec scan_match(String.t()) :: {:ok, [String.t()]} | {:error, term()} def scan_match(pattern) do do_scan_match(pattern, "0", []) end defp do_scan_match(pattern, cursor, acc) do case command(["SCAN", cursor, "MATCH", pattern, "COUNT", "500"]) do {:ok, [next_cursor, keys]} -> new_acc = acc ++ keys case next_cursor do "0" -> {:ok, new_acc} other -> do_scan_match(pattern, other, new_acc) end {:error, _} = err -> err end end defp adapter, do: Application.get_env(:microwaveprop, :valkey_adapter, Microwaveprop.Valkey.RedixAdapter) defp command(args) do if configured?() do try do adapter().command(@conn, args, timeout: 1500) catch :exit, reason -> {:error, {:exit, reason}} end else {:error, :not_configured} end end defp pipeline(cmds) do if configured?() do try do adapter().pipeline(@conn, cmds, timeout: 3000) catch :exit, reason -> {:error, {:exit, reason}} end else {:error, :not_configured} end end defp safe_decode(bin) do {:ok, :erlang.binary_to_term(bin, [:safe])} rescue e -> {:error, {:decode, Exception.message(e)}} end defp elem_or_nil({:ok, term}), do: term defp elem_or_nil(_), do: nil defp url, do: Application.get_env(:microwaveprop, :valkey_url) end