Feature/26 deactivate inactve users (#477)

* inactivity worker

* remove unused migration

* add settings migration

* make setting a list of timestamps

* inactive contacts job handler

* get rid of dependency

* resolve todos

* make formatter happy

* fix reminder query of only one timespan is given

* resolve mr discussion

pool/app/contact/contact.mli

@@ -126,6 +126,7 @@ val find_full_cell_phone_verification_by_contact
   -> (Pool_user.UnverifiedCellPhone.full, Pool_message.Error.t) Lwt_result.t
 
 val has_terms_accepted : Database.Label.t -> t -> bool Lwt.t
+val find_last_signin_at : Database.Label.t -> t -> Ptime.t Lwt.t
 
 type create =
   { user_id : Id.t
@@ -158,6 +159,7 @@ type event =
   | RegistrationAttemptNotificationSent of t
   | Updated of t
   | SignInCounterUpdated of t
+  | NotifiedAbountInactivity of t
 
 val created : create -> event
 val updated : t -> event

pool/app/contact/dune

@@ -9,6 +9,7 @@
   ptime
   role
   settings
+  schedule
   sihl
   utils
   query)

pool/app/contact/event.ml

@@ -41,6 +41,7 @@ type event =
   | RegistrationAttemptNotificationSent of t
   | Updated of t
   | SignInCounterUpdated of t
+  | NotifiedAbountInactivity of t
 [@@deriving eq, show, variants]
 
 let handle_event ?tags pool : event -> unit Lwt.t =
@@ -121,5 +122,9 @@ let handle_event ?tags pool : event -> unit Lwt.t =
       Pool_user.update pool ~email ~lastname ~firstname contact.user
     in
     Lwt.return_unit
-  | SignInCounterUpdated contact -> Repo.update_sign_in_count pool contact
+  | SignInCounterUpdated contact ->
+    let%lwt () = Repo.update_sign_in_count pool contact in
+    let%lwt () = Repo.remove_deactivation_notifications pool contact in
+    Lwt.return_unit
+  | NotifiedAbountInactivity t -> Repo.InactivityNotification.insert pool t
 ;;

pool/app/contact/repo/repo.ml

@@ -1,4 +1,5 @@
 open CCFun.Infix
+open Utils.Lwt_result.Infix
 open Repo_entity
 module Dynparam = Database.Dynparam
 
@@ -69,7 +70,6 @@ let find_request =
 ;;
 
 let find pool id =
-  let open Utils.Lwt_result.Infix in
   Database.find_opt pool find_request id
   ||> CCOption.to_result Pool_message.(Error.NotFound Field.Contact)
 ;;
@@ -85,7 +85,6 @@ let find_admin_comment_request =
 ;;
 
 let find_admin_comment pool id =
-  let open Utils.Lwt_result.Infix in
   Database.find_opt pool find_admin_comment_request id ||> CCOption.flatten
 ;;
 
@@ -100,7 +99,6 @@ let find_by_email_request =
 ;;
 
 let find_by_email pool email =
-  let open Utils.Lwt_result.Infix in
   Database.find_opt pool find_by_email_request email
   ||> CCOption.to_result Pool_message.(Error.NotFound Field.Contact)
 ;;
@@ -117,7 +115,6 @@ let find_confirmed_request =
 ;;
 
 let find_confirmed pool email =
-  let open Utils.Lwt_result.Infix in
   Database.find_opt pool find_confirmed_request email
   ||> CCOption.to_result Pool_message.(Error.NotFound Field.Contact)
 ;;
@@ -160,7 +157,6 @@ let select_count where_fragment =
 ;;
 
 let find_all ?query ?actor ?permission pool () =
-  let open Utils.Lwt_result.Infix in
   let checks =
     [ Format.asprintf
         {sql|
@@ -474,7 +470,6 @@ let find_cell_phone_verification_by_contact_and_code_request =
 ;;
 
 let find_cell_phone_verification_by_contact_and_code pool contact code =
-  let open Utils.Lwt_result.Infix in
   Database.find_opt
     pool
     find_cell_phone_verification_by_contact_and_code_request
@@ -497,7 +492,6 @@ let find_full_cell_phone_verification_by_contact_request =
 ;;
 
 let find_full_cell_phone_verification_by_contact pool contact =
-  let open Utils.Lwt_result.Infix in
   Database.find_opt
     pool
     find_full_cell_phone_verification_by_contact_request
@@ -536,6 +530,18 @@ let update_sign_in_count pool =
   Entity.id %> Database.exec pool update_sign_in_count_request
 ;;
 
+let remove_deactivation_notifications pool contact =
+  let open Caqti_request.Infix in
+  let request =
+    {sql|
+      DELETE FROM pool_contact_deactivation_notification
+      WHERE contact_uuid = UNHEX(REPLACE(?, '-', ''))
+    |sql}
+    |> Id.t ->. Caqti_type.unit
+  in
+  Database.exec pool request (Entity.id contact)
+;;
+
 let set_inactive_request =
   let open Caqti_request.Infix in
   {sql|
@@ -550,3 +556,30 @@ let set_inactive_request =
 ;;
 
 let set_inactive pool = Entity.id %> Database.exec pool set_inactive_request
+
+let find_last_signin_at pool contact =
+  let request =
+    let open Caqti_request.Infix in
+    {sql|
+      SELECT last_sign_in_at
+      FROM pool_contacts
+      WHERE user_uuid = UNHEX(REPLACE($1, '-', ''))
+    |sql}
+    |> Repo_entity.Id.t ->! Caqti_type.ptime
+  in
+  contact |> Entity.id |> Database.find pool request
+;;
+
+module InactivityNotification = struct
+  let insert pool contact =
+    let open Caqti_request.Infix in
+    let request =
+      {sql|
+       INSERT INTO pool_contact_deactivation_notification (contact_uuid)
+       VALUES (UNHEX(REPLACE(?, '-', '')))
+      |sql}
+      |> Repo_entity.Id.t ->. Caqti_type.unit
+    in
+    Database.exec pool request (Entity.id contact)
+  ;;
+end

pool/app/email/email.mli

@@ -297,6 +297,8 @@ type dispatch =
   ; job_ctx : Pool_queue.job_ctx option
   }
 
+val equal_dispatch : dispatch -> dispatch -> bool
+val pp_dispatch : Format.formatter -> dispatch -> unit
 val yojson_of_dispatch : dispatch -> Yojson.Safe.t
 val job : dispatch -> Service.Job.t
 val id : dispatch -> Pool_queue.Id.t option

pool/app/job/contact_job/contact_job.ml

@@ -0,0 +1,2 @@
+include Contact_job_repo
+module Inactivity = Inactivity

pool/app/job/contact_job/contact_job.mli

@@ -0,0 +1,19 @@
+val find_to_warn_about_inactivity
+  :  Database.Label.t
+  -> Ptime.Span.t list
+  -> Contact.t list Lwt.t
+
+module Inactivity : sig
+  val handle_disable_contacts
+    :  Database.Label.t
+    -> Settings.InactiveUser.DisableAfter.t
+    -> Ptime.Span.t list
+    -> (Email.dispatch list * Contact.event list, Pool_message.Error.t) Lwt_result.t
+
+  val handle_contact_warnings
+    :  Database.Label.t
+    -> Ptime.Span.t list
+    -> (Email.dispatch list * Contact.event list, Pool_message.Error.t) result Lwt.t
+
+  val register : unit -> Sihl.Container.Service.t
+end

pool/app/job/contact_job/contact_job_repo.ml

@@ -0,0 +1,135 @@
+module Dynparam = Database.Dynparam
+open CCFun.Infix
+
+let additional_joins =
+  [ {sql|
+    LEFT JOIN (
+      SELECT
+        contact_uuid,
+        MAX(created_at) AS latest_notification,
+        COUNT(*) AS notification_count
+      FROM
+        pool_contact_deactivation_notification
+      GROUP BY
+        contact_uuid
+    ) pcdn ON pool_contacts.user_uuid = pcdn.contact_uuid
+  |sql}
+  ]
+;;
+
+let find_to_warn_about_inactivity_request latest_notification_timestamps =
+  let where =
+    Format.asprintf
+      {sql|
+        WHERE
+          (
+            pool_contacts.paused = 0
+            AND pool_contacts.disabled = 0
+            AND user_users.status = "active"
+            AND pool_contacts.email_verified IS NOT NULL
+            AND pool_contacts.import_pending = 0
+          )
+          AND 
+          (
+            (
+              last_sign_in_at <= NOW() - INTERVAL ? SECOND
+              AND 
+                (
+                  pcdn.notification_count = 0 
+                  OR 
+                  pcdn.notification_count IS NULL
+                )
+            ) OR (
+              pcdn.latest_notification <= NOW() - INTERVAL %s SECOND 
+              AND
+              pcdn.notification_count < ?
+              )
+          ) LIMIT 100
+      |sql}
+      latest_notification_timestamps
+  in
+  Contact.Repo.find_request_sql ~additional_joins where
+;;
+
+let find_to_warn_about_inactivity pool warn_after =
+  match warn_after with
+  | [] -> Lwt.return []
+  | warn_after ->
+    let open Dynparam in
+    let open Caqti_request.Infix in
+    let warn_after_s =
+      CCList.map
+        (Ptime.Span.to_int_s %> CCOption.get_exn_or "Invalid time span")
+        warn_after
+    in
+    let dyn =
+      CCList.fold_left (fun dyn span -> dyn |> add Caqti_type.int span) empty warn_after_s
+    in
+    let sql =
+      match warn_after with
+      | [] -> failwith "Emtpy list provided"
+      | [ _ ] -> "?"
+      | _ :: tl_warn ->
+        (* Ignoring the first element, as this case is hardcoded in the first condition *)
+        tl_warn
+        |> CCList.mapi (fun i _ ->
+          Format.asprintf "WHEN pcdn.notification_count = %i THEN ?" (i + 1))
+        |> CCString.concat "\n"
+        |> Format.asprintf "( CASE %s ELSE ? END )"
+    in
+    (* Adding the last timestampt as the default case *)
+    let dyn =
+      add Caqti_type.int (CCList.rev warn_after_s |> CCList.hd) dyn
+      |> add Caqti_type.int (CCList.length warn_after)
+    in
+    let request = find_to_warn_about_inactivity_request sql in
+    let (Pack (pt, pv)) = dyn in
+    let request = request |> pt ->* Contact.Repo.t in
+    Database.collect pool request pv
+;;
+
+let find_to_disable pool disable_after n_reminders =
+  let open Dynparam in
+  let request where =
+    Format.asprintf
+      {sql|
+        WHERE
+          (
+            pool_contacts.paused = 0
+            AND pool_contacts.disabled = 0
+            AND user_users.status = "active"
+            AND pool_contacts.email_verified IS NOT NULL
+            AND pool_contacts.import_pending = 0
+          )
+          AND %s
+          LIMIT 100
+      |sql}
+      where
+    |> Contact.Repo.find_request_sql ~additional_joins
+  in
+  let needs_reminders =
+    {sql|
+      pcdn.latest_notification <= NOW() - INTERVAL $1 SECOND
+      AND pcdn.notification_count = $2
+  |sql}
+  in
+  let check_last_login =
+    {sql|
+      last_sign_in_at <= NOW() - INTERVAL $1 SECOND
+   |sql}
+  in
+  let Pack (pt, pv), where =
+    let disable_after =
+      disable_after |> Ptime.Span.to_int_s |> CCOption.get_exn_or "Invalid time span"
+    in
+    let dyn = empty |> add Caqti_type.int disable_after in
+    match n_reminders with
+    | 0 -> dyn, check_last_login
+    | _ -> add Caqti_type.int n_reminders dyn, needs_reminders
+  in
+  let request =
+    let open Caqti_request.Infix in
+    request where |> pt ->* Contact.Repo.t
+  in
+  Database.collect pool request pv
+;;

pool/app/job/contact_job/dune

@@ -0,0 +1,7 @@
+(library
+ (name contact_job)
+ (libraries contact message_template pool_common utils)
+ (preprocess
+  (pps lwt_ppx ppx_deriving.eq ppx_deriving.show ppx_yojson_conv)))
+
+(include_subdirs unqualified)