Refactor firehose demo to use library's subscribe function

Major architectural change:
- Use Firehose.subscribe from the library instead of manual event loop
- Move WebSocket implementation into Ws_handler module as effect handler
- Demo now focuses on filtering/formatting, not protocol details

The demo provides effect handlers for:
- Firehose.Ws_connect: TLS connection with WebSocket upgrade
- Firehose.Ws_recv: Frame parsing with fragmentation support
- Firehose.Ws_close: Connection cleanup

This separates concerns:
- Library: protocol logic (subscribe, decode_frame, event types)
- Demo: presentation (filtering, formatting, statistics)
- Ws_handler: I/O implementation (TLS, WebSocket framing)

Line count: 795 (down from 1204, -34%)

examples/firehose_demo/firehose_demo.ml

···
 module Dag_cbor = Atproto_ipld.Dag_cbor
 module Cid = Atproto_ipld.Cid
 
-(* Record content types *)
+(* ============================================================================
+   Record Content Extraction
+   ============================================================================ *)
+
 type record_content =
   | Post of {
       text : string;
···
   | Profile of { display_name : string option; description : string option }
   | Unknown of string
 
-(* CBOR helpers extending the library *)
 let get_string_array key pairs =
   Firehose.get_array key pairs
   |> Option.map
···
   | Some cid -> List.assoc_opt (Cid.to_string cid) records
   | None -> None
 
-(* Formatting helpers *)
+(* ============================================================================
+   Formatting
+   ============================================================================ *)
+
 let truncate n s =
   if String.length s <= n then s else String.sub s 0 (n - 3) ^ "..."
 
···
 
 let opt_or d = Option.value ~default:d
 
-(* Text formatting *)
 let format_record = function
   | Post { text; reply_to; _ } ->
       Printf.sprintf "\027[36m\"%s\"\027[0m%s"
···
   | Firehose.StreamError err ->
       Printf.sprintf "{\"type\":\"error\",\"message\":%s}" (json_str err)
 
-(* Statistics *)
+(* ============================================================================
+   Statistics
+   ============================================================================ *)
+
 type stats = {
   mutable commits : int;
   mutable identities : int;
···
     elapsed
     (if elapsed > 0. then float_of_int total /. elapsed else 0.)
 
-(* WebSocket implementation *)
-let ws_nonce () =
-  let b = Bytes.create 16 in
-  for i = 0 to 15 do
-    Bytes.set b i (Char.chr (Random.int 256))
-  done;
-  Base64.encode_exn (Bytes.to_string b)
-
-let parse_ws_header buf off len =
-  if len < 2 then None
-  else
-    let b0, b1 =
-      (Char.code (Bytes.get buf off), Char.code (Bytes.get buf (off + 1)))
-    in
-    let fin, opcode = (b0 land 0x80 <> 0, b0 land 0x0f) in
-    let masked, plen = (b1 land 0x80 <> 0, b1 land 0x7f) in
-    let hlen, plen =
-      if plen = 126 then
-        if len < 4 then (0, -1)
-        else
-          ( 4,
-            (Char.code (Bytes.get buf (off + 2)) lsl 8)
-            lor Char.code (Bytes.get buf (off + 3)) )
-      else if plen = 127 then
-        if len < 10 then (0, -1)
-        else
-          ( 10,
-            (Char.code (Bytes.get buf (off + 6)) lsl 24)
-            lor (Char.code (Bytes.get buf (off + 7)) lsl 16)
-            lor (Char.code (Bytes.get buf (off + 8)) lsl 8)
-            lor Char.code (Bytes.get buf (off + 9)) )
-      else (2, plen)
-    in
-    if plen < 0 then None
-    else
-      let mlen = if masked then 4 else 0 in
-      if len < hlen + mlen then None
-      else
-        Some
-          ( fin,
-            opcode,
-            plen,
-            (if masked then Some (Bytes.sub buf (off + hlen) 4) else None),
-            hlen + mlen )
-
-let ws_frame opcode payload =
-  let plen = String.length payload in
-  let mask = Bytes.init 4 (fun _ -> Char.chr (Random.int 256)) in
-  let hdr =
-    if plen < 126 then (
-      let h = Bytes.create 6 in
-      Bytes.set h 0 (Char.chr (0x80 lor opcode));
-      Bytes.set h 1 (Char.chr (0x80 lor plen));
-      Bytes.blit mask 0 h 2 4;
-      Bytes.to_string h)
-    else if plen < 65536 then (
-      let h = Bytes.create 8 in
-      Bytes.set h 0 (Char.chr (0x80 lor opcode));
-      Bytes.set h 1 (Char.chr (0x80 lor 126));
-      Bytes.set h 2 (Char.chr ((plen lsr 8) land 0xff));
-      Bytes.set h 3 (Char.chr (plen land 0xff));
-      Bytes.blit mask 0 h 4 4;
-      Bytes.to_string h)
-    else
-      let h = Bytes.create 14 in
-      Bytes.set h 0 (Char.chr (0x80 lor opcode));
-      Bytes.set h 1 (Char.chr (0x80 lor 127));
-      for i = 2 to 5 do
-        Bytes.set h i '\000'
-      done;
-      Bytes.set h 6 (Char.chr ((plen lsr 24) land 0xff));
-      Bytes.set h 7 (Char.chr ((plen lsr 16) land 0xff));
-      Bytes.set h 8 (Char.chr ((plen lsr 8) land 0xff));
-      Bytes.set h 9 (Char.chr (plen land 0xff));
-      Bytes.blit mask 0 h 10 4;
-      Bytes.to_string h
-  in
-  let masked =
-    Bytes.mapi
-      (fun i c ->
-        Char.chr (Char.code c lxor Char.code (Bytes.get mask (i mod 4))))
-      (Bytes.of_string payload)
-  in
-  hdr ^ Bytes.to_string masked
+(* ============================================================================
+   Event Filtering
+   ============================================================================ *)
 
-(* Event filtering *)
 type filter =
   | Posts
   | Likes
···
   | Tombstones -> "tombstones"
 
 let prefix_match prefix path =
-  let n = String.length prefix in
-  String.length path >= n && String.sub path 0 n = prefix
+  String.length path >= String.length prefix
+  && String.sub path 0 (String.length prefix) = prefix
 
 let op_matches filter (op : Firehose.operation) =
   match filter with
···
       | Firehose.Tombstone _ -> List.mem Tombstones filters
       | Firehose.Info _ | Firehose.StreamError _ -> true)
 
-(* Firehose connection *)
-let run_firehose ~net ~filters ~json ~rich ?cursor ?limit () =
-  let host, port = ("bsky.network", 443) in
-  let resource =
-    "/xrpc/com.atproto.sync.subscribeRepos"
-    ^ (cursor |> Option.map (Printf.sprintf "?cursor=%Ld") |> opt_or "")
-  in
+(* ============================================================================
+   WebSocket Effect Handler (Eio-based implementation)
+   ============================================================================ *)
 
-  if not json then Printf.printf "Connecting to wss://%s%s...\n%!" host resource;
+module Ws_handler = struct
+  type conn = {
+    socket : Tls_eio.t;
+    buf : bytes;
+    mutable buf_len : int;
+    frag_buf : Buffer.t;
+    mutable frag_opcode : int;
+  }
+  (** WebSocket connection state *)
+
+  let ws_nonce () =
+    let b = Bytes.create 16 in
+    for i = 0 to 15 do
+      Bytes.set b i (Char.chr (Random.int 256))
+    done;
+    Base64.encode_exn (Bytes.to_string b)
+
+  let parse_frame_header buf off len =
+    if len < 2 then None
+    else
+      let b0, b1 =
+        (Char.code (Bytes.get buf off), Char.code (Bytes.get buf (off + 1)))
+      in
+      let fin, opcode = (b0 land 0x80 <> 0, b0 land 0x0f) in
+      let masked, plen = (b1 land 0x80 <> 0, b1 land 0x7f) in
+      let hlen, plen =
+        if plen = 126 then
+          if len < 4 then (0, -1)
+          else
+            ( 4,
+              (Char.code (Bytes.get buf (off + 2)) lsl 8)
+              lor Char.code (Bytes.get buf (off + 3)) )
+        else if plen = 127 then
+          if len < 10 then (0, -1)
+          else
+            ( 10,
+              (Char.code (Bytes.get buf (off + 6)) lsl 24)
+              lor (Char.code (Bytes.get buf (off + 7)) lsl 16)
+              lor (Char.code (Bytes.get buf (off + 8)) lsl 8)
+              lor Char.code (Bytes.get buf (off + 9)) )
+        else (2, plen)
+      in
+      if plen < 0 then None
+      else
+        let mlen = if masked then 4 else 0 in
+        if len < hlen + mlen then None
+        else
+          Some
+            ( fin,
+              opcode,
+              plen,
+              (if masked then Some (Bytes.sub buf (off + hlen) 4) else None),
+              hlen + mlen )
 
-  let auth =
-    match Ca_certs_nss.authenticator () with
-    | Ok a -> a
-    | Error (`Msg m) -> failwith ("CA certs: " ^ m)
-  in
-  let tls =
-    match
-      Tls.Config.client ~authenticator:auth ~alpn_protocols:[ "http/1.1" ] ()
-    with
-    | Ok c -> c
-    | Error (`Msg m) -> failwith ("TLS config: " ^ m)
-  in
-  let hostname =
-    match Domain_name.of_string host with
-    | Error _ -> None
-    | Ok dn -> (
-        match Domain_name.host dn with Ok h -> Some h | Error _ -> None)
-  in
+  let make_frame opcode payload =
+    let plen = String.length payload in
+    let mask = Bytes.init 4 (fun _ -> Char.chr (Random.int 256)) in
+    let hdr =
+      if plen < 126 then (
+        let h = Bytes.create 6 in
+        Bytes.set h 0 (Char.chr (0x80 lor opcode));
+        Bytes.set h 1 (Char.chr (0x80 lor plen));
+        Bytes.blit mask 0 h 2 4;
+        Bytes.to_string h)
+      else if plen < 65536 then (
+        let h = Bytes.create 8 in
+        Bytes.set h 0 (Char.chr (0x80 lor opcode));
+        Bytes.set h 1 (Char.chr (0x80 lor 126));
+        Bytes.set h 2 (Char.chr ((plen lsr 8) land 0xff));
+        Bytes.set h 3 (Char.chr (plen land 0xff));
+        Bytes.blit mask 0 h 4 4;
+        Bytes.to_string h)
+      else
+        let h = Bytes.create 14 in
+        Bytes.set h 0 (Char.chr (0x80 lor opcode));
+        Bytes.set h 1 (Char.chr (0x80 lor 127));
+        for i = 2 to 5 do
+          Bytes.set h i '\000'
+        done;
+        Bytes.set h 6 (Char.chr ((plen lsr 24) land 0xff));
+        Bytes.set h 7 (Char.chr ((plen lsr 16) land 0xff));
+        Bytes.set h 8 (Char.chr ((plen lsr 8) land 0xff));
+        Bytes.set h 9 (Char.chr (plen land 0xff));
+        Bytes.blit mask 0 h 10 4;
+        Bytes.to_string h
+    in
+    let masked =
+      Bytes.mapi
+        (fun i c ->
+          Char.chr (Char.code c lxor Char.code (Bytes.get mask (i mod 4))))
+        (Bytes.of_string payload)
+    in
+    hdr ^ Bytes.to_string masked
 
-  let stats, count = (create_stats (), ref 0) in
-  let addr =
-    match Eio.Net.getaddrinfo_stream net host ~service:(string_of_int port) with
-    | [] -> failwith ("DNS failed: " ^ host)
-    | a :: _ -> a
-  in
+  let connect ~net ~sw uri =
+    let host = Uri.host uri |> Option.value ~default:"localhost" in
+    let port = Uri.port uri |> Option.value ~default:443 in
+    let path = Uri.path_and_query uri in
 
-  if not json then Printf.printf "Resolved, connecting...\n%!";
+    (* TLS setup *)
+    let auth =
+      match Ca_certs_nss.authenticator () with
+      | Ok a -> a
+      | Error (`Msg m) -> failwith ("CA certs: " ^ m)
+    in
+    let tls_config =
+      match
+        Tls.Config.client ~authenticator:auth ~alpn_protocols:[ "http/1.1" ] ()
+      with
+      | Ok c -> c
+      | Error (`Msg m) -> failwith ("TLS config: " ^ m)
+    in
+    let hostname =
+      match Domain_name.of_string host with
+      | Error _ -> None
+      | Ok dn -> (
+          match Domain_name.host dn with Ok h -> Some h | Error _ -> None)
+    in
 
-  Eio.Switch.run @@ fun sw ->
-  let sock = Eio.Net.connect ~sw net addr in
-  if not json then Printf.printf "TCP connected, TLS handshake...\n%!";
-  let tls_sock = Tls_eio.client_of_flow tls ?host:hostname sock in
-  if not json then Printf.printf "TLS done, WebSocket upgrade...\n%!";
+    (* Connect *)
+    let addr =
+      match
+        Eio.Net.getaddrinfo_stream net host ~service:(string_of_int port)
+      with
+      | [] -> failwith ("DNS failed: " ^ host)
+      | a :: _ -> a
+    in
+    let socket = Eio.Net.connect ~sw net addr in
+    let tls_socket = Tls_eio.client_of_flow tls_config ?host:hostname socket in
 
-  let nonce = ws_nonce () in
-  Eio.Flow.copy_string
-    (Printf.sprintf
-       "GET %s HTTP/1.1\r\n\
-        Host: %s\r\n\
-        Upgrade: websocket\r\n\
-        Connection: Upgrade\r\n\
-        Sec-WebSocket-Key: %s\r\n\
-        Sec-WebSocket-Version: 13\r\n\
-        \r\n"
-       resource host nonce)
-    tls_sock;
+    (* WebSocket handshake *)
+    let nonce = ws_nonce () in
+    Eio.Flow.copy_string
+      (Printf.sprintf
+         "GET %s HTTP/1.1\r\n\
+          Host: %s\r\n\
+          Upgrade: websocket\r\n\
+          Connection: Upgrade\r\n\
+          Sec-WebSocket-Key: %s\r\n\
+          Sec-WebSocket-Version: 13\r\n\
+          \r\n"
+         path host nonce)
+      tls_socket;
 
-  let resp_buf = Cstruct.create 4096 in
-  let n = Eio.Flow.single_read tls_sock resp_buf in
-  let resp = Cstruct.to_string (Cstruct.sub resp_buf 0 n) in
-  if not (String.length resp >= 12 && String.sub resp 0 12 = "HTTP/1.1 101")
-  then
-    failwith
-      ("WebSocket upgrade failed: "
-      ^ String.sub resp 0 (min 50 (String.length resp)));
+    let resp_buf = Cstruct.create 4096 in
+    let n = Eio.Flow.single_read tls_socket resp_buf in
+    let resp = Cstruct.to_string (Cstruct.sub resp_buf 0 n) in
+    if not (String.length resp >= 12 && String.sub resp 0 12 = "HTTP/1.1 101")
+    then
+      failwith
+        ("WebSocket upgrade failed: "
+        ^ String.sub resp 0 (min 50 (String.length resp)));
 
-  if not json then Printf.printf "Connected! (Ctrl+C to stop)\n\n%!";
+    {
+      socket = tls_socket;
+      buf = Bytes.create 1048576;
+      buf_len = 0;
+      frag_buf = Buffer.create 65536;
+      frag_opcode = 0;
+    }
 
-  let buf, buf_len = (Bytes.create 1048576, ref 0) in
-  let frag_buf, frag_op = (Buffer.create 65536, ref 0) in
-  let read_more () =
+  let read_more conn =
     let cs = Cstruct.create 65536 in
-    let n = Eio.Flow.single_read tls_sock cs in
-    Cstruct.blit_to_bytes cs 0 buf !buf_len n;
-    buf_len := !buf_len + n
-  in
-  let stop = ref false in
+    let n = Eio.Flow.single_read conn.socket cs in
+    Cstruct.blit_to_bytes cs 0 conn.buf conn.buf_len n;
+    conn.buf_len <- conn.buf_len + n
 
-  let fmt = if json then format_event_json ~rich else format_event ~rich in
-  let process evt =
-    if event_matches filters evt then begin
-      print_endline (fmt evt);
-      update_stats stats evt;
-      incr count;
-      match limit with
-      | Some m when !count >= m ->
-          if not json then (
-            Printf.printf "\nLimit reached (%d)\n%!" m;
-            print_stats stats);
-          stop := true
-      | _ -> ()
-    end
-    else update_stats stats evt
-  in
+  let recv conn =
+    (* Read until we have a complete message *)
+    let rec read_frame () =
+      if conn.buf_len < 2 then (
+        read_more conn;
+        read_frame ())
+      else
+        match parse_frame_header conn.buf 0 conn.buf_len with
+        | None ->
+            read_more conn;
+            read_frame ()
+        | Some (fin, opcode, plen, mask, hlen) -> (
+            let total = hlen + plen in
+            while conn.buf_len < total do
+              read_more conn
+            done;
+            let payload = Bytes.sub conn.buf hlen plen in
+            (* Unmask *)
+            (match mask with
+            | Some k ->
+                for i = 0 to plen - 1 do
+                  Bytes.set payload i
+                    (Char.chr
+                       (Char.code (Bytes.get payload i)
+                       lxor Char.code (Bytes.get k (i mod 4))))
+                done
+            | None -> ());
+            (* Shift buffer *)
+            let rem = conn.buf_len - total in
+            if rem > 0 then Bytes.blit conn.buf total conn.buf 0 rem;
+            conn.buf_len <- rem;
+            (* Handle opcode *)
+            match opcode with
+            | 0x0 (* continuation *) ->
+                Buffer.add_bytes conn.frag_buf payload;
+                if fin then (
+                  let data = Buffer.contents conn.frag_buf in
+                  Buffer.clear conn.frag_buf;
+                  if conn.frag_opcode = 0x2 then Ok data else read_frame ())
+                else read_frame ()
+            | 0x1 (* text *) ->
+                if not fin then (
+                  Buffer.clear conn.frag_buf;
+                  Buffer.add_bytes conn.frag_buf payload;
+                  conn.frag_opcode <- 0x1);
+                read_frame ()
+            | 0x2 (* binary *) ->
+                if fin then Ok (Bytes.to_string payload)
+                else (
+                  Buffer.clear conn.frag_buf;
+                  Buffer.add_bytes conn.frag_buf payload;
+                  conn.frag_opcode <- 0x2;
+                  read_frame ())
+            | 0x8 (* close *) -> Error "Connection closed by server"
+            | 0x9 (* ping *) ->
+                Eio.Flow.copy_string
+                  (make_frame 0xA (Bytes.to_string payload))
+                  conn.socket;
+                read_frame ()
+            | 0xA (* pong *) -> read_frame ()
+            | _ -> read_frame ())
+    in
+    try read_frame () with exn -> Error (Printexc.to_string exn)
 
-  while not !stop do
-    if !buf_len < 2 then read_more ();
-    match parse_ws_header buf 0 !buf_len with
-    | None -> read_more ()
-    | Some (fin, opcode, plen, mask, hlen) -> (
-        let total = hlen + plen in
-        while !buf_len < total do
-          read_more ()
-        done;
-        let payload = Bytes.sub buf hlen plen in
-        (match mask with
-        | Some k ->
-            for i = 0 to plen - 1 do
-              Bytes.set payload i
-                (Char.chr
-                   (Char.code (Bytes.get payload i)
-                   lxor Char.code (Bytes.get k (i mod 4))))
-            done
-        | None -> ());
-        let rem = !buf_len - total in
-        if rem > 0 then Bytes.blit buf total buf 0 rem;
-        buf_len := rem;
-        match opcode with
-        | 0x0 ->
-            Buffer.add_bytes frag_buf payload;
-            if fin then (
-              let d = Buffer.contents frag_buf in
-              Buffer.clear frag_buf;
-              if !frag_op = 0x2 then
-                match Firehose.decode_frame d with
-                | Ok e -> process e
-                | Error e ->
-                    Printf.eprintf "\027[31mDecode: %s\027[0m\n%!"
-                      (Firehose.error_to_string e);
-                    stats.errors <- stats.errors + 1)
-        | 0x1 ->
-            if not fin then (
-              Buffer.clear frag_buf;
-              Buffer.add_bytes frag_buf payload;
-              frag_op := 0x1)
-        | 0x2 ->
-            if fin then (
-              match Firehose.decode_frame (Bytes.to_string payload) with
-              | Ok e -> process e
-              | Error e ->
-                  Printf.eprintf "\027[31mDecode: %s\027[0m\n%!"
-                    (Firehose.error_to_string e);
-                  stats.errors <- stats.errors + 1)
-            else (
-              Buffer.clear frag_buf;
-              Buffer.add_bytes frag_buf payload;
-              frag_op := 0x2)
-        | 0x8 ->
-            if not json then (
-              Printf.printf "\nServer closed\n%!";
-              print_stats stats);
-            stop := true
-        | 0x9 ->
-            Eio.Flow.copy_string
-              (ws_frame 0xA (Bytes.to_string payload))
-              tls_sock
-        | 0xA -> ()
-        | _ -> if not json then Printf.printf "Unknown opcode: 0x%x\n%!" opcode)
-  done
+  let close _conn = () (* TLS socket closed by switch *)
+end
+
+(** Run a computation with WebSocket effect handlers *)
+let with_websocket ~net ~sw f =
+  let open Effect.Deep in
+  try_with f ()
+    {
+      effc =
+        (fun (type a) (eff : a Effect.t) ->
+          match eff with
+          | Firehose.Ws_connect uri ->
+              Some
+                (fun (k : (a, _) continuation) ->
+                  try
+                    let c = Ws_handler.connect ~net ~sw uri in
+                    continue k (Ok (Obj.magic c : Firehose.websocket))
+                  with exn -> continue k (Error (Printexc.to_string exn)))
+          | Firehose.Ws_recv ws ->
+              Some
+                (fun k ->
+                  let c : Ws_handler.conn = Obj.magic ws in
+                  continue k (Ws_handler.recv c))
+          | Firehose.Ws_close ws ->
+              Some
+                (fun k ->
+                  let c : Ws_handler.conn = Obj.magic ws in
+                  Ws_handler.close c;
+                  continue k ())
+          | _ -> None);
+    }
+
+(* ============================================================================
+   Main
+   ============================================================================ *)
 
-(* CLI *)
 type config = {
   cursor : int64 option;
   limit : int option;
···
     rich = !rich;
   }
 
+let run ~net ~sw config =
+  let uri =
+    Uri.of_string "wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos"
+  in
+  let firehose_config = Firehose.config ~uri ?cursor:config.cursor () in
+
+  if not config.json then
+    Printf.printf "Connecting to %s...\n%!"
+      (Uri.to_string (Firehose.build_uri firehose_config));
+
+  let stats = create_stats () in
+  let count = ref 0 in
+  let stats_printed = ref false in
+  let fmt =
+    if config.json then format_event_json ~rich:config.rich
+    else format_event ~rich:config.rich
+  in
+
+  let handler event =
+    update_stats stats event;
+    if event_matches config.filters event then begin
+      print_endline (fmt event);
+      incr count;
+      match config.limit with
+      | Some max when !count >= max ->
+          if not config.json then (
+            Printf.printf "\nLimit reached (%d)\n%!" max;
+            print_stats stats;
+            stats_printed := true);
+          false
+      | _ -> true
+    end
+    else true
+  in
+
+  with_websocket ~net ~sw (fun () ->
+      if not config.json then Printf.printf "Connected! (Ctrl+C to stop)\n\n%!";
+      match Firehose.subscribe firehose_config ~handler with
+      | Ok () ->
+          if (not config.json) && not !stats_printed then print_stats stats
+      | Error e ->
+          Printf.eprintf "\027[31mError: %s\027[0m\n%!"
+            (Firehose.error_to_string e))
+
 let () =
-  let c = parse_args () in
-  if not c.json then begin
+  let config = parse_args () in
+  if not config.json then begin
     Printf.printf
       "\027[1mAT Protocol Firehose Demo\027[0m\n=========================\n\n%!";
-    if c.filters <> [] then
+    if config.filters <> [] then
       Printf.printf "Filtering: %s\n\n%!"
-        (String.concat ", " (List.map filter_to_string c.filters))
+        (String.concat ", " (List.map filter_to_string config.filters))
   end;
   Mirage_crypto_rng_unix.use_default ();
   Random.self_init ();
   Eio_main.run @@ fun env ->
-  try
-    run_firehose ~net:(Eio.Stdenv.net env) ~filters:c.filters ~json:c.json
-      ~rich:c.rich ?cursor:c.cursor ?limit:c.limit ()
-  with
+  Eio.Switch.run @@ fun sw ->
+  try run ~net:(Eio.Stdenv.net env) ~sw config with
   | Failure m -> Printf.eprintf "\027[31mError: %s\027[0m\n%!" m
   | e -> Printf.eprintf "\027[31mError: %s\027[0m\n%!" (Printexc.to_string e)