···
-
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