Add Lwt-based firehose demo using websocket-lwt-unix

- Uses websocket-lwt-unix library for WebSocket handling
- Implements Firehose effects with Lwt_main.run in handlers
- 380 lines vs 528 for Eio version (28% reduction)
- Same CLI options: --filter, --limit, --json, --content, --cursor

examples/firehose_demo/dune

···
   mirage-crypto-rng.unix
   base64
   cstruct))
+
+(executable
+ (name firehose_demo_lwt)
+ (public_name firehose_demo_lwt)
+ (package atproto)
+ (libraries
+  atproto-sync
+  atproto-ipld
+  climate
+  uri
+  lwt
+  lwt.unix
+  websocket-lwt-unix
+  conduit-lwt-unix
+  tls-lwt
+  mirage-crypto-rng.unix))

examples/firehose_demo/firehose_demo_lwt.ml

···
+(** Firehose Demo (Lwt version) - AT Protocol real-time events. Uses
+    websocket-lwt-unix for WebSocket handling. Use --help for options. *)
+
+open Lwt.Infix
+module Firehose = Atproto_sync.Firehose
+module Car = Atproto_ipld.Car
+module Dag_cbor = Atproto_ipld.Dag_cbor
+module Dag_json = Atproto_ipld.Dag_json
+module Cid = Atproto_ipld.Cid
+
+let truncate n s =
+  if String.length s <= n then s else String.sub s 0 (n - 3) ^ "..."
+
+let single_line s =
+  String.map (fun c -> if c = '\n' || c = '\r' then ' ' else c) s
+
+let opt_or d = Option.value ~default:d
+
+let extract_blocks blocks_data =
+  if String.length blocks_data = 0 then []
+  else
+    match Car.read blocks_data with
+    | Error _ -> []
+    | Ok (_, blocks) ->
+        List.filter_map
+          (fun (b : Car.block) ->
+            match Dag_cbor.decode b.data with
+            | Ok cbor -> Some (Cid.to_string b.cid, cbor)
+            | _ -> None)
+          blocks
+
+let find_block blocks (op : Firehose.operation) =
+  match op.cid with
+  | Some cid -> List.assoc_opt (Cid.to_string cid) blocks
+  | None -> None
+
+let get_text cbor =
+  match cbor with
+  | Dag_cbor.Map pairs -> Firehose.get_string "text" pairs
+  | _ -> None
+
+let format_op blocks (op : Firehose.operation) =
+  let action =
+    match op.action with `Create -> "+" | `Update -> "~" | `Delete -> "-"
+  in
+  let content =
+    match find_block blocks op with
+    | Some cbor -> (
+        match get_text cbor with
+        | Some t -> " \"" ^ truncate 70 (single_line t) ^ "\""
+        | None -> "")
+    | None -> ""
+  in
+  Printf.sprintf "%s%s%s" action op.path content
+
+let format_event ?(rich = false) = function
+  | Firehose.Commit evt ->
+      let blocks = if rich then extract_blocks evt.blocks else [] in
+      let ops =
+        String.concat
+          (if rich then "\n        " else " ")
+          (List.map (format_op blocks) evt.ops)
+      in
+      Printf.sprintf "#commit seq=%Ld repo=%s%s%s" evt.seq evt.repo
+        (if rich then "\n        " else " ")
+        ops
+  | Firehose.Identity e ->
+      Printf.sprintf "#identity seq=%Ld did=%s handle=%s" e.seq e.did
+        (opt_or "-" e.handle)
+  | Firehose.Account e ->
+      Printf.sprintf "#account seq=%Ld did=%s active=%b" e.seq e.did e.active
+  | Firehose.Handle e ->
+      Printf.sprintf "#handle seq=%Ld did=%s handle=%s" e.seq e.did e.handle
+  | Firehose.Tombstone e ->
+      Printf.sprintf "#tombstone seq=%Ld did=%s" e.seq e.did
+  | Firehose.Info m -> Printf.sprintf "#info name=%s" m.name
+  | Firehose.StreamError err -> Printf.sprintf "#error %s" err
+
+let json_of_op_with_record blocks (op : Firehose.operation) =
+  let base_json = Firehose.operation_to_json op in
+  match find_block blocks op with
+  | Some cbor ->
+      let record_json = Yojson.Safe.from_string (Dag_json.encode_string cbor) in
+      Firehose.add_record_to_op_json base_json record_json
+  | None -> base_json
+
+let json_of_event ?(rich = false) event =
+  let json =
+    match event with
+    | Firehose.Commit evt when rich -> (
+        let blocks = extract_blocks evt.blocks in
+        let ops = `List (List.map (json_of_op_with_record blocks) evt.ops) in
+        let base = Firehose.commit_event_to_json evt in
+        match base with
+        | `Assoc fields ->
+            `Assoc
+              (List.map
+                 (fun (k, v) -> if k = "ops" then (k, ops) else (k, v))
+                 fields)
+        | _ -> base)
+    | _ -> Firehose.event_to_json event
+  in
+  Yojson.Safe.to_string json
+
+type filter =
+  | Posts
+  | Likes
+  | Follows
+  | Reposts
+  | Blocks
+  | Lists
+  | Profiles
+  | Feeds
+  | Commits
+  | Identities
+  | Accounts
+  | Handles
+  | Tombstones
+
+let filters =
+  [
+    ("posts", Posts);
+    ("likes", Likes);
+    ("follows", Follows);
+    ("reposts", Reposts);
+    ("blocks", Blocks);
+    ("lists", Lists);
+    ("profiles", Profiles);
+    ("feeds", Feeds);
+    ("commits", Commits);
+    ("identities", Identities);
+    ("accounts", Accounts);
+    ("handles", Handles);
+    ("tombstones", Tombstones);
+  ]
+
+let filter_of_string s = List.assoc_opt s filters
+let filter_to_string f = fst (List.find (fun (_, v) -> v = f) filters)
+
+let prefix_match prefix path =
+  String.length path >= String.length prefix
+  && String.sub path 0 (String.length prefix) = prefix
+
+let op_matches filter (op : Firehose.operation) =
+  match filter with
+  | Posts -> prefix_match "app.bsky.feed.post" op.path
+  | Likes -> prefix_match "app.bsky.feed.like" op.path
+  | Follows -> prefix_match "app.bsky.graph.follow" op.path
+  | Reposts -> prefix_match "app.bsky.feed.repost" op.path
+  | Blocks -> prefix_match "app.bsky.graph.block" op.path
+  | Lists -> prefix_match "app.bsky.graph.list" op.path
+  | Profiles -> prefix_match "app.bsky.actor.profile" op.path
+  | Feeds -> prefix_match "app.bsky.feed.generator" op.path
+  | Commits -> true
+  | _ -> false
+
+let event_matches filters event =
+  match filters with
+  | [] -> true
+  | _ -> (
+      match event with
+      | Firehose.Commit evt ->
+          List.exists
+            (fun f -> f = Commits || List.exists (op_matches f) evt.ops)
+            filters
+      | Firehose.Identity _ -> List.mem Identities filters
+      | Firehose.Account _ -> List.mem Accounts filters
+      | Firehose.Handle _ -> List.mem Handles filters
+      | Firehose.Tombstone _ -> List.mem Tombstones filters
+      | Firehose.Info _ | Firehose.StreamError _ -> true)
+
+(** WebSocket client using websocket-lwt-unix *)
+module Ws = struct
+  type conn = Websocket_lwt_unix.conn
+
+  let connect uri : (conn, string) result Lwt.t =
+    let host = Uri.host uri |> Option.value ~default:"localhost" in
+    let scheme = Uri.scheme uri |> Option.value ~default:"wss" in
+    let port =
+      Uri.port uri |> Option.value ~default:(if scheme = "wss" then 443 else 80)
+    in
+    let resource = Uri.path_and_query uri in
+    (* Resolve hostname to IP *)
+    Lwt.catch
+      (fun () ->
+        Lwt_unix.getaddrinfo host (string_of_int port)
+          [ Unix.AI_SOCKTYPE Unix.SOCK_STREAM ]
+        >>= function
+        | [] -> Lwt.return (Error ("DNS failed: " ^ host))
+        | ai :: _ ->
+            let ip =
+              match ai.Unix.ai_addr with
+              | Unix.ADDR_INET (addr, _) -> Ipaddr_unix.of_inet_addr addr
+              | _ -> failwith "Unexpected address type"
+            in
+            let endp = `TLS_native (`Hostname host, `IP ip, `Port port) in
+            Websocket_lwt_unix.connect endp (Uri.with_path uri resource)
+            >|= fun conn -> Ok conn)
+      (fun exn -> Lwt.return (Error (Printexc.to_string exn)))
+
+  let recv conn : (string, string) result Lwt.t =
+    let open Websocket in
+    let rec read_binary () =
+      Lwt.catch
+        (fun () ->
+          Websocket_lwt_unix.read conn >>= fun frame ->
+          match frame.Frame.opcode with
+          | Frame.Opcode.Binary -> Lwt.return (Ok frame.Frame.content)
+          | Frame.Opcode.Text -> read_binary () (* Skip text frames *)
+          | Frame.Opcode.Close -> Lwt.return (Error "Connection closed")
+          | Frame.Opcode.Ping ->
+              (* Respond to ping with pong *)
+              let pong =
+                Frame.create ~opcode:Frame.Opcode.Pong
+                  ~content:frame.Frame.content ()
+              in
+              Websocket_lwt_unix.write conn pong >>= fun () -> read_binary ()
+          | Frame.Opcode.Pong -> read_binary ()
+          | _ -> read_binary ())
+        (fun exn -> Lwt.return (Error (Printexc.to_string exn)))
+    in
+    read_binary ()
+
+  let close conn =
+    Lwt.catch
+      (fun () ->
+        let close_frame = Websocket.Frame.close 1000 in
+        Websocket_lwt_unix.write conn close_frame >>= fun () ->
+        Websocket_lwt_unix.close_transport conn)
+      (fun _ -> Lwt.return_unit)
+end
+
+(** Effect handler that uses Lwt for WebSocket operations *)
+let with_websocket_lwt f =
+  let open Effect.Deep in
+  (* We need to track Lwt state across effect invocations *)
+  let ws_conn : Ws.conn option ref = ref None in
+  try_with f ()
+    {
+      effc =
+        (fun (type a) (eff : a Effect.t) ->
+          match eff with
+          | Firehose.Ws_connect uri ->
+              Some
+                (fun (k : (a, _) continuation) ->
+                  let result = Lwt_main.run (Ws.connect uri) in
+                  (match result with
+                  | Ok conn -> ws_conn := Some conn
+                  | Error _ -> ());
+                  continue k (Result.map Obj.magic result))
+          | Firehose.Ws_recv _ws ->
+              Some
+                (fun k ->
+                  match !ws_conn with
+                  | None -> continue k (Error "Not connected")
+                  | Some conn ->
+                      let result = Lwt_main.run (Ws.recv conn) in
+                      continue k result)
+          | Firehose.Ws_close _ws ->
+              Some
+                (fun k ->
+                  (match !ws_conn with
+                  | None -> ()
+                  | Some conn ->
+                      Lwt_main.run (Ws.close conn);
+                      ws_conn := None);
+                  continue k ())
+          | _ -> None);
+    }
+
+type config = {
+  cursor : int64 option;
+  limit : int option;
+  filters : filter list;
+  json : bool;
+  rich : bool;
+}
+
+let parse_filters s =
+  String.split_on_char ',' s
+  |> List.filter_map (fun p -> filter_of_string (String.trim p))
+
+let int64_conv =
+  let open Climate.Arg_parser in
+  make_conv
+    ~parse:(fun s ->
+      match Int64.of_string_opt s with
+      | Some n -> Ok n
+      | None -> Error (`Msg "invalid integer"))
+    ~print:(fun fmt n -> Format.fprintf fmt "%Ld" n)
+    ()
+
+let config_parser =
+  let open Climate.Arg_parser in
+  let+ cursor = named_opt [ "cursor" ] int64_conv ~doc:"Resume from sequence"
+  and+ limit = named_opt [ "limit" ] int ~doc:"Stop after N events"
+  and+ filter_str =
+    named_opt [ "filter" ] string
+      ~doc:"Filter (posts,likes,follows,reposts,etc.)"
+  and+ json = flag [ "json" ] ~doc:"JSON output"
+  and+ rich = flag [ "content" ] ~doc:"Show record content" in
+  {
+    cursor;
+    limit;
+    filters = Option.fold ~none:[] ~some:parse_filters filter_str;
+    json;
+    rich;
+  }
+
+let cli =
+  Climate.Command.singleton ~doc:"AT Protocol firehose demo (Lwt)" config_parser
+
+type stats = {
+  mutable total : int;
+  mutable matched : int;
+  mutable last_seq : int64;
+  mutable start : float;
+}
+
+let print_stats s =
+  let elapsed = Unix.gettimeofday () -. s.start in
+  let rate = if elapsed > 0. then float_of_int s.total /. elapsed else 0. in
+  Printf.printf
+    "\n--- Stats: %d events (%d matched) | seq=%Ld | %.1fs | %.1f evt/s ---\n%!"
+    s.total s.matched s.last_seq elapsed rate
+
+let stats = { total = 0; matched = 0; last_seq = 0L; start = 0. }
+let interrupted = ref false
+
+let run config =
+  let uri =
+    Uri.of_string "wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos"
+  in
+  let cfg = Firehose.config ~uri ?cursor:config.cursor () in
+  if not config.json then
+    Printf.printf "Connecting to %s...\n%!"
+      (Uri.to_string (Firehose.build_uri cfg));
+  let fmt =
+    if config.json then json_of_event ~rich:config.rich
+    else format_event ~rich:config.rich
+  in
+  let handler event =
+    stats.total <- stats.total + 1;
+    (match Firehose.event_seq event with
+    | Some s -> stats.last_seq <- s
+    | None -> ());
+    if event_matches config.filters event then begin
+      print_endline (fmt event);
+      stats.matched <- stats.matched + 1;
+      match config.limit with
+      | Some max when stats.matched >= max -> false
+      | _ -> not !interrupted
+    end
+    else not !interrupted
+  in
+  with_websocket_lwt (fun () ->
+      if not config.json then Printf.printf "Connected!\n\n%!";
+      match Firehose.subscribe cfg ~handler with
+      | Ok () -> ()
+      | Error e -> Printf.eprintf "Error: %s\n%!" (Firehose.error_to_string e))
+
+let () =
+  let config =
+    Climate.Command.run
+      ~program_name:(Climate.Program_name.Literal "firehose_demo_lwt") cli
+  in
+  if not config.json then begin
+    Printf.printf
+      "AT Protocol Firehose Demo (Lwt)\n===============================\n\n%!";
+    if config.filters <> [] then
+      Printf.printf "Filtering: %s\n\n%!"
+        (String.concat ", " (List.map filter_to_string config.filters))
+  end;
+  Sys.set_signal Sys.sigint (Sys.Signal_handle (fun _ -> interrupted := true));
+  stats.start <- Unix.gettimeofday ();
+  Mirage_crypto_rng_unix.use_default ();
+  (try run config with
+  | Failure m -> Printf.eprintf "Error: %s\n%!" m
+  | e -> Printf.eprintf "Error: %s\n%!" (Printexc.to_string e));
+  if not config.json then print_stats stats