diff --git a/AGENTS.md b/AGENTS.md
index e6d04ca..09a7185 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -24,14 +24,34 @@
 
 ## Project Structure
 
-- **Namespace**: All firehose functionality under `Drinkup.Firehose.*`
-  - `Drinkup.Firehose` - Main supervisor
-  - `Drinkup.Firehose.Consumer` - Behaviour for handling all events
-  - `Drinkup.Firehose.RecordConsumer` - Macro for handling commit record events with filtering
-  - `Drinkup.Firehose.Event` - Event types (`Commit`, `Sync`, `Identity`, `Account`, `Info`)
-  - `Drinkup.Firehose.Socket` - `:gen_statem` WebSocket connection manager
-- **Consumer Pattern**: Implement `@behaviour Drinkup.Firehose.Consumer` with `handle_event/1`
-- **RecordConsumer Pattern**: `use Drinkup.Firehose.RecordConsumer, collections: [~r/app\.bsky\.graph\..+/, "app.bsky.feed.post"]` with `handle_create/1`, `handle_update/1`, `handle_delete/1` overrides
+Each namespace (`Drinkup.Firehose`, `Drinkup.Jetstream`, `Drinkup.Tap`) follows a common architecture:
+
+- **Core Modules**:
+  - `Consumer` - Behaviour/macro for handling events; `use Namespace` with `handle_event/1` implementation
+  - `Event` - Typed event structs specific to the protocol
+  - `Socket` - `:gen_statem` WebSocket connection manager
+  - `Options` (or top-level utility module) - Configuration and runtime utilities
+
+- **Consumer Pattern**: `use Namespace, opts...` with `handle_event/1` callback; consumer module becomes a supervisor
+
+### Namespace-Specific Details
+
+- **Firehose** (`Drinkup.Firehose.*`): Full AT Protocol firehose
+  - Events: `Commit`, `Sync`, `Identity`, `Account`, `Info`
+  - Additional: `RecordConsumer` macro for filtered commit records with `handle_create/1`, `handle_update/1`, `handle_delete/1` callbacks
+  - Pattern: `use Drinkup.Firehose.RecordConsumer, collections: [~r/app\.bsky\.graph\..+/, "app.bsky.feed.post"]`
+
+- **Jetstream** (`Drinkup.Jetstream.*`): Simplified JSON event stream
+  - Events: `Commit`, `Identity`, `Account`
+  - Config: `:wanted_collections`, `:wanted_dids`, `:compress` (zstd)
+  - Utility: `Drinkup.Jetstream.update_options/2` for dynamic filtering
+  - Semantics: Fire-and-forget (no acks)
+
+- **Tap** (`Drinkup.Tap.*`): HTTP API + WebSocket indexer/backfill service
+  - Events: `Record`, `Identity`
+  - Config: `:host`, `:admin_password`, `:disable_acks`
+  - Utility: `Drinkup.Tap` HTTP API functions (`add_repos/2`, `remove_repos/2`, `get_repo_info/2`)
+  - Semantics: Ack/nack - return `:ok`/`{:ok, _}`/`nil` to ack, `{:error, _}` to nack (Tap retries)
 
 ## Important Notes
 
diff --git a/CHANGELOG.md b/CHANGELOG.md
index f05a521..caac5b3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,8 @@ and this project adheres to
 
 ### Breaking Changes
 
+- Simplify usage by removing the concept of a separate "consumer", integrating
+  it directly into the socket's behaviour.
 - Existing behaviour moved to `Drinkup.Firehose` namespace, to make way for
   alternate sync systems.
 
diff --git a/README.md b/README.md
index 23e8e1e..b134568 100644
--- a/README.md
+++ b/README.md
@@ -31,8 +31,9 @@ Documentation can be found on HexDocs at https://hexdocs.pm/drinkup.
 
 ```elixir
 defmodule MyApp.FirehoseConsumer do
-  @behaviour Drinkup.Firehose.Consumer
+  use Drinkup.Firehose
 
+  @impl true
   def handle_event(%Drinkup.Firehose.Event.Commit{} = event) do
     IO.inspect(event, label: "Commit")
   end
@@ -41,15 +42,31 @@ defmodule MyApp.FirehoseConsumer do
 end
 
 # In your supervision tree:
-children = [{Drinkup.Firehose, %{consumer: MyApp.FirehoseConsumer}}]
+children = [MyApp.FirehoseConsumer]
+```
+
+For filtered commit events by collection:
+
+```elixir
+defmodule MyApp.PostConsumer do
+  use Drinkup.Firehose.RecordConsumer,
+    collections: ["app.bsky.feed.post"]
+
+  @impl true
+  def handle_create(record) do
+    IO.inspect(record, label: "New post")
+  end
+end
 ```
 
 ### Jetstream
 
 ```elixir
 defmodule MyApp.JetstreamConsumer do
-  @behaviour Drinkup.Jetstream.Consumer
+  use Drinkup.Jetstream,
+    wanted_collections: ["app.bsky.feed.post"]
 
+  @impl true
   def handle_event(%Drinkup.Jetstream.Event.Commit{} = event) do
     IO.inspect(event, label: "Commit")
   end
@@ -58,37 +75,35 @@ defmodule MyApp.JetstreamConsumer do
 end
 
 # In your supervision tree:
-children = [
-  {Drinkup.Jetstream, %{
-    consumer: MyApp.JetstreamConsumer,
-    wanted_collections: ["app.bsky.feed.post"]
-  }}
-]
+children = [MyApp.JetstreamConsumer]
+
+# Update filters dynamically:
+Drinkup.Jetstream.update_options(MyApp.JetstreamConsumer, %{
+  wanted_collections: ["app.bsky.graph.follow"]
+})
 ```
 
 ### Tap
 
 ```elixir
 defmodule MyApp.TapConsumer do
-  @behaviour Drinkup.Tap.Consumer
+  use Drinkup.Tap,
+    host: "http://localhost:2480"
 
+  @impl true
   def handle_event(%Drinkup.Tap.Event.Record{} = event) do
     IO.inspect(event, label: "Record")
+    :ok
   end
 
-  def handle_event(_), do: :noop
+  def handle_event(_), do: :ok
 end
 
 # In your supervision tree:
-children = [
-  {Drinkup.Tap, %{
-    consumer: MyApp.TapConsumer,
-    host: "http://localhost:2480"
-  }}
-]
+children = [MyApp.TapConsumer]
 
 # Track specific repos:
-Drinkup.Tap.add_repos(Drinkup.Tap, ["did:plc:abc123"])
+Drinkup.Tap.add_repos(MyTap.TapConsumer, ["did:plc:abc123"])
 ```
 
 See [the examples](./examples) for some more complete samples.
diff --git a/examples/firehose/basic_consumer.ex b/examples/firehose/basic_consumer.ex
index 0f114bf..f6e7389 100644
--- a/examples/firehose/basic_consumer.ex
+++ b/examples/firehose/basic_consumer.ex
@@ -1,6 +1,7 @@
 defmodule BasicConsumer do
-  @behaviour Drinkup.Firehose.Consumer
+  use Drinkup.Firehose
 
+  @impl true
   def handle_event(%Drinkup.Firehose.Event.Commit{} = event) do
     IO.inspect(event, label: "Got commit event")
   end
@@ -17,9 +18,7 @@ defmodule ExampleSupervisor do
 
   @impl true
   def init(_) do
-    children = [
-      {Drinkup.Firehose, %{consumer: BasicConsumer}}
-    ]
+    children = [BasicConsumer]
 
     Supervisor.init(children, strategy: :one_for_one)
   end
diff --git a/examples/firehose/multiple_consumers.ex b/examples/firehose/multiple_consumers.ex
index 3fa41a3..d5ef109 100644
--- a/examples/firehose/multiple_consumers.ex
+++ b/examples/firehose/multiple_consumers.ex
@@ -7,8 +7,9 @@ defmodule PostDeleteConsumer do
 end
 
 defmodule IdentityConsumer do
-  @behaviour Drinkup.Firehose.Consumer
+  use Drinkup.Firehose, name: :identities
 
+  @impl true
   def handle_event(%Drinkup.Firehose.Event.Identity{} = event) do
     IO.inspect(event, label: "identity event")
   end
@@ -26,8 +27,8 @@ defmodule ExampleSupervisor do
   @impl true
   def init(_) do
     children = [
-      {Drinkup.Firehose, %{consumer: PostDeleteConsumer}},
-      {Drinkup.Firehose, %{consumer: IdentityConsumer, name: :identities}}
+      PostDeleteConsumer,
+      IdentityConsumer
     ]
 
     Supervisor.init(children, strategy: :one_for_one)
diff --git a/examples/firehose/record_consumer.ex b/examples/firehose/record_consumer.ex
index 945b7d3..6340739 100644
--- a/examples/firehose/record_consumer.ex
+++ b/examples/firehose/record_consumer.ex
@@ -2,14 +2,17 @@ defmodule ExampleRecordConsumer do
   use Drinkup.Firehose.RecordConsumer,
     collections: [~r/app\.bsky\.graph\..+/, "app.bsky.feed.post"]
 
+  @impl true
   def handle_create(record) do
     IO.inspect(record, label: "create")
   end
 
+  @impl true
   def handle_update(record) do
     IO.inspect(record, label: "update")
   end
 
+  @impl true
   def handle_delete(record) do
     IO.inspect(record, label: "delete")
   end
@@ -24,9 +27,7 @@ defmodule ExampleSupervisor do
 
   @impl true
   def init(_) do
-    children = [
-      {Drinkup.Firehose, %{consumer: ExampleRecordConsumer}}
-    ]
+    children = [ExampleRecordConsumer]
 
     Supervisor.init(children, strategy: :one_for_one)
   end
diff --git a/examples/jetstream/jetstream_consumer.ex b/examples/jetstream/jetstream_consumer.ex
index 1cc2777..d7033ed 100644
--- a/examples/jetstream/jetstream_consumer.ex
+++ b/examples/jetstream/jetstream_consumer.ex
@@ -8,8 +8,11 @@ defmodule JetstreamConsumer do
   - Account events (status changes)
   """
 
-  @behaviour Drinkup.Jetstream.Consumer
+  use Drinkup.Jetstream,
+    name: MyJetstream,
+    wanted_collections: ["app.bsky.feed.post", "app.bsky.feed.like"]
 
+  @impl true
   def handle_event(%Drinkup.Jetstream.Event.Commit{operation: :create} = event) do
     IO.inspect(event, label: "New record created")
     :ok
@@ -72,15 +75,7 @@ defmodule ExampleJetstreamSupervisor do
 
   @impl true
   def init(_) do
-    children = [
-      # Connect to public Jetstream instance and filter for posts and likes
-      {Drinkup.Jetstream,
-       %{
-         consumer: JetstreamConsumer,
-         name: MyJetstream,
-         wanted_collections: ["app.bsky.feed.post", "app.bsky.feed.like"]
-       }}
-    ]
+    children = [JetstreamConsumer]
 
     Supervisor.init(children, strategy: :one_for_one)
   end
@@ -88,8 +83,11 @@ end
 
 # Example: Filter for all graph operations (follows, blocks, etc.)
 defmodule GraphEventsConsumer do
-  @behaviour Drinkup.Jetstream.Consumer
+  use Drinkup.Jetstream,
+    name: :graph_events,
+    wanted_collections: ["app.bsky.graph.*"]
 
+  @impl true
   def handle_event(%Drinkup.Jetstream.Event.Commit{collection: "app.bsky.graph." <> _} = event) do
     IO.puts("Graph event: #{event.collection} - #{event.operation}")
     :ok
@@ -100,13 +98,16 @@ end
 
 # Example: Filter for specific DIDs
 defmodule SpecificDIDConsumer do
-  @behaviour Drinkup.Jetstream.Consumer
+  use Drinkup.Jetstream,
+    name: :specific_dids,
+    wanted_dids: ["did:plc:abc123", "did:plc:def456"]
 
   @watched_dids [
     "did:plc:abc123",
     "did:plc:def456"
   ]
 
+  @impl true
   def handle_event(%Drinkup.Jetstream.Event.Commit{did: did} = event)
       when did in @watched_dids do
     IO.puts("Activity from watched DID: #{did}")
diff --git a/examples/tap/tap_consumer.ex b/examples/tap/tap_consumer.ex
index d3d8110..f15ed30 100644
--- a/examples/tap/tap_consumer.ex
+++ b/examples/tap/tap_consumer.ex
@@ -1,16 +1,21 @@
 defmodule TapConsumer do
-  @behaviour Drinkup.Tap.Consumer
+  use Drinkup.Tap,
+    name: MyTap,
+    host: "http://localhost:2480"
 
+  @impl true
   def handle_event(%Drinkup.Tap.Event.Record{} = record) do
     IO.inspect(record, label: "Tap record event")
+    :ok
   end
 
   def handle_event(%Drinkup.Tap.Event.Identity{} = identity) do
     IO.inspect(identity, label: "Tap identity event")
+    :ok
   end
 end
 
-defmodule TapExampleSupervisor do
+defmodule ExampleTapConsumer do
   use Supervisor
 
   def start_link(arg \\ []) do
@@ -19,14 +24,7 @@ defmodule TapExampleSupervisor do
 
   @impl true
   def init(_) do
-    children = [
-      {Drinkup.Tap,
-       %{
-         consumer: TapConsumer,
-         name: MyTap,
-         host: "http://localhost:2480"
-       }}
-    ]
+    children = [TapConsumer]
 
     Supervisor.init(children, strategy: :one_for_one)
   end
diff --git a/lib/firehose.ex b/lib/firehose.ex
index 93b7dab..b2441b1 100644
--- a/lib/firehose.ex
+++ b/lib/firehose.ex
@@ -1,32 +1,140 @@
 defmodule Drinkup.Firehose do
-  use Supervisor
-  alias Drinkup.Firehose.Options
-
-  @dialyzer nowarn_function: {:init, 1}
-  @impl true
-  def init({%Options{name: name} = drinkup_options, supervisor_options}) do
-    children = [
-      {Task.Supervisor, name: {:via, Registry, {Drinkup.Registry, {name, Tasks}}}},
-      {Drinkup.Firehose.Socket, drinkup_options}
-    ]
-
-    Supervisor.start_link(
-      children,
-      supervisor_options ++ [name: {:via, Registry, {Drinkup.Registry, {name, Supervisor}}}]
-    )
-  end
+  @moduledoc """
+  Module for handling events from the AT Protocol [firehose](https://docs.bsky.app/docs/advanced-guides/firehose).
+
+  Due to the nature of the firehose, this will result in a lot of incoming
+  traffic as it receives every repo and identity event within the network. If
+  you're concerened about bandwidth constaints or just don't need a
+  whole-network sync, you may be better off using `Drinkup.Jetstream` or
+  `Drinkup.Tap`.
+
+  ## Usage
+
+      defmodule MyFirehoseConsumer do
+        use Drinkup.Firehose,
+          name: :my_firehose,
+          host: "https://bsky.network",
+          cursor: nil
+
+        @impl true
+        def handle_event(%Drinkup.Firehose.Event.Commit{} = event) do
+          IO.inspect(event, label: "Commit")
+          :ok
+        end
+
+        def handle_event(_event), do: :ok
+      end
+
+      # In your application supervision tree:
+      children = [MyFirehoseConsumer]
+
+  Exceptions raised by `handle_event/1` will be logged instead of killing and
+  restarting the socket process.
+
+  ## Options
+
+  - `:name` - Unique name for this Firehose instance (default: the module name)
+  - `:host` - Firehose relay URL (default: `"https://bsky.network"`)
+  - `:cursor` - Optional sequence number to resume streaming from
+
+  ## Runtime Configuration
+
+  You can override options at runtime by providing them to `child_spec/1`:
+
+      children = [
+        {MyFirehoseConsumer, name: :runtime_name, cursor: 12345}
+      ]
+
+  ## Event Types
+
+  `handle_event/1` will receive the following event structs:
+
+  - `Drinkup.Firehose.Event.Commit` - Repository commits
+  - `Drinkup.Firehose.Event.Sync` - Sync events
+  - `Drinkup.Firehose.Event.Identity` - Identity updates
+  - `Drinkup.Firehose.Event.Account` - Account status changes
+  - `Drinkup.Firehose.Event.Info` - Info messages
+  """
+
+  defmacro __using__(opts) do
+    quote location: :keep, bind_quoted: [opts: opts] do
+      use Supervisor
+      @behaviour Drinkup.Firehose.Consumer
+
+      alias Drinkup.Firehose.Options
+
+      # Store compile-time options as module attributes
+      @name Keyword.get(opts, :name)
+      @host Keyword.get(opts, :host, "https://bsky.network")
+      @cursor Keyword.get(opts, :cursor)
+
+      @doc """
+      Starts the Firehose consumer supervisor.
+
+      Accepts optional runtime configuration that overrides compile-time options.
+      """
+      def start_link(runtime_opts \\ []) do
+        # Merge compile-time and runtime options
+        opts = build_options(runtime_opts)
+        Supervisor.start_link(__MODULE__, opts, name: via_tuple(opts.name))
+      end
+
+      @impl true
+      def init(%Options{name: name} = options) do
+        children = [
+          {Task.Supervisor, name: {:via, Registry, {Drinkup.Registry, {name, Tasks}}}},
+          {Drinkup.Firehose.Socket, options}
+        ]
+
+        Supervisor.init(children, strategy: :one_for_one)
+      end
+
+      @doc """
+      Returns a child spec for adding this consumer to a supervision tree.
+
+      Runtime options override compile-time options.
+      """
+      def child_spec(runtime_opts) when is_list(runtime_opts) do
+        opts = build_options(runtime_opts)
+
+        %{
+          id: opts.name,
+          start: {__MODULE__, :start_link, [runtime_opts]},
+          type: :supervisor,
+          restart: :permanent,
+          shutdown: 500
+        }
+      end
+
+      def child_spec(_opts) do
+        raise ArgumentError, "child_spec expects a keyword list of options"
+      end
+
+      defoverridable child_spec: 1
+
+      # Build Options struct from compile-time and runtime options
+      defp build_options(runtime_opts) do
+        # Compile-time defaults
+        compile_opts = [
+          name: @name || __MODULE__,
+          host: @host,
+          cursor: @cursor
+        ]
+
+        # Merge with runtime opts (runtime takes precedence)
+        merged =
+          compile_opts
+          |> Keyword.merge(runtime_opts)
+          |> Enum.reject(fn {_k, v} -> is_nil(v) end)
+          |> Map.new()
+          |> Map.put(:consumer, __MODULE__)
+
+        Options.from(merged)
+      end
 
-  @spec child_spec(Options.options()) :: Supervisor.child_spec()
-  def child_spec(%{} = options), do: child_spec({options, [strategy: :one_for_one]})
-
-  @spec child_spec({Options.options(), Keyword.t()}) :: Supervisor.child_spec()
-  def child_spec({drinkup_options, supervisor_options}) do
-    %{
-      id: Map.get(drinkup_options, :name, __MODULE__),
-      start: {__MODULE__, :init, [{Options.from(drinkup_options), supervisor_options}]},
-      type: :supervisor,
-      restart: :permanent,
-      shutdown: 500
-    }
+      defp via_tuple(name) do
+        {:via, Registry, {Drinkup.Registry, {name, Supervisor}}}
+      end
+    end
   end
 end
diff --git a/lib/firehose/consumer.ex b/lib/firehose/consumer.ex
index 68203bc..26bca17 100644
--- a/lib/firehose/consumer.ex
+++ b/lib/firehose/consumer.ex
@@ -1,9 +1,9 @@
 defmodule Drinkup.Firehose.Consumer do
   @moduledoc """
-  An unopinionated consumer of the Firehose. Will receive all events, not just commits.
-  """
+  Behaviour for handling Firehose events.
 
-  alias Drinkup.Firehose.Event
+  Implemented by `Drinkup.Firehose`, you'll likely want to be using that instead.
+  """
 
-  @callback handle_event(Event.t()) :: any()
+  @callback handle_event(Drinkup.Firehose.Event.t()) :: any()
 end
diff --git a/lib/firehose/record_consumer.ex b/lib/firehose/record_consumer.ex
index aaf917c..d167905 100644
--- a/lib/firehose/record_consumer.ex
+++ b/lib/firehose/record_consumer.ex
@@ -1,6 +1,50 @@
 defmodule Drinkup.Firehose.RecordConsumer do
   @moduledoc """
-  An opinionated consumer of the Firehose that eats consumers
+  Opinionated consumer of the Firehose focused on record operations.
+
+  This is an abstraction over the core `Drinkup.Firehose` implementation
+  designed for easily handling `commit` events, with the ability to filter by
+  collection. It's similiar to `Drinkup.Jetstream`, but using the Firehose
+  directly (and currently more naive).
+
+  ## Example
+
+      defmodule MyRecordConsumer do
+        use Drinkup.Firehose.RecordConsumer,
+          collections: ["app.bsky.feed.post", ~r/app\\.bsky\\.graph\\..+/],
+          name: :my_records,
+          host: "https://bsky.network"
+
+        @impl true
+        def handle_create(record) do
+          IO.inspect(record, label: "New record")
+        end
+
+        @impl true
+        def handle_delete(record) do
+          IO.inspect(record, label: "Deleted record")
+        end
+      end
+
+      # In your application supervision tree:
+      children = [MyRecordConsumer]
+
+  ## Options
+
+  All options from `Drinkup.Firehose` are supported, plus:
+
+  - `:collections` - List of collection NSIDs (strings or regexes) to filter. If
+    empty or not provided, all collections are processed.
+
+  ## Callbacks
+
+  Implement these callbacks to handle different record actions:
+
+  - `handle_create/1` - Called when a record is created
+  - `handle_update/1` - Called when a record is updated
+  - `handle_delete/1` - Called when a record is deleted
+
+  All callbacks receive a `Drinkup.Firehose.RecordConsumer.Record` struct.
   """
 
   @callback handle_create(any()) :: any()
@@ -8,12 +52,13 @@ defmodule Drinkup.Firehose.RecordConsumer do
   @callback handle_delete(any()) :: any()
 
   defmacro __using__(opts) do
-    {collections, _opts} = Keyword.pop(opts, :collections, [])
+    {collections, firehose_opts} = Keyword.pop(opts, :collections, [])
 
     quote location: :keep do
-      @behaviour Drinkup.Firehose.Consumer
+      use Drinkup.Firehose, unquote(firehose_opts)
       @behaviour Drinkup.Firehose.RecordConsumer
 
+      @impl true
       def handle_event(%Drinkup.Firehose.Event.Commit{} = event) do
         event.ops
         |> Enum.filter(fn %{path: path} ->
diff --git a/lib/jetstream.ex b/lib/jetstream.ex
index eba2507..fe2f9f4 100644
--- a/lib/jetstream.ex
+++ b/lib/jetstream.ex
@@ -1,32 +1,43 @@
 defmodule Drinkup.Jetstream do
   @moduledoc """
-  Supervisor for Jetstream event stream connections.
+  Module for handling events from an AT Protocol
+  [Jetstream](https://github.com/bluesky-social/jetstream) instance.
 
-  Jetstream is a simplified JSON event stream that converts the CBOR-encoded
-  ATProto Firehose into lightweight, friendly JSON events. It provides zstd
-  compression and filtering capabilities for collections and DIDs.
+  Jetstream is an abstraction over the raw AT Protocol firehose that converts
+  the CBOR-encoded events into easier to handle JSON objects, and also provides
+  the ability to filter the events received by repository DID or collection
+  NSID. This is useful when you know specifically which repos or collections you
+  want events from, and thus reduces the amount of bandwidth consumed vs
+  consuming the raw firehose directly.
+
+  If you need a solution for easy backfilling from repositories and not just a
+  firehose translation layer, check out `Drinkup.Tap`.
 
   ## Usage
 
-  Add Jetstream to your supervision tree:
+      defmodule MyJetstreamConsumer do
+        use Drinkup.Jetstream,
+          name: :my_jetstream,
+          wanted_collections: ["app.bsky.feed.post"]
+
+        @impl true
+        def handle_event(event) do
+          IO.inspect(event)
+        end
+      end
 
-      children = [
-        {Drinkup.Jetstream, %{
-          consumer: MyJetstreamConsumer,
-          name: MyJetstream,
-          wanted_collections: ["app.bsky.feed.post", "app.bsky.feed.like"]
-        }}
-      ]
+      # In your application supervision tree:
+      children = [MyJetstreamConsumer]
 
   ## Configuration
 
-  See `Drinkup.Jetstream.Options` for all available configuration options.
+  See `Drinkup.Jetstream.Consumer` for all available configuration options.
 
   ## Dynamic Filter Updates
 
   You can update filters after the connection is established:
 
-      Drinkup.Jetstream.update_options(MyJetstream, %{
+      Drinkup.Jetstream.update_options(:my_jetstream, %{
         wanted_collections: ["app.bsky.graph.follow"],
         wanted_dids: ["did:plc:abc123"]
       })
@@ -36,49 +47,106 @@ defmodule Drinkup.Jetstream do
   By default Drinkup connects to `jetstream2.us-east.bsky.network`.
 
   Bluesky operates a few different Jetstream instances:
-  - `jetstream1.us-east.bsky.network`
-  - `jetstream2.us-east.bsky.network`
-  - `jetstream1.us-west.bsky.network`
-  - `jetstream2.us-west.bsky.network`
-
-  There also some third-party instances not run by Bluesky PBC:
-  - `jetstream.fire.hose.cam`
-  - `jetstream2.fr.hose.cam`
-  - `jetstream1.us-east.fire.hose.cam`
+  - `wss://jetstream1.us-east.bsky.network`
+  - `wss://jetstream2.us-east.bsky.network`
+  - `wss://jetstream1.us-west.bsky.network`
+  - `wss://jetstream2.us-west.bsky.network`
+
+  There also some third-party instances not run by Bluesky PBC, including but not limited to:
+  - `wss://jetstream.fire.hose.cam`
+  - `wss://jetstream2.fr.hose.cam`
+  - `wss://jetstream1.us-east.fire.hose.cam`
+
+  https://firehose.stream/ also hosts several instances around the world.
   """
 
-  use Supervisor
   require Logger
-  alias Drinkup.Jetstream.Options
-
-  @dialyzer nowarn_function: {:init, 1}
-
-  @impl true
-  def init({%Options{name: name} = drinkup_options, supervisor_options}) do
-    children = [
-      {Task.Supervisor, name: {:via, Registry, {Drinkup.Registry, {name, JetstreamTasks}}}},
-      {Drinkup.Jetstream.Socket, drinkup_options}
-    ]
-
-    Supervisor.start_link(
-      children,
-      supervisor_options ++
-        [name: {:via, Registry, {Drinkup.Registry, {name, JetstreamSupervisor}}}]
-    )
-  end
 
-  @spec child_spec(Options.options()) :: Supervisor.child_spec()
-  def child_spec(%{} = options), do: child_spec({options, [strategy: :one_for_one]})
-
-  @spec child_spec({Options.options(), Keyword.t()}) :: Supervisor.child_spec()
-  def child_spec({drinkup_options, supervisor_options}) do
-    %{
-      id: Map.get(drinkup_options, :name, __MODULE__),
-      start: {__MODULE__, :init, [{Options.from(drinkup_options), supervisor_options}]},
-      type: :supervisor,
-      restart: :permanent,
-      shutdown: 500
-    }
+  defmacro __using__(opts) do
+    quote location: :keep, bind_quoted: [opts: opts] do
+      use Supervisor
+      @behaviour Drinkup.Jetstream.Consumer
+
+      alias Drinkup.Jetstream.Options
+
+      # Store compile-time options as module attributes
+      @name Keyword.get(opts, :name)
+      @host Keyword.get(opts, :host, "wss://jetstream2.us-east.bsky.network")
+      @wanted_collections Keyword.get(opts, :wanted_collections, [])
+      @wanted_dids Keyword.get(opts, :wanted_dids, [])
+      @cursor Keyword.get(opts, :cursor)
+      @require_hello Keyword.get(opts, :require_hello, false)
+      @max_message_size_bytes Keyword.get(opts, :max_message_size_bytes)
+
+      @doc """
+      Starts the Jetstream consumer supervisor.
+
+      Accepts optional runtime configuration that overrides compile-time options.
+      """
+      def start_link(runtime_opts \\ []) do
+        opts = build_options(runtime_opts)
+        Supervisor.start_link(__MODULE__, opts, name: via_tuple(opts.name))
+      end
+
+      @impl true
+      def init(%Options{name: name} = options) do
+        children = [
+          {Task.Supervisor, name: {:via, Registry, {Drinkup.Registry, {name, JetstreamTasks}}}},
+          {Drinkup.Jetstream.Socket, options}
+        ]
+
+        Supervisor.init(children, strategy: :one_for_one)
+      end
+
+      @doc """
+      Returns a child spec for adding this consumer to a supervision tree.
+
+      Runtime options override compile-time options.
+      """
+      def child_spec(runtime_opts) when is_list(runtime_opts) do
+        opts = build_options(runtime_opts)
+
+        %{
+          id: opts.name,
+          start: {__MODULE__, :start_link, [runtime_opts]},
+          type: :supervisor,
+          restart: :permanent,
+          shutdown: 500
+        }
+      end
+
+      def child_spec(_opts) do
+        raise ArgumentError, "child_spec expects a keyword list of options"
+      end
+
+      defoverridable child_spec: 1
+
+      # Build Options struct from compile-time and runtime options
+      defp build_options(runtime_opts) do
+        compile_opts = [
+          name: @name || __MODULE__,
+          host: @host,
+          wanted_collections: @wanted_collections,
+          wanted_dids: @wanted_dids,
+          cursor: @cursor,
+          require_hello: @require_hello,
+          max_message_size_bytes: @max_message_size_bytes
+        ]
+
+        merged =
+          compile_opts
+          |> Keyword.merge(runtime_opts)
+          |> Enum.reject(fn {_k, v} -> is_nil(v) end)
+          |> Map.new()
+          |> Map.put(:consumer, __MODULE__)
+
+        Options.from(merged)
+      end
+
+      defp via_tuple(name) do
+        {:via, Registry, {Drinkup.Registry, {name, JetstreamSupervisor}}}
+      end
+    end
   end
 
   # Options Update API
@@ -107,7 +175,7 @@ defmodule Drinkup.Jetstream do
 
   ## Parameters
 
-  - `name` - The name of the Jetstream instance (default: `Drinkup.Jetstream`)
+  - `name` - The name of the Jetstream consumer (the `:name` option passed to `use Drinkup.Jetstream`)
   - `opts` - Map with optional fields:
     - `:wanted_collections` - List of collection NSIDs or prefixes (max 100)
     - `:wanted_dids` - List of DIDs to filter (max 10,000)
@@ -116,17 +184,17 @@ defmodule Drinkup.Jetstream do
   ## Examples
 
       # Filter to only posts
-      Drinkup.Jetstream.update_options(MyJetstream, %{
+      Drinkup.Jetstream.update_options(:my_jetstream, %{
         wanted_collections: ["app.bsky.feed.post"]
       })
 
       # Filter to specific DIDs
-      Drinkup.Jetstream.update_options(MyJetstream, %{
+      Drinkup.Jetstream.update_options(:my_jetstream, %{
         wanted_dids: ["did:plc:abc123", "did:plc:def456"]
       })
 
       # Disable all filters (receive all events)
-      Drinkup.Jetstream.update_options(MyJetstream, %{
+      Drinkup.Jetstream.update_options(:my_jetstream, %{
         wanted_collections: [],
         wanted_dids: []
       })
@@ -140,7 +208,7 @@ defmodule Drinkup.Jetstream do
   Invalid updates will result in the connection being closed by the server.
   """
   @spec update_options(atom(), update_opts()) :: :ok | {:error, term()}
-  def update_options(name \\ Drinkup.Jetstream, opts) when is_map(opts) do
+  def update_options(name, opts) when is_atom(name) and is_map(opts) do
     case find_connection(name) do
       {:ok, {conn, stream}} ->
         message = build_options_update_message(opts)
@@ -154,8 +222,6 @@ defmodule Drinkup.Jetstream do
     end
   end
 
-  # Private functions
-
   @spec find_connection(atom()) :: {:ok, {pid(), :gun.stream_ref()}} | {:error, :not_connected}
   defp find_connection(name) do
     # Look up the connection details from Registry
diff --git a/lib/jetstream/consumer.ex b/lib/jetstream/consumer.ex
index e4cade2..46fbbca 100644
--- a/lib/jetstream/consumer.ex
+++ b/lib/jetstream/consumer.ex
@@ -1,58 +1,8 @@
 defmodule Drinkup.Jetstream.Consumer do
   @moduledoc """
-  Consumer behaviour for handling Jetstream events.
+  Behaviour for handling Jetstream events.
 
-  Implement this behaviour to process events from a Jetstream instance.
-  Events are dispatched asynchronously via `Task.Supervisor`.
-
-  Unlike Tap, Jetstream does not require event acknowledgments. Events are
-  processed in a fire-and-forget manner.
-
-  ## Example
-
-      defmodule MyJetstreamConsumer do
-        @behaviour Drinkup.Jetstream.Consumer
-
-        def handle_event(%Drinkup.Jetstream.Event.Commit{operation: :create} = event) do
-          # Handle new record creation
-          IO.inspect(event, label: "New record")
-          :ok
-        end
-
-        def handle_event(%Drinkup.Jetstream.Event.Commit{operation: :delete} = event) do
-          # Handle record deletion
-          IO.inspect(event, label: "Deleted record")
-          :ok
-        end
-
-        def handle_event(%Drinkup.Jetstream.Event.Identity{} = event) do
-          # Handle identity changes
-          IO.inspect(event, label: "Identity update")
-          :ok
-        end
-
-        def handle_event(%Drinkup.Jetstream.Event.Account{active: false} = event) do
-          # Handle account deactivation
-          IO.inspect(event, label: "Account inactive")
-          :ok
-        end
-
-        def handle_event(_event), do: :ok
-      end
-
-  ## Event Types
-
-  The consumer will receive one of three event types:
-
-  - `Drinkup.Jetstream.Event.Commit` - Repository commits (create, update, delete)
-  - `Drinkup.Jetstream.Event.Identity` - Identity updates (handle changes, etc.)
-  - `Drinkup.Jetstream.Event.Account` - Account status changes (active, taken down, etc.)
-
-  ## Error Handling
-
-  If your `handle_event/1` implementation raises an exception, it will be logged
-  but will not affect the stream. The error is caught and logged by the event
-  dispatcher.
+  Implemented by `Drinkup.Jetstream`, you'll likely want to be using that instead.
   """
 
   alias Drinkup.Jetstream.Event
diff --git a/lib/tap.ex b/lib/tap.ex
index bea13da..4e4862a 100644
--- a/lib/tap.ex
+++ b/lib/tap.ex
@@ -1,72 +1,128 @@
 defmodule Drinkup.Tap do
   @moduledoc """
-  Supervisor and HTTP API for Tap indexer/backfill service.
+  Module for handling events from a
+  [Tap](https://github.com/bluesky-social/indigo/tree/main/cmd/tap) instance.
 
-  Tap simplifies AT sync by handling the firehose connection, verification,
-  backfill, and filtering. Your application connects to a Tap service and
-  receives simple JSON events for only the repos and collections you care about.
+  Tap is a complete sync and backfill solution which handles the firehose
+  connection itself, and automatically searches for repositories to backfill
+  from based on the options given to it. It's great for building an app that
+  wants all of a certain set of records within the AT Protocol network.
 
-  ## Usage
+  This module requires you to be running a properly configured Tap instance, it
+  doesn't spawn one for itself.
 
-  Add Tap to your supervision tree:
+  ## Usage
 
-      children = [
-        {Drinkup.Tap, %{
-          consumer: MyTapConsumer,
-          name: MyTap,
+      defmodule MyTapConsumer do
+        use Drinkup.Tap,
+          name: :my_tap,
           host: "http://localhost:2480",
-          admin_password: "secret"  # optional
-        }}
-      ]
+          admin_password: System.get_env("TAP_PASSWORD")
 
-  Then interact with the Tap HTTP API:
+        @impl true
+        def handle_event(event) do
+          # Process event
+          :ok
+        end
+      end
 
-      # Add repos to track (triggers backfill)
-      Drinkup.Tap.add_repos(MyTap, ["did:plc:abc123"])
+      # In your application supervision tree:
+      children = [MyTapConsumer]
 
-      # Get stats
-      {:ok, count} = Drinkup.Tap.get_repo_count(MyTap)
+  You can also interact with the Tap HTTP API to manually start tracking
+  specific repositories or get information about what's going on.
 
-  ## Configuration
+      # Add repos to track (triggers backfill)
+      Drinkup.Tap.add_repos(:my_tap, ["did:plc:abc123"])
 
-  Tap itself is configured via environment variables. See the Tap documentation
-  for details on configuring collection filters, signal collections, and other
-  operational settings:
-  https://github.com/bluesky-social/indigo/blob/main/cmd/tap/README.md
+      # Get stats
+      {:ok, count} = Drinkup.Tap.get_repo_count(:my_tap)
   """
 
-  use Supervisor
   alias Drinkup.Tap.Options
 
-  @dialyzer nowarn_function: {:init, 1}
-  @impl true
-  def init({%Options{name: name} = drinkup_options, supervisor_options}) do
-    # Register options in Registry for HTTP API access
-    Registry.register(Drinkup.Registry, {name, TapOptions}, drinkup_options)
+  defmacro __using__(opts) do
+    quote location: :keep, bind_quoted: [opts: opts] do
+      use Supervisor
+      @behaviour Drinkup.Tap.Consumer
 
-    children = [
-      {Task.Supervisor, name: {:via, Registry, {Drinkup.Registry, {name, TapTasks}}}},
-      {Drinkup.Tap.Socket, drinkup_options}
-    ]
+      alias Drinkup.Tap.Options
 
-    Supervisor.start_link(
-      children,
-      supervisor_options ++ [name: {:via, Registry, {Drinkup.Registry, {name, TapSupervisor}}}]
-    )
-  end
+      # Store compile-time options as module attributes
+      @name Keyword.get(opts, :name)
+      @host Keyword.get(opts, :host, "http://localhost:2480")
+      @admin_password Keyword.get(opts, :admin_password)
+      @disable_acks Keyword.get(opts, :disable_acks, false)
+
+      @doc """
+      Starts the Tap consumer supervisor.
+
+      Accepts optional runtime configuration that overrides compile-time options.
+      """
+      def start_link(runtime_opts \\ []) do
+        opts = build_options(runtime_opts)
+        Supervisor.start_link(__MODULE__, opts, name: via_tuple(opts.name))
+      end
 
-  @spec child_spec(Options.options()) :: Supervisor.child_spec()
-  def child_spec(%{} = options), do: child_spec({options, [strategy: :one_for_one]})
-
-  @spec child_spec({Options.options(), Keyword.t()}) :: Supervisor.child_spec()
-  def child_spec({drinkup_options, supervisor_options}) do
-    %{
-      id: Map.get(drinkup_options, :name, __MODULE__),
-      start: {__MODULE__, :init, [{Options.from(drinkup_options), supervisor_options}]},
-      type: :supervisor,
-      restart: :permanent,
-      shutdown: 500
-    }
+      @impl true
+      def init(%Options{name: name} = options) do
+        # Register options in Registry for HTTP API access
+        Registry.register(Drinkup.Registry, {name, TapOptions}, options)
+
+        children = [
+          {Task.Supervisor, name: {:via, Registry, {Drinkup.Registry, {name, TapTasks}}}},
+          {Drinkup.Tap.Socket, options}
+        ]
+
+        Supervisor.init(children, strategy: :one_for_one)
+      end
+
+      @doc """
+      Returns a child spec for adding this consumer to a supervision tree.
+
+      Runtime options override compile-time options.
+      """
+      def child_spec(runtime_opts) when is_list(runtime_opts) do
+        opts = build_options(runtime_opts)
+
+        %{
+          id: opts.name,
+          start: {__MODULE__, :start_link, [runtime_opts]},
+          type: :supervisor,
+          restart: :permanent,
+          shutdown: 500
+        }
+      end
+
+      def child_spec(_opts) do
+        raise ArgumentError, "child_spec expects a keyword list of options"
+      end
+
+      defoverridable child_spec: 1
+
+      # Build Options struct from compile-time and runtime options
+      defp build_options(runtime_opts) do
+        compile_opts = [
+          name: @name || __MODULE__,
+          host: @host,
+          admin_password: @admin_password,
+          disable_acks: @disable_acks
+        ]
+
+        merged =
+          compile_opts
+          |> Keyword.merge(runtime_opts)
+          |> Enum.reject(fn {_k, v} -> is_nil(v) end)
+          |> Map.new()
+          |> Map.put(:consumer, __MODULE__)
+
+        Options.from(merged)
+      end
+
+      defp via_tuple(name) do
+        {:via, Registry, {Drinkup.Registry, {name, TapSupervisor}}}
+      end
+    end
   end
 
   # HTTP API Functions
@@ -76,9 +132,14 @@ defmodule Drinkup.Tap do
 
   Triggers backfill for the specified DIDs. Historical events will be fetched
   from each repo's PDS, followed by live events from the firehose.
+
+  ## Parameters
+
+  - `name` - The name of the Tap consumer (the `:name` option passed to `use Drinkup.Tap`)
+  - `dids` - List of DID strings to add
   """
   @spec add_repos(atom(), [String.t()]) :: {:ok, term()} | {:error, term()}
-  def add_repos(name \\ Drinkup.Tap, dids) when is_list(dids) do
+  def add_repos(name, dids) when is_atom(name) and is_list(dids) do
     with {:ok, options} <- get_options(name),
          {:ok, response} <- make_request(options, :post, "/repos/add", %{dids: dids}) do
       {:ok, response}
@@ -90,9 +151,14 @@ defmodule Drinkup.Tap do
 
   Stops syncing the specified repos and deletes tracked repo metadata. Does not
   delete buffered events in the outbox.
+
+  ## Parameters
+
+  - `name` - The name of the Tap consumer (the `:name` option passed to `use Drinkup.Tap`)
+  - `dids` - List of DID strings to remove
   """
   @spec remove_repos(atom(), [String.t()]) :: {:ok, term()} | {:error, term()}
-  def remove_repos(name \\ Drinkup.Tap, dids) when is_list(dids) do
+  def remove_repos(name, dids) when is_atom(name) and is_list(dids) do
     with {:ok, options} <- get_options(name),
          {:ok, response} <- make_request(options, :post, "/repos/remove", %{dids: dids}) do
       {:ok, response}
@@ -101,9 +167,14 @@ defmodule Drinkup.Tap do
 
   @doc """
   Resolve a DID to its DID document.
+
+  ## Parameters
+
+  - `name` - The name of the Tap consumer
+  - `did` - DID string to resolve
   """
   @spec resolve_did(atom(), String.t()) :: {:ok, term()} | {:error, term()}
-  def resolve_did(name \\ Drinkup.Tap, did) when is_binary(did) do
+  def resolve_did(name, did) when is_atom(name) and is_binary(did) do
     with {:ok, options} <- get_options(name),
          {:ok, response} <- make_request(options, :get, "/resolve/#{did}") do
       {:ok, response}
@@ -114,9 +185,14 @@ defmodule Drinkup.Tap do
   Get info about a tracked repo.
 
   Returns repo state, repo rev, record count, error info, and retry count.
+
+  ## Parameters
+
+  - `name` - The name of the Tap consumer
+  - `did` - DID string to get info for
   """
   @spec get_repo_info(atom(), String.t()) :: {:ok, term()} | {:error, term()}
-  def get_repo_info(name \\ Drinkup.Tap, did) when is_binary(did) do
+  def get_repo_info(name, did) when is_atom(name) and is_binary(did) do
     with {:ok, options} <- get_options(name),
          {:ok, response} <- make_request(options, :get, "/info/#{did}") do
       {:ok, response}
@@ -125,9 +201,13 @@ defmodule Drinkup.Tap do
 
   @doc """
   Get the total number of tracked repos.
+
+  ## Parameters
+
+  - `name` - The name of the Tap consumer
   """
   @spec get_repo_count(atom()) :: {:ok, integer()} | {:error, term()}
-  def get_repo_count(name \\ Drinkup.Tap) do
+  def get_repo_count(name) when is_atom(name) do
     with {:ok, options} <- get_options(name),
          {:ok, response} <- make_request(options, :get, "/stats/repo-count") do
       {:ok, response}
@@ -136,9 +216,13 @@ defmodule Drinkup.Tap do
 
   @doc """
   Get the total number of tracked records.
+
+  ## Parameters
+
+  - `name` - The name of the Tap consumer
   """
   @spec get_record_count(atom()) :: {:ok, integer()} | {:error, term()}
-  def get_record_count(name \\ Drinkup.Tap) do
+  def get_record_count(name) when is_atom(name) do
     with {:ok, options} <- get_options(name),
          {:ok, response} <- make_request(options, :get, "/stats/record-count") do
       {:ok, response}
@@ -147,9 +231,13 @@ defmodule Drinkup.Tap do
 
   @doc """
   Get the number of events in the outbox buffer.
+
+  ## Parameters
+
+  - `name` - The name of the Tap consumer
   """
   @spec get_outbox_buffer(atom()) :: {:ok, integer()} | {:error, term()}
-  def get_outbox_buffer(name \\ Drinkup.Tap) do
+  def get_outbox_buffer(name) when is_atom(name) do
     with {:ok, options} <- get_options(name),
          {:ok, response} <- make_request(options, :get, "/stats/outbox-buffer") do
       {:ok, response}
@@ -158,9 +246,13 @@ defmodule Drinkup.Tap do
 
   @doc """
   Get the number of events in the resync buffer.
+
+  ## Parameters
+
+  - `name` - The name of the Tap consumer
   """
   @spec get_resync_buffer(atom()) :: {:ok, integer()} | {:error, term()}
-  def get_resync_buffer(name \\ Drinkup.Tap) do
+  def get_resync_buffer(name) when is_atom(name) do
     with {:ok, options} <- get_options(name),
          {:ok, response} <- make_request(options, :get, "/stats/resync-buffer") do
       {:ok, response}
@@ -169,9 +261,13 @@ defmodule Drinkup.Tap do
 
   @doc """
   Get current firehose and list repos cursors.
+
+  ## Parameters
+
+  - `name` - The name of the Tap consumer
   """
   @spec get_cursors(atom()) :: {:ok, map()} | {:error, term()}
-  def get_cursors(name \\ Drinkup.Tap) do
+  def get_cursors(name) when is_atom(name) do
     with {:ok, options} <- get_options(name),
          {:ok, response} <- make_request(options, :get, "/stats/cursors") do
       {:ok, response}
@@ -182,17 +278,19 @@ defmodule Drinkup.Tap do
   Check Tap health status.
 
   Returns `{:ok, %{"status" => "ok"}}` if healthy.
+
+  ## Parameters
+
+  - `name` - The name of the Tap consumer
   """
   @spec health(atom()) :: {:ok, map()} | {:error, term()}
-  def health(name \\ Drinkup.Tap) do
+  def health(name) when is_atom(name) do
     with {:ok, options} <- get_options(name),
          {:ok, response} <- make_request(options, :get, "/health") do
       {:ok, response}
     end
   end
 
-  # Private Functions
-
   @spec get_options(atom()) :: {:ok, Options.t()} | {:error, :not_found}
   defp get_options(name) do
     case Registry.lookup(Drinkup.Registry, {name, TapOptions}) do
diff --git a/lib/tap/consumer.ex b/lib/tap/consumer.ex
index 2d34ad9..ec0c071 100644
--- a/lib/tap/consumer.ex
+++ b/lib/tap/consumer.ex
@@ -1,43 +1,8 @@
 defmodule Drinkup.Tap.Consumer do
   @moduledoc """
-  Consumer behaviour for handling Tap events.
+  Behaviour for handling Tap events.
 
-  Implement this behaviour to process events from a Tap indexer/backfill service.
-  Events are dispatched asynchronously via `Task.Supervisor` and acknowledged
-  to Tap based on the return value of `handle_event/1`.
-
-  ## Event Acknowledgment
-
-  By default, events are acknowledged to Tap based on your return value:
-
-  - `:ok`, `{:ok, any()}`, or `nil` → Success, event is acked to Tap
-  - `{:error, reason}` → Failure, event is NOT acked (Tap will retry after timeout)
-  - Exception raised → Failure, event is NOT acked (Tap will retry after timeout)
-
-  Any other value will log a warning and acknowledge the event anyway.
-
-  If you set `disable_acks: true` in your Tap options, no acks are sent regardless
-  of the return value. This matches Tap's `TAP_DISABLE_ACKS` environment variable.
-
-  ## Example
-
-      defmodule MyTapConsumer do
-        @behaviour Drinkup.Tap.Consumer
-
-        def handle_event(%Drinkup.Tap.Event.Record{action: :create} = record) do
-          # Handle new record creation
-          case save_to_database(record) do
-            :ok -> :ok  # Success - event will be acked
-            {:error, reason} -> {:error, reason}  # Failure - Tap will retry
-          end
-        end
-
-        def handle_event(%Drinkup.Tap.Event.Identity{} = identity) do
-          # Handle identity changes
-          update_identity(identity)
-          :ok  # Success - event will be acked
-        end
-      end
+  Implemented by `Drinkup.Tap`, you'll likely want to be using that instead.
   """
 
   alias Drinkup.Tap.Event