defmodule PhoenixGenApi do @moduledoc """ PhoenixGenApi is a framework for building distributed API systems with Phoenix. This library provides a comprehensive solution for handling API requests with support for multiple execution modes (sync, async, streaming), distributed node selection, permission checking, and automatic argument validation. ## Features - **Multiple Execution Modes**: Support for synchronous, asynchronous, streaming, and fire-and-forget requests - **Distributed Execution**: Execute functions on remote nodes with automatic node selection - **Node Selection Strategies**: Random, hash-based, round-robin, and custom selection strategies - **Automatic Argument Validation**: Type checking and conversion for request arguments - **Permission Control**: Built-in permission checking for requests - **Streaming Support**: Handle long-running operations with streaming responses - **Configuration Caching**: Efficient caching of function configurations with automatic updates - **Configuration Push**: Remote nodes can actively push their service and function configs to the gateway - **Rate Limiting**: Global and per-API rate limiting with sliding window algorithm - **Relay Messages**: Group-based message relaying with public, private, and strict_private group types - **Diagnostics**: Runtime health checks, statistics, debug reports, and admin-gated tracing utilities ## Architecture The library consists of several key components: - `PhoenixGenApi.Executor` - Core execution engine for processing requests - `PhoenixGenApi.ConfigDb` - Caches function configurations for fast lookup - `PhoenixGenApi.ConfigPuller` - Pulls and updates configurations from remote services - `PhoenixGenApi.ConfigReceiver` - Receives pushed configurations from remote nodes (server-side) - `PhoenixGenApi.ConfigPusher` - Pushes configurations to the gateway node (client-side) - `PhoenixGenApi.NodeSelector` - Selects target nodes based on configured strategies - `PhoenixGenApi.Permission` - Handles permission checking for requests - `PhoenixGenApi.ArgumentHandler` - Validates and converts request arguments - `PhoenixGenApi.StreamCall` - Manages streaming function calls - `PhoenixGenApi.RateLimiter` - Rate limiting for global and per-API requests - `PhoenixGenApi.Relay` - Group-based message relaying (public, private, strict_private) - `PhoenixGenApi.Diagnostics` - Runtime health, statistics, debug, and tracing utilities ## Usage Example ### Basic Setup First, define your function configurations: config = %PhoenixGenApi.Structs.FunConfig{ request_type: "get_user", service: "user_service", nodes: ["user@node1", "user@node2"], choose_node_mode: :random, timeout: 5000, mfa: {UserService, :get_user, []}, arg_types: %{"user_id" => :string}, arg_orders: ["user_id"], response_type: :sync, check_permission: {:arg, "user_id"}, request_info: false } # Add configuration to cache PhoenixGenApi.ConfigDb.add(config) ### Execute Requests use PhoenixGenApi # Create a request request = %PhoenixGenApi.Structs.Request{ request_id: "req_123", request_type: "get_user", user_id: "user_456", device_id: "device_789", args: %{"user_id" => "user_123"} } # Execute the request response = PhoenixGenApi.Executor.execute!(request) ### Streaming Requests For long-running operations, use streaming mode: stream_config = %PhoenixGenApi.Structs.FunConfig{ request_type: "process_data", service: "processing_service", nodes: :local, choose_node_mode: :random, timeout: :infinity, mfa: {DataProcessor, :process_large_dataset, []}, arg_types: %{"dataset_id" => :string}, arg_orders: ["dataset_id"], response_type: :stream, check_permission: false, request_info: true } # The streaming function should send results using StreamHelper: # StreamHelper.send_result(stream, chunk_data) # StreamHelper.send_last_result(stream, final_data) # Or: StreamHelper.send_complete(stream) ## Channel Options When using `use PhoenixGenApi` in a channel, the following options are available: - `:event` (default: `"phoenix_gen_api"`) — the event name to handle. - `:override_user_id` (default: `true`) — when `true`, the `user_id` from `socket.assigns` is injected into the request payload, but **only** when `socket.assigns.user_id` is a verified non-empty binary. This prevents a client-supplied `user_id` from overriding the server-side value. The `user_id` in assigns must be set by a verified authentication step in `Phoenix.Socket.connect/3`, never from client payload. - `:require_verified_user_id` (default: `true`) — when `true`, requests are rejected immediately with `"Authentication required"` if `socket.assigns.user_id` is nil or empty. This prevents unauthenticated requests from reaching permission checks or function execution. Set to `false` for public endpoints that use `check_permission: false`. **Security note**: Setting this to `false` disables the early rejection of unauthenticated requests. Only do this for channels that serve public data. ## Configuration Add to your `config.exs`: config :phoenix_gen_api, :gen_api, pull_timeout: 5_000, pull_interval: 30_000, detail_error: false, service_configs: [ %{ service: "user_service", nodes: ["user@node1", "user@node2"], module: "UserService", function: "get_config", args: [] } ] ## Rate Limiting Configuration Configure rate limits in your `config.exs`: config :phoenix_gen_api, :rate_limiter, enabled: true, fail_open: true, global_limits: [ %{key: :user_id, max_requests: 2000, window_ms: 60_000}, %{key: :device_id, max_requests: 10000, window_ms: 60_000} ], api_limits: [ %{ service: "data_service", request_type: "export_data", key: :user_id, max_requests: 10, window_ms: 60_000 } ] ## Learn More For detailed information about specific components, see: - `PhoenixGenApi.Executor` - Request execution - `PhoenixGenApi.Structs.FunConfig` - Function configuration - `PhoenixGenApi.Structs.Request` - Request structure - `PhoenixGenApi.Structs.Response` - Response structure - `PhoenixGenApi.NodeSelector` - Node selection strategies """ alias PhoenixGenApi.RateLimiter alias PhoenixGenApi.StreamCall alias PhoenixGenApi.Diagnostics @spec stop_stream(pid()) :: :ok @doc """ Stops an active streaming call. This function gracefully terminates a streaming call process and sends a completion message to the receiver. The stream call process is identified by its PID. ## Parameters - `stream_pid` - The PID of the streaming call process to stop ## Returns - `:ok` - The stop signal was sent successfully ## Examples # Start a stream {:ok, stream_pid} = StreamCall.start_link(%{ request: request, fun_config: config, receiver: self() }) # Later, stop the stream PhoenixGenApi.stop_stream(stream_pid) # Receive the completion message receive do {:stream_response, response} -> assert response.has_more == false end ## Notes - The stream call will send a completion response to its receiver before terminating - This does not notify the data generator process; it only stops the stream relay - If you need to stop the data generation itself, handle that in your generator function """ def stop_stream(request_id) do StreamCall.stop(request_id) end @doc """ Checks rate limit for a request. This function checks both global and per-API rate limits. It is automatically called during request execution, but can also be called manually for custom rate limiting logic. ## Parameters - `request` - The `Request` struct to check ## Returns - `:ok` - Request is within all rate limits - `{:error, :rate_limited, details}` - Request exceeds a rate limit ## Examples request = %Request{user_id: "user_123", service: "my_service", request_type: "my_api"} case PhoenixGenApi.check_rate_limit(request) do :ok -> # Proceed with execution {:error, :rate_limited, details} -> # Handle rate limit exceeded end """ @spec check_rate_limit(PhoenixGenApi.Structs.Request.t()) :: :ok | {:error, :rate_limited, map()} def check_rate_limit(request) do RateLimiter.check_rate_limit(request) end @doc """ Resets rate limit counters for a specific key. ## Parameters - `key_value` - The key value to reset (e.g., user ID) - `scope` - Either `:global` or `{service, request_type}` tuple - `rate_limit_key` - The type of key (`:user_id`, `:device_id`, etc.) ## Returns - `:ok` - Counters were reset ## Examples # Reset all rate limits for a user PhoenixGenApi.reset_rate_limit("user_123", :global, :user_id) # Reset API-specific rate limit PhoenixGenApi.reset_rate_limit("user_123", {"my_service", "my_api"}, :user_id) """ @spec reset_rate_limit( String.t(), :global | {String.t() | atom(), String.t()}, atom() | String.t() ) :: :ok def reset_rate_limit(key_value, scope, rate_limit_key) do RateLimiter.reset_rate_limit(key_value, scope, rate_limit_key) end @doc """ Gets current rate limit status for a key. ## Returns A list of maps with current usage information for all applicable rate limits. """ @spec get_rate_limit_status( String.t(), :global | {String.t() | atom(), String.t()}, atom() | String.t() ) :: map() def get_rate_limit_status(key_value, scope, rate_limit_key) do RateLimiter.get_rate_limit_status(key_value, scope, rate_limit_key) end @doc """ Updates rate limit configuration at runtime. ## Parameters - `config` - A map with `:global_limits` and/or `:api_limits` keys ## Returns - `:ok` - Configuration was updated ## Examples PhoenixGenApi.update_rate_limit_config(%{ global_limits: [ %{key: :user_id, max_requests: 2000, window_ms: 60_000} ] }) """ @spec update_rate_limit_config(map()) :: :ok def update_rate_limit_config(config) do RateLimiter.update_config(config) end @doc """ Gets all configured rate limits. ## Returns A map with `:global` and `:api` keys containing the configured limits. """ @spec get_rate_limit_config() :: %{global: list(), api: list()} def get_rate_limit_config() do RateLimiter.get_configured_limits() end @doc """ Gets the current global rate limits (may differ from config.exs if changed at runtime). ## Returns A list of global rate limit maps. ## Examples PhoenixGenApi.get_global_limits() # => [%{key: :user_id, max_requests: 2000, window_ms: 60_000}] """ @spec get_global_limits() :: [map()] def get_global_limits() do RateLimiter.get_global_limits() end @doc """ Sets (replaces) all global rate limits at runtime. ## Parameters - `limits` - A list of global rate limit maps, each with: - `:key` - The rate limit key (`:user_id`, `:device_id`, `:ip_address`, or custom string) - `:max_requests` - Maximum requests allowed in the window - `:window_ms` - Window duration in milliseconds ## Returns - `:ok` - Limits were updated ## Examples PhoenixGenApi.set_global_limits([ %{key: :user_id, max_requests: 2000, window_ms: 60_000}, %{key: :device_id, max_requests: 10000, window_ms: 60_000} ]) """ @spec set_global_limits([map()]) :: :ok def set_global_limits(limits) when is_list(limits) do RateLimiter.set_global_limits(limits) end @doc """ Adds a single global rate limit at runtime. If a limit with the same `:key` already exists, it will be replaced. ## Parameters - `limit` - A map with `:key`, `:max_requests`, and `:window_ms` ## Returns - `:ok` - Limit was added ## Examples PhoenixGenApi.add_global_limit(%{ key: :ip_address, max_requests: 100, window_ms: 60_000 }) """ @spec add_global_limit(map()) :: :ok def add_global_limit(limit) when is_map(limit) do RateLimiter.add_global_limit(limit) end @doc """ Removes a global rate limit by key at runtime. ## Parameters - `key` - The rate limit key to remove (`:user_id`, `:device_id`, etc.) ## Returns - `:ok` - Limit was removed (or didn't exist) ## Examples PhoenixGenApi.remove_global_limit(:ip_address) """ @spec remove_global_limit(atom() | String.t()) :: :ok def remove_global_limit(key) do RateLimiter.remove_global_limit(key) end @doc """ Attaches a telemetry handler to all PhoenixGenApi events. This is a convenience function that attaches handlers to both executor and rate limiter events with a single call. ## Events ### Executor Events - `[:phoenix_gen_api, :executor, :request, :start]` - `[:phoenix_gen_api, :executor, :request, :stop]` - `[:phoenix_gen_api, :executor, :request, :exception]` ### Rate Limiter Events - `[:phoenix_gen_api, :rate_limiter, :check]` - `[:phoenix_gen_api, :rate_limiter, :exceeded]` - `[:phoenix_gen_api, :rate_limiter, :reset]` - `[:phoenix_gen_api, :rate_limiter, :cleanup]` ## Parameters - `handler_id` - A unique string identifier for the handler - `function` - A 4-arity function: fn(event, measurements, metadata, config) -> any end - `config` - Optional configuration map passed to the handler (default: %{}) ## Examples # Attach a simple logging handler PhoenixGenApi.attach_telemetry("my-app", fn event, measurements, metadata, _config -> ... end) # Attach with custom config PhoenixGenApi.attach_telemetry("metrics", &MyApp.Metrics.handle_event/4, %{prefix: "phoenix_gen_api"}) """ @spec attach_telemetry(String.t(), function(), map()) :: :ok def attach_telemetry(handler_id, function, config \\ %{}) when is_binary(handler_id) and is_function(function, 4) do PhoenixGenApi.Executor.attach_telemetry(handler_id, function, config) PhoenixGenApi.RateLimiter.attach_telemetry(handler_id, function, config) :ok end @doc """ Detaches all telemetry handlers with the given ID. ## Parameters - `handler_id` - The handler ID used when attaching ## Examples PhoenixGenApi.detach_telemetry("my-app") """ @spec detach_telemetry(String.t()) :: :ok def detach_telemetry(handler_id) when is_binary(handler_id) do PhoenixGenApi.Executor.detach_telemetry(handler_id) PhoenixGenApi.RateLimiter.detach_telemetry(handler_id) :ok end # ============================================================================ # Shell Helper Functions (for easy inspection in IEx) # ============================================================================ @doc """ [Shell Helper] Quick view of rate limit status for a user. ## Usage in IEx iex> PhoenixGenApi.rl_status("user_123") """ def rl_status(user_id) when is_binary(user_id) do IO.puts("\n=== Rate Limit Status for #{user_id} ===\n") IO.puts("Global Limits:") get_rate_limit_status(user_id, :global, :user_id) |> Enum.each(fn info -> IO.puts( " Scope: #{inspect(info.scope)} | Used: #{info.current_requests}/#{info.max_requests} | Remaining: #{info.remaining}" ) end) IO.puts("\nAPI Limits:") get_rate_limit_config().api |> Enum.filter(&(&1.key == :user_id)) |> Enum.each(fn limit -> scope = {limit.service, limit.request_type} status = get_rate_limit_status(user_id, scope, :user_id) Enum.each(status, fn info -> IO.puts( " #{limit.service}/#{limit.request_type} | Used: #{info.current_requests}/#{info.max_requests} | Remaining: #{info.remaining}" ) end) end) :ok end @doc """ [Shell Helper] Quick view and management of global rate limits. ## Usage in IEx # View current global limits iex> PhoenixGenApi.rl_global() # Set new global limits iex> PhoenixGenApi.rl_global([%{key: :user_id, max_requests: 2000, window_ms: 60_000}]) # Add a single limit iex> PhoenixGenApi.rl_global(:add, %{key: :ip_address, max_requests: 100, window_ms: 60_000}) # Remove a limit by key iex> PhoenixGenApi.rl_global(:remove, :ip_address) """ def rl_global() do IO.puts("\n=== Global Rate Limits ===\n") get_global_limits() |> Enum.each(fn l -> IO.puts(" Key: #{inspect(l.key)} | Max: #{l.max_requests} | Window: #{l.window_ms}ms") end) :ok end def rl_global(limits) when is_list(limits) do set_global_limits(limits) IO.puts("\nGlobal rate limits updated.\n") rl_global() end def rl_global(:add, limit) when is_map(limit) do add_global_limit(limit) IO.puts("\nGlobal rate limit added/updated: #{inspect(limit)}\n") rl_global() end def rl_global(:remove, key) do remove_global_limit(key) IO.puts("\nGlobal rate limit removed for key: #{inspect(key)}\n") rl_global() end @doc """ [Shell Helper] Quick view of current rate limit configuration. ## Usage in IEx iex> PhoenixGenApi.rl_config() """ def rl_config() do config = get_rate_limit_config() IO.puts("\n=== Rate Limit Configuration ===\n") IO.puts("Global Limits:") Enum.each(config.global, fn l -> IO.puts(" Key: #{inspect(l.key)} | Max: #{l.max_requests} | Window: #{l.window_ms}ms") end) IO.puts("\nAPI Limits:") Enum.each(config.api, fn l -> IO.puts( " #{l.service}/#{l.request_type} | Key: #{inspect(l.key)} | Max: #{l.max_requests} | Window: #{l.window_ms}ms" ) end) :ok end @doc """ [Shell Helper] Quick view of ConfigDb cache status. ## Usage in IEx iex> PhoenixGenApi.cache_status() """ def cache_status() do IO.puts("\n=== ConfigDb Cache Status ===\n") IO.puts("Total cached configs: #{PhoenixGenApi.ConfigDb.count()}") IO.puts("Services: #{inspect(PhoenixGenApi.ConfigDb.get_all_services())}") :ok end @doc """ [Shell Helper] Quick view of Worker Pool status. ## Usage in IEx iex> PhoenixGenApi.pool_status() """ def pool_status() do IO.puts("\n=== Worker Pool Status ===\n") async_status = PhoenixGenApi.WorkerPool.status(:async_pool) stream_status = PhoenixGenApi.WorkerPool.status(:stream_pool) IO.puts("Async Pool:") IO.puts( " Idle: #{async_status.idle_workers} | Busy: #{async_status.busy_workers} | Queued: #{async_status.queued_tasks}" ) IO.puts(" Circuit Open: #{async_status.circuit_open}") IO.puts("\nStream Pool:") IO.puts( " Idle: #{stream_status.idle_workers} | Busy: #{stream_status.busy_workers} | Queued: #{stream_status.queued_tasks}" ) IO.puts(" Circuit Open: #{stream_status.circuit_open}") :ok end @doc """ [Shell Helper] Print a formatted health check summary to the console. ## Usage in IEx iex> PhoenixGenApi.health_print() iex> PhoenixGenApi.health_print(max_memory_bytes: 100_000_000) """ def health_print(opts \\ []) do report = Diagnostics.health_check(opts) status_icon = fn :ok -> "✅" :degraded -> "⚠️ " :error -> "❌" end IO.puts("\n=== PhoenixGenApi Health Check ===") IO.puts("Node: #{report.node}") IO.puts("Status: #{status_icon.(report.status)} #{report.status}") IO.puts("Checked at: #{DateTime.from_unix!(report.checked_at_ms, :millisecond)}") vm = report.checks.vm IO.puts("\n--- VM ---") IO.puts(" Status: #{status_icon.(vm.status)} #{vm.status}") IO.puts(" Processes: #{vm.process_count} / #{vm.process_limit}") IO.puts(" Schedulers: #{vm.schedulers_online} / #{vm.schedulers}") IO.puts(" Memory: #{format_bytes(vm.memory[:total])} total") IO.puts(" Processes: #{format_bytes(vm.memory[:processes])}") IO.puts(" System: #{format_bytes(vm.memory[:system])}") IO.puts(" Uptime: #{format_uptime(vm.uptime)}") node = report.checks.node IO.puts("\n--- Node ---") IO.puts(" Self: #{node.node}") IO.puts(" Alive: #{node.alive?}") IO.puts(" Connected: #{inspect(node.connected_nodes)}") pga = report.checks.phoenix_gen_api IO.puts("\n--- PhoenixGenApi ---") IO.puts(" Status: #{status_icon.(pga.status)} #{pga.status}") IO.puts(" Mode: #{pga.mode}") case pga do %{checks: checks} -> IO.puts(" Processes:") Enum.each(checks, fn {name, check} -> icon = if check.status == :ok, do: "✅", else: "❌" pid_str = if check[:pid], do: "#{inspect(check[:pid])}", else: "N/A" extra = if check[:instance_count], do: " instances=#{check[:instance_count]}", else: "" IO.puts(" #{icon} #{name} pid=#{pid_str}#{extra}") end) %{} -> :ok end IO.puts("") :ok end @doc """ [Shell Helper] Print formatted VM and PhoenixGenApi statistics to the console. ## Usage in IEx iex> PhoenixGenApi.stats_print() """ def stats_print() do stats = Diagnostics.statistics() IO.puts("\n=== PhoenixGenApi Statistics ===") IO.puts("Node: #{stats.node}") IO.puts("Collected at: #{DateTime.from_unix!(stats.collected_at_ms, :millisecond)}") vm = stats.vm IO.puts("\n--- VM ---") IO.puts(" Processes: #{vm.process_count} / #{vm.process_limit}") IO.puts(" Ports: #{vm.port_count}") IO.puts(" ETS tables: #{vm.ets_count}") IO.puts(" Schedulers: #{vm.schedulers_online} / #{vm.schedulers}") IO.puts(" Memory: #{format_bytes(vm.memory[:total])} total") IO.puts( " Processes: #{format_bytes(vm.memory[:processes])} (#{format_bytes(vm.memory[:processes_used])} used)" ) IO.puts(" System: #{format_bytes(vm.memory[:system])}") IO.puts( " Atoms: #{format_bytes(vm.memory[:atom])} (#{format_bytes(vm.memory[:atom_used])} used)" ) IO.puts(" Binary: #{format_bytes(vm.memory[:binary])}") IO.puts(" Code: #{format_bytes(vm.memory[:code])}") IO.puts(" ETS: #{format_bytes(vm.memory[:ets])}") {red_total, red_since} = vm.reductions IO.puts(" Reductions: #{red_total} total, #{red_since} since last call") {gc_count, gc_words, _} = vm.garbage_collection IO.puts(" GC: #{gc_count} collections, #{gc_words} words reclaimed") {cs_count, _} = vm.context_switches IO.puts(" Context sw: #{cs_count}") IO.puts(" Uptime: #{format_uptime(vm.uptime)}") if vm.scheduler_wall_time && vm.scheduler_wall_time != :undefined do IO.puts(" Scheduler wall time:") Enum.each(vm.scheduler_wall_time, fn {id, active, total} -> pct = if total > 0, do: Float.round(active / total * 100, 1), else: 0.0 IO.puts(" Scheduler #{id}: #{pct}% active (#{active}ms / #{total}ms)") end) end pga = stats.phoenix_gen_api IO.puts("\n--- PhoenixGenApi ---") IO.puts(" Client mode: #{pga.client_mode}") IO.puts(" Telemetry events: #{pga.telemetry_events}") case pga do %{config_db: db} -> IO.puts("\n ConfigDb:") IO.puts(" Status: #{db.status}") IO.puts(" Count: #{db.count}") IO.puts(" Services: #{inspect(db.services)}") %{} -> :ok end case pga do %{rate_limiter: rl} -> IO.puts("\n Rate Limiter:") IO.puts(" Status: #{rl.status}") if rl.data do IO.puts(" Instances: #{length(rl.data.instances)}") end %{} -> :ok end case pga do %{worker_pool: worker_pool} -> IO.puts("\n Worker Pools:") Enum.each(worker_pool, fn {name, pool} -> IO.puts(" #{name}:") IO.puts(" Status: #{pool.status}") if pool.data do d = pool.data IO.puts( " Idle: #{d.idle_workers} Busy: #{d.busy_workers} Queued: #{d.queued_tasks}" ) IO.puts( " Circuit: #{d.circuit_open} Executed: #{d.total_tasks_executed} Failed: #{d.total_tasks_failed}" ) end end) %{} -> :ok end case pga do %{relay: relay} -> IO.puts("\n Relay:") IO.puts(" Status: #{relay.status}") if relay.data do IO.puts(" Groups: #{relay.data.group_count}") IO.puts(" Monitored memberships: #{relay.data.monitored_memberships}") end %{} -> :ok end IO.puts("") :ok end @doc """ [Shell Helper] Print a formatted debug report to the console. ## Usage in IEx iex> PhoenixGenApi.debug_print(process_limit: 10) """ def debug_print(opts \\ []) do report = Diagnostics.debug_report(opts) limit = Keyword.get(opts, :process_limit, 20) IO.puts("\n=== Debug Report ===") IO.puts("Node: #{report.node}") IO.puts("Collected at: #{DateTime.from_unix!(report.collected_at_ms, :millisecond)}") IO.puts("\n--- Top #{limit} Processes by Memory ---") IO.puts( String.pad_trailing("PID", 15) <> String.pad_trailing("Name", 30) <> String.pad_trailing("Memory", 12) <> "Status" ) IO.puts(String.duplicate("-", 70)) Enum.each(report.processes, fn p -> pid_str = String.pad_trailing(inspect(p.pid), 15) name_str = String.pad_trailing(inspect(p.registered_name || :unnamed), 30) mem_str = String.pad_trailing(format_bytes(p.memory), 12) status_str = inspect(p.status) IO.puts(pid_str <> name_str <> mem_str <> status_str) end) IO.puts("\n--- ETS Tables ---") Enum.each(report.ets_tables, fn {name, info} -> if info[:exists] do size = Map.get(info, :size, "?") mem_words = Map.get(info, :memory, "?") IO.puts(" #{name}: size=#{size} memory=#{mem_words} words") else IO.puts(" #{name}: (not found)") end end) IO.puts("\n--- Trace ---") IO.puts(" Control word: #{report.trace.trace_control_word}") IO.puts("") :ok end @doc """ [Shell Helper] Print a formatted call flow trace to the console. ## Usage in IEx iex> PhoenixGenApi.call_flow_print("user_service", "get_user") iex> PhoenixGenApi.call_flow_print("user_service", "get_user", "1.0.0") """ def call_flow_print(service, request_type, version \\ nil) do flow = Diagnostics.call_flow(service, request_type, version) IO.puts("\n=== Call Flow: #{service}/#{request_type} ===") case flow do %{error: error} -> IO.puts("❌ Config not found: #{inspect(error)}") IO.puts("") :ok %{service: service_name, request_type: request_type_name, version: version} -> IO.puts("Service: #{service_name}") IO.puts("Request Type: #{request_type_name}") IO.puts("Version: #{inspect(version)}") IO.puts("Response: #{flow.response_type}") IO.puts("Local: #{flow.local?}") IO.puts("Node Mode: #{flow.choose_node_mode}") IO.puts("Timeout: #{flow.timeout}ms") IO.puts("MFA: #{inspect(flow.mfa)}") IO.puts("\n--- Nodes ---") IO.puts(" All: #{inspect(flow.nodes)}") IO.puts(" Reachable: #{inspect(flow.reachable_nodes)}") if flow.unreachable_nodes != [] do IO.puts(" ❌ Unreachable: #{inspect(flow.unreachable_nodes)}") end IO.puts("\n--- Rate Limits ---") if flow.rate_limit.global != [] do IO.puts(" Global:") Enum.each(flow.rate_limit.global, fn l -> IO.puts(" #{l.key}: #{l.max_requests} req / #{l.window_ms}ms") end) else IO.puts(" Global: (none)") end if flow.rate_limit.api != [] do IO.puts(" API:") Enum.each(flow.rate_limit.api, fn l -> IO.puts(" #{l.key}: #{l.max_requests} req / #{l.window_ms}ms") end) else IO.puts(" API: (none)") end IO.puts("\n--- Permission ---") IO.puts(" #{flow.permission.description}") IO.puts("\n--- Hooks ---") IO.puts(" Before: #{flow.hooks.before_execute.description}") IO.puts(" After: #{flow.hooks.after_execute.description}") IO.puts("\n--- Retry ---") IO.puts(" #{flow.retry.description}") IO.puts("\n--- Execution Steps ---") Enum.with_index(flow.steps, 1) |> Enum.each(fn {step, idx} -> IO.puts(" #{idx}. [#{step.phase}] #{step.desc}") end) IO.puts("") :ok end end @doc """ [Shell Helper] Print a formatted cluster topology view to the console. ## Usage in IEx iex> PhoenixGenApi.cluster_print() """ def cluster_print() do view = Diagnostics.cluster_view() IO.puts("\n=== Cluster View ===") IO.puts("Self: #{view.self}") IO.puts("Connected nodes (#{view.connected_count}): #{inspect(view.connected)}") IO.puts("\n--- Registered Processes ---") Enum.each(view.registered_processes, fn {node, names} -> IO.puts(" #{node}: #{length(names)} processes") Enum.each(names, fn name -> IO.puts(" - #{name}") end) end) IO.puts("\n--- PhoenixGenApi Services by Node ---") Enum.each(view.phoenix_gen_api_services, fn {node, services} -> IO.puts(" #{node}: #{inspect(services)}") end) IO.puts("\n--- Node Selection Strategies ---") IO.puts(" #{Enum.join(Enum.map(view.node_selection.strategies, &inspect/1), ", ")}") IO.puts(" #{view.node_selection.description}") IO.puts("") :ok end @doc """ [Shell Helper] Print a formatted list of all registered call flows. ## Usage in IEx iex> PhoenixGenApi.flows_print() iex> PhoenixGenApi.flows_print(include_disabled: true) """ def flows_print(opts \\ []) do flows = Diagnostics.list_call_flows(opts) IO.puts("\n=== Registered Call Flows ===") IO.puts("Total: #{length(flows)}") if flows == [] do IO.puts(" (no registered configs)") else header = String.pad_trailing("Service", 25) <> String.pad_trailing("Request Type", 25) <> String.pad_trailing("Version", 12) <> String.pad_trailing("Local", 8) <> "Nodes" IO.puts("\n" <> header) IO.puts(String.duplicate("-", 90)) Enum.each(flows, fn flow -> status = if flow.disabled, do: " [DISABLED]", else: "" service = String.pad_trailing(inspect(flow.service), 25) req_type = String.pad_trailing(inspect(flow.request_type), 25) version = String.pad_trailing(inspect(flow.version), 12) mode = String.pad_trailing(to_string(flow.local?), 8) nodes = inspect(flow.nodes) IO.puts(service <> req_type <> version <> mode <> nodes <> status) end) end IO.puts("") :ok end @doc """ [Shell Helper] Print a formatted request inspection to the console. ## Usage in IEx iex> PhoenixGenApi.inspect_print(%{service: "user_service", request_type: "get_user"}) """ def inspect_print(request) do plan = Diagnostics.inspect_request(request) IO.puts("\n=== Request Inspection ===") if plan.request do r = plan.request IO.puts("Service: #{inspect(r.service)}") IO.puts("Request Type: #{inspect(r.request_type)}") IO.puts("Version: #{inspect(r.version)}") IO.puts("User ID: #{inspect(r.user_id)}") IO.puts("Device ID: #{inspect(r.device_id)}") IO.puts("Request ID: #{inspect(r.request_id)}") end if plan.error do IO.puts("\n❌ Config not found: #{inspect(plan.error)}") else IO.puts("\n--- Execution Plan ---") Enum.with_index(plan.steps, 1) |> Enum.each(fn {step, idx} -> IO.puts(" #{idx}. [#{step.phase}] #{step.desc}") end) end IO.puts("") :ok end # ────────────────────────────────────────────────────────────────────── # Private formatters # ────────────────────────────────────────────────────────────────────── defp format_bytes(bytes) when is_integer(bytes) when bytes < 1024 do "#{bytes} B" end defp format_bytes(bytes) when is_integer(bytes) when bytes < 1024 * 1024 do "#{Float.round(bytes / 1024, 1)} KB" end defp format_bytes(bytes) when is_integer(bytes) when bytes < 1024 * 1024 * 1024 do "#{Float.round(bytes / (1024 * 1024), 1)} MB" end defp format_bytes(bytes) when is_integer(bytes) do "#{Float.round(bytes / (1024 * 1024 * 1024), 1)} GB" end defp format_bytes(_), do: "?" defp format_uptime({ms, _}) when is_integer(ms) do seconds = div(ms, 1000) hours = div(seconds, 3600) minutes = rem(div(seconds, 60), 60) secs = rem(seconds, 60) "#{hours}h #{minutes}m #{secs}s" end defp format_uptime(_), do: "?" @doc """ Returns a runtime health check for the VM, distribution, and PhoenixGenApi processes. Delegates to `PhoenixGenApi.Diagnostics.health_check/1`. """ @spec health_check(keyword()) :: map() def health_check(opts \\ []) do PhoenixGenApi.Diagnostics.health_check(opts) end @doc """ Returns runtime VM and PhoenixGenApi statistics. Delegates to `PhoenixGenApi.Diagnostics.statistics/1`. """ @spec statistics(keyword()) :: map() def statistics(opts \\ []) do PhoenixGenApi.Diagnostics.statistics(opts) end @doc """ Returns a debug-oriented runtime report. Delegates to `PhoenixGenApi.Diagnostics.debug_report/1`. """ @spec debug_report(keyword()) :: map() def debug_report(opts \\ []) do PhoenixGenApi.Diagnostics.debug_report(opts) end @doc """ Enables legacy Erlang tracing for processes or ports. Requires the `:enable_tracing` admin action. """ @spec trace_processes(term(), keyword()) :: {:ok, map()} | {:error, term()} def trace_processes(targets, opts \\ []) do PhoenixGenApi.Diagnostics.trace_processes(targets, opts) end @doc """ Enables call tracing for specific MFAs. Requires the `:enable_tracing` admin action. """ @spec trace_functions(term(), keyword()) :: {:ok, map()} | {:error, term()} def trace_functions(mfas, opts \\ []) do PhoenixGenApi.Diagnostics.trace_functions(mfas, opts) end @doc """ Disables legacy Erlang tracing for processes or ports. Requires the `:disable_tracing` admin action. """ @spec stop_trace(term(), keyword()) :: {:ok, map()} | {:error, term()} def stop_trace(targets \\ :all, opts \\ []) do PhoenixGenApi.Diagnostics.stop_trace(targets, opts) end @doc """ Disables call tracing for specific MFAs. Requires the `:disable_tracing` admin action. """ @spec stop_trace_functions(term()) :: {:ok, map()} | {:error, term()} def stop_trace_functions(mfas \\ :all) do PhoenixGenApi.Diagnostics.stop_trace_functions(mfas) end @doc """ Returns a small trace status snapshot. """ @spec trace_status() :: map() def trace_status do PhoenixGenApi.Diagnostics.trace_status() end @doc """ [Shell Helper] Traces the call flow for a service/request_type from the gateway to its target nodes. ## Usage in IEx iex> PhoenixGenApi.call_flow("user_service", "get_user") """ @spec call_flow(String.t() | atom(), String.t(), String.t() | nil) :: map() def call_flow(service, request_type, version \\ nil) do PhoenixGenApi.Diagnostics.call_flow(service, request_type, version) end @doc """ [Shell Helper] Inspects a request and returns its full execution plan. ## Usage in IEx iex> PhoenixGenApi.inspect_request(%{service: "user_service", request_type: "get_user"}) """ @spec inspect_request(map()) :: map() def inspect_request(request) do PhoenixGenApi.Diagnostics.inspect_request(request) end @doc """ [Shell Helper] Returns the cluster topology view from this node. ## Usage in IEx iex> PhoenixGenApi.cluster_view() """ @spec cluster_view() :: map() def cluster_view do PhoenixGenApi.Diagnostics.cluster_view() end @doc """ [Shell Helper] Lists all registered call flows across all services. ## Usage in IEx iex> PhoenixGenApi.list_call_flows() iex> PhoenixGenApi.list_call_flows(include_disabled: true) """ @spec list_call_flows(keyword()) :: [map()] def list_call_flows(opts \\ []) do PhoenixGenApi.Diagnostics.list_call_flows(opts) end @doc """ Pushes a `PushConfig` to this server node. This is the server-side API for receiving pushed configs from remote nodes. Remote nodes should use `ConfigPusher.push/2` or `ConfigPusher.push_on_startup/3` instead, which make RPC calls to this function. ## Parameters - `push_config` - A `%PushConfig{}` struct or map that can be decoded into one - `opts` - Options keyword list: - `:force` - Force push even if version matches (default: `false`) ## Returns - `{:ok, :accepted}` - New configs were stored successfully - `{:ok, :skipped, reason}` - Push was skipped (e.g., version matches) - `{:error, reason}` - Push failed (validation error, etc.) ## Examples alias PhoenixGenApi.Structs.PushConfig push_config = %PushConfig{ service: "my_service", nodes: [:"node1@host"], config_version: "1.0.0", fun_configs: [%FunConfig{...}] } {:ok, :accepted} = PhoenixGenApi.push_config(push_config) {:ok, :skipped, :version_matches} = PhoenixGenApi.push_config(push_config) {:ok, :accepted} = PhoenixGenApi.push_config(push_config, force: true) """ @spec push_config(PhoenixGenApi.Structs.PushConfig.t() | map(), keyword()) :: {:ok, :accepted} | {:ok, :skipped, term()} | {:error, term()} def push_config(push_config, opts \\ []) do PhoenixGenApi.ConfigReceiver.push(push_config, opts) end @doc """ Verifies that the server has the given service and config version. Useful for checking whether a push is necessary before sending the full configuration. Remote nodes should use `ConfigPusher.verify/3` instead. ## Parameters - `service` - The service name (string or atom) - `config_version` - The config version string to verify ## Returns - `{:ok, :matched}` - Version matches what is stored - `{:ok, :mismatch, stored_version}` - Version differs from what is stored - `{:error, :not_found}` - Service is not known ## Examples {:ok, :matched} = PhoenixGenApi.verify_config("my_service", "1.0.0") {:ok, :mismatch, "0.9.0"} = PhoenixGenApi.verify_config("my_service", "1.0.0") {:error, :not_found} = PhoenixGenApi.verify_config("unknown_service", "1.0.0") """ @spec verify_config(String.t() | atom(), String.t()) :: {:ok, :matched} | {:ok, :mismatch, String.t()} | {:error, :not_found} def verify_config(service, config_version) do PhoenixGenApi.ConfigReceiver.verify(service, config_version) end @doc """ [Shell Helper] Quick view of pushed services status. Shows all services that have been registered via push, along with their config versions and auto-pull registration status. ## Usage in IEx iex> PhoenixGenApi.pushed_services_status() """ def pushed_services_status() do IO.puts("\n=== Pushed Services Status ===\n") pushed_services = PhoenixGenApi.ConfigReceiver.get_all_pushed_services() if map_size(pushed_services) == 0 do IO.puts("No pushed services registered.") else Enum.each(pushed_services, fn {service, version} -> push_config = PhoenixGenApi.ConfigReceiver.get_pushed_config(service) auto_pull = if push_config.module && push_config.function, do: "Yes", else: "No" IO.puts(" #{inspect(service)}: version=#{inspect(version)}, auto_pull=#{auto_pull}") end) end IO.puts("") :ok end @doc """ [Shell Helper] Quick view of failed FunConfig entries. Shows configs that failed validation during pull or push, with their service, request_type, version, source, node, and reason. ## Usage in IEx iex> PhoenixGenApi.failed_configs() iex> PhoenixGenApi.failed_configs(source: :pull) iex> PhoenixGenApi.failed_configs(source: :push, limit: 20) """ def failed_configs(opts \\ []) do PhoenixGenApi.ConfigFailed.list(opts) end @doc """ [Shell Helper] Print a formatted table of failed FunConfig entries to the console. ## Usage in IEx iex> PhoenixGenApi.failed_configs_print() iex> PhoenixGenApi.failed_configs_print(source: :pull, limit: 10) """ def failed_configs_print(opts \\ []) do entries = PhoenixGenApi.ConfigFailed.list(opts) source_filter = Keyword.get(opts, :source) limit = Keyword.get(opts, :limit, 100) IO.puts("\n=== Failed FunConfig Entries ===") if source_filter do IO.puts("Source: #{source_filter}") end IO.puts("Showing: #{length(entries)} entries (limit: #{limit})") if entries == [] do IO.puts(" (no failed entries)") else header = String.pad_trailing("ID", 8) <> String.pad_trailing("Service", 20) <> String.pad_trailing("Request Type", 22) <> String.pad_trailing("Version", 10) <> String.pad_trailing("Source", 8) <> "Reason" IO.puts("\n" <> header) IO.puts(String.duplicate("-", 90)) Enum.each(entries, fn entry -> id_str = String.pad_trailing("#{entry.id}", 8) svc_str = String.pad_trailing(inspect(entry.service), 20) req_str = String.pad_trailing(inspect(entry.request_type), 22) ver_str = String.pad_trailing(inspect(entry.version), 10) src_str = String.pad_trailing(inspect(entry.source), 8) reason_str = Enum.join(entry.reason, "; ") IO.puts(id_str <> svc_str <> req_str <> ver_str <> src_str <> reason_str) end) end IO.puts("") :ok end @doc """ [Shell Helper] Print a summary of failed FunConfig entries to the console. ## Usage in IEx iex> PhoenixGenApi.failed_configs_summary() """ def failed_configs_summary() do summary = PhoenixGenApi.ConfigFailed.summary() IO.puts("\n=== Failed Configs Summary ===") IO.puts("Total: #{summary.total}") IO.puts("Pull: #{summary.pull}") IO.puts("Push: #{summary.push}") if summary.by_service != %{} do IO.puts("\n--- By Service ---") Enum.each(summary.by_service, fn {service, entries} -> IO.puts(" #{inspect(service)}: #{length(entries)} failed") end) end if summary.newest do IO.puts("\n--- Newest ---") n = summary.newest IO.puts( " ##{n.id} #{inspect(n.service)}/#{inspect(n.request_type)} [#{n.source}] #{Enum.join(n.reason, "; ")}" ) end IO.puts("") :ok end @doc """ Cleans up expired failed config entries (older than 24h). Returns the number of entries removed. """ @spec cleanup_failed_configs() :: non_neg_integer() def cleanup_failed_configs do PhoenixGenApi.ConfigFailed.cleanup() end @doc """ Clears all failed config entries regardless of expiry. """ @spec clear_failed_configs() :: :ok def clear_failed_configs do PhoenixGenApi.ConfigFailed.clear() end defmacro __using__(opts) do quote location: :keep, bind_quoted: [opts: opts] do use PhoenixGenApi.ImplHelper, encoder: Module.concat(Application.compile_env(:phoenix, :json_library, JSON), Encoder), impl: [ PhoenixGenApi.Structs.Response ] @phoenix_gen_api_override_user_id Keyword.get(opts, :override_user_id, true) @phoenix_gen_api_require_verified_user_id Keyword.get(opts, :require_verified_user_id, true) @phoenix_gen_api_event Keyword.get(opts, :event, "phoenix_gen_api") require Logger @impl true @doc false def handle_in(@phoenix_gen_api_event, payload, socket) do Logger.debug(fn -> "[PhoenixGenApi] incoming request, module: #{__MODULE__}, payload: #{inspect(payload)}" end) # Only override user_id from socket assigns if it's a valid non-empty string. # Previously, nil user_ids from socket.assigns would be set in the payload, # which could bypass :any_authenticated permission checks. payload = if @phoenix_gen_api_override_user_id do case Map.get(socket.assigns, :user_id) do user_id when is_binary(user_id) and byte_size(user_id) > 0 -> Map.put(payload, "user_id", user_id) _ -> payload end else payload end # When require_verified_user_id is true, reject requests immediately # if socket.assigns does not contain a verified non-empty user_id. # This prevents unauthenticated requests from reaching permission checks # or function execution. Set to false for public endpoints. {reply_result, push_response} = if @phoenix_gen_api_require_verified_user_id do case Map.get(socket.assigns, :user_id) do user_id when is_binary(user_id) and byte_size(user_id) > 0 -> do_handle_request(payload, socket) _ -> request_id = Map.get(payload, "request_id", "unknown") Logger.warning( "[PhoenixGenApi] rejected unauthenticated request, module: #{__MODULE__}, request_id: #{inspect(request_id)}" ) error_response = PhoenixGenApi.Structs.Response.error_response( request_id, "Authentication required" ) {{:error, "Authentication required"}, error_response} end else do_handle_request(payload, socket) end if push_response do push(socket, @phoenix_gen_api_event, push_response) end {:reply, reply_result, socket} end defp do_handle_request(payload, _socket) do try do request = PhoenixGenApi.Structs.Request.decode!(payload) case PhoenixGenApi.Executor.execute!(request) do %PhoenixGenApi.Structs.Response{} = result -> {{:ok, request.request_type}, result} {:ok, :no_response} -> {{:ok, request.request_type}, nil} end rescue e in PhoenixGenApi.Errors.DecodeError -> request_id = Map.get(payload, "request_id", "unknown") Logger.warning( "[PhoenixGenApi] decode error, module: #{__MODULE__}, request_id: #{inspect(request_id)}, code: #{inspect(e.code)}, error: #{e.message}" ) error_response = PhoenixGenApi.Structs.Response.error_response( request_id, "Invalid request: #{e.message}" ) {{:error, e.message}, error_response} e -> request_id = Map.get(payload, "request_id", "unknown") Logger.error( "[PhoenixGenApi] request processing failed, module: #{__MODULE__}, request_id: #{inspect(request_id)}, error: #{Exception.message(e)}" ) error_response = PhoenixGenApi.Structs.Response.error_response( request_id, "Request processing failed" ) {{:error, Exception.message(e)}, error_response} end end @doc false def handle_info({:stream_started, request_id, pid}, socket) do Process.put({:phoenix_gen_api, :stream_call_pid, request_id}, pid) {:noreply, socket} end @doc false @impl true def handle_info({:push, result}, socket) do Logger.debug(fn -> "[PhoenixGenApi] push result, module: #{__MODULE__}, result: #{inspect(result)}" end) push(socket, @phoenix_gen_api_event, result) {:noreply, socket} end @doc false def handle_info({:stream_response, result}, socket) do Logger.debug(fn -> "[PhoenixGenApi] stream response, module: #{__MODULE__}, result: #{inspect(result)}" end) push(socket, @phoenix_gen_api_event, result) {:noreply, socket} end @doc false def handle_info({:async_call, result}, socket) do Logger.debug(fn -> "[PhoenixGenApi] async call result, module: #{__MODULE__}, result: #{inspect(result)}" end) push(socket, @phoenix_gen_api_event, result) {:noreply, socket} end @doc false def handle_info({:relay_message, result}, socket) do Logger.debug(fn -> "[PhoenixGenApi] relay message, module: #{__MODULE__}, result: #{inspect(result)}" end) push(socket, @phoenix_gen_api_event, result) {:noreply, socket} end end end end