towerops/docs/references/elixir-best-practices.md

Elixir Best Practices Reference

A comprehensive guide covering OTP patterns, Phoenix LiveView, Ecto, testing, performance, code organization, and deployment.

---

Table of Contents

1. OTP Patterns
2. Phoenix LiveView
3. Ecto Best Practices
4. Testing
5. Performance
6. Code Organization
7. Deployment

---

1. OTP Patterns

GenServer Best Practices

Keep GenServer callbacks thin

The GenServer module should be a thin wrapper. Push business logic into pure functions in separate modules.

# ✅ Good: Thin GenServer, logic in pure module
defmodule MyApp.RateLimiter do
  use GenServer

  def start_link(opts) do
    name = Keyword.fetch!(opts, :name)
    config = Keyword.fetch!(opts, :config)
    GenServer.start_link(__MODULE__, config, name: name)
  end

  def check(server, key), do: GenServer.call(server, {:check, key})

  @impl true
  def init(config) do
    {:ok, RateLimiter.State.new(config)}
  end

  @impl true
  def handle_call({:check, key}, _from, state) do
    {result, new_state} = RateLimiter.Logic.check(state, key)
    {:reply, result, new_state}
  end
end

defmodule MyApp.RateLimiter.Logic do
  # Pure functions — easy to test without starting a process
  def check(state, key) do
    now = System.monotonic_time(:millisecond)
    # ... pure logic returning {result, new_state}
  end
end

# ❌ Bad: Fat GenServer with all logic inline
defmodule MyApp.RateLimiter do
  use GenServer

  @impl true
  def handle_call({:check, key}, _from, state) do
    now = System.monotonic_time(:millisecond)
    bucket = Map.get(state.buckets, key, %{tokens: state.max, last: now})
    elapsed = now - bucket.last
    refilled = min(bucket.tokens + elapsed * state.rate, state.max)
    # ... 40 more lines of logic mixed with GenServer plumbing
  end
end

Always define a client API

# ✅ Good: Client API hides GenServer details
defmodule MyApp.Cache do
  use GenServer

  # --- Client API ---
  def get(server \\ __MODULE__, key) do
    GenServer.call(server, {:get, key})
  end

  def put(server \\ __MODULE__, key, value, ttl \\ :infinity) do
    GenServer.call(server, {:put, key, value, ttl})
  end

  # --- Server Callbacks ---
  @impl true
  def handle_call({:get, key}, _from, state) do
    {:reply, Map.get(state, key), state}
  end
end

# Callers just do:
MyApp.Cache.get(:user_123)

Use handle_continue for post-init work

defmodule MyApp.DataLoader do
  use GenServer

  @impl true
  def init(config) do
    # Don't block the supervisor — return immediately, do heavy work in continue
    {:ok, %{config: config, data: nil}, {:continue, :load_data}}
  end

  @impl true
  def handle_continue(:load_data, state) do
    data = expensive_load(state.config)
    {:noreply, %{state | data: data}}
  end
end

Timeouts and :hibernate

# For processes that are mostly idle, hibernate to reclaim memory
@impl true
def handle_call(:get_status, _from, state) do
  {:reply, state.status, state, :hibernate}
end

# Use Process.send_after for periodic work instead of :timer
@impl true
def init(state) do
  schedule_cleanup()
  {:ok, state}
end

@impl true
def handle_info(:cleanup, state) do
  new_state = do_cleanup(state)
  schedule_cleanup()
  {:noreply, new_state}
end

defp schedule_cleanup do
  Process.send_after(self(), :cleanup, :timer.minutes(5))
end

Reply later with handle_call + manual reply

@impl true
def handle_call(:expensive_op, from, state) do
  # Don't block the GenServer — spawn the work
  Task.start(fn ->
    result = do_expensive_thing()
    GenServer.reply(from, result)
  end)

  {:noreply, state}
end

Supervisor Trees

Use granular supervision strategies

defmodule MyApp.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [
      # Infrastructure first
      MyApp.Repo,
      {Phoenix.PubSub, name: MyApp.PubSub},

      # Application-specific supervisors
      MyApp.WorkerSupervisor,
      MyApp.SchedulerSupervisor,

      # Web layer last (depends on everything above)
      MyAppWeb.Endpoint
    ]

    opts = [strategy: :one_for_one, name: MyApp.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

Choose the right strategy

# :one_for_one — children are independent
# Most common. Use when children don't depend on each other.
Supervisor.start_link(children, strategy: :one_for_one)

# :one_for_all — if one dies, restart all
# Use when children are tightly coupled (e.g., producer + consumer).
Supervisor.start_link(children, strategy: :one_for_all)

# :rest_for_one — if one dies, restart it and all children started after it
# Use when later children depend on earlier ones.
Supervisor.start_link([
  MyApp.Database,        # if this crashes...
  MyApp.Cache,           # ...restart this
  MyApp.API              # ...and this
], strategy: :rest_for_one)

DynamicSupervisor for runtime-spawned children

defmodule MyApp.SessionSupervisor do
  use DynamicSupervisor

  def start_link(init_arg) do
    DynamicSupervisor.start_link(__MODULE__, init_arg, name: __MODULE__)
  end

  @impl true
  def init(_init_arg) do
    DynamicSupervisor.init(strategy: :one_for_one, max_children: 1000)
  end

  def start_session(session_id, opts) do
    spec = {MyApp.Session, [id: session_id] ++ opts}
    DynamicSupervisor.start_child(__MODULE__, spec)
  end

  def stop_session(pid) do
    DynamicSupervisor.terminate_child(__MODULE__, pid)
  end
end

When to Use Agent vs GenServer vs Task

Use Case	Choice	Why
Simple state wrapper	Agent	No custom messages needed
State + complex message handling	GenServer	Need handle_info, timeouts, multi-step logic
Fire-and-forget work	Task.start	Don't need result
Async work, need result	Task.async + Task.await	Short-lived, supervised via caller
Background work under supervisor	Task.Supervisor.start_child	Crash isolation, can be async_nolink
Long-running background job	GenServer	Needs lifecycle management, restart
Pool of concurrent workers	Task.Supervisor	Bounded concurrency with max_children

# Agent: dead simple state. Use sparingly — often a GenServer is clearer.
{:ok, counter} = Agent.start_link(fn -> 0 end, name: MyApp.Counter)
Agent.update(MyApp.Counter, &(&1 + 1))
Agent.get(MyApp.Counter, & &1) #=> 1

# Task: fire-and-forget
Task.start(fn -> send_welcome_email(user) end)

# Task: async/await (short-lived, caller-linked)
task = Task.async(fn -> fetch_external_data(url) end)
result = Task.await(task, 5_000)

# Task.Supervisor: supervised tasks
{:ok, result} = Task.Supervisor.async_nolink(MyApp.TaskSup, fn ->
  risky_external_call()
end) |> Task.yield(10_000) || {:error, :timeout}

Anti-pattern: Using Agent for complex state management

# ❌ Bad: Agent doing too much
Agent.update(pid, fn state ->
  # Complex multi-step logic that should be in a GenServer
  state
  |> validate_transition()
  |> apply_side_effects()  # Side effects inside Agent.update = bad
  |> update_counters()
end)

# ✅ Good: Use GenServer when logic is non-trivial

Process Linking vs Monitoring

# Link: bidirectional. If either process crashes, the other gets an EXIT signal.
# Use for: tightly coupled processes that should die together.
Process.link(pid)

# Monitor: unidirectional. Observer gets a :DOWN message, monitored process is unaffected.
# Use for: watching a process you don't own / shouldn't crash with.
ref = Process.monitor(pid)

receive do
  {:DOWN, ^ref, :process, ^pid, reason} ->
    Logger.warning("Process #{inspect(pid)} went down: #{inspect(reason)}")
end

# In GenServer, handle :DOWN in handle_info:
@impl true
def handle_info({:DOWN, _ref, :process, pid, _reason}, state) do
  {:noreply, remove_worker(state, pid)}
end

Rule of thumb:

• Link = "we live and die together" (parent-child, co-dependent)
• Monitor = "I want to know when you die" (observer pattern)

Registry Patterns

# Start registry in supervision tree
children = [
  {Registry, keys: :unique, name: MyApp.Registry},
  # ...
]

# Register a GenServer via name
defmodule MyApp.Session do
  use GenServer

  def start_link(session_id) do
    GenServer.start_link(__MODULE__, session_id,
      name: via_tuple(session_id)
    )
  end

  def get_state(session_id) do
    GenServer.call(via_tuple(session_id), :get_state)
  end

  defp via_tuple(session_id) do
    {:via, Registry, {MyApp.Registry, {:session, session_id}}}
  end
end

# Lookup
case Registry.lookup(MyApp.Registry, {:session, "abc123"}) do
  [{pid, _value}] -> {:ok, pid}
  [] -> {:error, :not_found}
end

# Dispatch to all registered processes (use :duplicate keys)
{Registry, keys: :duplicate, name: MyApp.PubSubRegistry}

Registry.dispatch(MyApp.PubSubRegistry, :event_type, fn entries ->
  for {pid, _} <- entries, do: send(pid, {:event, payload})
end)

Partitioned Registry for high concurrency

{Registry, keys: :unique, name: MyApp.Registry, partitions: System.schedulers_online()}

---

2. Phoenix LiveView

Assigns Optimization

temporary_assigns — free memory after render

# ✅ Good: Large lists that only need to render once
def mount(_params, _session, socket) do
  {:ok,
   assign(socket, :messages, list_messages()),
   temporary_assigns: [messages: []]}
end

# After the first render, socket.assigns.messages is reset to []
# The client retains the DOM. New messages are appended via handle_event/handle_info.

Streams — the modern replacement for temporary_assigns

Streams (LiveView 0.19+) are the preferred way to handle large collections. They track items by ID and send only diffs.

def mount(_params, _session, socket) do
  {:ok,
   socket
   |> stream(:messages, list_messages())
   |> assign(:page, 1)}
end

# Adding items
def handle_info({:new_message, message}, socket) do
  {:noreply, stream_insert(socket, :messages, message)}
end

# Removing items
def handle_event("delete", %{"id" => id}, socket) do
  message = Messages.get!(id)
  Messages.delete(message)
  {:noreply, stream_delete(socket, :messages, message)}
end

# Bulk reset
def handle_event("refresh", _, socket) do
  {:noreply, stream(socket, :messages, list_messages(), reset: true)}
end

In the template:

<div id="messages" phx-update="stream">
  <div :for={{dom_id, message} <- @streams.messages} id={dom_id}>
    <p><%= message.body %></p>
  </div>
</div>

Anti-pattern: Storing huge lists in assigns without streaming

# ❌ Bad: 10,000 items sitting in process memory permanently
def mount(_, _, socket) do
  {:ok, assign(socket, :items, Repo.all(Item))}  # All in memory, all diffed every render
end

# ✅ Good: Use streams
def mount(_, _, socket) do
  {:ok, stream(socket, :items, Repo.all(Item))}
end

handle_params vs mount

# mount/3 — called ONCE when the LiveView process starts.
# Use for: one-time setup, subscriptions, session-derived state.

# handle_params/3 — called on EVERY URL change (including initial mount).
# Use for: loading data based on URL params, pagination, filtering.

defmodule MyAppWeb.ProductLive.Index do
  use MyAppWeb, :live_view

  # Called once
  @impl true
  def mount(_params, _session, socket) do
    if connected?(socket) do
      Phoenix.PubSub.subscribe(MyApp.PubSub, "products")
    end

    {:ok, socket}
  end

  # Called on every navigation (including initial)
  @impl true
  def handle_params(params, _uri, socket) do
    page = String.to_integer(params["page"] || "1")
    category = params["category"]

    {:noreply,
     socket
     |> assign(:page, page)
     |> assign(:category, category)
     |> stream(:products, list_products(page, category), reset: true)}
  end
end

Anti-pattern: Loading URL-dependent data in mount

# ❌ Bad: Data loaded in mount won't refresh on live_patch navigation
def mount(%{"id" => id}, _session, socket) do
  product = Products.get!(id)
  {:ok, assign(socket, :product, product)}
end

# The user live_patches to /products/456 — mount is NOT called again,
# so the old product stays. Use handle_params instead.

live_session Boundaries

# live_session groups LiveViews that share authentication/layout.
# Navigating between different live_sessions triggers a full page load (dead render + reconnect).
# Within the same live_session, live_patch/live_redirect keeps the WebSocket alive.

scope "/", MyAppWeb do
  pipe_through [:browser]

  live_session :public,
    on_mount: [{MyAppWeb.InitAssigns, :default}] do
    live "/", HomeLive.Index
    live "/about", AboutLive
  end

  live_session :authenticated,
    on_mount: [{MyAppWeb.UserAuth, :ensure_authenticated}] do
    live "/dashboard", DashboardLive
    live "/settings", SettingsLive
    live "/projects/:id", ProjectLive.Show
  end

  live_session :admin,
    on_mount: [{MyAppWeb.UserAuth, :ensure_admin}] do
    live "/admin", AdminLive.Index
    live "/admin/users", AdminLive.Users
  end
end

# on_mount hook for authentication
defmodule MyAppWeb.UserAuth do
  import Phoenix.LiveView
  import Phoenix.Component

  def on_mount(:ensure_authenticated, _params, %{"user_token" => token}, socket) do
    case Accounts.get_user_by_session_token(token) do
      nil -> {:halt, redirect(socket, to: "/login")}
      user -> {:cont, assign(socket, :current_user, user)}
    end
  end

  def on_mount(:ensure_authenticated, _params, _session, socket) do
    {:halt, redirect(socket, to: "/login")}
  end
end

JS Hooks + Server Coordination

# In the template
<div id="chart" phx-hook="Chart" data-points={Jason.encode!(@points)}></div>

# phx-hook value must match the hook name in app.js

// assets/js/hooks/chart.js
const Chart = {
  mounted() {
    this.chart = new ChartLib(this.el, {
      data: JSON.parse(this.el.dataset.points)
    });

    // Receive events from server
    this.handleEvent("update-chart", ({ points }) => {
      this.chart.update(points);
    });

    // Send events to server
    this.el.addEventListener("click", (e) => {
      this.pushEvent("chart-clicked", { x: e.offsetX, y: e.offsetY });
    });
  },

  updated() {
    // Called when the element's attributes change from server
    const points = JSON.parse(this.el.dataset.points);
    this.chart.update(points);
  },

  destroyed() {
    this.chart.destroy();
  }
};

export default Chart;

// assets/js/app.js
import Chart from "./hooks/chart";
import InfiniteScroll from "./hooks/infinite_scroll";

let Hooks = { Chart, InfiniteScroll };
let liveSocket = new LiveSocket("/live", Socket, {
  params: { _csrf_token: csrfToken },
  hooks: Hooks
});

# Server-side: push events to the hook
def handle_info({:data_updated, points}, socket) do
  {:noreply, push_event(socket, "update-chart", %{points: points})}
end

# Handle events from the hook
def handle_event("chart-clicked", %{"x" => x, "y" => y}, socket) do
  # Process click coordinates
  {:noreply, socket}
end

phx-update Strategies

<%!-- Default: replace entire container contents --%>
<div id="notifications">
  <%= for n <- @notifications do %>
    <div id={"notification-#{n.id}"}><%= n.text %></div>
  <% end %>
</div>

<%!-- Stream: managed by stream_insert/stream_delete --%>
<div id="messages" phx-update="stream">
  <div :for={{dom_id, msg} <- @streams.messages} id={dom_id}>
    <%= msg.body %>
  </div>
</div>

<%!-- Append: new children appended, existing preserved (legacy, prefer streams) --%>
<div id="log" phx-update="append">
  <%= for entry <- @log_entries do %>
    <div id={"log-#{entry.id}"}><%= entry.text %></div>
  <% end %>
</div>

<%!-- Ignore: server never updates this container (for JS-managed DOM) --%>
<div id="js-widget" phx-update="ignore">
  <div>JS manages this subtree</div>
</div>

Dead Views vs Live Views

# Dead view: traditional request/response. Use for:
# - Static pages (about, terms, privacy)
# - SEO-critical content that doesn't need interactivity
# - Simple forms that don't need real-time validation

# In router:
get "/terms", PageController, :terms

# Live view: persistent WebSocket. Use for:
# - Real-time updates (dashboards, notifications)
# - Complex interactive forms
# - Collaborative features
# - Any UI that benefits from server push

# Rule of thumb: if it doesn't need a WebSocket, a dead view is simpler and cheaper.

---

3. Ecto Best Practices

Composable Queries

defmodule MyApp.Devices.Query do
  import Ecto.Query

  def base, do: from(d in MyApp.Devices.Device, as: :device)

  def by_site(query, site_id) do
    where(query, [device: d], d.site_id == ^site_id)
  end

  def by_status(query, status) do
    where(query, [device: d], d.status == ^status)
  end

  def with_interfaces(query) do
    preload(query, [:interfaces])
  end

  def search(query, nil), do: query
  def search(query, ""), do: query
  def search(query, term) do
    term = "%#{term}%"
    where(query, [device: d], ilike(d.hostname, ^term) or ilike(d.ip_address, ^term))
  end

  def paginate(query, page, per_page \\ 25) do
    offset = (page - 1) * per_page

    query
    |> limit(^per_page)
    |> offset(^offset)
  end

  def order_by_hostname(query) do
    order_by(query, [device: d], asc: d.hostname)
  end
end

# Usage — compose like pipes:
import MyApp.Devices.Query

base()
|> by_site(site_id)
|> by_status(:active)
|> search(params["q"])
|> order_by_hostname()
|> paginate(page)
|> with_interfaces()
|> Repo.all()

Multi-Tenancy Patterns

Option A: Foreign Key (simpler, recommended for most apps)

# Every query scopes by tenant_id
defmodule MyApp.Devices do
  def list_devices(%Tenant{id: tenant_id}) do
    Device
    |> where(tenant_id: ^tenant_id)
    |> Repo.all()
  end
end

# Enforce at the changeset level too
def changeset(device, attrs, tenant) do
  device
  |> cast(attrs, [:hostname, :ip_address])
  |> put_assoc(:tenant, tenant)
  |> unique_constraint([:hostname, :tenant_id])
end

Option B: Postgres Schema Prefix (stronger isolation)

# Each tenant gets a separate Postgres schema
defmodule MyApp.Repo do
  def put_tenant_prefix(tenant_slug) do
    # Call this at the start of each request
    prefix = "tenant_#{tenant_slug}"
    put_dynamic_repo({__MODULE__, prefix})
    prefix
  end
end

# In queries
def list_devices(tenant_prefix) do
  Device
  |> Repo.all(prefix: tenant_prefix)
end

# Migrations must run per-tenant
def migrate_tenant(tenant_prefix) do
  Ecto.Migrator.run(Repo, :up, all: true, prefix: tenant_prefix)
end

Trade-offs:

• Foreign key: simpler ops, cross-tenant queries easy, must remember to scope every query
• Prefix: stronger isolation, harder cross-tenant queries, migration complexity

Changesets

Embedded schemas for complex nested data

defmodule MyApp.Devices.SNMPConfig do
  use Ecto.Schema
  import Ecto.Changeset

  # Not a database table — embedded in a JSONB column
  embedded_schema do
    field :community, :string
    field :version, Ecto.Enum, values: [:v1, :v2c, :v3]
    field :port, :integer, default: 161
    field :timeout_ms, :integer, default: 5000
  end

  def changeset(config, attrs) do
    config
    |> cast(attrs, [:community, :version, :port, :timeout_ms])
    |> validate_required([:community, :version])
    |> validate_number(:port, greater_than: 0, less_than: 65536)
    |> validate_number(:timeout_ms, greater_than: 0)
  end
end

defmodule MyApp.Devices.Device do
  use Ecto.Schema
  import Ecto.Changeset

  schema "devices" do
    field :hostname, :string
    embeds_one :snmp_config, MyApp.Devices.SNMPConfig, on_replace: :update
    timestamps()
  end

  def changeset(device, attrs) do
    device
    |> cast(attrs, [:hostname])
    |> cast_embed(:snmp_config, required: true)
    |> validate_required([:hostname])
  end
end

cast_assoc vs put_assoc

# cast_assoc: user-facing forms where you cast nested params
# Expects params like %{"addresses" => [%{"street" => "123 Main"}]}
def changeset(user, attrs) do
  user
  |> cast(attrs, [:name])
  |> cast_assoc(:addresses, with: &Address.changeset/2)
end

# put_assoc: programmatic. You already have the struct/changeset.
def add_address(user, address) do
  user
  |> change()
  |> put_assoc(:addresses, [address | user.addresses])
end

Repo.transaction Patterns

Multi for composable transactions

alias Ecto.Multi

def create_team_with_owner(team_attrs, user) do
  Multi.new()
  |> Multi.insert(:team, Team.changeset(%Team{}, team_attrs))
  |> Multi.insert(:membership, fn %{team: team} ->
    Membership.changeset(%Membership{}, %{
      team_id: team.id,
      user_id: user.id,
      role: :owner
    })
  end)
  |> Multi.run(:audit, fn repo, %{team: team} ->
    repo.insert(AuditLog.entry(:team_created, team, user))
  end)
  |> Repo.transaction()
  |> case do
    {:ok, %{team: team, membership: membership}} ->
      {:ok, team}

    {:error, :team, changeset, _changes} ->
      {:error, changeset}

    {:error, :membership, changeset, _changes} ->
      {:error, changeset}
  end
end

Manual transactions with early returns

Repo.transaction(fn ->
  with {:ok, user} <- create_user(attrs),
       {:ok, _profile} <- create_profile(user, profile_attrs),
       :ok <- send_welcome_email(user) do
    user
  else
    {:error, reason} -> Repo.rollback(reason)
  end
end)

Preload Strategies (Avoid N+1)

# ❌ Bad: N+1 queries
devices = Repo.all(Device)
Enum.map(devices, fn d ->
  interfaces = Repo.all(from i in Interface, where: i.device_id == ^d.id)
  %{d | interfaces: interfaces}
end)

# ✅ Good: Preload in the query (single JOIN)
Repo.all(
  from d in Device,
    join: i in assoc(d, :interfaces),
    preload: [interfaces: i]
)

# ✅ Good: Separate preload query (two queries total, often faster than JOIN for large datasets)
Device
|> Repo.all()
|> Repo.preload(:interfaces)

# ✅ Good: Nested preloads
Repo.all(
  from d in Device,
    preload: [interfaces: :ip_addresses, site: :organization]
)

# ✅ Good: Custom preload query
active_interfaces_query = from i in Interface, where: i.status == :up

Repo.all(
  from d in Device,
    preload: [interfaces: ^active_interfaces_query]
)

Upserts

# Insert or update on conflict
def upsert_device(attrs) do
  %Device{}
  |> Device.changeset(attrs)
  |> Repo.insert(
    on_conflict: {:replace, [:hostname, :ip_address, :updated_at]},
    conflict_target: :serial_number,
    returning: true
  )
end

# Bulk upsert
def upsert_metrics(metrics) do
  now = NaiveDateTime.utc_now() |> NaiveDateTime.truncate(:second)

  entries = Enum.map(metrics, fn m ->
    %{
      device_id: m.device_id,
      metric_name: m.name,
      value: m.value,
      inserted_at: now,
      updated_at: now
    }
  end)

  Repo.insert_all(Metric, entries,
    on_conflict: {:replace, [:value, :updated_at]},
    conflict_target: [:device_id, :metric_name],
    returning: true
  )
end

Advisory Locks

# Use PostgreSQL advisory locks for distributed locking
defmodule MyApp.Lock do
  @doc """
  Runs `fun` only if the advisory lock is acquired.
  Lock is automatically released when the transaction commits.
  """
  def with_lock(lock_key, fun) when is_integer(lock_key) do
    Repo.transaction(fn ->
      case Repo.query("SELECT pg_try_advisory_xact_lock($1)", [lock_key]) do
        {:ok, %{rows: [[true]]}} ->
          fun.()

        {:ok, %{rows: [[false]]}} ->
          Repo.rollback(:lock_not_acquired)
      end
    end)
  end

  # Helper to convert string keys to integers
  def key(name) when is_binary(name) do
    :erlang.phash2(name)
  end
end

# Usage
case MyApp.Lock.with_lock(Lock.key("sync:site:#{site_id}"), fn ->
  perform_sync(site_id)
end) do
  {:ok, result} -> {:ok, result}
  {:error, :lock_not_acquired} -> {:ok, :already_running}
end

---

4. Testing

ExUnit Best Practices

Describe blocks and clear naming

defmodule MyApp.DevicesTest do
  use MyApp.DataCase, async: true

  alias MyApp.Devices

  describe "create_device/1" do
    test "creates device with valid attributes" do
      attrs = %{hostname: "switch-01", ip_address: "10.0.0.1"}
      assert {:ok, device} = Devices.create_device(attrs)
      assert device.hostname == "switch-01"
    end

    test "returns error changeset with invalid attributes" do
      assert {:error, %Ecto.Changeset{}} = Devices.create_device(%{})
    end

    test "enforces unique hostname per site" do
      site = insert_site()
      attrs = %{hostname: "switch-01", site_id: site.id}
      {:ok, _} = Devices.create_device(attrs)
      assert {:error, changeset} = Devices.create_device(attrs)
      assert "has already been taken" in errors_on(changeset).hostname
    end
  end
end

Setup blocks

describe "device operations" do
  setup do
    site = insert_site()
    device = insert_device(site: site)
    %{site: site, device: device}
  end

  test "updates device hostname", %{device: device} do
    assert {:ok, updated} = Devices.update_device(device, %{hostname: "new-name"})
    assert updated.hostname == "new-name"
  end
end

# Named setup for reuse
setup :create_user

defp create_user(_context) do
  %{user: insert_user()}
end

Property-Based Testing with StreamData

defmodule MyApp.ParserTest do
  use ExUnit.Case, async: true
  use ExUnitProperties

  property "parsing and formatting are inverse operations" do
    check all ip <- ip_address_generator() do
      assert ip == ip |> MyApp.IP.format() |> MyApp.IP.parse!()
    end
  end

  property "all valid SNMP communities pass validation" do
    check all community <- string(:alphanumeric, min_length: 1, max_length: 32) do
      changeset = SNMPConfig.changeset(%SNMPConfig{}, %{
        community: community,
        version: :v2c
      })
      assert changeset.valid?
    end
  end

  property "rate limiter never allows more than max requests" do
    check all requests <- list_of(constant(:request), min_length: 1, max_length: 100),
              max <- integer(1..10) do
      state = RateLimiter.Logic.new(max_requests: max, window_ms: 1000)

      {allowed, _state} =
        Enum.reduce(requests, {0, state}, fn _, {count, s} ->
          case RateLimiter.Logic.check(s, "key") do
            {:ok, new_s} -> {count + 1, new_s}
            {:error, new_s} -> {count, new_s}
          end
        end)

      assert allowed <= max
    end
  end

  # Custom generators
  defp ip_address_generator do
    gen all a <- integer(0..255),
            b <- integer(0..255),
            c <- integer(0..255),
            d <- integer(0..255) do
      {a, b, c, d}
    end
  end
end

Mox for Behaviours

# 1. Define a behaviour
defmodule MyApp.SNMP.Client do
  @callback get(String.t(), String.t(), keyword()) :: {:ok, term()} | {:error, term()}
  @callback walk(String.t(), String.t(), keyword()) :: {:ok, [term()]} | {:error, term()}
end

# 2. Real implementation
defmodule MyApp.SNMP.NetSNMP do
  @behaviour MyApp.SNMP.Client

  @impl true
  def get(host, oid, opts \\ []) do
    # Real SNMP call
  end
end

# 3. Define mock in test/support/mocks.ex
Mox.defmock(MyApp.SNMP.MockClient, for: MyApp.SNMP.Client)

# 4. Configure in config/test.exs
config :my_app, :snmp_client, MyApp.SNMP.MockClient

# 5. Use in application code
defmodule MyApp.Poller do
  @snmp_client Application.compile_env(:my_app, :snmp_client)

  def poll_device(device) do
    @snmp_client.get(device.ip_address, @sys_descr_oid)
  end
end

# 6. Test with expectations
defmodule MyApp.PollerTest do
  use MyApp.DataCase, async: true

  import Mox

  setup :verify_on_exit!

  test "polls device and stores result" do
    device = insert_device(ip_address: "10.0.0.1")

    expect(MyApp.SNMP.MockClient, :get, fn "10.0.0.1", _oid, _opts ->
      {:ok, "Cisco IOS"}
    end)

    assert {:ok, result} = MyApp.Poller.poll_device(device)
    assert result.sys_descr == "Cisco IOS"
  end

  test "handles SNMP timeout" do
    device = insert_device()

    expect(MyApp.SNMP.MockClient, :get, fn _, _, _ ->
      {:error, :timeout}
    end)

    assert {:error, :timeout} = MyApp.Poller.poll_device(device)
  end
end

Async Test Isolation

# DataCase uses Ecto SQL Sandbox for async isolation
defmodule MyApp.DataCase do
  use ExUnit.CaseTemplate

  using do
    quote do
      alias MyApp.Repo
      import Ecto
      import Ecto.Changeset
      import Ecto.Query
      import MyApp.DataCase
    end
  end

  setup tags do
    MyApp.DataCase.setup_sandbox(tags)
    :ok
  end

  def setup_sandbox(tags) do
    pid = Ecto.Adapters.SQL.Sandbox.start_owner!(MyApp.Repo,
      shared: !tags[:async]
    )
    on_exit(fn -> Ecto.Adapters.SQL.Sandbox.stop_owner(pid) end)
  end
end

# Mark test modules as async when they don't share state
defmodule MyApp.AccountsTest do
  use MyApp.DataCase, async: true  # Runs in parallel
end

Factory Patterns (Without ExMachina)

# Simple factory module — no library needed
defmodule MyApp.Factory do
  alias MyApp.Repo

  def build(:user) do
    %MyApp.Accounts.User{
      email: "user-#{System.unique_integer([:positive])}@example.com",
      name: "Test User",
      hashed_password: Bcrypt.hash_pwd_salt("password123")
    }
  end

  def build(:device) do
    %MyApp.Devices.Device{
      hostname: "device-#{System.unique_integer([:positive])}",
      ip_address: "10.0.0.#{:rand.uniform(254)}",
      status: :active
    }
  end

  def build(:site) do
    %MyApp.Sites.Site{
      name: "Site #{System.unique_integer([:positive])}",
      slug: "site-#{System.unique_integer([:positive])}"
    }
  end

  # Override fields
  def build(factory, attrs) do
    factory
    |> build()
    |> struct!(attrs)
  end

  # Insert into DB
  def insert!(factory, attrs \\ []) do
    factory
    |> build(attrs)
    |> Repo.insert!()
  end

  # Convenience helpers for tests
  def insert_user(attrs \\ []), do: insert!(:user, attrs)
  def insert_device(attrs \\ []), do: insert!(:device, attrs)
  def insert_site(attrs \\ []), do: insert!(:site, attrs)
end

# Import in DataCase:
# import MyApp.Factory

# Usage in tests:
test "deactivates device" do
  device = insert_device(status: :active)
  assert {:ok, updated} = Devices.deactivate(device)
  assert updated.status == :inactive
end

Trait-like patterns

def build(:device, :with_interfaces) do
  device = build(:device)
  interfaces = for i <- 1..4, do: build(:interface, device_id: device.id, index: i)
  %{device | interfaces: interfaces}
end

def insert_device_with_interfaces(attrs \\ []) do
  device = insert_device(attrs)
  for i <- 1..4 do
    insert!(:interface, device_id: device.id, index: i)
  end
  Repo.preload(device, :interfaces)
end

ConnCase vs DataCase

# DataCase: tests that need the database but NOT HTTP/LiveView
# - Context module tests (Accounts.create_user/1)
# - Business logic tests
# - Query tests

# ConnCase: tests that need HTTP connection or LiveView
# - Controller tests
# - LiveView tests
# - Plug tests
# - API endpoint tests

defmodule MyAppWeb.DeviceLiveTest do
  use MyAppWeb.ConnCase, async: true

  import Phoenix.LiveViewTest

  test "lists devices", %{conn: conn} do
    device = insert_device(hostname: "switch-01")
    {:ok, view, html} = live(conn, ~p"/devices")
    assert html =~ "switch-01"
  end

  test "creates device via form", %{conn: conn} do
    {:ok, view, _} = live(conn, ~p"/devices/new")

    view
    |> form("#device-form", device: %{hostname: "new-switch"})
    |> render_submit()

    assert_redirect(view, ~p"/devices")
  end
end

---

5. Performance

ETS for Caching

defmodule MyApp.Cache do
  use GenServer

  @table __MODULE__
  @default_ttl :timer.minutes(5)

  def start_link(_opts) do
    GenServer.start_link(__MODULE__, [], name: __MODULE__)
  end

  @impl true
  def init(_) do
    table = :ets.new(@table, [
      :set,
      :public,            # Any process can read (fast, no bottleneck)
      :named_table,
      read_concurrency: true
    ])

    schedule_cleanup()
    {:ok, %{table: table}}
  end

  # Client API — direct ETS access, no GenServer bottleneck for reads
  def get(key) do
    case :ets.lookup(@table, key) do
      [{^key, value, expires_at}] ->
        if System.monotonic_time(:millisecond) < expires_at do
          {:ok, value}
        else
          :ets.delete(@table, key)
          :miss
        end

      [] ->
        :miss
    end
  end

  def put(key, value, ttl \\ @default_ttl) do
    expires_at = System.monotonic_time(:millisecond) + ttl
    :ets.insert(@table, {key, value, expires_at})
    :ok
  end

  def delete(key), do: :ets.delete(@table, key)

  # Fetch-or-compute pattern
  def fetch(key, ttl \\ @default_ttl, fun) do
    case get(key) do
      {:ok, value} -> value
      :miss ->
        value = fun.()
        put(key, value, ttl)
        value
    end
  end

  # Periodic cleanup of expired entries
  @impl true
  def handle_info(:cleanup, state) do
    now = System.monotonic_time(:millisecond)

    :ets.foldl(fn {key, _value, expires_at}, acc ->
      if now >= expires_at, do: :ets.delete(@table, key)
      acc
    end, nil, @table)

    schedule_cleanup()
    {:noreply, state}
  end

  defp schedule_cleanup, do: Process.send_after(self(), :cleanup, :timer.minutes(1))
end

# Usage
MyApp.Cache.fetch("device:#{device_id}:interfaces", :timer.minutes(10), fn ->
  Repo.preload(device, :interfaces).interfaces
end)

:persistent_term for Config

# persistent_term is optimized for reads (zero-copy), expensive for writes.
# Perfect for: config, feature flags, compiled schemas.

defmodule MyApp.Config do
  def load! do
    # Called once at startup or on config reload
    :persistent_term.put({__MODULE__, :features}, %{
      new_dashboard: true,
      beta_api: false
    })

    :persistent_term.put({__MODULE__, :limits}, %{
      max_devices_per_site: 10_000,
      poll_interval_ms: 30_000
    })
  end

  # Lightning-fast reads — no message passing, no copying
  def feature_enabled?(flag) do
    features = :persistent_term.get({__MODULE__, :features})
    Map.get(features, flag, false)
  end

  def get_limit(key) do
    limits = :persistent_term.get({__MODULE__, :limits})
    Map.fetch!(limits, key)
  end
end

# ❌ Don't: Write to persistent_term frequently
# Every write triggers a global GC of all copies. Use ETS for frequently changing data.

Telemetry Instrumentation

# Emit telemetry events from your application
defmodule MyApp.Devices do
  def poll_device(device) do
    start_time = System.monotonic_time()

    result = do_poll(device)

    :telemetry.execute(
      [:my_app, :device, :poll],
      %{duration: System.monotonic_time() - start_time},
      %{device_id: device.id, site_id: device.site_id, result: elem(result, 0)}
    )

    result
  end

  # Or use :telemetry.span/3 for automatic start/stop/exception events
  def sync_site(site) do
    :telemetry.span([:my_app, :site, :sync], %{site_id: site.id}, fn ->
      result = do_sync(site)
      {result, %{device_count: length(result.devices)}}
    end)
  end
end

# Attach handlers in application startup
defmodule MyApp.Telemetry do
  def setup do
    :telemetry.attach_many("my-app-metrics", [
      [:my_app, :device, :poll],
      [:my_app, :site, :sync, :stop],
      [:my_app, :site, :sync, :exception],
      [:phoenix, :endpoint, :stop],
      [:my_app, :repo, :query]
    ], &handle_event/4, nil)
  end

  def handle_event([:my_app, :device, :poll], measurements, metadata, _config) do
    duration_ms = System.convert_time_unit(measurements.duration, :native, :millisecond)

    if duration_ms > 5_000 do
      Logger.warning("Slow poll for device #{metadata.device_id}: #{duration_ms}ms")
    end

    # Push to metrics system (Prometheus, StatsD, etc.)
    :telemetry.execute([:prometheus, :histogram, :observe], %{
      name: "device_poll_duration_ms",
      value: duration_ms,
      labels: %{result: metadata.result}
    })
  end
end

GenStage / Broadway for Backpressure

# Broadway: higher-level, batteries-included pipeline
defmodule MyApp.SNMPPollPipeline do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {MyApp.PollProducer, []},
        transformer: {__MODULE__, :transform, []},
        concurrency: 1
      ],
      processors: [
        default: [concurrency: 50, max_demand: 10]
      ],
      batchers: [
        default: [batch_size: 100, batch_timeout: 1_000, concurrency: 5]
      ]
    )
  end

  def transform(event, _opts) do
    %Broadway.Message{
      data: event,
      acknowledger: {__MODULE__, :ack_id, :ack_data}
    }
  end

  @impl true
  def handle_message(_, %Broadway.Message{data: device} = message, _) do
    case poll_device(device) do
      {:ok, result} ->
        Broadway.Message.put_data(message, result)

      {:error, reason} ->
        Broadway.Message.failed(message, reason)
    end
  end

  @impl true
  def handle_batch(_, messages, _, _) do
    # Bulk insert results
    results = Enum.map(messages, & &1.data)
    MyApp.Metrics.bulk_insert(results)
    messages
  end

  @impl true
  def handle_failed(messages, _context) do
    Enum.each(messages, fn msg ->
      Logger.error("Poll failed: #{inspect(msg.data)} — #{inspect(msg.status)}")
    end)
    messages
  end
end

Connection Pooling with Finch

# In application supervisor
children = [
  {Finch, name: MyApp.Finch,
    pools: %{
      "https://api.example.com" => [size: 25, count: 2],
      :default => [size: 10]
    }
  }
]

# Usage
def get_device_info(device_id) do
  Finch.build(:get, "https://api.example.com/devices/#{device_id}",
    [{"authorization", "Bearer #{token()}"}]
  )
  |> Finch.request(MyApp.Finch)
  |> case do
    {:ok, %{status: 200, body: body}} -> {:ok, Jason.decode!(body)}
    {:ok, %{status: status}} -> {:error, {:http_error, status}}
    {:error, reason} -> {:error, reason}
  end
end

# Streaming large responses
Finch.build(:get, url)
|> Finch.stream(MyApp.Finch, acc, fn
  {:status, status}, acc -> %{acc | status: status}
  {:headers, headers}, acc -> %{acc | headers: headers}
  {:data, chunk}, acc -> %{acc | body: acc.body <> chunk}
end)

Binary Optimization

# Binaries > 64 bytes are reference-counted (shared, not copied).
# Sub-binaries point into the original — beware keeping references to large binaries.

# ❌ Bad: Keeping a small slice alive holds the entire large binary in memory
def extract_header(<<header::binary-size(4), _rest::binary>>) do
  header  # This is a sub-binary — the entire original stays in memory
end

# ✅ Good: Copy when you need only a small part of a large binary
def extract_header(<<header::binary-size(4), _rest::binary>>) do
  :binary.copy(header)  # New, independent 4-byte binary
end

# Pattern matching on binaries is very efficient
def parse_tlv(<<type::8, length::16, value::binary-size(length), rest::binary>>) do
  {type, value, rest}
end

# Use iodata/iolists to avoid concatenation
# ❌ Bad: Repeated concatenation creates intermediate binaries
def build_response(items) do
  Enum.reduce(items, "", fn item, acc ->
    acc <> "<li>" <> item.name <> "</li>"
  end)
end

# ✅ Good: iolists — zero-copy, sent directly to IO
def build_response(items) do
  Enum.map(items, fn item ->
    ["<li>", item.name, "</li>"]
  end)
end
# Pass to IO.iodata_to_binary/1 only if you truly need a flat binary

---

6. Code Organization

Context Boundaries (DDD-lite)

Phoenix contexts are your primary code organization tool. Think of them as bounded contexts.

lib/my_app/
├── accounts/           # User management
│   ├── accounts.ex     # Public API (the context module)
│   ├── user.ex         # Schema
│   ├── user_token.ex   # Schema
│   └── user_notifier.ex
├── devices/            # Device management
│   ├── devices.ex      # Public API
│   ├── device.ex       # Schema
│   ├── interface.ex    # Schema
│   └── query.ex        # Composable queries
├── monitoring/         # Polling & metrics
│   ├── monitoring.ex
│   ├── poller.ex
│   ├── metric.ex
│   └── alert_rule.ex
└── sites/
    ├── sites.ex
    └── site.ex

The context module is the public API

defmodule MyApp.Devices do
  @moduledoc """
  The Devices context — manages network devices and their interfaces.
  """
  alias MyApp.Repo
  alias MyApp.Devices.{Device, Interface, Query}

  # --- Public API ---

  def list_devices(site_id, opts \\ []) do
    Query.base()
    |> Query.by_site(site_id)
    |> Query.maybe_search(opts[:search])
    |> Query.paginate(opts[:page] || 1, opts[:per_page] || 25)
    |> Repo.all()
  end

  def get_device!(id), do: Repo.get!(Device, id)

  def create_device(attrs) do
    %Device{}
    |> Device.changeset(attrs)
    |> Repo.insert()
  end

  # Cross-context: accept IDs, not structs from other contexts
  def assign_to_site(device_id, site_id) do
    device_id
    |> get_device!()
    |> Device.changeset(%{site_id: site_id})
    |> Repo.update()
  end
end

Anti-pattern: Reaching into another context's schemas

# ❌ Bad: Controller directly queries another context's schema
def index(conn, params) do
  devices = Repo.all(
    from d in MyApp.Devices.Device,
      join: s in MyApp.Sites.Site, on: d.site_id == s.id,
      where: s.organization_id == ^conn.assigns.org_id
  )
end

# ✅ Good: Go through context APIs
def index(conn, params) do
  devices = MyApp.Devices.list_devices_for_org(conn.assigns.org_id, params)
end

When to Split Contexts

Split when:

• Two areas of the codebase change for different reasons
• You're tempted to build circular dependencies between schemas
• A context module exceeds ~300-400 lines
• Different areas need different data access patterns (e.g., one is read-heavy, another write-heavy)

Don't split when:

• It would force every operation to cross context boundaries
• The schemas are tightly coupled and always change together
• You'd just be creating pass-through functions

Service Modules vs Context Functions

# Keep simple CRUD in the context module:
defmodule MyApp.Devices do
  def create_device(attrs), do: ...
  def update_device(device, attrs), do: ...
  def delete_device(device), do: ...
end

# Extract complex operations into dedicated service modules:
defmodule MyApp.Devices.Discovery do
  @moduledoc "Handles SNMP-based device auto-discovery."

  def discover(subnet, opts \\ []) do
    # 50+ lines of complex discovery logic
  end
end

defmodule MyApp.Devices.BulkImport do
  @moduledoc "Imports devices from CSV/spreadsheet."

  def import_csv(path, site_id) do
    # Complex parsing, validation, deduplication
  end
end

# The context still exposes them if needed:
defmodule MyApp.Devices do
  defdelegate discover(subnet, opts \\ []), to: MyApp.Devices.Discovery
  defdelegate import_csv(path, site_id), to: MyApp.Devices.BulkImport
end

Avoiding God Modules

Signs of a god module:

• 500+ lines
• Mix of CRUD, business logic, notifications, reporting
• Functions that don't share state or purpose

# ❌ Bad: God module
defmodule MyApp.Devices do
  def create_device(attrs), do: ...
  def update_device(device, attrs), do: ...
  def poll_device(device), do: ...          # Should be in Monitoring
  def generate_report(site_id), do: ...     # Should be in Reports
  def send_alert(device, message), do: ...  # Should be in Notifications
  def calculate_uptime(device_id), do: ...  # Should be in Monitoring/Analytics
  def export_to_csv(devices), do: ...       # Should be in Exports
  # ... 800 more lines
end

# ✅ Good: Split by responsibility
defmodule MyApp.Devices do        # CRUD + device-specific logic
defmodule MyApp.Monitoring do     # Polling, metrics, uptime
defmodule MyApp.Notifications do  # Alerts, emails
defmodule MyApp.Reports do        # Report generation, exports

---

7. Deployment

Releases with mix release

# mix.exs
def project do
  [
    app: :my_app,
    version: "1.0.0",
    elixir: "~> 1.17",
    releases: [
      my_app: [
        include_executables_for: [:unix],
        steps: [:assemble, :tar],
        cookie: "my-secure-cookie-#{Mix.env()}"
      ]
    ]
  ]
end

# Build a release
MIX_ENV=prod mix release

# Run it
_build/prod/rel/my_app/bin/my_app start

# Remote console
_build/prod/rel/my_app/bin/my_app remote

Runtime Config

# config/runtime.exs — evaluated at boot time, not compile time
import Config

config :my_app, MyApp.Repo,
  url: System.get_env("DATABASE_URL") || raise("DATABASE_URL not set"),
  pool_size: String.to_integer(System.get_env("POOL_SIZE") || "10"),
  ssl: System.get_env("DATABASE_SSL") == "true",
  ssl_opts: [verify: :verify_none]

config :my_app, MyAppWeb.Endpoint,
  url: [host: System.get_env("PHX_HOST") || "localhost", port: 443, scheme: "https"],
  http: [
    ip: {0, 0, 0, 0},
    port: String.to_integer(System.get_env("PORT") || "4000")
  ],
  secret_key_base: System.get_env("SECRET_KEY_BASE") || raise("SECRET_KEY_BASE not set")

# ❌ Bad: Using Application.get_env at compile time for runtime values
# config/config.exs
config :my_app, api_key: System.get_env("API_KEY")  # Baked in at compile time!

# ✅ Good: Use runtime.exs or Application.compile_env only for truly static config

Migrations in Production

# rel/overlays/bin/migrate — migration script for releases
defmodule MyApp.Release do
  @moduledoc "Release tasks (run without Mix)."

  @app :my_app

  def migrate do
    load_app()

    for repo <- repos() do
      {:ok, _, _} =
        Ecto.Migrator.with_repo(repo, &Ecto.Migrator.run(&1, :up, all: true))
    end
  end

  def rollback(repo, version) do
    load_app()
    {:ok, _, _} = Ecto.Migrator.with_repo(repo, &Ecto.Migrator.run(&1, :down, to: version))
  end

  defp repos do
    Application.fetch_env!(@app, :ecto_repos)
  end

  defp load_app do
    Application.ensure_all_started(:ssl)
    Application.load(@app)
  end
end

# Run from release:
# bin/my_app eval "MyApp.Release.migrate()"

Safe migration practices

# ✅ Safe: Add columns as nullable first, backfill, then add constraint
def change do
  alter table(:devices) do
    add :firmware_version, :string  # nullable, no default
  end
end

# Later migration:
def change do
  alter table(:devices) do
    modify :firmware_version, :string, null: false, default: "unknown"
  end
end

# ✅ Safe: Create index concurrently (no table lock)
@disable_ddl_transaction true
@disable_migration_lock true

def change do
  create index(:devices, [:site_id, :status], concurrently: true)
end

# ❌ Dangerous: Renaming columns (breaks running code during deploy)
# Instead: add new column, backfill, deploy code using new column, drop old column

Health Checks

defmodule MyAppWeb.HealthController do
  use MyAppWeb, :controller

  def check(conn, _params) do
    checks = %{
      database: check_database(),
      migrations: check_migrations(),
      memory: check_memory()
    }

    status = if Enum.all?(Map.values(checks), &(&1 == :ok)), do: 200, else: 503

    json(conn, %{
      status: if(status == 200, do: "healthy", else: "unhealthy"),
      checks: checks,
      version: Application.spec(:my_app, :vsn) |> to_string(),
      node: Node.self()
    })
  end

  # Liveness: "is the process alive?" (for k8s liveness probe)
  def liveness(conn, _params) do
    send_resp(conn, 200, "ok")
  end

  # Readiness: "can it serve traffic?" (for k8s readiness probe / load balancer)
  def readiness(conn, _params) do
    if ready?() do
      send_resp(conn, 200, "ready")
    else
      send_resp(conn, 503, "not ready")
    end
  end

  defp check_database do
    case Ecto.Adapters.SQL.query(MyApp.Repo, "SELECT 1", []) do
      {:ok, _} -> :ok
      {:error, _} -> :error
    end
  end

  defp check_migrations do
    # Ensure no pending migrations
    case Ecto.Migrator.migrations(MyApp.Repo) do
      migrations when is_list(migrations) ->
        if Enum.any?(migrations, fn {status, _, _} -> status == :down end),
          do: :pending,
          else: :ok

      _ ->
        :error
    end
  end

  defp check_memory do
    mem = :erlang.memory(:total)
    # Alert if using more than 1GB
    if mem > 1_073_741_824, do: :warning, else: :ok
  end

  defp ready? do
    # Check if all required services are up
    check_database() == :ok
  end
end

# Router
scope "/health", MyAppWeb do
  pipe_through :api
  get "/", HealthController, :check
  get "/live", HealthController, :liveness
  get "/ready", HealthController, :readiness
end

Graceful Shutdown

# BEAM handles SIGTERM gracefully by default in releases.
# Customize shutdown behavior:

defmodule MyApp.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [
      MyApp.Repo,
      {Phoenix.PubSub, name: MyApp.PubSub},
      # Workers that need graceful shutdown
      {MyApp.Poller, shutdown: :timer.seconds(30)},
      MyAppWeb.Endpoint
    ]

    opts = [strategy: :one_for_one, name: MyApp.Supervisor]
    Supervisor.start_link(children, opts)
  end

  @impl true
  def prep_stop(_state) do
    # Called before supervision tree shuts down
    # Deregister from load balancer, stop accepting new work
    Logger.info("Application shutting down, draining connections...")
    MyAppWeb.Endpoint.config_change([], [])
  end

  @impl true
  def stop(_state) do
    Logger.info("Application stopped")
  end
end

# In GenServer, handle terminate for cleanup
defmodule MyApp.Poller do
  use GenServer

  @impl true
  def terminate(reason, state) do
    Logger.info("Poller terminating: #{inspect(reason)}")
    # Flush pending results, close connections, etc.
    flush_pending(state)
    :ok
  end
end

# vm.args.eex — configure shutdown timeout
# In rel/vm.args.eex:
# +sbwt none
# -shutdown_time 30000

Rolling Deploys

# Dockerfile for releases
# syntax=docker/dockerfile:1
FROM hexpm/elixir:1.17.3-erlang-27.1-debian-bookworm-20240904 AS build

RUN apt-get update && apt-get install -y git build-essential

WORKDIR /app
ENV MIX_ENV=prod

COPY mix.exs mix.lock ./
RUN mix deps.get --only $MIX_ENV
RUN mix deps.compile

COPY config/config.exs config/prod.exs config/
COPY lib lib
COPY priv priv
COPY assets assets

RUN mix assets.deploy
RUN mix compile
COPY config/runtime.exs config/
RUN mix release

# --- Runtime ---
FROM debian:bookworm-slim

RUN apt-get update && apt-get install -y libssl3 libncurses6 locales \
  && rm -rf /var/lib/apt/lists/*

RUN sed -i '/en_US.UTF-8/s/^# //g' /etc/locale.gen && locale-gen
ENV LANG=en_US.UTF-8

WORKDIR /app
COPY --from=build /app/_build/prod/rel/my_app ./

ENV PHX_SERVER=true
CMD ["bin/my_app", "eval", "MyApp.Release.migrate()"]  # Run migrations
CMD ["bin/my_app", "start"]

Blue-green / rolling deploy checklist

1. Database migrations first — run migrate before deploying new code
2. Backward-compatible migrations only — new code must work with old schema during transition
3. Health check gates — don't route traffic until /health/ready returns 200
4. Graceful drain — send SIGTERM, wait for in-flight requests to complete (30s timeout)
5. Connection draining — WebSocket/LiveView connections reconnect automatically
6. Feature flags — for risky changes, deploy code behind a flag, enable after verification

# Example k8s deployment spec
apiVersion: apps/v1
kind: Deployment
spec:
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  template:
    spec:
      terminationGracePeriodSeconds: 30
      containers:
        - name: app
          livenessProbe:
            httpGet:
              path: /health/live
              port: 4000
            initialDelaySeconds: 5
            periodSeconds: 10
          readinessProbe:
            httpGet:
              path: /health/ready
              port: 4000
            initialDelaySeconds: 10
            periodSeconds: 5

---

Quick Reference: Common Anti-Patterns

Anti-Pattern	Better Approach
Fat GenServer callbacks	Extract logic into pure-function modules
Application.get_env in module body	Application.compile_env or runtime.exs
Loading URL-dependent data in mount	Use handle_params
Storing large lists in assigns	Use stream/3
N+1 Ecto queries in loops	Preload or batch queries
God context module (500+ lines)	Split by responsibility
Agent for complex state	Use GenServer
Renaming DB columns in one migration	Add → backfill → migrate code → drop
System.get_env in config.exs	Move to runtime.exs
String.concat in loops	Use iolists
Frequent :persistent_term writes	Use ETS for mutable cache
Blocking work in GenServer.init	Use handle_continue
Process.link for observing	Use Process.monitor
Repo.all then filter in Elixir	Filter in the query