diff --git a/CHANGELOG.md b/CHANGELOG.md
index fa1674c..c087e89 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,11 +8,16 @@ and this project adheres to
 
 ## [Unreleased]
 
-### Breaking Change
+### Breaking Changes
 
 - Existing behaviour moved to `Drinkup.Firehose` namespace, to make way for
   alternate sync systems.
 
+### Changed
+
+- Refactor core connection logic for websockets into `Drinkup.Socket` to make it
+  easy to use across multiple different services.
+
 ## [0.1.0] - 2025-05-26
 
 Initial release.
diff --git a/lib/firehose/options.ex b/lib/firehose/options.ex
index 53e4bae..172b186 100644
--- a/lib/firehose/options.ex
+++ b/lib/firehose/options.ex
@@ -1,20 +1,75 @@
 defmodule Drinkup.Firehose.Options do
+  @moduledoc """
+  Configuration options for ATProto Firehose relay subscriptions.
+
+  This module defines the configuration structure for connecting to and
+  consuming events from an ATProto Firehose relay. The Firehose streams
+  real-time repository events from the AT Protocol network.
+
+  ## Options
+
+  - `:consumer` (required) - Module implementing `Drinkup.Firehose.Consumer` behaviour
+  - `:name` - Unique name for this Firehose instance in the supervision tree (default: `Drinkup.Firehose`)
+  - `:host` - Firehose relay URL (default: `"https://bsky.network"`)
+  - `:cursor` - Optional sequence number to resume streaming from
+
+  ## Example
+
+      %{
+        consumer: MyFirehoseConsumer,
+        name: MyFirehose,
+        host: "https://bsky.network",
+        cursor: 12345
+      }
+  """
+
   use TypedStruct
 
   @default_host "https://bsky.network"
 
+  @typedoc """
+  Map of configuration options accepted by `Drinkup.Firehose.child_spec/1`.
+  """
   @type options() :: %{
-          required(:consumer) => module(),
-          optional(:name) => atom(),
-          optional(:host) => String.t(),
-          optional(:cursor) => pos_integer()
+          required(:consumer) => consumer(),
+          optional(:name) => name(),
+          optional(:host) => host(),
+          optional(:cursor) => cursor()
         }
 
+  @typedoc """
+  Module implementing the `Drinkup.Firehose.Consumer` behaviour.
+  """
+  @type consumer() :: module()
+
+  @typedoc """
+  Unique identifier for this Firehose instance in the supervision tree.
+
+  Used for Registry lookups and naming child processes.
+  """
+  @type name() :: atom()
+
+  @typedoc """
+  HTTP/HTTPS URL of the ATProto Firehose relay.
+
+  Defaults to `"https://bsky.network"` which is the public Bluesky relay.
+  """
+  @type host() :: String.t()
+
+  @typedoc """
+  Optional sequence number to resume streaming from.
+
+  When provided, the Firehose will replay events starting from this sequence
+  number. Useful for resuming after a restart without missing events. The
+  cursor is automatically tracked and updated as events are received.
+  """
+  @type cursor() :: pos_integer() | nil
+
   typedstruct do
-    field :consumer, module(), enforce: true
-    field :name, atom(), default: Drinkup
-    field :host, String.t(), default: @default_host
-    field :cursor, pos_integer() | nil
+    field :consumer, consumer(), enforce: true
+    field :name, name(), default: Drinkup.Firehose
+    field :host, host(), default: @default_host
+    field :cursor, cursor()
   end
 
   @spec from(options()) :: t()
diff --git a/lib/firehose/socket.ex b/lib/firehose/socket.ex
index 61fed13..b3922fc 100644
--- a/lib/firehose/socket.ex
+++ b/lib/firehose/socket.ex
@@ -1,119 +1,52 @@
 defmodule Drinkup.Firehose.Socket do
   @moduledoc """
-  gen_statem process for managing the websocket connection to an ATProto relay.
+  WebSocket connection handler for ATProto relay subscriptions.
+
+  Implements the Drinkup.Socket behaviour to manage connections to an ATProto
+  Firehose relay, handling CAR/CBOR-encoded frames and dispatching events to
+  the configured consumer.
   """
 
+  use Drinkup.Socket
+
   require Logger
   alias Drinkup.Firehose.{Event, Options}
 
-  @behaviour :gen_statem
-  @timeout :timer.seconds(5)
-  # TODO: `flow` determines messages in buffer. Determine ideal value?
-  @flow 10
-
   @op_regular 1
   @op_error -1
 
-  defstruct [:options, :seq, :conn, :stream]
-
   @impl true
-  def callback_mode, do: [:state_functions, :state_enter]
-
-  def child_spec(opts) do
-    %{
-      id: __MODULE__,
-      start: {__MODULE__, :start_link, [opts, []]},
-      type: :worker,
-      restart: :permanent,
-      shutdown: 500
-    }
+  def init(opts) do
+    options = Keyword.fetch!(opts, :options)
+    {:ok, %{seq: options.cursor, options: options, host: options.host}}
   end
 
   def start_link(%Options{} = options, statem_opts) do
-    :gen_statem.start_link(__MODULE__, options, statem_opts)
-  end
-
-  @impl true
-  def init(%{cursor: seq} = options) do
-    data = %__MODULE__{seq: seq, options: options}
-    {:ok, :disconnected, data, [{:next_event, :internal, :connect}]}
-  end
-
-  def disconnected(:enter, _from, data) do
-    Logger.debug("Initial connection")
-    # TODO: differentiate between initial & reconnects, probably stuff to do with seq
-    {:next_state, :disconnected, data}
-  end
-
-  def disconnected(:internal, :connect, data) do
-    {:next_state, :connecting_http, data}
-  end
-
-  def connecting_http(:enter, _from, %{options: options} = data) do
-    Logger.debug("Connecting to http")
-
-    %{host: host, port: port} = URI.new!(options.host)
-
-    {:ok, conn} =
-      :gun.open(:binary.bin_to_list(host), port, %{
-        retry: 0,
-        protocols: [:http],
-        connect_timeout: @timeout,
-        domain_lookup_timeout: @timeout,
-        tls_handshake_timeout: @timeout,
-        tls_opts: [
-          verify: :verify_peer,
-          cacerts: :certifi.cacerts(),
-          depth: 3,
-          customize_hostname_check: [
-            match_fun: :public_key.pkix_verify_hostname_match_fun(:https)
-          ]
-        ]
-      })
-
-    {:keep_state, %{data | conn: conn}, [{:state_timeout, @timeout, :connect_timeout}]}
-  end
-
-  def connecting_http(:info, {:gun_up, _conn, :http}, data) do
-    {:next_state, :connecting_ws, data}
-  end
+    # Build opts for Drinkup.Socket from Options struct
+    socket_opts = [
+      host: options.host,
+      cursor: options.cursor,
+      options: options
+    ]
 
-  def connecting_http(:state_timeout, :connect_timeout, _data) do
-    {:stop, :connect_http_timeout}
+    Drinkup.Socket.start_link(__MODULE__, socket_opts, statem_opts)
   end
 
-  def connecting_ws(:enter, _from, %{conn: conn, seq: seq} = data) do
-    Logger.debug("Upgrading connection to websocket")
-    path = "/xrpc/com.atproto.sync.subscribeRepos?" <> URI.encode_query(%{cursor: seq})
-    stream = :gun.ws_upgrade(conn, path, [], %{flow: @flow})
-    {:keep_state, %{data | stream: stream}, [{:state_timeout, @timeout, :upgrade_timeout}]}
-  end
-
-  def connecting_ws(:info, {:gun_upgrade, _conn, _stream, ["websocket"], _headers}, data) do
-    {:next_state, :connected, data}
-  end
-
-  def connecting_ws(:state_timeout, :upgrade_timeout, _data) do
-    {:stop, :connect_ws_timeout}
-  end
-
-  def connected(:enter, _from, _data) do
-    Logger.debug("Connected to websocket")
-    :keep_state_and_data
+  @impl true
+  def build_path(%{seq: seq}) do
+    cursor_param = if seq, do: %{cursor: seq}, else: %{}
+    "/xrpc/com.atproto.sync.subscribeRepos?" <> URI.encode_query(cursor_param)
   end
 
-  def connected(:info, {:gun_ws, conn, stream, {:binary, frame}}, %{options: options} = data) do
-    # TODO: let clients specify a handler for raw* (*decoded) packets to support any atproto subscription
-    # Will also need support for JSON frames
+  @impl true
+  def handle_frame({:binary, frame}, %{seq: seq, options: options} = data) do
     with {:ok, header, next} <- CAR.DagCbor.decode(frame),
          {:ok, payload, _} <- CAR.DagCbor.decode(next),
          {%{"op" => @op_regular, "t" => type}, _} <- {header, payload},
-         true <- Event.valid_seq?(data.seq, payload["seq"]) do
-      data = %{data | seq: payload["seq"] || data.seq}
-      message = Event.from(type, payload)
-      :ok = :gun.update_flow(conn, stream, @flow)
+         true <- Event.valid_seq?(seq, payload["seq"]) do
+      new_seq = payload["seq"] || seq
 
-      case message do
+      case Event.from(type, payload) do
         nil ->
           Logger.warning("Received unrecognised event from firehose: #{inspect({type, payload})}")
 
@@ -121,49 +54,37 @@ defmodule Drinkup.Firehose.Socket do
           Event.dispatch(message, options)
       end
 
-      {:keep_state, data}
+      {:ok, %{data | seq: new_seq}}
     else
       false ->
         Logger.error("Got out of sequence or invalid `seq` from Firehose")
-        {:keep_state, data}
+        :noop
 
       {%{"op" => @op_error, "t" => type}, payload} ->
         Logger.error("Got error from Firehose: #{inspect({type, payload})}")
-        {:keep_state, data}
+        :noop
 
       {:error, reason} ->
         Logger.warning("Failed to decode frame from Firehose: #{inspect(reason)}")
-        {:keep_state, data}
+        :noop
     end
   end
 
-  def connected(:info, {:gun_ws, _conn, _stream, :close}, _data) do
+  @impl true
+  def handle_frame(:close, _data) do
     Logger.info("Websocket closed, reason unknown")
-    {:keep_state_and_data, [{:next_event, :internal, :reconnect}]}
+    nil
   end
 
-  def connected(:info, {:gun_ws, _conn, _stream, {:close, errno, reason}}, _data) do
+  @impl true
+  def handle_frame({:close, errno, reason}, _data) do
     Logger.info("Websocket closed, errno: #{errno}, reason: #{inspect(reason)}")
-    {:keep_state_and_data, [{:next_event, :internal, :reconnect}]}
-  end
-
-  def connected(:info, {:gun_down, old_conn, _proto, _reason, _killed_streams}, %{conn: new_conn})
-      when old_conn != new_conn do
-    Logger.debug("Ignoring received :gun_down for a previous connection.")
-    :keep_state_and_data
+    nil
   end
 
-  def connected(:info, {:gun_down, _conn, _proto, _reason, _killed_streams}, _data) do
-    Logger.info("Websocket connection killed. Attempting to reconnect")
-    {:keep_state_and_data, [{:next_event, :internal, :reconnect}]}
-  end
-
-  def connected(:internal, :reconnect, %{conn: conn} = data) do
-    :ok = :gun.close(conn)
-    :ok = :gun.flush(conn)
-
-    # TODO: reconnect backoff
-    {:next_state, :disconnected, %{data | conn: nil, stream: nil},
-     [{:next_event, :internal, :connect}]}
+  @impl true
+  def handle_frame({:text, _text}, _data) do
+    Logger.warning("Received unexpected text frame from Firehose")
+    :noop
   end
 end
diff --git a/lib/socket.ex b/lib/socket.ex
new file mode 100644
index 0000000..119b3cb
--- /dev/null
+++ b/lib/socket.ex
@@ -0,0 +1,341 @@
+defmodule Drinkup.Socket do
+  # TODO: talk about how to implment, but that it's for internal use
+  @moduledoc false
+
+  require Logger
+
+  @behaviour :gen_statem
+
+  @type frame ::
+          {:binary, binary()}
+          | {:text, String.t()}
+          | :close
+          | {:close, errno :: integer(), reason :: binary()}
+
+  @type user_data :: term()
+
+  @type reconnect_strategy ::
+          :exponential
+          | {:exponential, max_backoff :: pos_integer()}
+          | {:custom, (attempt :: pos_integer() -> delay_ms :: pos_integer())}
+
+  @type option ::
+          {:host, String.t()}
+          | {:flow, pos_integer()}
+          | {:timeout, pos_integer()}
+          | {:tls_opts, keyword()}
+          | {:gun_opts, map()}
+          | {:reconnect_strategy, reconnect_strategy()}
+          | {atom(), term()}
+
+  @callback init(opts :: keyword()) :: {:ok, user_data()} | {:error, reason :: term()}
+
+  @callback build_path(data :: user_data()) :: String.t()
+
+  @callback handle_frame(frame :: frame(), data :: user_data()) ::
+              {:ok, new_data :: user_data()} | :noop | nil | {:error, reason :: term()}
+
+  @callback handle_connected(data :: user_data()) :: {:ok, new_data :: user_data()}
+
+  @callback handle_disconnected(reason :: term(), data :: user_data()) ::
+              {:ok, new_data :: user_data()}
+
+  @optional_callbacks handle_connected: 1, handle_disconnected: 2
+
+  defstruct [
+    :module,
+    :user_data,
+    :options,
+    :conn,
+    :stream,
+    reconnect_attempts: 0
+  ]
+
+  defmacro __using__(_opts) do
+    quote do
+      @behaviour Drinkup.Socket
+
+      def start_link(opts, statem_opts \\ [])
+
+      def start_link(opts, statem_opts) do
+        Drinkup.Socket.start_link(__MODULE__, opts, statem_opts)
+      end
+
+      defoverridable start_link: 2
+
+      def child_spec(opts) do
+        %{
+          id: __MODULE__,
+          start: {__MODULE__, :start_link, [opts, []]},
+          type: :worker,
+          restart: :permanent,
+          shutdown: 500
+        }
+      end
+
+      defoverridable child_spec: 1
+
+      @impl true
+      def handle_connected(data), do: {:ok, data}
+
+      @impl true
+      def handle_disconnected(_reason, data), do: {:ok, data}
+
+      defoverridable handle_connected: 1, handle_disconnected: 2
+    end
+  end
+
+  @impl true
+  def callback_mode, do: [:state_functions, :state_enter]
+
+  @doc """
+  Start a WebSocket connection process.
+
+  ## Parameters
+
+    * `module` - The module implementing the Drinkup.Socket behaviour
+    * `opts` - Keyword list of options (see module documentation)
+    * `statem_opts` - Options passed to `:gen_statem.start_link/3`
+  """
+  def start_link(module, opts, statem_opts) do
+    :gen_statem.start_link(__MODULE__, {module, opts}, statem_opts)
+  end
+
+  @impl true
+  def init({module, opts}) do
+    case module.init(opts) do
+      {:ok, user_data} ->
+        options = parse_options(opts)
+
+        data = %__MODULE__{
+          module: module,
+          user_data: user_data,
+          options: options,
+          reconnect_attempts: 0
+        }
+
+        {:ok, :disconnected, data, [{:next_event, :internal, :connect}]}
+
+      {:error, reason} ->
+        {:stop, {:init_failed, reason}}
+    end
+  end
+
+  # :disconnected state - waiting to connect or reconnect
+
+  def disconnected(:enter, _from, _data) do
+    Logger.debug("[Drinkup.Socket] Entering disconnected state")
+    :keep_state_and_data
+  end
+
+  def disconnected(:internal, :connect, data) do
+    {:next_state, :connecting_http, data}
+  end
+
+  def disconnected(:timeout, :reconnect, data) do
+    {:next_state, :connecting_http, data}
+  end
+
+  # :connecting_http state - establishing HTTP connection with TLS
+
+  def connecting_http(:enter, _from, %{options: options} = data) do
+    Logger.debug("[Drinkup.Socket] Connecting to HTTP")
+
+    %{host: host, port: port} = URI.new!(options.host)
+
+    gun_opts =
+      Map.merge(
+        %{
+          retry: 0,
+          protocols: [:http],
+          connect_timeout: options.timeout,
+          domain_lookup_timeout: options.timeout,
+          tls_handshake_timeout: options.timeout,
+          tls_opts: options.tls_opts
+        },
+        options.gun_opts
+      )
+
+    case :gun.open(:binary.bin_to_list(host), port, gun_opts) do
+      {:ok, conn} ->
+        {:keep_state, %{data | conn: conn}, [{:state_timeout, options.timeout, :connect_timeout}]}
+
+      {:error, reason} ->
+        Logger.error("[Drinkup.Socket] Failed to open connection: #{inspect(reason)}")
+        {:stop, {:connect_failed, reason}}
+    end
+  end
+
+  def connecting_http(:info, {:gun_up, _conn, :http}, data) do
+    {:next_state, :connecting_ws, data}
+  end
+
+  def connecting_http(:state_timeout, :connect_timeout, data) do
+    Logger.error("[Drinkup.Socket] HTTP connection timeout")
+    trigger_reconnect(data)
+  end
+
+  # :connecting_ws state - upgrading to WebSocket
+
+  def connecting_ws(
+        :enter,
+        _from,
+        %{module: module, user_data: user_data, options: options} = data
+      ) do
+    Logger.debug("[Drinkup.Socket] Upgrading connection to WebSocket")
+
+    path = module.build_path(user_data)
+    stream = :gun.ws_upgrade(data.conn, path, [], %{flow: options.flow})
+
+    {:keep_state, %{data | stream: stream}, [{:state_timeout, options.timeout, :upgrade_timeout}]}
+  end
+
+  def connecting_ws(:info, {:gun_upgrade, _conn, _stream, ["websocket"], _headers}, data) do
+    {:next_state, :connected, data}
+  end
+
+  def connecting_ws(:info, {:gun_response, _conn, _stream, _fin, status, _headers}, data) do
+    Logger.error("[Drinkup.Socket] WebSocket upgrade failed with status: #{status}")
+    trigger_reconnect(data)
+  end
+
+  def connecting_ws(:info, {:gun_error, _conn, _stream, reason}, data) do
+    Logger.error("[Drinkup.Socket] WebSocket upgrade error: #{inspect(reason)}")
+    trigger_reconnect(data)
+  end
+
+  def connecting_ws(:state_timeout, :upgrade_timeout, data) do
+    Logger.error("[Drinkup.Socket] WebSocket upgrade timeout")
+    trigger_reconnect(data)
+  end
+
+  # :connected state - active WebSocket connection
+
+  def connected(:enter, _from, %{module: module, user_data: user_data} = data) do
+    Logger.debug("[Drinkup.Socket] WebSocket connected")
+
+    case module.handle_connected(user_data) do
+      {:ok, new_user_data} ->
+        {:keep_state, %{data | user_data: new_user_data, reconnect_attempts: 0}}
+
+      _ ->
+        {:keep_state, %{data | reconnect_attempts: 0}}
+    end
+  end
+
+  def connected(
+        :info,
+        {:gun_ws, conn, _stream, frame},
+        %{module: module, user_data: user_data, options: options} = data
+      ) do
+    result = module.handle_frame(frame, user_data)
+
+    :ok = :gun.update_flow(conn, frame, options.flow)
+
+    case result do
+      {:ok, new_user_data} ->
+        {:keep_state, %{data | user_data: new_user_data}}
+
+      result when result in [:noop, nil] ->
+        :keep_state_and_data
+
+      {:error, reason} ->
+        Logger.error("[Drinkup.Socket] Frame handler error: #{inspect(reason)}")
+        :keep_state_and_data
+    end
+  end
+
+  def connected(:info, {:gun_ws, _conn, _stream, :close}, data) do
+    Logger.info("[Drinkup.Socket] WebSocket closed by remote")
+    trigger_reconnect(data, :remote_close)
+  end
+
+  def connected(:info, {:gun_ws, _conn, _stream, {:close, errno, reason}}, data) do
+    Logger.info("[Drinkup.Socket] WebSocket closed: #{errno} - #{inspect(reason)}")
+    trigger_reconnect(data, {:remote_close, errno, reason})
+  end
+
+  def connected(:info, {:gun_down, old_conn, _proto, _reason, _killed_streams}, %{conn: new_conn})
+      when old_conn != new_conn do
+    Logger.debug("[Drinkup.Socket] Ignoring :gun_down for old connection")
+    :keep_state_and_data
+  end
+
+  def connected(:info, {:gun_down, _conn, _proto, reason, _killed_streams}, data) do
+    Logger.info("[Drinkup.Socket] Connection down: #{inspect(reason)}")
+    trigger_reconnect(data, {:connection_down, reason})
+  end
+
+  def connected(
+        :internal,
+        :reconnect,
+        %{conn: conn, options: options, reconnect_attempts: attempts} = data
+      ) do
+    :ok = :gun.close(conn)
+    :ok = :gun.flush(conn)
+
+    backoff = calculate_backoff(attempts, options.reconnect_strategy)
+
+    Logger.info("[Drinkup.Socket] Reconnecting in #{backoff}ms (attempt #{attempts + 1})")
+
+    {:next_state, :disconnected,
+     %{data | conn: nil, stream: nil, reconnect_attempts: attempts + 1},
+     [{{:timeout, :reconnect}, backoff, :reconnect}]}
+  end
+
+  # Helper functions
+
+  defp trigger_reconnect(data, reason \\ :unknown) do
+    %{module: module, user_data: user_data} = data
+
+    case module.handle_disconnected(reason, user_data) do
+      {:ok, new_user_data} ->
+        {:keep_state, %{data | user_data: new_user_data}, [{:next_event, :internal, :reconnect}]}
+
+      _ ->
+        {:keep_state_and_data, [{:next_event, :internal, :reconnect}]}
+    end
+  end
+
+  defp parse_options(opts) do
+    %{
+      host: Keyword.fetch!(opts, :host),
+      flow: Keyword.get(opts, :flow, 10),
+      timeout: Keyword.get(opts, :timeout, :timer.seconds(5)),
+      tls_opts: Keyword.get(opts, :tls_opts, default_tls_opts()),
+      gun_opts: Keyword.get(opts, :gun_opts, %{}),
+      reconnect_strategy: Keyword.get(opts, :reconnect_strategy, :exponential)
+    }
+  end
+
+  defp default_tls_opts do
+    [
+      verify: :verify_peer,
+      cacerts: :certifi.cacerts(),
+      depth: 3,
+      customize_hostname_check: [
+        match_fun: :public_key.pkix_verify_hostname_match_fun(:https)
+      ]
+    ]
+  end
+
+  defp calculate_backoff(attempt, strategy) do
+    case strategy do
+      :exponential ->
+        exponential_backoff(attempt, :timer.seconds(60))
+
+      {:exponential, max_backoff} ->
+        exponential_backoff(attempt, max_backoff)
+
+      {:custom, func} when is_function(func, 1) ->
+        func.(attempt)
+    end
+  end
+
+  defp exponential_backoff(attempt, max_backoff) do
+    base = :timer.seconds(1)
+    delay = min(base * :math.pow(2, attempt), max_backoff)
+    jitter = :rand.uniform(trunc(delay * 0.1))
+    trunc(delay) + jitter
+  end
+end