···
+
last_river_post_date: float option; (** Timestamp of last River post for this user *)
(** Parse a user registration from JSON-like string format *)
let user_registration_of_string s : user_registration option =
+
(* Format: "email|zulip_id|full_name|timestamp|is_admin|last_river_post_date" *)
match String.split_on_char '|' s with
+
| [email; zulip_id_str; full_name; timestamp_str; is_admin_str; last_river_str] ->
+
let last_river_post_date =
+
if last_river_str = "" || last_river_str = "none" then None
+
else Some (float_of_string last_river_str)
+
zulip_id = int_of_string zulip_id_str;
+
registered_at = float_of_string timestamp_str;
+
is_admin = bool_of_string is_admin_str;
| [email; zulip_id_str; full_name; timestamp_str; is_admin_str] ->
+
(* Backward compatibility - old format without last_river_post_date *)
zulip_id = int_of_string zulip_id_str;
registered_at = float_of_string timestamp_str;
is_admin = bool_of_string is_admin_str;
+
last_river_post_date = None;
| [email; zulip_id_str; full_name; timestamp_str] ->
+
(* Backward compatibility - old format without is_admin and last_river_post_date *)
zulip_id = int_of_string zulip_id_str;
registered_at = float_of_string timestamp_str;
+
last_river_post_date = None;
(** Convert a user registration to string format *)
let user_registration_to_string (reg : user_registration) : string =
+
let last_river_str = match reg.last_river_post_date with
+
| Some t -> string_of_float t
+
Printf.sprintf "%s|%d|%s|%f|%b|%s"
(** Storage key for a user registration by Zulip ID - this is the only storage key we use *)
let storage_key_for_id zulip_id = Printf.sprintf "user:id:%d" zulip_id
···
| None -> is_admin || (zulip_id = default_admin_id)
+
(* Preserve last_river_post_date if user already exists *)
+
let last_river_post_date = match existing_by_id with
+
| Some existing -> existing.last_river_post_date
registered_at = Unix.gettimeofday ();
is_admin = final_is_admin;
let reg_str = user_registration_to_string reg in
···
String.contains domain '.'
+
(** {1 River Integration Helper Functions} *)
+
(** Configuration storage keys for River *)
+
let river_feeds_key = "river:feeds:list"
+
let river_channel_key = "river:channel"
+
let river_polling_enabled_key = "river:polling:enabled"
+
let river_last_sync_key = "river:last_sync"
+
let river_default_channel = "Sandbox-test"
+
(** Feed source codec *)
+
let feed_source_jsont =
+
let make name url = { River.name; url } in
+
Jsont.Object.map ~kind:"FeedSource" make
+
|> Jsont.Object.mem "name" Jsont.string ~enc:(fun s -> s.River.name)
+
|> Jsont.Object.mem "url" Jsont.string ~enc:(fun s -> s.River.url)
+
(** Load feed sources from bot storage *)
+
let load_feed_sources storage =
+
match Bot_storage.get storage ~key:river_feeds_key with
+
| Some json_str when json_str <> "" ->
+
(match Jsont_bytesrw.decode_string' (Jsont.list feed_source_jsont) json_str with
+
Log.debug (fun m -> m "Loaded %d feed sources" (List.length feeds));
+
Log.err (fun m -> m "Failed to parse feed sources: %s" (Jsont.Error.to_string err));
+
Log.debug (fun m -> m "No feed sources configured");
+
(** Save feed sources to bot storage *)
+
let save_feed_sources storage feeds =
+
match Jsont_bytesrw.encode_string' ~format:Jsont.Indent (Jsont.list feed_source_jsont) feeds with
+
Bot_storage.put storage ~key:river_feeds_key ~value:json_str
+
let msg = Printf.sprintf "Failed to encode feed sources: %s" (Jsont.Error.to_string err) in
+
Error (Zulip.create_error ~code:(Other "encoding_error") ~msg ())
+
(** Add a feed source *)
+
let add_feed storage ~name ~url =
+
let feeds = load_feed_sources storage in
+
if List.exists (fun f -> f.River.url = url) feeds then
+
Error (Zulip.create_error ~code:(Other "already_exists")
+
~msg:(Printf.sprintf "Feed with URL %s already exists" url) ())
+
let new_feed = { River.name; url } in
+
save_feed_sources storage (new_feed :: feeds)
+
(** Remove a feed source *)
+
let remove_feed storage ~name =
+
let feeds = load_feed_sources storage in
+
let updated_feeds = List.filter (fun f -> f.River.name <> name) feeds in
+
if List.length updated_feeds = List.length feeds then
+
Error (Zulip.create_error ~code:(Other "not_found")
+
~msg:(Printf.sprintf "Feed '%s' not found" name) ())
+
save_feed_sources storage updated_feeds
+
(** Get target Zulip channel *)
+
let get_river_channel storage =
+
match Bot_storage.get storage ~key:river_channel_key with
+
| Some ch when ch <> "" -> ch
+
| _ -> river_default_channel
+
(** Set target Zulip channel *)
+
let set_river_channel storage channel =
+
Bot_storage.put storage ~key:river_channel_key ~value:channel
+
(** Check if polling is enabled *)
+
let is_river_polling_enabled storage =
+
match Bot_storage.get storage ~key:river_polling_enabled_key with
+
let enable_river_polling storage =
+
Bot_storage.put storage ~key:river_polling_enabled_key ~value:"true"
+
let disable_river_polling storage =
+
Bot_storage.put storage ~key:river_polling_enabled_key ~value:"false"
+
(** Get last sync timestamp *)
+
let get_river_last_sync storage =
+
match Bot_storage.get storage ~key:river_last_sync_key with
+
| Some ts_str when ts_str <> "" ->
+
(try Some (float_of_string ts_str) with _ -> None)
+
(** Update last sync timestamp *)
+
let update_river_last_sync storage timestamp =
+
Bot_storage.put storage ~key:river_last_sync_key ~value:(string_of_float timestamp)
+
(** {1 Command Handlers} *)
(** Handle the 'register' command *)
let handle_register storage sender_email sender_id sender_name custom_email_opt =
(* First, try to fetch the user's profile from the Zulip API to get delivery_email and email *)
···
(String.concat "\n" user_lines)
+
(** Handle River 'feeds' command *)
+
let handle_river_feeds storage =
+
let feeds = load_feed_sources storage in
+
"📡 No River feeds configured yet.\n\nUse `river add-feed <name> <url>` to add a feed."
+
let feed_lines = List.mapi (fun i feed ->
+
Printf.sprintf "%d. **%s**\n URL: `%s`" (i + 1) feed.River.name feed.River.url
+
Printf.sprintf "📡 Configured River feeds (%d):\n\n%s\n\nChannel: #%s"
+
(String.concat "\n\n" feed_lines)
+
(get_river_channel storage)
+
(** Handle River 'add-feed' command *)
+
let handle_river_add_feed storage args =
+
match String.split_on_char ' ' args |> List.filter (fun s -> s <> "") with
+
let url = String.concat " " url_parts in
+
(match add_feed storage ~name ~url with
+
Printf.sprintf "✅ Added feed **%s**\n URL: `%s`\n\nUse `river sync` to fetch posts." name url
+
Printf.sprintf "❌ Failed to add feed: %s" (Zulip.error_message e))
+
"Usage: `river add-feed <name> <url>`\n\nExample: `river add-feed \"OCaml Blog\" https://ocaml.org/blog/feed.xml`"
+
(** Handle River 'remove-feed' command *)
+
let handle_river_remove_feed storage args =
+
let name = String.trim args in
+
"Usage: `river remove-feed <name>`\n\nExample: `river remove-feed \"OCaml Blog\"`"
+
match remove_feed storage ~name with
+
Printf.sprintf "✅ Removed feed: **%s**" name
+
Printf.sprintf "❌ Failed to remove feed: %s" (Zulip.error_message e)
+
(** Handle River 'set-channel' command *)
+
let handle_river_set_channel storage args =
+
let channel = String.trim args in
+
Printf.sprintf "Current channel: #%s\n\nUsage: `river set-channel <channel-name>`\n\nExample: `river set-channel general`"
+
(get_river_channel storage)
+
match set_river_channel storage channel with
+
Printf.sprintf "✅ River posts will now go to #%s" channel
+
Printf.sprintf "❌ Failed to set channel: %s" (Zulip.error_message e)
+
(** Handle River 'start' command *)
+
let handle_river_start storage =
+
match enable_river_polling storage with
+
| Ok () -> "✅ River polling enabled. Feeds will be checked every 5 minutes."
+
| Error e -> Printf.sprintf "❌ Failed to enable polling: %s" (Zulip.error_message e)
+
(** Handle River 'stop' command *)
+
let handle_river_stop storage =
+
match disable_river_polling storage with
+
| Ok () -> "⏸️ River polling disabled. Use `river start` to resume."
+
| Error e -> Printf.sprintf "❌ Failed to disable polling: %s" (Zulip.error_message e)
+
(** Handle River 'status' command *)
+
let handle_river_status storage =
+
let feeds = load_feed_sources storage in
+
let polling_status = if is_river_polling_enabled storage then "✅ Enabled" else "⏸️ Disabled" in
+
let last_sync = match get_river_last_sync storage with
+
| Some ts -> format_timestamp ts
+
Printf.sprintf "📊 River Feed Integration Status:\n\
+
• Target channel: #%s\n\
+
• Feeds configured: %d\n\
+
(get_river_channel storage)
(** Handle the 'help' command *)
let handle_help sender_name sender_email =
+
Printf.sprintf "👋 Hi %s! I'm **Vicuna**, your user registration and feed aggregation assistant.\n\n\
+
**User Registration Commands:**\n\
• `register` - Auto-detect your real email or use Zulip email\n\
• `register <your-email@example.com>` - Register with a specific email\n\
• `whoami` - Show your registration status\n\
• `whois <email|id>` - Look up a registered user\n\
+
• `list` - List all registered users\n\n\
+
**River Feed Commands:**\n\
+
• `river feeds` - List all configured feeds\n\
+
• `river add-feed <name> <url>` - Add a new feed\n\
+
• `river remove-feed <name>` - Remove a feed\n\
+
• `river sync` - Force immediate feed sync\n\
+
• `river status` - Show River integration status\n\
+
• `river set-channel <name>` - Set target Zulip channel\n\
+
• `river start` - Enable automatic polling\n\
+
• `river stop` - Disable automatic polling\n\
• `help` - Show this help message\n\n\
• `register` - Auto-detect your email (your Zulip email is `%s`)\n\
+
• `river add-feed \"OCaml Weekly\" https://ocaml.org/feed.xml`\n\
+
• `river set-channel sandbox-test`\n\n\
Send me a direct message to get started!"
···
handle_whois storage args
+
(* Parse river subcommand *)
+
let (subcmd, subargs) = parse_command args in
+
let subcmd_lower = String.lowercase_ascii subcmd in
+
(match subcmd_lower with
+
| "" | "feeds" | "list" ->
+
handle_river_feeds storage
+
| "add-feed" | "add" ->
+
handle_river_add_feed storage subargs
+
| "remove-feed" | "remove" | "rm" ->
+
handle_river_remove_feed storage subargs
+
| "set-channel" | "channel" ->
+
handle_river_set_channel storage subargs
+
| "start" | "enable" ->
+
handle_river_start storage
+
| "stop" | "disable" ->
+
handle_river_stop storage
+
handle_river_status storage
+
"⏳ Syncing River feeds... (Note: sync requires environment access, use CLI for now)"
+
Printf.sprintf "Unknown river command: `%s`\n\nAvailable: feeds, add-feed, remove-feed, set-channel, start, stop, status, sync" subcmd)
Printf.sprintf "Unknown command: `%s`. Use `help` to see available commands." command
···
| Ok () -> Bot_storage.remove storage ~key
+
(** Normalize a name for fuzzy matching *)
+
let normalize_name name =
+
|> String.lowercase_ascii
+
|> Str.global_replace (Str.regexp "[ \t\n\r]+") " "
+
(** Match user by exact name *)
+
let lookup_user_by_name_exact storage name =
+
let all_ids = get_all_user_ids storage in
+
List.find_map (fun id ->
+
match lookup_user_by_id storage id with
+
| Some user when user.full_name = name -> Some user
+
(** Match user by fuzzy name *)
+
let lookup_user_by_name_fuzzy storage name =
+
let normalized_query = normalize_name name in
+
let all_ids = get_all_user_ids storage in
+
List.find_map (fun id ->
+
match lookup_user_by_id storage id with
+
| Some user when normalize_name user.full_name = normalized_query -> Some user
+
(** Smart user matching for a River post *)
+
let match_user_for_post storage (post : River.post) =
+
let author_email = River.email post in
+
let author_name = River.author post in
+
Log.debug (fun m -> m "Matching user for post by %s (%s)" author_name author_email);
+
(* Try email → name exact → name fuzzy *)
+
match lookup_user_by_email storage author_email with
+
Log.debug (fun m -> m "Matched by email: %s" user.email);
+
(match lookup_user_by_name_exact storage author_name with
+
Log.debug (fun m -> m "Matched by exact name: %s" user.full_name);
+
match lookup_user_by_name_fuzzy storage author_name with
+
Log.debug (fun m -> m "Matched by fuzzy name: %s" user.full_name);
+
Log.debug (fun m -> m "No user match found");
+
(** Convert HTML content to markdown summary *)
+
let content_to_summary content_html ~max_length =
+
let markdown = Markdown_converter.to_markdown content_html in
+
if String.length markdown <= max_length then markdown
+
else String.sub markdown 0 (max_length - 3) ^ "..."
+
(** Format a River post for Zulip *)
+
let format_river_post ~user_match (post : River.post) =
+
match River.summary post with
+
| None -> content_to_summary (River.content post) ~max_length:200
+
| Some user -> Printf.sprintf "By @**%s**" user.full_name
+
| None -> Printf.sprintf "By %s" (River.author post)
+
match River.link post with
+
| Some uri -> Printf.sprintf "\n\n[Read more](%s)" (Uri.to_string uri)
+
Printf.sprintf "%s\n\n%s%s" author_line summary link_line
+
(** Update user's last_river_post_date *)
+
let update_user_river_date storage user new_date =
+
let updated = { user with last_river_post_date = Some new_date } in
+
let reg_str = user_registration_to_string updated in
+
Bot_storage.put storage ~key:(storage_key_for_id user.zulip_id) ~value:reg_str
+
(** Get latest post date from a list of posts *)
+
let get_latest_post_date posts =
+
List.fold_left (fun acc post ->
+
match River.date post with
+
let timestamp = Ptime.to_float_s ptime in
+
| None -> Some timestamp
+
| Some existing -> Some (max existing timestamp))
+
(** Filter posts newer than a timestamp *)
+
let filter_posts_since posts since_opt =
+
List.filter (fun post ->
+
match River.date post with
+
Ptime.to_float_s ptime > since
+
(** Post to Zulip channel *)
+
let post_to_zulip client ~channel ~topic ~content =
+
let stream_message = Zulip.Message.create ~type_:`Channel ~to_:[channel] ~topic ~content () in
+
Zulip.Messages.send client stream_message
+
(** Sync feeds and post new items *)
+
let sync_river_and_post ~env ~storage ~client () =
+
Log.info (fun m -> m "Starting River feed sync");
+
let feeds = load_feed_sources storage in
+
Log.info (fun m -> m "No feeds configured, skipping sync");
+
River.with_session env (fun session ->
+
Log.debug (fun m -> m "Fetching %d feeds" (List.length feeds));
+
let fetched_feeds = List.map (fun source ->
+
Log.debug (fun m -> m "Fetching: %s" source.River.name);
+
River.fetch session source
+
let all_posts = River.posts fetched_feeds in
+
Log.info (fun m -> m "Fetched %d total posts" (List.length all_posts));
+
let users = List.filter_map (lookup_user_by_id storage) (get_all_user_ids storage) in
+
let posted_count = ref 0 in
+
let channel = get_river_channel storage in
+
let new_posts = filter_posts_since all_posts user.last_river_post_date in
+
let user_match = match_user_for_post storage post in
+
let topic = River.title post in
+
let content = format_river_post ~user_match post in
+
Log.info (fun m -> m "Posting to #%s: %s" channel topic);
+
match post_to_zulip client ~channel ~topic ~content with
+
| Ok _response -> incr posted_count
+
| Error e -> Log.err (fun m -> m "Failed to post: %s" (Zulip.error_message e))
+
match get_latest_post_date all_posts with
+
| Some latest -> let _ = update_user_river_date storage user latest in ()
+
let _ = update_river_last_sync storage (Unix.gettimeofday ()) in
+
Log.info (fun m -> m "Sync complete, posted %d items" !posted_count);
+
let msg = Printf.sprintf "Sync failed: %s" (Printexc.to_string exn) in
+
Log.err (fun m -> m "%s" msg);
+
Error (Zulip.create_error ~code:(Other "sync_error") ~msg ())
(** Create the bot handler instance *)
let create_handler config storage identity =
anil.recoil.org