state: persist pipeline_commits state

lib/action.ml

@@ -140,6 +140,7 @@ module Action (Github_api : Api.Github) (Slack_api : Api.Slack) = struct
         if notify_dm then begin
           match%lwt Slack_api.lookup_user ~ctx ~cfg ~email:n.commit.commit.author.email () with
           | Ok res ->
+            State.set_repo_pipeline_commit ctx.state repo.url ~pipeline ~commit:n.sha;
             (* To send a DM, channel parameter is set to the user id of the recipient *)
             Lwt.return [ res.user.id ]
           | Error e ->
@@ -190,12 +191,13 @@ module Action (Github_api : Api.Github) (Slack_api : Api.Slack) = struct
               branches
             | None -> n.branches
           in
-          let notify_dm = notify_dm && not (State.In_memory.Dm_commits.mem n.sha) in
+          let notify_dm =
+            notify_dm && not (State.mem_repo_pipeline_commits ctx.state repo.url ~pipeline ~commit:n.sha)
+          in
           action_on_match branches ~notify_channels ~notify_dm
       end
       else Lwt.return []
     in
-    State.In_memory.Dm_commits.add n.sha;
     State.set_repo_pipeline_status ctx.state repo.url ~pipeline ~branches:n.branches ~status:current_status;
     Lwt.return recipients
 

lib/common.atd

@@ -5,3 +5,7 @@ type 'v map_as_object =
 type 'v table_as_object =
   (string * 'v) list <json repr="object">
     wrap <ocaml module="Common.Stringtbl" t="'v Common.Stringtbl.t">
+
+type string_set =
+  string list
+    wrap <ocaml module="Common.StringSet">

lib/common.ml

@@ -1,4 +1,10 @@
-module StringSet = Set.Make (String)
+module StringSet = struct
+  include Set.Make (String)
+
+  let to_list set : string list = elements set
+  let wrap = of_list
+  let unwrap = to_list
+end
 
 module StringMap = struct
   include Map.Make (String)

lib/state.atd

@@ -1,6 +1,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
 
 (* A map from branch names to build statuses *)
 type branch_statuses = status_state map_as_object
@@ -10,13 +11,23 @@ type branch_statuses = status_state map_as_object
    branch *)
 type pipeline_statuses = branch_statuses map_as_object
 
+type commit_sets = {
+  s1: string_set;
+  s2: string_set;
+}
+
+(* A map from pipeline names to a set of commits. This tracks the commits
+   that have triggered a direct message notification. *)
+type pipeline_commits = commit_sets map_as_object
+
 (* The runtime state of a given GitHub repository *)
 type repo_state = {
-  pipeline_statuses <ocaml mutable>: pipeline_statuses
+  pipeline_statuses <ocaml mutable>: pipeline_statuses;
+  pipeline_commits <ocaml mutable>: pipeline_commits;
 }
 
 (* The serializable runtime state of the bot *)
 type state = {
   repos : repo_state table_as_object;
   ?bot_user_id <ocaml mutable>: string nullable;
-}
+}

lib/state.ml

@@ -5,7 +5,8 @@ let log = Log.from "state"
 
 type t = { state : State_t.state }
 
-let empty_repo_state () : State_t.repo_state = { pipeline_statuses = StringMap.empty }
+let empty_repo_state () : State_t.repo_state =
+  { pipeline_statuses = StringMap.empty; pipeline_commits = StringMap.empty }
 
 let empty () : t =
   let state = State_t.{ repos = Stringtbl.empty (); bot_user_id = None } in
@@ -31,6 +32,23 @@ let set_repo_pipeline_status { state } repo_url ~pipeline ~(branches : Github_t.
   let repo_state = find_or_add_repo' state repo_url in
   repo_state.pipeline_statuses <- StringMap.update pipeline set_branch_status repo_state.pipeline_statuses
 
+let set_repo_pipeline_commit { state } repo_url ~pipeline ~commit =
+  let rotation_threshold = 1000 in
+  let repo_state = find_or_add_repo' state repo_url in
+  let set_commit commits =
+    let { State_t.s1; s2 } = Option.default { State_t.s1 = StringSet.empty; s2 = StringSet.empty } commits in
+    let s1 = StringSet.add commit s1 in
+    let s1, s2 = if StringSet.cardinal s1 > rotation_threshold then StringSet.empty, s1 else s1, s2 in
+    Some { State_t.s1; s2 }
+  in
+  repo_state.pipeline_commits <- StringMap.update pipeline set_commit repo_state.pipeline_commits
+
+let mem_repo_pipeline_commits { state } repo_url ~pipeline ~commit =
+  let repo_state = find_or_add_repo' state repo_url in
+  match StringMap.find_opt pipeline repo_state.pipeline_commits with
+  | None -> false
+  | Some { State_t.s1; s2 } -> StringSet.mem commit s1 || StringSet.mem commit s2
+
 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
 
@@ -40,20 +58,3 @@ let save { state; _ } path =
     Files.save_as path (fun oc -> output_string oc data);
     Ok ()
   with exn -> fmt_error ~exn "failed to save state to file %s" path
-
-module In_memory = struct
-  module Dm_commits = struct
-    let state = ref (StringSet.empty, StringSet.empty)
-    let rotation_threshold = 1000
-
-    let add (sha : string) =
-      let s1, s2 = !state in
-      let s1 = StringSet.add sha s1 in
-      let s1, s2 = if StringSet.cardinal s1 > rotation_threshold then StringSet.empty, s1 else s1, s2 in
-      state := s1, s2
-
-    let mem (sha : string) =
-      let s1, s2 = !state in
-      StringSet.mem sha s1 || StringSet.mem sha s2
-  end
-end

mock_states/status.commit1-02-failed.json

@@ -0,0 +1,15 @@
+{
+  "pipeline_statuses": {
+    "buildkite/pipeline2": {
+      "master": "failure"
+    }
+  },
+  "pipeline_commits": {
+    "buildkite/pipeline2": {
+      "s1": [
+        "7e0a933e9c71b4ca107680ca958ca1888d5e479b"
+      ],
+      "s2": []
+    }
+  }
+}

mock_states/status.state_hide_success_test.json

@@ -1,4 +1,3 @@
-
 {
   "pipeline_statuses": {
     "default": {
@@ -7,5 +6,6 @@
     "buildkite/pipeline2": {
       "master": "success"
     }
-  }
-}
+  },
+  "pipeline_commits": {}
+}

mock_states/status.state_hide_success_test_disallowed_pipeline.json

@@ -3,5 +3,6 @@
     "buildkite/pipeline2": {
       "master": "failure"
     }
-  }
-}
+  },
+  "pipeline_commits": {}
+}