use stratus pkg instead of gun for websocket client, remove logging/retry options

.gitignore

···
 *.ez
 **/build
 erl_crash.dump
+.claude

CHANGELOG.md

···
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [2.0.0] - 2025-11-04
+
+### Changed
+- **BREAKING**: Migrated from Erlang's `gun` library to Gleam's `stratus` WebSocket client
+- **BREAKING**: Removed three configuration fields from `JetstreamConfig`:
+  - `max_backoff_seconds` (retry logic still works internally, capped at 60s)
+  - `log_connection_events` (connection logging removed)
+  - `log_retry_attempts` (retry logging removed)
+- Reduced Erlang FFI code from ~290 lines to ~40 lines
+- Improved code maintainability with more idiomatic Gleam implementation
+
+### Added
+- Vendored `stratus` WebSocket client internally (moved to `goose/stratus`) until a new version is published
+- New dependency: `simplifile` (>= 2.0.0 and < 3.0.0)
+- New dependencies (from stratus): `gleam_otp`, `gleam_crypto`, `logging`, `exception`, `gramps`
+
+### Removed
+- Dependency: `gun` (>= 2.2.0 and < 3.0.0)
+
+### Migration Guide
+If you're upgrading from v1.x, remove the following fields from your `JetstreamConfig`:
+```gleam
+// Before (v1.x)
+let config = goose.JetstreamConfig(
+  endpoint: "wss://jetstream2.us-east.bsky.network/subscribe",
+  wanted_collections: [],
+  wanted_dids: [],
+  cursor: option.None,
+  max_message_size_bytes: option.None,
+  compress: True,
+  require_hello: False,
+  max_backoff_seconds: 60,        // Remove this
+  log_connection_events: True,    // Remove this
+  log_retry_attempts: True,       // Remove this
+)
+
+// After (v2.x)
+let config = goose.JetstreamConfig(
+  endpoint: "wss://jetstream2.us-east.bsky.network/subscribe",
+  wanted_collections: [],
+  wanted_dids: [],
+  cursor: option.None,
+  max_message_size_bytes: option.None,
+  compress: True,
+  require_hello: False,
+)
+```
+
+All other functionality remains the same. Automatic retry with exponential backoff still works internally.
+
 ## [1.1.0] - 2025-01-29
 
 ### Added

README.md

···
     wanted_dids: [],
     cursor: option.None,
     max_message_size_bytes: option.None,
-    compress: False,
+    compress: True,
     require_hello: False,
-    // Retry configuration
-    max_backoff_seconds: 60,      // Max wait between retries
-    log_connection_events: True,  // Log connects/disconnects
-    log_retry_attempts: False,    // Skip verbose retry logs
   )
 
   goose.start_consumer(config, handle_event)
···
 
 ## Configuration Options
 
-**Note:** Goose automatically handles connection failures with exponential backoff retry logic (1s, 2s, 4s, 8s, 16s, 32s, up to max). All connections automatically retry on failure, reconnect on disconnection, and distinguish between harmless timeouts and real errors.
+**Note:** Goose automatically handles connection failures with exponential backoff retry logic (1s, 2s, 4s, 8s, 16s, 32s, capped at 60s). All connections automatically retry on failure and reconnect on disconnection.
 
 ### `wanted_collections`
 An array of Collection NSIDs to filter which records you receive (default: empty = all collections)
···
 require_hello: True
 ```
 
-### `max_backoff_seconds`
-Maximum wait time in seconds between retry attempts
-
-- Uses exponential backoff: 1s, 2s, 4s, 8s, 16s, 32s, then capped at this value
-- Default: `60`
-
-**Example:**
-```gleam
-max_backoff_seconds: 120  // Allow up to 2 minute waits between retries
-```
-
-### `log_connection_events`
-Whether to log connection state changes (connected, disconnected)
-
-- Set to `True` to log important connection events
-- Default: `True`
-- Recommended: `True` for production (know when disconnects happen)
-
-**Example:**
-```gleam
-log_connection_events: True
-```
-
-### `log_retry_attempts`
-Whether to log detailed retry attempt information
-
-- Set to `True` to log attempt numbers, errors, and backoff times
-- Default: `True`
-- Recommended: `False` for production (reduces log noise)
-
-**Example:**
-```gleam
-log_retry_attempts: False  // Production: skip verbose retry logs
-```
-
 ## Full Configuration Example
 
 ```gleam
···
   max_message_size_bytes: option.Some(2097152),  // 2MB
   compress: True,
   require_hello: False,
-  max_backoff_seconds: 60,
-  log_connection_events: True,
-  log_retry_attempts: False,
 )
 
 goose.start_consumer(config, handle_event)

example/manifest.toml

···
 # You typically do not need to edit this file
 
 packages = [
-  { name = "cowlib", version = "2.16.0", build_tools = ["make", "rebar3"], requirements = [], otp_app = "cowlib", source = "hex", outer_checksum = "7F478D80D66B747344F0EA7708C187645CFCC08B11AA424632F78E25BF05DB51" },
+  { name = "exception", version = "2.1.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "exception", source = "hex", outer_checksum = "329D269D5C2A314F7364BD2711372B6F2C58FA6F39981572E5CA68624D291F8C" },
   { name = "ezstd", version = "1.2.3", build_tools = ["rebar3"], requirements = [], otp_app = "ezstd", source = "hex", outer_checksum = "DE32E0B41BA36A9ED46DB8215DA74777D2F141BB75F67BFC05DBB4B7C3386DEE" },
+  { name = "filepath", version = "1.1.2", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "filepath", source = "hex", outer_checksum = "B06A9AF0BF10E51401D64B98E4B627F1D2E48C154967DA7AF4D0914780A6D40A" },
+  { name = "gleam_crypto", version = "1.5.1", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_crypto", source = "hex", outer_checksum = "50774BAFFF1144E7872814C566C5D653D83A3EBF23ACC3156B757A1B6819086E" },
   { name = "gleam_erlang", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_erlang", source = "hex", outer_checksum = "1124AD3AA21143E5AF0FC5CF3D9529F6DB8CA03E43A55711B60B6B7B3874375C" },
   { name = "gleam_http", version = "4.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_http", source = "hex", outer_checksum = "82EA6A717C842456188C190AFB372665EA56CE13D8559BF3B1DD9E40F619EE0C" },
   { name = "gleam_json", version = "3.0.2", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_json", source = "hex", outer_checksum = "874FA3C3BB6E22DD2BB111966BD40B3759E9094E05257899A7C08F5DE77EC049" },
+  { name = "gleam_otp", version = "1.2.0", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_stdlib"], otp_app = "gleam_otp", source = "hex", outer_checksum = "BA6A294E295E428EC1562DC1C11EA7530DCB981E8359134BEABC8493B7B2258E" },
   { name = "gleam_stdlib", version = "0.65.0", build_tools = ["gleam"], requirements = [], otp_app = "gleam_stdlib", source = "hex", outer_checksum = "7C69C71D8C493AE11A5184828A77110EB05A7786EBF8B25B36A72F879C3EE107" },
-  { name = "gleeunit", version = "1.7.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleeunit", source = "hex", outer_checksum = "CD701726CBCE5588B375D157B4391CFD0F2F134CD12D9B6998A395484DE05C58" },
-  { name = "goose", version = "1.0.0", build_tools = ["gleam"], requirements = ["ezstd", "gleam_erlang", "gleam_http", "gleam_json", "gleam_stdlib", "gun"], source = "local", path = ".." },
-  { name = "gun", version = "2.2.0", build_tools = ["make", "rebar3"], requirements = ["cowlib"], otp_app = "gun", source = "hex", outer_checksum = "76022700C64287FEB4DF93A1795CFF6741B83FB37415C40C34C38D2A4645261A" },
+  { name = "gleeunit", version = "1.8.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleeunit", source = "hex", outer_checksum = "7AE0F64B26CC065ED705FF7CA5F4EDAB8015E72A883736FE251E46FACCCE1E08" },
+  { name = "goose", version = "2.0.0", build_tools = ["gleam"], requirements = ["exception", "ezstd", "gleam_crypto", "gleam_erlang", "gleam_http", "gleam_json", "gleam_otp", "gleam_stdlib", "gramps", "logging", "simplifile"], source = "local", path = ".." },
+  { name = "gramps", version = "6.0.0", build_tools = ["gleam"], requirements = ["gleam_crypto", "gleam_erlang", "gleam_http", "gleam_stdlib"], otp_app = "gramps", source = "hex", outer_checksum = "8B7195978FBFD30B43DF791A8A272041B81E45D245314D7A41FC57237AA882A0" },
+  { name = "logging", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "logging", source = "hex", outer_checksum = "1098FBF10B54B44C2C7FDF0B01C1253CAFACDACABEFB4B0D027803246753E06D" },
+  { name = "simplifile", version = "2.3.0", build_tools = ["gleam"], requirements = ["filepath", "gleam_stdlib"], otp_app = "simplifile", source = "hex", outer_checksum = "0A868DAC6063D9E983477981839810DC2E553285AB4588B87E3E9C96A7FB4CB4" },
 ]
 
 [requirements]

example/src/example.gleam

···
       max_message_size_bytes: option.None,
       compress: True,
       require_hello: False,
-      max_backoff_seconds: 60,
-      log_connection_events: True,
-      log_retry_attempts: True,
     )
 
   io.println("Starting Jetstream consumer...")

gleam.toml

···
 name = "goose"
-version = "1.1.0"
+version = "2.0.0"
 description = "A Gleam WebSocket consumer for AT Protocol Jetstream events"
 licences = ["Apache-2.0"]
 repository = { type = "github", user = "bigmoves", repo = "goose" }
···
 [dependencies]
 gleam_stdlib = ">= 0.44.0 and < 2.0.0"
 gleam_erlang = ">= 1.0.0 and < 2.0.0"
+gleam_otp = ">= 1.0.0 and < 2.0.0"
 gleam_http = ">= 4.0.0 and < 5.0.0"
 gleam_json = ">= 3.0.2 and < 4.0.0"
-gun = ">= 2.2.0 and < 3.0.0"
+gleam_crypto = ">= 1.5.0 and < 2.0.0"
 ezstd = ">= 1.2.0 and < 2.0.0"
+simplifile = ">= 2.0.0 and < 3.0.0"
+logging = ">= 1.3.0 and < 2.0.0"
+exception = ">= 2.1.0 and < 3.0.0"
+gramps = ">= 6.0.0 and < 7.0.0"
 
 [dev-dependencies]
 gleeunit = ">= 1.0.0 and < 2.0.0"

manifest.toml

···
 # You typically do not need to edit this file
 
 packages = [
-  { name = "cowlib", version = "2.16.0", build_tools = ["make", "rebar3"], requirements = [], otp_app = "cowlib", source = "hex", outer_checksum = "7F478D80D66B747344F0EA7708C187645CFCC08B11AA424632F78E25BF05DB51" },
+  { name = "exception", version = "2.1.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "exception", source = "hex", outer_checksum = "329D269D5C2A314F7364BD2711372B6F2C58FA6F39981572E5CA68624D291F8C" },
   { name = "ezstd", version = "1.2.3", build_tools = ["rebar3"], requirements = [], otp_app = "ezstd", source = "hex", outer_checksum = "DE32E0B41BA36A9ED46DB8215DA74777D2F141BB75F67BFC05DBB4B7C3386DEE" },
+  { name = "filepath", version = "1.1.2", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "filepath", source = "hex", outer_checksum = "B06A9AF0BF10E51401D64B98E4B627F1D2E48C154967DA7AF4D0914780A6D40A" },
+  { name = "gleam_crypto", version = "1.5.1", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_crypto", source = "hex", outer_checksum = "50774BAFFF1144E7872814C566C5D653D83A3EBF23ACC3156B757A1B6819086E" },
   { name = "gleam_erlang", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_erlang", source = "hex", outer_checksum = "1124AD3AA21143E5AF0FC5CF3D9529F6DB8CA03E43A55711B60B6B7B3874375C" },
   { name = "gleam_http", version = "4.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_http", source = "hex", outer_checksum = "82EA6A717C842456188C190AFB372665EA56CE13D8559BF3B1DD9E40F619EE0C" },
   { name = "gleam_json", version = "3.0.2", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_json", source = "hex", outer_checksum = "874FA3C3BB6E22DD2BB111966BD40B3759E9094E05257899A7C08F5DE77EC049" },
+  { name = "gleam_otp", version = "1.2.0", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_stdlib"], otp_app = "gleam_otp", source = "hex", outer_checksum = "BA6A294E295E428EC1562DC1C11EA7530DCB981E8359134BEABC8493B7B2258E" },
   { name = "gleam_stdlib", version = "0.65.0", build_tools = ["gleam"], requirements = [], otp_app = "gleam_stdlib", source = "hex", outer_checksum = "7C69C71D8C493AE11A5184828A77110EB05A7786EBF8B25B36A72F879C3EE107" },
-  { name = "gleeunit", version = "1.6.1", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleeunit", source = "hex", outer_checksum = "FDC68A8C492B1E9B429249062CD9BAC9B5538C6FBF584817205D0998C42E1DAC" },
-  { name = "gun", version = "2.2.0", build_tools = ["make", "rebar3"], requirements = ["cowlib"], otp_app = "gun", source = "hex", outer_checksum = "76022700C64287FEB4DF93A1795CFF6741B83FB37415C40C34C38D2A4645261A" },
+  { name = "gleeunit", version = "1.8.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleeunit", source = "hex", outer_checksum = "7AE0F64B26CC065ED705FF7CA5F4EDAB8015E72A883736FE251E46FACCCE1E08" },
+  { name = "gramps", version = "6.0.0", build_tools = ["gleam"], requirements = ["gleam_crypto", "gleam_erlang", "gleam_http", "gleam_stdlib"], otp_app = "gramps", source = "hex", outer_checksum = "8B7195978FBFD30B43DF791A8A272041B81E45D245314D7A41FC57237AA882A0" },
+  { name = "logging", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "logging", source = "hex", outer_checksum = "1098FBF10B54B44C2C7FDF0B01C1253CAFACDACABEFB4B0D027803246753E06D" },
+  { name = "simplifile", version = "2.3.0", build_tools = ["gleam"], requirements = ["filepath", "gleam_stdlib"], otp_app = "simplifile", source = "hex", outer_checksum = "0A868DAC6063D9E983477981839810DC2E553285AB4588B87E3E9C96A7FB4CB4" },
 ]
 
 [requirements]
+exception = { version = ">= 2.1.0 and < 3.0.0" }
 ezstd = { version = ">= 1.2.0 and < 2.0.0" }
+gleam_crypto = { version = ">= 1.5.0 and < 2.0.0" }
 gleam_erlang = { version = ">= 1.0.0 and < 2.0.0" }
 gleam_http = { version = ">= 4.0.0 and < 5.0.0" }
 gleam_json = { version = ">= 3.0.2 and < 4.0.0" }
+gleam_otp = { version = ">= 1.0.0 and < 2.0.0" }
 gleam_stdlib = { version = ">= 0.44.0 and < 2.0.0" }
 gleeunit = { version = ">= 1.0.0 and < 2.0.0" }
-gun = { version = ">= 2.2.0 and < 3.0.0" }
+gramps = { version = ">= 6.0.0 and < 7.0.0" }
+logging = { version = ">= 1.3.0 and < 2.0.0" }
+simplifile = { version = ">= 2.0.0 and < 3.0.0" }

src/goose.gleam

···
+import gleam/bit_array
 import gleam/dynamic.{type Dynamic}
 import gleam/dynamic/decode
-import gleam/erlang/atom
-import gleam/erlang/process.{type Pid}
-import gleam/int
-import gleam/io
+import gleam/erlang/process
+import gleam/http/request
 import gleam/json
 import gleam/list
 import gleam/option.{type Option}
+import gleam/result
 import gleam/string
+import goose/internal/zstd
+import goose/stratus
+import simplifile
 
 /// Jetstream event types
 pub type JetstreamEvent {
···
     max_message_size_bytes: Option(Int),
     compress: Bool,
     require_hello: Bool,
-    /// Maximum backoff time in seconds for retry logic (default: 60)
-    max_backoff_seconds: Int,
-    /// Whether to log connection events (connected, disconnected) (default: True)
-    log_connection_events: Bool,
-    /// Whether to log retry attempts and errors (default: True)
-    log_retry_attempts: Bool,
+  )
+}
+
+/// Internal state for WebSocket connection
+type ConnectionState {
+  ConnectionState(
+    config: JetstreamConfig,
+    on_event: fn(String) -> Nil,
+    decompressor: Option(Decompressor),
+  )
+}
+
+/// Decompression context holder
+type Decompressor {
+  Decompressor(
+    dctx: zstd.DecompressionContext,
+    ddict: zstd.DecompressionDict,
   )
 }
 
···
     max_message_size_bytes: option.None,
     compress: False,
     require_hello: False,
-    max_backoff_seconds: 60,
-    log_connection_events: True,
-    log_retry_attempts: True,
   )
 }
 
···
   }
 }
 
-/// Connect to Jetstream WebSocket using Erlang gun library
-@external(erlang, "goose_ws_ffi", "connect")
-pub fn connect(
-  url: String,
-  handler_pid: Pid,
-  compress: Bool,
-) -> Result(Pid, Dynamic)
+/// Load the zstd decompression dictionary
+fn load_decompressor() -> Result(Decompressor, String) {
+  // Get priv directory
+  let priv_dir = get_priv_dir()
+  let dict_path = priv_dir <> "/zstd_dictionary"
+
+  // Read dictionary file
+  use dict_data <- result.try(
+    simplifile.read_bits(dict_path)
+    |> result.map_error(fn(_) {
+      "Failed to load zstd dictionary from " <> dict_path
+    }),
+  )
+
+  // Create decompression context and dictionary
+  let dctx = zstd.create_decompression_context(1024 * 1024)
+  let ddict = zstd.create_ddict(dict_data)
+
+  // Select dictionary for context
+  use _ <- result.try(zstd.select_ddict(dctx, ddict))
+
+  Ok(Decompressor(dctx: dctx, ddict: ddict))
+}
+
+/// Get the priv directory path for this application
+@external(erlang, "goose_ffi", "priv_dir")
+fn get_priv_dir() -> String
+
+/// Decompress zstd-compressed data
+fn decompress_data(
+  data: BitArray,
+  decompressor: Decompressor,
+) -> Result(String, String) {
+  // Try decompress_using_ddict first (works for frames with content size)
+  case zstd.decompress_using_ddict(data, decompressor.ddict) {
+    Ok(decompressed) ->
+      bit_array.to_string(decompressed)
+      |> result.replace_error("Failed to decode decompressed data as UTF-8")
+    Error(msg) -> {
+      // Check if error is due to unknown content size
+      case string.contains(msg, "ZSTD_CONTENTSIZE_UNKNOWN") {
+        True -> {
+          // Frame doesn't have content size, use streaming with dictionary-loaded context
+          case zstd.decompress_streaming(decompressor.dctx, data) {
+            Ok(decompressed) ->
+              bit_array.to_string(decompressed)
+              |> result.replace_error(
+                "Failed to decode streaming decompressed data as UTF-8",
+              )
+            Error(_stream_err) -> Error("Streaming decompression failed")
+          }
+        }
+        False -> Error("Decompression failed")
+      }
+    }
+  }
+}
 
 /// Start consuming the Jetstream feed with automatic retry logic
 ///
···
   retry_count: Int,
 ) -> Nil {
   let url = build_url(config)
-  let self = process.self()
-  let result = connect(url, self, config.compress)
+
+  // Convert wss:// to https:// and ws:// to http:// for request parsing
+  let http_url =
+    url
+    |> string.replace("wss://", "https://")
+    |> string.replace("ws://", "http://")
+
+  // Parse URL into HTTP request
+  let assert Ok(req) = request.to(http_url)
+
+  // Load decompressor if compression is enabled
+  let decompressor = case config.compress {
+    True ->
+      case load_decompressor() {
+        Ok(dec) -> option.Some(dec)
+        Error(_err) -> option.None
+      }
+    False -> option.None
+  }
+
+  // Create initial state
+  let state =
+    ConnectionState(
+      config: config,
+      on_event: on_event,
+      decompressor: decompressor,
+    )
+
+  // Start WebSocket connection
+  let result =
+    stratus.new_with_initialiser(
+      request: req,
+      init: fn() { Ok(stratus.initialised(state)) },
+    )
+    |> stratus.on_message(handle_message)
+    |> stratus.on_close(handle_close)
+    |> stratus.with_connect_timeout(30_000)
+    |> stratus.start()
 
   case result {
-    Ok(_conn_pid) -> {
-      case config.log_connection_events {
-        True -> io.println("Connected to Jetstream successfully")
-        False -> Nil
-      }
-      // Start receiving with retry support
-      receive_with_retry(config, on_event)
+    Ok(_websocket) -> {
+      // Keep process alive indefinitely
+      process.sleep_forever()
     }
-    Error(err) -> {
+    Error(_err) -> {
       // Connection failed, calculate backoff and retry
-      let backoff_seconds = calculate_backoff(retry_count, config)
-      case config.log_retry_attempts {
-        True -> {
-          io.println(
-            "Failed to connect to Jetstream (attempt "
-            <> int.to_string(retry_count + 1)
-            <> "): "
-            <> string.inspect(err),
-          )
-          io.println(
-            "Retrying in " <> int.to_string(backoff_seconds) <> " seconds...",
-          )
-        }
-        False -> Nil
-      }
+      let backoff_seconds = calculate_backoff(retry_count)
 
       // Sleep for backoff period
       process.sleep(backoff_seconds * 1000)
···
   }
 }
 
-/// Calculate exponential backoff with configurable maximum
-fn calculate_backoff(retry_count: Int, config: JetstreamConfig) -> Int {
-  let backoff = case retry_count {
-    0 -> 1
-    1 -> 2
-    2 -> 4
-    3 -> 8
-    4 -> 16
-    5 -> 32
-    _ -> config.max_backoff_seconds
-  }
-
-  // Cap at max_backoff_seconds
-  case backoff > config.max_backoff_seconds {
-    True -> config.max_backoff_seconds
-    False -> backoff
-  }
-}
-
-/// Receive messages with retry logic
-fn receive_with_retry(
-  config: JetstreamConfig,
-  on_event: fn(String) -> Nil,
-) -> Nil {
-  case receive_ws_message() {
-    Ok(text) -> {
-      on_event(text)
-      receive_with_retry(config, on_event)
+/// Handle incoming WebSocket messages
+fn handle_message(
+  state: ConnectionState,
+  msg: stratus.Message(Nil),
+  _conn: stratus.Connection,
+) -> stratus.Next(ConnectionState, Nil) {
+  case msg {
+    stratus.Text(text) -> {
+      state.on_event(text)
+      stratus.continue(state)
     }
-    Error(error_dynamic) -> {
-      // Decode error type
-      let atm = atom.cast_from_dynamic(error_dynamic)
-      let error_type = atom.to_string(atm)
-
-      case error_type {
-        "timeout" -> {
-          // No messages in 60s, connection is alive - continue
-          receive_with_retry(config, on_event)
-        }
-        "closed" -> {
-          // Connection closed - log if configured
-          case config.log_connection_events {
-            True -> io.println("Jetstream connection closed, reconnecting...")
-            False -> Nil
-          }
-          start_with_retry_internal(config, on_event, 0)
-        }
-        "connection_error" -> {
-          // Connection error - log if configured
-          case config.log_connection_events {
-            True -> io.println("Jetstream connection error, reconnecting...")
-            False -> Nil
-          }
-          start_with_retry_internal(config, on_event, 0)
-        }
-        _ -> {
-          // Unknown error - log if retry logging is enabled
-          case config.log_retry_attempts {
-            True -> {
-              io.println("Unknown Jetstream error: " <> error_type)
-              io.println("Reconnecting...")
+    stratus.Binary(data) -> {
+      // Handle compressed binary data if decompressor is available
+      case state.decompressor {
+        option.Some(decompressor) -> {
+          case decompress_data(data, decompressor) {
+            Ok(text) -> {
+              state.on_event(text)
+              stratus.continue(state)
             }
-            False -> Nil
+            Error(_err) -> {
+              // Skip frames that fail to decompress
+              stratus.continue(state)
+            }
           }
-          start_with_retry_internal(config, on_event, 0)
+        }
+        option.None -> {
+          // No decompressor, ignore binary messages
+          stratus.continue(state)
         }
       }
     }
+    stratus.User(_) -> {
+      // No custom user messages in this implementation
+      stratus.continue(state)
+    }
   }
 }
 
-/// Receive a WebSocket message from the message queue
-/// Returns Ok(text) for messages, or Error with one of: timeout, closed, connection_error
-@external(erlang, "goose_ffi", "receive_ws_message")
-fn receive_ws_message() -> Result(String, Dynamic)
+/// Handle WebSocket connection close
+fn handle_close(state: ConnectionState) -> Nil {
+  // Reconnect from the beginning
+  start_with_retry_internal(state.config, state.on_event, 0)
+}
+
+/// Calculate exponential backoff capped at 60 seconds
+fn calculate_backoff(retry_count: Int) -> Int {
+  case retry_count {
+    0 -> 1
+    1 -> 2
+    2 -> 4
+    3 -> 8
+    4 -> 16
+    5 -> 32
+    _ -> 60
+  }
+}
 
 /// Parse a JSON event string into a JetstreamEvent
 pub fn parse_event(json_string: String) -> JetstreamEvent {

src/goose/internal/zstd.gleam

···
+import gleam/erlang/atom
+
+/// Opaque type for decompression context
+pub type DecompressionContext
+
+/// Opaque type for decompression dictionary
+pub type DecompressionDict
+
+/// Create a decompression context with a given buffer size
+@external(erlang, "ezstd", "create_decompression_context")
+pub fn create_decompression_context(buffer_size: Int) -> DecompressionContext
+
+/// Create a decompression dictionary from binary data
+@external(erlang, "ezstd", "create_ddict")
+pub fn create_ddict(dict_data: BitArray) -> DecompressionDict
+
+/// Select a dictionary for the decompression context
+/// Note: ezstd returns 'ok' atom, not {ok, undefined}
+@external(erlang, "ezstd", "select_ddict")
+fn select_ddict_ffi(
+  dctx: DecompressionContext,
+  ddict: DecompressionDict,
+) -> atom.Atom
+
+/// Select a dictionary for the decompression context (safe wrapper)
+pub fn select_ddict(
+  dctx: DecompressionContext,
+  ddict: DecompressionDict,
+) -> Result(Nil, String) {
+  let result_atom = select_ddict_ffi(dctx, ddict)
+  case atom.to_string(result_atom) {
+    "ok" -> Ok(Nil)
+    err -> Error("Failed to select dictionary: " <> err)
+  }
+}
+
+/// Decompress data using a dictionary
+/// Note: ezstd can return either BitArray or {error, Binary}
+@external(erlang, "goose_ffi", "decompress_using_ddict_safe")
+pub fn decompress_using_ddict(
+  data: BitArray,
+  ddict: DecompressionDict,
+) -> Result(BitArray, String)
+
+/// Decompress data using a streaming context (for frames without content size)
+/// Note: ezstd can return either BitArray or {error, Binary}
+@external(erlang, "goose_ffi", "decompress_streaming_safe")
+pub fn decompress_streaming(
+  dctx: DecompressionContext,
+  data: BitArray,
+) -> Result(BitArray, String)

src/goose/stratus.gleam

···
+import exception
+import gleam/bit_array
+import gleam/bool
+import gleam/bytes_tree.{type BytesTree}
+import gleam/crypto
+import gleam/erlang/charlist
+import gleam/erlang/process.{type Selector, type Subject}
+import gleam/http.{Http, Https}
+import gleam/http/request.{type Request}
+import gleam/http/response.{type Response}
+import gleam/int
+import gleam/list
+import gleam/option.{type Option, None, Some}
+import gleam/otp/actor
+import gleam/result
+import gleam/string
+import gleam/uri
+import gramps/http as gramps_http
+import gramps/websocket.{
+  BinaryFrame, CloseFrame, Continuation, Control, Data as DataFrame, PingFrame,
+  PongFrame, TextFrame,
+}
+import gramps/websocket/compression
+import logging
+import goose/stratus/internal/socket.{
+  type Socket, type SocketMessage, Cacerts, Once, Pull, Receive,
+}
+import goose/stratus/internal/ssl
+import goose/stratus/internal/tcp
+import goose/stratus/internal/transport.{type Transport, Ssl, Tcp}
+
+/// This holds some information needed to communicate with the WebSocket.
+pub opaque type Connection {
+  Connection(
+    socket: Socket,
+    transport: Transport,
+    context: Option(compression.Context),
+  )
+}
+
+pub type SocketReason {
+  SocketClosed
+  Timeout
+  Badarg
+  Terminated
+  Eaddrinuse
+  Eaddrnotavail
+  Eafnosupport
+  Ealready
+  Econnaborted
+  Econnrefused
+  Econnreset
+  Edestaddrreq
+  Ehostdown
+  Ehostunreach
+  Einprogress
+  Eisconn
+  Emsgsize
+  Enetdown
+  Enetunreach
+  Enopkg
+  Enoprotoopt
+  Enotconn
+  Enotty
+  Enotsock
+  Eproto
+  Eprotonosupport
+  Eprototype
+  Esocktnosupport
+  Etimedout
+  Ewouldblock
+  Exbadport
+  Exbadseq
+}
+
+pub type CustomCloseError {
+  SocketFail(SocketReason)
+  InvalidCode
+}
+
+fn convert_socket_reason(reason: socket.SocketReason) -> SocketReason {
+  case reason {
+    socket.Badarg -> Badarg
+    socket.Closed -> SocketClosed
+    socket.Eaddrinuse -> Eaddrinuse
+    socket.Eaddrnotavail -> Eaddrnotavail
+    socket.Eafnosupport -> Eafnosupport
+    socket.Ealready -> Ealready
+    socket.Econnaborted -> Econnaborted
+    socket.Econnrefused -> Econnrefused
+    socket.Econnreset -> Econnreset
+    socket.Edestaddrreq -> Edestaddrreq
+    socket.Ehostdown -> Ehostdown
+    socket.Ehostunreach -> Ehostunreach
+    socket.Einprogress -> Einprogress
+    socket.Eisconn -> Eisconn
+    socket.Emsgsize -> Emsgsize
+    socket.Enetdown -> Enetdown
+    socket.Enetunreach -> Enetunreach
+    socket.Enopkg -> Enopkg
+    socket.Enoprotoopt -> Enoprotoopt
+    socket.Enotconn -> Enotconn
+    socket.Enotsock -> Enotsock
+    socket.Enotty -> Enotty
+    socket.Eproto -> Eproto
+    socket.Eprotonosupport -> Eprotonosupport
+    socket.Eprototype -> Eprototype
+    socket.Esocktnosupport -> Esocktnosupport
+    socket.Etimedout -> Etimedout
+    socket.Ewouldblock -> Ewouldblock
+    socket.Exbadport -> Exbadport
+    socket.Exbadseq -> Exbadseq
+    socket.Terminated -> Terminated
+    socket.Timeout -> Timeout
+  }
+}
+
+fn from_socket_message(msg: SocketMessage) -> InternalMessage(user_message) {
+  case msg {
+    socket.Data(bits) -> Data(bits)
+    socket.Err(reason) -> Err(convert_socket_reason(reason))
+  }
+}
+
+pub opaque type Next(state, user_message) {
+  Continue(state: state, selector: Option(Selector(user_message)))
+  NormalStop
+  AbnormalStop(reason: String)
+}
+
+pub fn continue(state: state) -> Next(state, user_message) {
+  Continue(state, None)
+}
+
+pub fn with_selector(
+  next: Next(state, user_message),
+  selector: Selector(user_message),
+) -> Next(state, user_message) {
+  case next {
+    Continue(state, _) -> Continue(state, Some(selector))
+    _ -> next
+  }
+}
+
+pub fn stop() -> Next(state, user_message) {
+  NormalStop
+}
+
+pub fn stop_abnormal(reason: String) -> Next(state, user_message) {
+  AbnormalStop(reason)
+}
+
+/// These are the messages emitted or received by the underlying process.  You
+/// should only need to interact with `Message` below.
+pub opaque type InternalMessage(user_message) {
+  UserMessage(user_message)
+  Err(SocketReason)
+  Data(BitArray)
+  Closed
+  Shutdown
+}
+
+/// This is the type of message your handler might receive.
+pub type Message(user_message) {
+  Text(String)
+  Binary(BitArray)
+  User(user_message)
+}
+
+pub opaque type Builder(state, user_message) {
+  Builder(
+    request: Request(String),
+    connect_timeout: Int,
+    init: fn() -> Result(Initialised(state, user_message), String),
+    loop: fn(state, Message(user_message), Connection) ->
+      Next(state, user_message),
+    on_close: fn(state) -> Nil,
+  )
+}
+
+//  and `on_close`.
+/// This creates a builder to set up a WebSocket actor. This will use default
+/// values for the connection initialization timeout, and provide an empty
+/// function to be called when the server closes the connection. If you want to
+/// customize either of those, see the helper functions `with_connect_timeout`
+pub fn new(
+  request req: Request(String),
+  state state: state,
+) -> Builder(state, user_message) {
+  Builder(
+    request: req,
+    connect_timeout: 5000,
+    init: fn() { Ok(initialised(state)) },
+    loop: fn(state, _msg, _conn) { continue(state) },
+    on_close: fn(_state) { Nil },
+  )
+}
+
+pub type Initialised(state, user_message) {
+  Initialised(state: state, selector: Option(Selector(user_message)))
+}
+
+pub fn initialised(state: state) -> Initialised(state, user_message) {
+  Initialised(state:, selector: None)
+}
+
+pub fn selecting(
+  initialised: Initialised(state, old_message),
+  selector: Selector(user_message),
+) -> Initialised(state, user_message) {
+  Initialised(state: initialised.state, selector: Some(selector))
+}
+
+pub fn new_with_initialiser(
+  request req: Request(String),
+  init init: fn() -> Result(Initialised(state, user_message), String),
+) -> Builder(state, user_message) {
+  Builder(
+    request: req,
+    connect_timeout: 5000,
+    init: init,
+    loop: fn(state, _msg, _conn) { continue(state) },
+    on_close: fn(_state) { Nil },
+  )
+}
+
+pub fn on_message(
+  builder: Builder(state, user_message),
+  on_message: fn(state, Message(user_message), Connection) ->
+    Next(state, user_message),
+) -> Builder(state, user_message) {
+  Builder(..builder, loop: on_message)
+}
+
+/// This sets the maximum amount of time you are willing to wait for both
+/// connecting to the server and receiving the upgrade response.  This means
+/// that it may take up to `timeout * 2` to begin sending or receiving messages.
+/// This value defaults to 5 seconds.
+pub fn with_connect_timeout(
+  builder: Builder(state, user_message),
+  timeout: Int,
+) -> Builder(state, user_message) {
+  Builder(..builder, connect_timeout: timeout)
+}
+
+/// You can provide a function to be called when the connection is closed. This
+/// function receives the last value for the state of the WebSocket.
+///
+/// NOTE:  If you manually call `stratus.close`, this function will not be
+/// called. I'm unsure right now if this is a bug or working as intended. But
+/// you will be in the loop with the state value handy.
+pub fn on_close(
+  builder: Builder(state, user_message),
+  on_close: fn(state) -> Nil,
+) -> Builder(state, user_message) {
+  Builder(..builder, on_close: on_close)
+}
+
+type State(state, user_message) {
+  State(
+    buffer: BitArray,
+    incomplete: Option(websocket.Frame),
+    self: Subject(InternalMessage(user_message)),
+    socket: Socket,
+    user_state: state,
+    compression: Option(compression.Compression),
+  )
+}
+
+/// These are the possible failures when calling `stratus.initialize`.
+pub type InitializationError {
+  // The WebSocket handshake failed for the provided reason
+  HandshakeFailed(HandshakeError)
+  // The actor failed to start, most likely due to a timeout in your `init`
+  ActorFailed(actor.StartError)
+  //
+  FailedToTransferSocket(SocketReason)
+}
+
+/// This opens the WebSocket connection with the provided `Builder`. It makes
+/// some assumptions about the request if you do not provide it.  It will use
+/// ports 80 or 443 for `ws` or `wss` respectively.
+///
+/// It will open the connection and perform the WebSocket handshake. If this
+/// fails, the actor will fail to start with the given reason as a string value.
+///
+/// After that, received messages will be passed to your loop, and you can use
+/// the helper functions to send messages to the server. The `close` method will
+/// send a close frame and end the connection.
+pub fn start(
+  builder: Builder(state, user_message),
+) -> Result(
+  actor.Started(Subject(InternalMessage(user_message))),
+  InitializationError,
+) {
+  let transport = case builder.request.scheme {
+    Https -> Ssl
+    _ -> Tcp
+  }
+
+  let handshake_result =
+    perform_handshake(builder.request, transport, builder.connect_timeout)
+    |> result.map_error(fn(reason) {
+      let msg = case reason {
+        UpgradeFailed(resp) ->
+          "WebSocket handshake failed with status "
+          <> int.to_string(resp.status)
+        reason -> "WebSocket handshake failed: " <> string.inspect(reason)
+      }
+      logging.log(logging.Error, msg)
+      HandshakeFailed(reason)
+    })
+
+  use handshake_response <- result.try(handshake_result)
+  logging.log(logging.Debug, "Handshake successful")
+
+  let extensions =
+    handshake_response.response
+    |> response.get_header("sec-websocket-extensions")
+    |> result.map(string.split(_, "; "))
+    |> result.unwrap([])
+
+  let context_takeovers = websocket.get_context_takeovers(extensions)
+
+  let assert Ok(_) =
+    transport.set_opts(
+      transport,
+      handshake_response.socket,
+      socket.convert_options([Receive(Once)]),
+    )
+
+  actor.new_with_initialiser(1000, fn(subject) {
+    let started_selector = process.select(process.new_selector(), subject)
+    logging.log(logging.Debug, "Calling user initializer")
+    use Initialised(user_state, user_selector) <- result.try(builder.init())
+    let selector = case user_selector {
+      Some(selector) -> {
+        selector
+        |> process.map_selector(UserMessage)
+        |> process.merge_selector(started_selector)
+        |> process.merge_selector(
+          process.map_selector(socket.selector(), fn(msg) {
+            let assert Ok(msg) = msg
+            from_socket_message(msg)
+          }),
+        )
+      }
+      _ ->
+        started_selector
+        |> process.merge_selector(
+          process.map_selector(socket.selector(), fn(msg) {
+            let assert Ok(msg) = msg
+            from_socket_message(msg)
+          }),
+        )
+    }
+    let context = case websocket.has_deflate(extensions) {
+      True -> Some(compression.init(context_takeovers))
+      False -> None
+    }
+    State(
+      buffer: <<>>,
+      incomplete: None,
+      self: subject,
+      socket: handshake_response.socket,
+      user_state: user_state,
+      compression: context,
+    )
+    |> actor.initialised
+    |> actor.selecting(selector)
+    |> actor.returning(subject)
+    |> Ok
+  })
+  |> actor.on_message(fn(state, message) {
+    case message {
+      UserMessage(user_message) -> {
+        let conn =
+          Connection(
+            state.socket,
+            transport,
+            option.map(state.compression, fn(context) { context.deflate }),
+          )
+        let res =
+          exception.rescue(fn() {
+            builder.loop(state.user_state, User(user_message), conn)
+          })
+        case res {
+          // TODO:  de-dupe this
+          Ok(Continue(user_state, user_selector)) -> {
+            let new_state = State(..state, user_state: user_state)
+            case user_selector {
+              Some(user_selector) -> {
+                let selector =
+                  user_selector
+                  |> process.map_selector(UserMessage)
+                  |> process.merge_selector(
+                    process.map_selector(socket.selector(), fn(msg) {
+                      let assert Ok(msg) = msg
+                      from_socket_message(msg)
+                    }),
+                  )
+                new_state
+                |> actor.continue
+                |> actor.with_selector(selector)
+              }
+              _ -> actor.continue(new_state)
+            }
+          }
+          Ok(NormalStop) -> actor.stop()
+          Ok(AbnormalStop(reason)) -> actor.stop_abnormal(reason)
+          Error(reason) -> {
+            logging.log(
+              logging.Error,
+              "Caught error in user handler: " <> string.inspect(reason),
+            )
+            actor.continue(state)
+          }
+        }
+      }
+      Err(reason) -> {
+        close_contexts(state.compression)
+        actor.stop_abnormal(string.inspect(reason))
+      }
+      Data(bits) -> {
+        let conn =
+          Connection(
+            state.socket,
+            transport,
+            option.map(state.compression, fn(context) { context.deflate }),
+          )
+        let #(frames, rest) =
+          websocket.decode_many_frames(
+            bit_array.append(state.buffer, bits),
+            option.map(state.compression, fn(context) { context.inflate }),
+            [],
+          )
+        let frames = websocket.aggregate_frames(frames, state.incomplete, [])
+        case frames {
+          Error(Nil) -> continue(state)
+          Ok(frames) -> {
+            list.fold_until(frames, continue(state), fn(acc, frame) {
+              let assert Continue(prev_state, _selector) = acc
+              case handle_frame(builder, prev_state, conn, frame) {
+                Continue(..) as next -> list.Continue(next)
+                err -> list.Stop(err)
+              }
+            })
+          }
+        }
+        |> fn(next) {
+          case next {
+            Continue(state, selector) -> {
+              let assert Ok(_) =
+                transport.set_opts(
+                  transport,
+                  state.socket,
+                  socket.convert_options([Receive(Once)]),
+                )
+              let next = actor.continue(State(..state, buffer: rest))
+              case selector {
+                Some(selector) -> actor.with_selector(next, selector)
+                _ -> next
+              }
+            }
+            NormalStop -> {
+              close_contexts(state.compression)
+              actor.stop()
+            }
+            AbnormalStop(reason) -> {
+              close_contexts(state.compression)
+              actor.stop_abnormal(reason)
+            }
+          }
+        }
+      }
+      Closed -> {
+        logging.log(logging.Debug, "Received closed frame")
+        builder.on_close(state.user_state)
+        close_contexts(state.compression)
+        actor.stop()
+      }
+      // TODO:  handle shutdown better?
+      Shutdown -> {
+        logging.log(logging.Debug, "Received shutdown messag")
+        close_contexts(state.compression)
+        actor.stop()
+      }
+    }
+  })
+  |> actor.start
+  |> result.map_error(ActorFailed)
+  |> result.try(fn(started) {
+    case transport {
+      Tcp -> tcp.controlling_process(handshake_response.socket, started.pid)
+      Ssl -> ssl.controlling_process(handshake_response.socket, started.pid)
+    }
+    |> result.map_error(fn(socket_reason) {
+      FailedToTransferSocket(convert_socket_reason(socket_reason))
+    })
+    |> result.map(fn(_nil) { started })
+  })
+  |> result.map(fn(started) {
+    let _ = case handshake_response.buffer {
+      <<>> -> Nil
+      data -> process.send(started.data, Data(data))
+    }
+    started
+  })
+}
+
+fn handle_frame(
+  builder: Builder(user_state, user_message),
+  state: State(user_state, user_message),
+  conn: Connection,
+  frame: websocket.Frame,
+) -> Next(State(user_state, user_message), InternalMessage(user_message)) {
+  case frame {
+    DataFrame(TextFrame(payload: data)) -> {
+      let assert Ok(str) = bit_array.to_string(data)
+      let res =
+        exception.rescue(fn() {
+          builder.loop(state.user_state, Text(str), conn)
+        })
+      case res {
+        // TODO:  de-dupe this
+        Ok(Continue(user_state, user_selector)) -> {
+          let new_state = State(..state, user_state: user_state)
+          case user_selector {
+            Some(user_selector) -> {
+              let selector =
+                user_selector
+                |> process.map_selector(UserMessage)
+                |> process.merge_selector(
+                  process.map_selector(socket.selector(), fn(msg) {
+                    let assert Ok(msg) = msg
+                    from_socket_message(msg)
+                  }),
+                )
+              Continue(new_state, Some(selector))
+            }
+            _ -> continue(new_state)
+          }
+        }
+        Ok(NormalStop) -> NormalStop
+        Ok(AbnormalStop(reason)) -> AbnormalStop(reason)
+        Error(reason) -> {
+          logging.log(
+            logging.Error,
+            "Caught error in user handler: " <> string.inspect(reason),
+          )
+          continue(state)
+        }
+      }
+    }
+    DataFrame(BinaryFrame(payload: data)) -> {
+      let res =
+        exception.rescue(fn() {
+          builder.loop(state.user_state, Binary(data), conn)
+        })
+      case res {
+        // TODO:  de-dupe this
+        Ok(Continue(user_state, user_selector)) -> {
+          let new_state = State(..state, user_state: user_state)
+          case user_selector {
+            Some(user_selector) -> {
+              let selector =
+                user_selector
+                |> process.map_selector(UserMessage)
+                |> process.merge_selector(
+                  process.map_selector(socket.selector(), fn(msg) {
+                    let assert Ok(msg) = msg
+                    from_socket_message(msg)
+                  }),
+                )
+              Continue(new_state, Some(selector))
+            }
+            _ -> continue(new_state)
+          }
+        }
+        Ok(NormalStop) -> NormalStop
+        Ok(AbnormalStop(reason)) -> AbnormalStop(reason)
+        Error(reason) -> {
+          logging.log(
+            logging.Error,
+            "Caught error in user handler: " <> string.inspect(reason),
+          )
+          continue(state)
+        }
+      }
+    }
+    Control(PingFrame(payload)) -> {
+      let mask = Some(<<0:unit(8)-size(4)>>)
+      let frame = websocket.encode_pong_frame(payload, mask)
+
+      let _ = transport.send(conn.transport, conn.socket, frame)
+      continue(state)
+    }
+    Control(PongFrame(..)) -> {
+      continue(state)
+    }
+    Control(CloseFrame(reason)) -> {
+      logging.log(
+        logging.Debug,
+        "WebSocket closing: " <> string.inspect(reason),
+      )
+      builder.on_close(state.user_state)
+      NormalStop
+    }
+    Continuation(..) -> {
+      continue(state)
+    }
+  }
+}
+
+/// The `Subject` returned from `initialize` is an opaque type.  In order to
+/// send custom messages to your process, you can do this mapping.
+///
+/// For example:
+/// ```gleam
+///   // using `process.send`
+///   MyMessage(some_data)
+///   |> stratus.to_user_message
+///   |> process.send(stratus_subject, _)
+///   // using `process.call`
+///   process.call(stratus_subject, fn(subj) {
+///     stratus.to_user_message(MyMessage(some_data, subj))
+///   })
+/// ```
+pub fn to_user_message(
+  user_message: user_message,
+) -> InternalMessage(user_message) {
+  UserMessage(user_message)
+}
+
+/// From within the actor loop, this is how you send a WebSocket text frame.
+/// This must be valid UTF-8, so it is a `String`.
+pub fn send_text_message(
+  conn: Connection,
+  msg: String,
+) -> Result(Nil, SocketReason) {
+  let frame =
+    websocket.encode_text_frame(
+      msg,
+      conn.context,
+      Some(crypto.strong_random_bytes(4)),
+    )
+  transport.send(conn.transport, conn.socket, frame)
+  |> result.map_error(convert_socket_reason)
+}
+
+/// From within the actor loop, this is how you send a WebSocket text frame.
+pub fn send_binary_message(
+  conn: Connection,
+  msg: BitArray,
+) -> Result(Nil, SocketReason) {
+  let frame =
+    websocket.encode_binary_frame(
+      msg,
+      conn.context,
+      Some(crypto.strong_random_bytes(4)),
+    )
+  transport.send(conn.transport, conn.socket, frame)
+  |> result.map_error(convert_socket_reason)
+}
+
+/// Send a ping frame with some data.
+pub fn send_ping(conn: Connection, data: BitArray) -> Result(Nil, SocketReason) {
+  let size = bit_array.byte_size(data)
+  let mask = case size {
+    0 -> Some(<<0:size(4)>>)
+    _n -> Some(crypto.strong_random_bytes(4))
+  }
+  let frame = websocket.encode_ping_frame(data, mask)
+  transport.send(conn.transport, conn.socket, frame)
+  |> result.map_error(convert_socket_reason)
+}
+
+pub type CloseReason {
+  NotProvided
+  /// Status code: 1000
+  Normal(body: BitArray)
+  /// Status code: 1001
+  GoingAway(body: BitArray)
+  /// Status code: 1002
+  ProtocolError(body: BitArray)
+  /// Status code: 1003
+  UnexpectedDataType(body: BitArray)
+  /// Status code: 1007
+  InconsistentDataType(body: BitArray)
+  /// Status code: 1008
+  PolicyViolation(body: BitArray)
+  /// Status code: 1009
+  MessageTooBig(body: BitArray)
+  /// Status code: 1010
+  MissingExtensions(body: BitArray)
+  /// Status code: 1011
+  UnexpectedCondition(body: BitArray)
+  /// Use [`close_custom`](#close_custom) to send a custom close code.
+  Custom(CustomCloseReason)
+}
+
+/// Use [`close_custom`](#close_custom) to send a custom close code.
+pub opaque type CustomCloseReason {
+  CustomCloseReason(code: Int, body: BitArray)
+}
+
+fn convert_close_reason(reason: CloseReason) -> websocket.CloseReason {
+  case reason {
+    NotProvided -> websocket.NotProvided
+    GoingAway(body:) -> websocket.GoingAway(body:)
+    InconsistentDataType(body:) -> websocket.InconsistentDataType(body:)
+    MessageTooBig(body:) -> websocket.MessageTooBig(body:)
+    MissingExtensions(body:) -> websocket.MissingExtensions(body:)
+    Normal(body:) -> websocket.Normal(body:)
+    PolicyViolation(body:) -> websocket.PolicyViolation(body:)
+    ProtocolError(body:) -> websocket.ProtocolError(body:)
+    UnexpectedCondition(body:) -> websocket.UnexpectedCondition(body:)
+    UnexpectedDataType(body:) -> websocket.UnexpectedDataType(body:)
+    Custom(CustomCloseReason(code:, body:)) ->
+      websocket.CustomCloseReason(code:, body:)
+  }
+}
+
+/// Closes the connection with a custom close code between 1000 and 4999.
+pub fn close_custom(
+  conn: Connection,
+  code code: Int,
+  body body: BitArray,
+) -> Result(Nil, CustomCloseError) {
+  use <- bool.guard(
+    when: code >= 5000 || code < 1000,
+    return: Error(InvalidCode),
+  )
+
+  close(conn, Custom(CustomCloseReason(code:, body:)))
+  |> result.map_error(SocketFail)
+}
+
+pub fn close(
+  conn: Connection,
+  because reason: CloseReason,
+) -> Result(Nil, SocketReason) {
+  let reason = convert_close_reason(reason)
+  let mask = crypto.strong_random_bytes(4)
+  let frame = websocket.encode_close_frame(reason, Some(mask))
+
+  transport.send(conn.transport, conn.socket, frame)
+  |> result.map_error(convert_socket_reason)
+}
+
+fn make_upgrade(req: Request(String)) -> BytesTree {
+  let user_headers = case req.headers {
+    [] -> ""
+    _ ->
+      req.headers
+      |> list.filter(fn(pair) {
+        let #(key, _value) = pair
+        key != "host"
+        && key != "upgrade"
+        && key != "connection"
+        && key != "sec-websocket-key"
+        && key != "sec-websocket-version"
+      })
+      |> list.map(fn(pair) {
+        let #(key, value) = pair
+        key <> ": " <> value
+      })
+      |> string.join("\r\n")
+      |> string.append("\r\n")
+  }
+
+  let path = case req.path {
+    "" -> "/"
+    path -> path
+  }
+
+  let query =
+    req
+    |> request.get_query
+    |> result.map(uri.query_to_string)
+    |> fn(str) {
+      case str {
+        Ok("") -> ""
+        Ok(str) -> "?" <> str
+        _ -> ""
+      }
+    }
+
+  let port =
+    req.port
+    |> option.map(fn(port) { ":" <> int.to_string(port) })
+    |> option.unwrap("")
+
+  bytes_tree.new()
+  |> bytes_tree.append_string("GET " <> path <> query <> " HTTP/1.1\r\n")
+  |> bytes_tree.append_string("host: " <> req.host <> port <> "\r\n")
+  |> bytes_tree.append_string("upgrade: websocket\r\n")
+  |> bytes_tree.append_string("connection: upgrade\r\n")
+  |> bytes_tree.append_string(
+    "sec-websocket-key: " <> websocket.make_client_key() <> "\r\n",
+  )
+  |> bytes_tree.append_string("sec-websocket-version: 13\r\n")
+  |> bytes_tree.append_string(
+    "sec-websocket-extensions: permessage-deflate\r\n",
+  )
+  |> bytes_tree.append_string(user_headers)
+  |> bytes_tree.append_string("\r\n")
+}
+
+pub type HandshakeError {
+  Sock(SocketReason)
+  Protocol(BitArray)
+  UpgradeFailed(Response(BitArray))
+}
+
+type HandshakeResponse {
+  HandshakeResponse(
+    socket: Socket,
+    response: Response(BitArray),
+    buffer: BitArray,
+  )
+}
+
+fn perform_handshake(
+  req: Request(String),
+  transport: Transport,
+  timeout: Int,
+) -> Result(HandshakeResponse, HandshakeError) {
+  let certs = case req.scheme {
+    Https -> {
+      let assert Ok(_ok) = ssl.start()
+      [Cacerts(socket.get_certs()), socket.get_custom_matcher()]
+    }
+    Http -> []
+  }
+
+  let opts =
+    socket.convert_options(
+      list.append(socket.default_options, [Receive(Pull), ..certs]),
+    )
+
+  let port =
+    option.lazy_unwrap(req.port, fn() {
+      case transport {
+        Ssl -> 443
+        Tcp -> 80
+      }
+    })
+
+  logging.log(
+    logging.Debug,
+    "Making request to " <> req.host <> " at " <> int.to_string(port),
+  )
+
+  use socket <- result.try(
+    result.map_error(
+      transport.connect(
+        transport,
+        charlist.from_string(req.host),
+        port,
+        opts,
+        timeout,
+      ),
+      fn(err) { Sock(convert_socket_reason(err)) },
+    ),
+  )
+
+  let upgrade_req = make_upgrade(req)
+
+  use _nil <- result.try(
+    result.map_error(transport.send(transport, socket, upgrade_req), fn(err) {
+      Sock(convert_socket_reason(err))
+    }),
+  )
+
+  logging.log(
+    logging.Debug,
+    "Sent upgrade request, waiting " <> int.to_string(timeout),
+  )
+
+  use resp <- result.try(
+    result.map_error(
+      transport.receive_timeout(transport, socket, 0, timeout),
+      fn(err) { Sock(convert_socket_reason(err)) },
+    ),
+  )
+
+  resp
+  |> gramps_http.read_response
+  |> result.map_error(fn(_err) { Protocol(resp) })
+  |> result.try(fn(pair) {
+    let #(resp, body) = pair
+    let body_size =
+      resp.headers
+      |> list.key_find("content-length")
+      |> result.try(int.parse)
+      |> result.unwrap(0)
+    case read_body(transport, socket, timeout, body_size, body) {
+      Ok(#(body, rest)) -> {
+        Ok(#(response.set_body(resp, body), rest))
+      }
+      Error(reason) -> Error(Sock(reason))
+    }
+  })
+  |> result.try(fn(pair) {
+    let #(resp, buffer) = pair
+    case resp.status {
+      101 -> Ok(HandshakeResponse(socket:, response: resp, buffer:))
+      _ -> Error(UpgradeFailed(resp))
+    }
+  })
+}
+
+fn read_body(
+  transport: Transport,
+  socket: Socket,
+  timeout: Int,
+  length: Int,
+  body: BitArray,
+) -> Result(#(BitArray, BitArray), SocketReason) {
+  case body {
+    <<data:bytes-size(length), rest:bits>> -> Ok(#(data, rest))
+    _ -> {
+      case transport.receive_timeout(transport, socket, 0, timeout) {
+        Ok(data) -> {
+          read_body(transport, socket, timeout, length, <<body:bits, data:bits>>)
+        }
+        Error(reason) -> Error(convert_socket_reason(reason))
+      }
+    }
+  }
+}
+
+fn close_contexts(contexts: Option(compression.Compression)) -> Nil {
+  case contexts {
+    Some(compression) -> {
+      compression.close(compression.deflate)
+      compression.close(compression.inflate)
+      Nil
+    }
+    _ -> Nil
+  }
+}

src/goose/stratus/internal/socket.gleam

···
+import gleam/dynamic.{type Dynamic}
+import gleam/dynamic/decode
+import gleam/erlang/atom.{type Atom}
+import gleam/erlang/process.{type Selector}
+import gleam/list
+
+pub type Socket
+
+pub type SocketReason {
+  Closed
+  Timeout
+  Badarg
+  Terminated
+  Eaddrinuse
+  Eaddrnotavail
+  Eafnosupport
+  Ealready
+  Econnaborted
+  Econnrefused
+  Econnreset
+  Edestaddrreq
+  Ehostdown
+  Ehostunreach
+  Einprogress
+  Eisconn
+  Emsgsize
+  Enetdown
+  Enetunreach
+  Enopkg
+  Enoprotoopt
+  Enotconn
+  Enotty
+  Enotsock
+  Eproto
+  Eprotonosupport
+  Eprototype
+  Esocktnosupport
+  Etimedout
+  Ewouldblock
+  Exbadport
+  Exbadseq
+}
+
+pub type TcpOption =
+  #(Atom, Dynamic)
+
+pub type ReceiveMode {
+  Count(Int)
+  Once
+  Pull
+  All
+}
+
+pub type PacketType {
+  Binary
+  List
+}
+
+pub type Options {
+  Receive(ReceiveMode)
+  PacketsOf(PacketType)
+  SendTimeout(Int)
+  SendTimeoutClose(Bool)
+  Reuseaddr(Bool)
+  Nodelay(Bool)
+  Cacerts(Dynamic)
+  CustomizeHostnameCheck(Dynamic)
+}
+
+pub const default_options = [
+  PacketsOf(Binary),
+  SendTimeout(30_000),
+  SendTimeoutClose(True),
+  Reuseaddr(True),
+  Nodelay(True),
+]
+
+@external(erlang, "gleam@function", "identity")
+fn from(value: a) -> Dynamic
+
+pub fn convert_options(options: List(Options)) -> List(TcpOption) {
+  let active = atom.create("active")
+  list.map(options, fn(opt) {
+    case opt {
+      Receive(Count(count)) -> #(active, dynamic.int(count))
+      Receive(Once) -> #(active, from(atom.create("once")))
+      Receive(Pull) -> #(active, dynamic.bool(False))
+      Receive(All) -> #(active, dynamic.bool(True))
+      PacketsOf(Binary) -> #(atom.create("mode"), from(Binary))
+      PacketsOf(List) -> #(atom.create("mode"), from(List))
+      Cacerts(data) -> #(atom.create("cacerts"), data)
+      Nodelay(bool) -> #(atom.create("nodelay"), dynamic.bool(bool))
+      Reuseaddr(bool) -> #(atom.create("reuseaddr"), dynamic.bool(bool))
+      SendTimeout(int) -> #(atom.create("send_timeout"), dynamic.int(int))
+      SendTimeoutClose(bool) -> #(
+        atom.create("send_timeout_close"),
+        dynamic.bool(bool),
+      )
+      CustomizeHostnameCheck(funcs) -> #(
+        atom.create("customize_hostname_check"),
+        funcs,
+      )
+    }
+  })
+}
+
+pub type Shutdown {
+  Read
+  Write
+  ReadWrite
+}
+
+pub type SocketMessage {
+  Data(BitArray)
+  Err(SocketReason)
+}
+
+type ErlangSocketMessage {
+  Ssl
+  SslClosed
+  SslError
+  Tcp
+  TcpClosed
+  TcpError
+}
+
+pub fn selector() -> Selector(Result(SocketMessage, List(decode.DecodeError))) {
+  process.new_selector()
+  |> process.select_record(Tcp, 2, fn(data) {
+    {
+      use msg <- decode.field(2, decode.bit_array)
+      decode.success(Data(msg))
+    }
+    |> decode.run(data, _)
+  })
+  |> process.select_record(Ssl, 2, fn(data) {
+    {
+      use msg <- decode.field(2, decode.bit_array)
+      decode.success(Data(msg))
+    }
+    |> decode.run(data, _)
+  })
+  |> process.select_record(SslClosed, 1, fn(_socket) { Ok(Err(Closed)) })
+  |> process.select_record(TcpClosed, 1, fn(_socket) { Ok(Err(Closed)) })
+  |> process.select_record(TcpError, 2, fn(data) {
+    {
+      use reason <- decode.field(2, atom.decoder())
+      case parse_known_socket_reason(reason) {
+        Ok(socket_reason) -> decode.success(Err(socket_reason))
+        Error(_) -> decode.failure(Err(Badarg), "SocketReason")
+      }
+    }
+    |> decode.run(data, _)
+  })
+  |> process.select_record(SslError, 2, fn(data) {
+    {
+      use reason <- decode.field(2, atom.decoder())
+      case parse_known_socket_reason(reason) {
+        Ok(socket_reason) -> decode.success(Err(socket_reason))
+        Error(_) -> decode.failure(Err(Badarg), "SocketReason")
+      }
+    }
+    |> decode.run(data, _)
+  })
+}
+
+@external(erlang, "public_key", "cacerts_get")
+pub fn get_certs() -> Dynamic
+
+@external(erlang, "stratus_ffi", "custom_sni_matcher")
+pub fn get_custom_matcher() -> Options
+
+@external(erlang, "stratus_ffi", "parse_known_socket_reason")
+fn parse_known_socket_reason(reason: Atom) -> Result(SocketReason, Dynamic)

src/goose/stratus/internal/ssl.gleam

···
+import gleam/bytes_tree.{type BytesTree}
+import gleam/erlang/charlist.{type Charlist}
+import gleam/erlang/process.{type Pid}
+import goose/stratus/internal/socket.{
+  type Shutdown, type Socket, type SocketReason, type TcpOption,
+}
+
+@external(erlang, "ssl", "connect")
+pub fn connect(
+  address: Charlist,
+  port: Int,
+  options: List(TcpOption),
+  timeout: Int,
+) -> Result(Socket, SocketReason)
+
+@external(erlang, "stratus_ffi", "ssl_shutdown")
+pub fn shutdown(socket: Socket, how: Shutdown) -> Result(Nil, SocketReason)
+
+@external(erlang, "stratus_ffi", "ssl_send")
+pub fn send(socket: Socket, packet: BytesTree) -> Result(Nil, SocketReason)
+
+@external(erlang, "ssl", "recv")
+pub fn receive(socket: Socket, length: Int) -> Result(BitArray, SocketReason)
+
+@external(erlang, "ssl", "recv")
+pub fn receive_timeout(
+  socket: Socket,
+  length: Int,
+  timeout: Int,
+) -> Result(BitArray, SocketReason)
+
+@external(erlang, "stratus_ffi", "ssl_set_opts")
+pub fn set_opts(
+  socket: Socket,
+  opts: List(TcpOption),
+) -> Result(Nil, SocketReason)
+
+@external(erlang, "stratus_ffi", "ssl_start")
+pub fn start() -> Result(Nil, Nil)
+
+@external(erlang, "stratus_ffi", "ssl_controlling_process")
+pub fn controlling_process(
+  socket: Socket,
+  new_owner: Pid,
+) -> Result(Nil, SocketReason)

src/goose/stratus/internal/tcp.gleam

···
+import gleam/bytes_tree.{type BytesTree}
+import gleam/erlang/charlist.{type Charlist}
+import gleam/erlang/process.{type Pid}
+import goose/stratus/internal/socket.{
+  type Shutdown, type Socket, type SocketReason, type TcpOption,
+}
+
+@external(erlang, "gen_tcp", "connect")
+pub fn connect(
+  address: Charlist,
+  port: Int,
+  options: List(TcpOption),
+  timeout: Int,
+) -> Result(Socket, SocketReason)
+
+@external(erlang, "stratus_ffi", "tcp_shutdown")
+pub fn shutdown(socket: Socket, how: Shutdown) -> Result(Nil, SocketReason)
+
+@external(erlang, "stratus_ffi", "tcp_send")
+pub fn send(socket: Socket, packet: BytesTree) -> Result(Nil, SocketReason)
+
+@external(erlang, "gen_tcp", "recv")
+pub fn receive(socket: Socket, length: Int) -> Result(BitArray, SocketReason)
+
+@external(erlang, "gen_tcp", "recv")
+pub fn receive_timeout(
+  socket: Socket,
+  length: Int,
+  timeout: Int,
+) -> Result(BitArray, SocketReason)
+
+@external(erlang, "stratus_ffi", "tcp_set_opts")
+pub fn set_opts(
+  socket: Socket,
+  opts: List(TcpOption),
+) -> Result(Nil, SocketReason)
+
+@external(erlang, "stratus_ffi", "tcp_controlling_process")
+pub fn controlling_process(
+  socket: Socket,
+  new_owner: Pid,
+) -> Result(Nil, SocketReason)

src/goose/stratus/internal/transport.gleam

···
+import gleam/bytes_tree.{type BytesTree}
+import gleam/erlang/charlist.{type Charlist}
+import goose/stratus/internal/socket.{
+  type Shutdown, type Socket, type SocketReason, type TcpOption,
+}
+import goose/stratus/internal/ssl
+import goose/stratus/internal/tcp
+
+pub type Transport {
+  Tcp
+  Ssl
+}
+
+pub fn connect(
+  transport: Transport,
+  host: Charlist,
+  port: Int,
+  options: List(TcpOption),
+  timeout: Int,
+) -> Result(Socket, SocketReason) {
+  case transport {
+    Ssl -> ssl.connect(host, port, options, timeout)
+    Tcp -> tcp.connect(host, port, options, timeout)
+  }
+}
+
+pub fn send(
+  transport: Transport,
+  socket: Socket,
+  data: BytesTree,
+) -> Result(Nil, SocketReason) {
+  case transport {
+    Ssl -> ssl.send(socket, data)
+    Tcp -> tcp.send(socket, data)
+  }
+}
+
+pub fn receive(
+  transport: Transport,
+  socket: Socket,
+  length: Int,
+) -> Result(BitArray, SocketReason) {
+  case transport {
+    Ssl -> ssl.receive(socket, length)
+    Tcp -> tcp.receive(socket, length)
+  }
+}
+
+pub fn receive_timeout(
+  transport: Transport,
+  socket: Socket,
+  length: Int,
+  timeout: Int,
+) -> Result(BitArray, SocketReason) {
+  case transport {
+    Ssl -> ssl.receive_timeout(socket, length, timeout)
+    Tcp -> tcp.receive_timeout(socket, length, timeout)
+  }
+}
+
+pub fn shutdown(
+  transport: Transport,
+  socket: Socket,
+  how: Shutdown,
+) -> Result(Nil, SocketReason) {
+  case transport {
+    Ssl -> ssl.shutdown(socket, how)
+    Tcp -> tcp.shutdown(socket, how)
+  }
+}
+
+pub fn set_opts(
+  transport: Transport,
+  socket: Socket,
+  opts: List(TcpOption),
+) -> Result(Nil, SocketReason) {
+  case transport {
+    Tcp -> tcp.set_opts(socket, opts)
+    Ssl -> ssl.set_opts(socket, opts)
+  }
+}

src/goose_ffi.erl

···
 -module(goose_ffi).
--export([receive_ws_message/0]).
+-export([priv_dir/0, decompress_using_ddict_safe/2, decompress_streaming_safe/2]).
 
-%% Receive a WebSocket text message from the process mailbox
-receive_ws_message() ->
-    receive
-        %% Handle messages forwarded from handler process
-        {ws_text, Text} ->
-            {ok, Text};
-        {ws_binary, _Binary} ->
-            %% Ignore binary messages, try again
-            receive_ws_message();
-        {ws_closed, _Reason} ->
-            {error, closed};
-        {ws_error, _Reason} ->
-            {error, connection_error};
-        _Other ->
-            %% Ignore unexpected messages
-            receive_ws_message()
-    after 60000 ->
-        %% Timeout - connection is still alive, just no messages
-        {error, timeout}
+%% Get the priv directory for the goose application
+priv_dir() ->
+    case code:priv_dir(goose) of
+        {error, _} -> "./priv";
+        Path when is_list(Path) -> list_to_binary(Path);
+        Path -> Path
+    end.
+
+%% Safe wrapper for ezstd:decompress_using_ddict that always returns Result type
+decompress_using_ddict_safe(Data, DDict) ->
+    case ezstd:decompress_using_ddict(Data, DDict) of
+        Result when is_binary(Result) ->
+            {ok, Result};
+        Result when is_list(Result) ->
+            {ok, iolist_to_binary(Result)};
+        {error, Err} when is_binary(Err) ->
+            %% Keep error as binary (Gleam String)
+            {error, Err};
+        {error, Err} when is_list(Err) ->
+            %% Convert list to binary
+            {error, list_to_binary(Err)};
+        {error, Err} ->
+            %% Convert any other type to binary
+            {error, list_to_binary(lists:flatten(io_lib:format("~p", [Err])))}
+    end.
+
+%% Safe wrapper for ezstd:decompress_streaming that always returns Result type
+decompress_streaming_safe(DCtx, Data) ->
+    case ezstd:decompress_streaming(DCtx, Data) of
+        Result when is_binary(Result) ->
+            {ok, Result};
+        Result when is_list(Result) ->
+            {ok, iolist_to_binary(Result)};
+        {error, Err} when is_binary(Err) ->
+            %% Keep error as binary (Gleam String)
+            {error, Err};
+        {error, Err} when is_list(Err) ->
+            %% Convert list to binary
+            {error, list_to_binary(Err)};
+        {error, Err} ->
+            %% Convert any other type to binary
+            {error, list_to_binary(lists:flatten(io_lib:format("~p", [Err])))}
     end.

src/goose_ws_ffi.erl

···
--module(goose_ws_ffi).
--export([connect/3]).
-
-%% Connect to WebSocket using gun
-connect(Url, HandlerPid, Compress) ->
-    %% Start gun application and dependencies
-    application:ensure_all_started(ssl),
-    application:ensure_all_started(gun),
-
-    %% Start ezstd if compression is enabled
-    case Compress of
-        true -> application:ensure_all_started(ezstd);
-        _ -> ok
-    end,
-
-    %% Spawn a connection process that will own the gun connection
-    Parent = self(),
-    spawn(fun() -> connect_worker(Url, HandlerPid, Compress, Parent) end),
-
-    %% Wait for the connection result
-    receive
-        {connection_result, Result} -> Result
-    after 60000 ->
-        {error, connection_timeout}
-    end.
-
-%% Worker process that owns the connection
-connect_worker(Url, HandlerPid, Compress, Parent) ->
-    %% Parse URL using uri_string
-    UriMap = uri_string:parse(Url),
-    #{scheme := SchemeStr, host := Host, path := Path} = UriMap,
-
-    %% Get query string if present and append to path
-    Query = maps:get(query, UriMap, undefined),
-    PathWithQuery = case Query of
-        undefined -> Path;
-        <<>> -> Path;
-        Q -> <<Path/binary, "?", Q/binary>>
-    end,
-
-    %% Get port, use defaults if not specified
-    Port = maps:get(port, uri_string:parse(Url),
-                    case SchemeStr of
-                        <<"wss">> -> 443;
-                        <<"ws">> -> 80;
-                        _ -> 443
-                    end),
-
-    %% Determine transport
-    Transport = case SchemeStr of
-        <<"wss">> -> tls;
-        <<"ws">> -> tcp;
-        _ -> tls
-    end,
-
-    %% TLS options for secure connections
-    TlsOpts = [{verify, verify_none}],  %% For simplicity, disable cert verification
-                                          %% In production, use proper CA certs
-
-    %% Connection options
-    Opts = case Transport of
-        tls ->
-            #{
-                transport => tls,
-                tls_opts => TlsOpts,
-                protocols => [http],
-                retry => 10,
-                retry_timeout => 1000
-            };
-        tcp ->
-            #{
-                transport => tcp,
-                protocols => [http],
-                retry => 10,
-                retry_timeout => 1000
-            }
-    end,
-
-    %% Convert host to list if needed
-    HostStr = case is_binary(Host) of
-        true -> binary_to_list(Host);
-        false -> Host
-    end,
-
-    %% Ensure path with query is binary
-    PathBin = case is_binary(PathWithQuery) of
-        true -> PathWithQuery;
-        false -> list_to_binary(PathWithQuery)
-    end,
-
-    %% Open connection (this process will be the owner)
-    case gun:open(HostStr, Port, Opts) of
-        {ok, ConnPid} ->
-            %% Monitor the connection
-            MRef = monitor(process, ConnPid),
-
-            %% Wait for connection
-            receive
-                {gun_up, ConnPid, _Protocol} ->
-                    %% Upgrade to WebSocket (compression is controlled via query string, not headers)
-                    StreamRef = gun:ws_upgrade(ConnPid, binary_to_list(PathBin), []),
-
-                    %% Wait for upgrade
-                    receive
-                        {gun_upgrade, ConnPid, StreamRef, [<<"websocket">>], _ResponseHeaders} ->
-                            %% Notify parent that connection is ready
-                            Parent ! {connection_result, {ok, ConnPid}},
-                            %% Now handle messages in this process (the connection owner)
-                            handle_messages(ConnPid, StreamRef, HandlerPid, Compress);
-                        {gun_response, ConnPid, _, _, Status, Headers} ->
-                            gun:close(ConnPid),
-                            Parent ! {connection_result, {error, {upgrade_failed, Status, Headers}}};
-                        {gun_error, ConnPid, _StreamRef, Reason} ->
-                            gun:close(ConnPid),
-                            Parent ! {connection_result, {error, {gun_error, Reason}}};
-                        {'DOWN', MRef, process, ConnPid, Reason} ->
-                            Parent ! {connection_result, {error, {connection_down, Reason}}};
-                        _Other ->
-                            gun:close(ConnPid),
-                            Parent ! {connection_result, {error, unexpected_message}}
-                    after 30000 ->
-                        gun:close(ConnPid),
-                        Parent ! {connection_result, {error, upgrade_timeout}}
-                    end;
-                {'DOWN', MRef, process, ConnPid, Reason} ->
-                    Parent ! {connection_result, {error, {connection_failed, Reason}}};
-                _Other ->
-                    gun:close(ConnPid),
-                    Parent ! {connection_result, {error, unexpected_message}}
-            after 30000 ->
-                gun:close(ConnPid),
-                Parent ! {connection_result, {error, connection_timeout}}
-            end;
-        {error, Reason} ->
-            Parent ! {connection_result, {error, {open_failed, Reason}}}
-    end.
-
-%% Handle incoming WebSocket messages
-handle_messages(ConnPid, StreamRef, HandlerPid, Compress) ->
-    %% Load zstd dictionary if compression is enabled
-    Decompressor = case Compress of
-        true ->
-            %% Load dictionary from priv directory
-            PrivDir = code:priv_dir(goose),
-            DictPath = filename:join(PrivDir, "zstd_dictionary"),
-            case file:read_file(DictPath) of
-                {ok, DictData} ->
-                    %% Create decompression context and dictionary
-                    DCtx = ezstd:create_decompression_context(1024 * 1024),
-                    DDict = ezstd:create_ddict(DictData),
-                    %% Select the dictionary for the decompression context
-                    ok = ezstd:select_ddict(DCtx, DDict),
-                    {ok, {DCtx, DDict}};
-                {error, Err} ->
-                    io:format("Failed to load zstd dictionary: ~p~n", [Err]),
-                    {error, Err}
-            end;
-        _ ->
-            none
-    end,
-    handle_messages_loop(ConnPid, StreamRef, HandlerPid, Compress, Decompressor).
-
-%% Message handling loop
-handle_messages_loop(ConnPid, StreamRef, HandlerPid, Compress, Decompressor) ->
-    receive
-        {gun_ws, _AnyConnPid, _AnyStreamRef, {text, Text}} ->
-            HandlerPid ! {ws_text, Text},
-            handle_messages_loop(ConnPid, StreamRef, HandlerPid, Compress, Decompressor);
-        {gun_ws, _AnyConnPid, _AnyStreamRef, {binary, Binary}} ->
-            %% If compression is enabled, decompress the binary data
-            case {Compress, Decompressor} of
-                {true, {ok, {DCtx, DDict}}} ->
-                    %% Try decompress_using_ddict first (works for frames with content size)
-                    case ezstd:decompress_using_ddict(Binary, DDict) of
-                        Result when is_binary(Result) ->
-                            HandlerPid ! {ws_text, Result};
-                        Result when is_list(Result) ->
-                            HandlerPid ! {ws_text, iolist_to_binary(Result)};
-                        {error, <<"failed to decompress: ZSTD_CONTENTSIZE_UNKNOWN">>} ->
-                            %% Frame doesn't have content size, use streaming with dictionary-loaded context
-                            case ezstd:decompress_streaming(DCtx, Binary) of
-                                StreamResult when is_binary(StreamResult) ->
-                                    HandlerPid ! {ws_text, StreamResult};
-                                StreamResult when is_list(StreamResult) ->
-                                    HandlerPid ! {ws_text, iolist_to_binary(StreamResult)};
-                                {error, _StreamReason} ->
-                                    %% Skip frames that fail to decompress
-                                    ok
-                            end;
-                        {error, _Reason} ->
-                            %% Skip frames that fail to decompress
-                            ok
-                    end;
-                _ ->
-                    %% No compression, ignore binary messages
-                    ok
-            end,
-            handle_messages_loop(ConnPid, StreamRef, HandlerPid, Compress, Decompressor);
-        {gun_ws, ConnPid, StreamRef, close} ->
-            HandlerPid ! {ws_closed, normal},
-            gun:close(ConnPid);
-        {gun_down, ConnPid, _Protocol, Reason, _KilledStreams} ->
-            HandlerPid ! {ws_error, Reason},
-            gun:close(ConnPid);
-        {gun_error, ConnPid, StreamRef, Reason} ->
-            HandlerPid ! {ws_error, Reason},
-            handle_messages_loop(ConnPid, StreamRef, HandlerPid, Compress, Decompressor);
-        stop ->
-            gun:close(ConnPid);
-        _Other ->
-            %% Ignore unexpected messages
-            handle_messages_loop(ConnPid, StreamRef, HandlerPid, Compress, Decompressor)
-    after 30000 ->
-        %% Heartbeat every 30 seconds to keep connection alive
-        handle_messages_loop(ConnPid, StreamRef, HandlerPid, Compress, Decompressor)
-    end.

src/stratus_ffi.erl

···
+-module(stratus_ffi).
+
+-export([tcp_shutdown/2, tcp_send/2, ssl_shutdown/2, ssl_send/2, tcp_set_opts/2,
+         ssl_set_opts/2, ssl_start/0, custom_sni_matcher/0,
+         ssl_controlling_process/2, tcp_controlling_process/2,
+         parse_known_socket_reason/1 ]).
+
+tcp_shutdown(Socket, How) ->
+  case gen_tcp:shutdown(Socket, How) of
+    ok ->
+      {ok, nil};
+    {error, Reason} ->
+      {error, Reason}
+  end.
+
+tcp_send(Socket, Packet) ->
+  case gen_tcp:send(Socket, Packet) of
+    ok ->
+      {ok, nil};
+    {error, Reason} ->
+      {error, Reason}
+  end.
+
+ssl_shutdown(Socket, How) ->
+  case ssl:shutdown(Socket, How) of
+    ok ->
+      {ok, nil};
+    {error, Reason} ->
+      {error, Reason}
+  end.
+
+ssl_send(Socket, Packet) ->
+  case ssl:send(Socket, Packet) of
+    ok ->
+      {ok, nil};
+    {error, Reason} ->
+      {error, Reason}
+  end.
+
+tcp_set_opts(Socket, Opts) ->
+  case inet:setopts(Socket, Opts) of
+    ok ->
+      {ok, nil};
+    {error, Reason} ->
+      {error, Reason}
+  end.
+
+ssl_set_opts(Socket, Opts) ->
+  case ssl:setopts(Socket, Opts) of
+    ok ->
+      {ok, nil};
+    {error, Reason} ->
+      {error, Reason}
+  end.
+
+ssl_start() ->
+  case ssl:start() of
+    ok ->
+      {ok, nil};
+    {error, Reason} ->
+      {error, Reason}
+  end.
+
+% Thank you!  https://github.com/erlang/otp/issues/4321
+custom_sni_matcher() ->
+  {customize_hostname_check,
+   [{match_fun, public_key:pkix_verify_hostname_match_fun(https)}]}.
+
+ssl_controlling_process(Socket, NewOwner) ->
+  case ssl:controlling_process(Socket, NewOwner) of
+    ok -> {ok, nil};
+    Error -> Error
+  end.
+
+tcp_controlling_process(Socket, NewOwner) ->
+  case gen_tcp:controlling_process(Socket, NewOwner) of
+    ok -> {ok, nil};
+    Error -> Error
+  end.
+
+parse_known_socket_reason(Reason) ->
+  case Reason of
+    closed -> {ok, closed};
+    timeout -> {ok, timeout};
+    badarg -> {ok, badarg};
+    terminated -> {ok, terminated};
+    eaddrinuse -> {ok, eaddrinuse};
+    eaddrnotavail -> {ok, eaddrnotavail};
+    eafnosupport -> {ok, eafnosupport};
+    ealready -> {ok, ealready};
+    econnaborted -> {ok, econnaborted};
+    econnrefused -> {ok, econnrefused};
+    econnreset -> {ok, econnreset};
+    edestaddrreq -> {ok, edestaddrreq};
+    ehostdown -> {ok, ehostdown};
+    ehostunreach -> {ok, ehostunreach};
+    einprogress -> {ok, einprogress};
+    eisconn -> {ok, eisconn};
+    emsgsize -> {ok, emsgsize};
+    enetdown -> {ok, enetdown};
+    enetunreach -> {ok, enetunreach};
+    enopkg -> {ok, enopkg};
+    enoprotoopt -> {ok, enoprotoopt};
+    enotconn -> {ok, enotconn};
+    enotty -> {ok, enotty};
+    enotsock -> {ok, enotsock};
+    eproto -> {ok, eproto};
+    eprotonosupport -> {ok, eprotonosupport};
+    eprototype -> {ok, eprototype};
+    esocktnosupport -> {ok, esocktnosupport};
+    etimedout -> {ok, etimedout};
+    ewouldblock -> {ok, ewouldblock};
+    exbadport -> {ok, exbadport};
+    exbadseq -> {ok, exbadseq};
+    Unknown -> {error, Unknown}
+  end.

test/goose_test.gleam

···
       max_message_size_bytes: option.None,
       compress: False,
       require_hello: False,
-      max_backoff_seconds: 60,
-      log_connection_events: True,
-      log_retry_attempts: True,
     )
   let url = goose.build_url(config)
 
···
       max_message_size_bytes: option.None,
       compress: False,
       require_hello: False,
-      max_backoff_seconds: 60,
-      log_connection_events: True,
-      log_retry_attempts: True,
     )
   let url = goose.build_url(config)
 
···
       max_message_size_bytes: option.None,
       compress: False,
       require_hello: False,
-      max_backoff_seconds: 60,
-      log_connection_events: True,
-      log_retry_attempts: True,
     )
   let url = goose.build_url(config)
 
···
       max_message_size_bytes: option.None,
       compress: False,
       require_hello: False,
-      max_backoff_seconds: 60,
-      log_connection_events: True,
-      log_retry_attempts: True,
     )
   let url = goose.build_url(config)
 
···
       max_message_size_bytes: option.Some(1_048_576),
       compress: False,
       require_hello: False,
-      max_backoff_seconds: 60,
-      log_connection_events: True,
-      log_retry_attempts: True,
     )
   let url = goose.build_url(config)
 
···
       max_message_size_bytes: option.None,
       compress: True,
       require_hello: False,
-      max_backoff_seconds: 60,
-      log_connection_events: True,
-      log_retry_attempts: True,
     )
   let url = goose.build_url(config)
 
···
       max_message_size_bytes: option.None,
       compress: False,
       require_hello: True,
-      max_backoff_seconds: 60,
-      log_connection_events: True,
-      log_retry_attempts: True,
     )
   let url = goose.build_url(config)
 
···
       max_message_size_bytes: option.Some(2_097_152),
       compress: True,
       require_hello: True,
-      max_backoff_seconds: 60,
-      log_connection_events: True,
-      log_retry_attempts: True,
     )
   let url = goose.build_url(config)