handle pr notifications on the same thread

lib/action.ml

@@ -5,8 +5,10 @@ open Common
 open Github_j
 
 exception Action_error of string
+exception Success_handler_error of string
 
 let action_error msg = raise (Action_error msg)
+let handler_error msg = raise (Success_handler_error msg)
 let log = Log.from "action"
 
 module Action (Github_api : Api.Github) (Slack_api : Api.Slack) = struct
@@ -41,12 +43,8 @@ module Action (Github_api : Api.Github) (Slack_api : Api.Slack) = struct
     (* Updates mapping every 24 hours *)
     refresh_username_to_slack_id_tbl_background_lwt ~ctx
 
-  let match_github_login_to_slack_id cfg_opt login =
-    let login =
-      match cfg_opt with
-      | None -> login
-      | Some cfg -> List.assoc_opt login cfg.user_mappings |> Option.default login
-    in
+  let match_github_login_to_slack_id cfg login =
+    let login = List.assoc_opt login cfg.user_mappings |> Option.default login in
     login |> canonicalize_email_username |> Stringtbl.find_opt username_to_slack_id_tbl
 
   let partition_push (cfg : Config_t.config) n =
@@ -81,10 +79,24 @@ module Action (Github_api : Api.Github) (Slack_api : Api.Slack) = struct
     labels |> List.concat_map (Rule.Label.match_rules ~rules) |> List.sort_uniq String.compare |> fun channel_names ->
     if channel_names = [] then Stdlib.Option.to_list default else channel_names
 
-  let partition_pr cfg (n : pr_notification) =
+  let partition_pr cfg (ctx : Context.t) (n : pr_notification) =
     match n.action with
-    | (Opened | Closed | Reopened | Labeled | Ready_for_review) when not n.pull_request.draft ->
+    | (Opened | Closed | Reopened | Ready_for_review) when not n.pull_request.draft ->
       partition_label cfg n.pull_request.labels
+    | Labeled when not n.pull_request.draft ->
+      (* labels get notified by gh in addition the pr notification itself, which means that when we open a pr
+          we have one `Open` notification and as many `Labeled` notifications as the pr has labels.
+          we want to avoid having many notifications for a single `Opened` event. *)
+      (match State.has_pr_thread ctx.state ~repo_url:n.repository.url ~pr_url:n.pull_request.html_url with
+      | false ->
+        (* we dont have a thread for the pr yet, these are the labels notifications before the PR *)
+        []
+      | true ->
+        (* if we already have a thread on a certain channel, we already have received an open PR notification.
+           If we have a new label that triggers a notification on a new channel, we'll notify the channel.
+           If the label triggers a notification on a channel with an existing thread, the notification will go
+           in the thread *)
+        partition_label cfg n.pull_request.labels)
     | _ -> []
 
   let partition_issue cfg (n : issue_notification) =
@@ -105,16 +117,17 @@ module Action (Github_api : Api.Github) (Slack_api : Api.Slack) = struct
   let partition_pr_review cfg (n : pr_review_notification) =
     let { review; action; _ } = n in
     match action, review.state, review.body with
-    | Submitted, "commented", (Some "" | None) -> []
-    (* the case (action = Submitted, review.state = "commented", review.body = "") happens when
-       a reviewer starts a review by commenting on particular sections of the code, which triggars a pull_request_review_comment event simultaneouly,
-       and then submits the review without submitting any general feedback or explicit approval/changes.
+    | Submitted, "commented", (Some "" | None) ->
+      (* the case (action = Submitted, review.state = "commented", review.body = "") happens when
+           a reviewer starts a review by commenting on particular sections of the code, which triggars a pull_request_review_comment event simultaneouly,
+           and then submits the review without submitting any general feedback or explicit approval/changes.
 
-       the case (action = Submitted, review.state = "commented", review.body = null) happens when
-       a reviewer adds a single comment on a particular section of the code, which triggars a pull_request_review_comment event simultaneouly.
+           the case (action = Submitted, review.state = "commented", review.body = null) happens when
+           a reviewer adds a single comment on a particular section of the code, which triggars a pull_request_review_comment event simultaneouly.
 
-       in both cases, since pull_request_review_comment is already handled by another type of event, information in the pull_request_review payload
-       does not provide any insightful information and will thus be ignored. *)
+         in both cases, since pull_request_review_comment is already handled by another type of event, information in the pull_request_review payload
+         does not provide any insightful information and will thus be ignored. *)
+      []
     | Submitted, _, _ -> partition_label cfg n.pull_request.labels
     | _ -> []
 
@@ -250,24 +263,21 @@ module Action (Github_api : Api.Github) (Slack_api : Api.Slack) = struct
   let generate_notifications (ctx : Context.t) (req : Github.t) =
     let repo = Github.repo_of_notification req in
     let cfg = Context.find_repo_config_exn ctx repo.url in
-    let slack_match_func = match_github_login_to_slack_id (Some cfg) in
+    let slack_match_func = match_github_login_to_slack_id cfg in
     match ignore_notifications_from_user cfg req with
     | true -> Lwt.return []
     | false ->
     match req with
     | Github.Push n ->
       partition_push cfg n |> List.map (fun (channel, n) -> generate_push_notification n channel) |> Lwt.return
     | Pull_request n ->
-      partition_pr cfg n |> List.map (generate_pull_request_notification ~slack_match_func n) |> Lwt.return
+      partition_pr cfg ctx n |> List.map (generate_pull_request_notification ~ctx ~slack_match_func n) |> Lwt.return
     | PR_review n ->
-      partition_pr_review cfg n |> List.map (generate_pr_review_notification ~slack_match_func n) |> Lwt.return
-    | PR_review_comment _n ->
-      (* we want to silence review comments and keep only the "main" review message
-         TODO: make this configurable? *)
-      Lwt.return []
-      (* partition_pr_review_comment cfg n
-         |> List.map (generate_pr_review_comment_notification ~slack_match_func n)
-         |> Lwt.return *)
+      partition_pr_review cfg n |> List.map (generate_pr_review_notification ~slack_match_func ~ctx n) |> Lwt.return
+    | PR_review_comment n ->
+      partition_pr_review_comment cfg n
+      |> List.map (generate_pr_review_comment_notification ~slack_match_func ~ctx n)
+      |> Lwt.return
     | Issue n -> partition_issue cfg n |> List.map (generate_issue_notification ~slack_match_func n) |> Lwt.return
     | Issue_comment n ->
       partition_issue_comment cfg n |> List.map (generate_issue_comment_notification ~slack_match_func n) |> Lwt.return
@@ -281,9 +291,17 @@ module Action (Github_api : Api.Github) (Slack_api : Api.Slack) = struct
       Lwt.return notifs
 
   let send_notifications (ctx : Context.t) notifications =
-    let notify (msg : Slack_t.post_message_req) =
+    let notify (msg, handler) =
       match%lwt Slack_api.send_notification ~ctx ~msg with
-      | Ok () -> Lwt.return_unit
+      | Ok (Some res) ->
+        (match handler with
+        | None -> Lwt.return_unit
+        | Some handler ->
+        try
+          handler res;
+          Lwt.return_unit
+        with exn -> handler_error (Printexc.to_string exn))
+      | Ok None -> Lwt.return_unit
       | Error e -> action_error e
     in
     Lwt_list.iter_s notify notifications
@@ -372,6 +390,9 @@ module Action (Github_api : Api.Github) (Slack_api : Api.Slack) = struct
     | Action_error msg ->
       log#error "action error %s" msg;
       Lwt.return_unit
+    | Success_handler_error msg ->
+      log#error "success handler error %s" msg;
+      Lwt.return_unit
     | Context.Context_error msg ->
       log#error "context error %s" msg;
       Lwt.return_unit

lib/api.ml

@@ -22,7 +22,7 @@ module type Slack = sig
     lookup_user_res slack_response Lwt.t
 
   val list_users : ?cursor:string -> ?limit:int -> ctx:Context.t -> unit -> list_users_res slack_response Lwt.t
-  val send_notification : ctx:Context.t -> msg:post_message_req -> unit slack_response Lwt.t
+  val send_notification : ctx:Context.t -> msg:post_message_req -> post_message_res option slack_response Lwt.t
 
   val send_chat_unfurl :
     ctx:Context.t ->

lib/api_local.ml

@@ -76,7 +76,7 @@ module Slack : Api.Slack = struct
     let json = msg |> Slack_j.string_of_post_message_req |> Yojson.Basic.from_string |> Yojson.Basic.pretty_to_string in
     Printf.printf "will notify #%s\n" msg.channel;
     Printf.printf "%s\n" json;
-    Lwt.return @@ Ok ()
+    Lwt.return @@ Ok None
 
   let send_chat_unfurl ~ctx:_ ~channel ~ts ~unfurls () =
     let req = Slack_j.{ channel; ts; unfurls } in
@@ -101,7 +101,7 @@ module Slack_simple : Api.Slack = struct
       (match msg.Slack_t.text with
       | None -> ""
       | Some s -> sprintf " with %S" s);
-    Lwt.return @@ Ok ()
+    Lwt.return @@ Ok None
 
   let send_chat_unfurl ~ctx:_ ~channel ~ts:_ ~(unfurls : Slack_t.message_attachment Common.StringMap.t) () =
     Printf.printf "will unfurl in #%s\n" channel;
@@ -129,7 +129,7 @@ module Slack_json : Api.Slack = struct
     let url = Uri.add_query_param url ("msg", [ json ]) in
     log#info "%s" (Uri.to_string url);
     log#info "%s" json;
-    Lwt.return @@ Ok ()
+    Lwt.return @@ Ok None
 
   let send_chat_unfurl ~ctx:_ ~channel ~ts:_ ~(unfurls : Slack_t.message_attachment Common.StringMap.t) () =
     log#info "will notify %s" channel;

lib/api_remote.ml

@@ -186,12 +186,14 @@ module Slack : Api.Slack = struct
       log#info "data: %s" data;
       if webhook_mode then begin
         match%lwt http_request ~body ~headers `POST url with
-        | Ok _res -> Lwt.return @@ Ok ()
+        | Ok _res ->
+          (* Webhooks reply only 200 `ok`. We can't generate anything useful for notification success handlers *)
+          Lwt.return @@ Ok None
         | Error e -> Lwt.return @@ build_error (query_error_msg url e)
       end
       else begin
         match%lwt slack_api_request ~body ~headers `POST url Slack_j.read_post_message_res with
-        | Ok _res -> Lwt.return @@ Ok ()
+        | Ok res -> Lwt.return @@ Ok (Some res)
         | Error e -> Lwt.return @@ build_error e
       end
 

lib/slack.atd

@@ -58,6 +58,7 @@ type message_block = [
 
 type post_message_req = {
   channel: string;
+  ?thread_ts: string nullable;
   ?username : string nullable;
   ?text: string nullable;
   ?attachments: message_attachment list nullable;
@@ -68,6 +69,7 @@ type post_message_req = {
 
 type post_message_res = {
   channel: string;
+  ts: string;
 }
 
 type lookup_user_res = {

lib/slack.ml

@@ -29,7 +29,7 @@ let pp_link ~url text = sprintf "<%s|%s>" url (Mrkdwn.escape_mrkdwn text)
 let show_labels = function
   | [] -> None
   | (labels : label list) ->
-    Some (sprintf "Labels: %s" @@ String.concat ", " (List.map (fun (x : label) -> x.name) labels))
+    Some (sprintf "*Labels*: %s" @@ String.concat ", " (List.map (fun (x : label) -> sprintf "*%s*" x.name) labels))
 
 let pluralize ~suf ~len name = if len = 1 then sprintf "%s" name else String.concat "" [ name; suf ]
 
@@ -44,8 +44,8 @@ let markdown_text_attachment ~footer markdown_body =
     };
   ]
 
-let make_message ?username ?text ?attachments ?blocks ~channel () =
-  { channel; text; attachments; blocks; username; unfurl_links = Some false; unfurl_media = None }
+let make_message ?username ?text ?attachments ?blocks ?thread_ts ?handler ~channel () =
+  { channel; thread_ts; text; attachments; blocks; username; unfurl_links = Some false; unfurl_media = None }, handler
 
 let github_handle_regex = Re2.create_exn {|\B@([[:alnum:]][[:alnum:]-]{1,38})\b|}
 (* Match GH handles in messages - a GitHub handle has at most 39 chars and no underscore *)
@@ -64,12 +64,17 @@ let format_attachments ~slack_match_func ~footer ~body =
   in
   Option.map (fun t -> markdown_text_attachment ~footer t |> List.map format_mention_in_markdown) body
 
-let generate_pull_request_notification ~slack_match_func notification channel =
+let generate_pull_request_notification ~slack_match_func ~(ctx : Context.t) notification channel =
   let { action; number; sender; pull_request; repository } = notification in
   let ({ body; title; html_url; labels; merged; _ } : pull_request) = pull_request in
   let action, body =
     match action with
-    | Opened | Ready_for_review -> "opened", body
+    | Opened | Ready_for_review ->
+      let labels_banner = show_labels labels in
+      ( "opened",
+        body
+        |> Option.map (fun body' ->
+               Option.map_default (fun labels' -> sprintf "%s\n%s" body' labels') body' labels_banner) )
     | Closed -> (if merged then "merged" else "closed"), None
     | Reopened -> "reopened", None
     | Labeled -> "labeled", show_labels labels
@@ -82,9 +87,20 @@ let generate_pull_request_notification ~slack_match_func notification channel =
     sprintf "<%s|[%s]> Pull request #%d %s %s by *%s*" repository.url repository.full_name number
       (pp_link ~url:html_url title) action sender.login
   in
-  make_message ~text:summary ?attachments:(format_attachments ~slack_match_func ~footer:None ~body) ~channel ()
+  (* if the message is already in a thread, post to that thread *)
+  let thread_ts = State.get_thread_ts ctx.state ~repo_url:repository.url ~pr_url:html_url channel in
+  let handler (res : Slack_t.post_message_res) =
+    match notification.action with
+    | Opened | Ready_for_review | Labeled ->
+      State.update_thread ctx.state ~repo_url:repository.url ~pr_url:html_url { channel; ts = res.ts }
+    | Closed -> State.delete_thread ctx.state ~repo_url:repository.url ~pr_url:html_url
+    | _ -> ()
+  in
+  make_message ~text:summary ?thread_ts
+    ?attachments:(format_attachments ~slack_match_func ~footer:None ~body)
+    ~handler ~channel ()
 
-let generate_pr_review_notification ~slack_match_func notification channel =
+let generate_pr_review_notification ~slack_match_func ~(ctx : Context.t) notification channel =
   let { action; sender; pull_request; review; repository } = notification in
   let ({ number; title; html_url; _ } : pull_request) = pull_request in
   let action_str =
@@ -104,11 +120,13 @@ let generate_pr_review_notification ~slack_match_func notification channel =
     sprintf "<%s|[%s]> *%s* <%s|%s> #%d %s" repository.url repository.full_name sender.login review.html_url action_str
       number (pp_link ~url:html_url title)
   in
-  make_message ~text:summary
+  (* if the message is already in a thread, post to that thread *)
+  let thread_ts = State.get_thread_ts ctx.state ~repo_url:repository.url ~pr_url:html_url channel in
+  make_message ~text:summary ?thread_ts
     ?attachments:(format_attachments ~slack_match_func ~footer:None ~body:review.body)
     ~channel ()
 
-let generate_pr_review_comment_notification ~slack_match_func notification channel =
+let generate_pr_review_comment_notification ~slack_match_func ~(ctx : Context.t) notification channel =
   let { action; pull_request; sender; comment; repository } = notification in
   let ({ number; title; html_url; _ } : pull_request) = pull_request in
   let action_str =
@@ -126,9 +144,11 @@ let generate_pr_review_comment_notification ~slack_match_func notification chann
   let file =
     match comment.path with
     | None -> None
-    | Some a -> Some (sprintf "New comment by %s in <%s|%s>" sender.login comment.html_url a)
+    | Some a -> Some (sprintf "Commented in file <%s|%s>" comment.html_url a)
   in
-  make_message ~text:summary
+  (* if the message is already in a thread, post to that thread *)
+  let thread_ts = State.get_thread_ts ctx.state ~repo_url:repository.url ~pr_url:html_url channel in
+  make_message ~text:summary ?thread_ts
     ?attachments:(format_attachments ~slack_match_func ~footer:file ~body:(Some comment.body))
     ~channel ()
 

lib/state.atd

@@ -2,6 +2,7 @@ type status_state <ocaml from="Github"> = abstract
 type 'v map_as_object <ocaml from="Common"> = abstract
 type 'v table_as_object <ocaml from="Common"> = abstract
 type string_set <ocaml from="Common"> = abstract
+type post_message_res <ocaml from="Slack"> = abstract
 
 type ci_commit = {
   sha: string;
@@ -35,10 +36,13 @@ type commit_sets = {
    that have triggered a direct message notification. *)
 type pipeline_commits = commit_sets map_as_object
 
+type slack_threads = post_message_res list map_as_object
+
 (* The runtime state of a given GitHub repository *)
 type repo_state = {
   pipeline_statuses <ocaml mutable>: pipeline_statuses;
   pipeline_commits <ocaml mutable>: pipeline_commits;
+  slack_threads <ocaml mutable>: slack_threads;
 }
 
 (* The serializable runtime state of the bot *)

lib/state.ml

@@ -6,7 +6,7 @@ let log = Log.from "state"
 type t = { state : State_t.state }
 
 let empty_repo_state () : State_t.repo_state =
-  { pipeline_statuses = StringMap.empty; pipeline_commits = StringMap.empty }
+  { pipeline_statuses = StringMap.empty; pipeline_commits = StringMap.empty; slack_threads = StringMap.empty }
 
 let empty () : t =
   let state = State_t.{ repos = Stringtbl.empty (); bot_user_id = None } in
@@ -104,6 +104,37 @@ let mem_repo_pipeline_commits { state } repo_url ~pipeline ~commit =
   | None -> false
   | Some { State_t.s1; s2 } -> StringSet.mem commit s1 || StringSet.mem commit s2
 
+let has_pr_thread { state } ~repo_url ~pr_url =
+  let repo_state = find_or_add_repo' state repo_url in
+  match StringMap.find_opt pr_url repo_state.slack_threads with
+  | None -> false
+  | Some _ -> true
+
+let get_thread_ts { state } ~repo_url ~pr_url channel =
+  let repo_state = find_or_add_repo' state repo_url in
+  match StringMap.find_opt pr_url repo_state.slack_threads with
+  | None -> None
+  | Some threads ->
+  match List.filter (fun (thread : Slack_t.post_message_res) -> String.equal channel thread.channel) threads with
+  | [] -> None
+  | thread :: _ -> Some thread.ts
+
+let update_thread { state } ~repo_url ~pr_url (msg : Slack_t.post_message_res) =
+  let repo_state = find_or_add_repo' state repo_url in
+  let set_threads threads =
+    match threads with
+    | None -> Some [ msg ]
+    | Some threads ->
+    match List.exists (fun (thread : Slack_t.post_message_res) -> String.equal msg.ts thread.ts) threads with
+    | true -> Some threads
+    | false -> Some (msg :: threads)
+  in
+  repo_state.slack_threads <- StringMap.update pr_url set_threads repo_state.slack_threads
+
+let delete_thread { state } ~repo_url ~pr_url =
+  let repo_state = find_or_add_repo' state repo_url in
+  repo_state.slack_threads <- StringMap.remove pr_url repo_state.slack_threads
+
 let set_bot_user_id { state; _ } user_id = state.State_t.bot_user_id <- Some user_id
 let get_bot_user_id { state; _ } = state.State_t.bot_user_id
 

src/dune

@@ -1,5 +1,15 @@
 (executable
- (libraries lib cmdliner devkit devkit.core extlib lwt.unix result uri unix)
+ (libraries
+  lib
+  cmdliner
+  devkit
+  devkit.core
+  extlib
+  lwt.unix
+  result
+  uri
+  yojson
+  unix)
  (preprocess
   (pps lwt_ppx))
  (public_name monorobot))