OPHKIOS-131: Uuden viestinvälityspalvelun käyttöönotto

dev/resources/dev.edn

@@ -96,5 +96,9 @@
                                           :access-log #ig/ref :yki.middleware.access-log/with-logging
                                           :auth       #ig/ref :yki.middleware.no-auth/with-authentication}
 
- :yki.env/environment {:environment "dev"}}
+ :yki.env/environment                    {:environment "dev"}
+
+ :yki.boundary.email/email-client        {:url-helper             #ig/ref :yki.util/url-helper
+                                          :cas-client             #ig/ref :yki.boundary.cas/cas-client
+                                          :use-new-email-service? true}}
 

oph-configuration/config.edn.template

@@ -72,6 +72,10 @@
                                 :cas-creds {:username "{{yki_cas_username}}"
                                             :password "{{yki_cas_password}}"}}
 
+  :yki.boundary.email/email-client {:url-helper             #ig/ref :yki.util/url-helper
+                                    :cas-client             #ig/ref :yki.boundary.cas/cas-client
+                                    :use-new-email-service? {{yki_use_new_email_service}}}
+
   :yki.middleware.auth/with-authentication
     {:url-helper #ig/ref :yki.util/url-helper
      :session-config {:key "{{yki_session_cookie_secret}}"

resources/yki/yki-oph.properties

@@ -47,6 +47,12 @@ ryhmasahkoposti-service = ${url-alb}/ryhmasahkoposti-service/email/firewall
 koodisto-service = ${url-virkailija}/koodisto-service/rest/json/$1/koodi?onlyValidKoodis=true
 koodisto-service.rinnasteinen = ${url-virkailija}/koodisto-service/rest/json/relaatio/rinnasteinen/$1
 
+new-email-service=https://viestinvalitys.${host-oppija}/lahetys
+new-email-service.cas-login=${new-email-service}/login
+new-email-service.submissions=${new-email-service}/v1/lahetykset
+new-email-service.attachments=${new-email-service}/v1/liitteet
+new-email-service.messages=${new-email-service}/v1/viestit
+
 yki-register.organizer = ${yki-register}/jarjestaja
 yki-register.exam-session = ${yki-register}/tutkintotilaisuus
 yki-register.participants = ${yki-register}/osallistujat

src/yki/boundary/cas.clj

@@ -10,11 +10,13 @@
     (fi.vm.sade.javautils.nio.cas CasClient CasConfig CasClientBuilder)
     (io.netty.handler.codec.http HttpHeaders)
     (org.asynchttpclient RequestBuilder)
-    (org.asynchttpclient Response)))
+    (org.asynchttpclient Request Response)
+    (org.asynchttpclient.request.body.multipart ByteArrayPart Part)))
 
 (defprotocol CasAccess
   (validate-ticket [this ticket])
   (cas-authenticated-post [this url body])
+  (cas-authenticated-post-multipart-form-data [this url body-parts])
   (cas-authenticated-get [this url]))
 
 (def csrf-token "csrf")
@@ -42,6 +44,15 @@
       (.setUrl url))
     (.build request-builder)))
 
+(defn- multipart-form-data-request [^String method url body-parts]
+  (let [request-builder (RequestBuilder. method)]
+    (doto request-builder
+      (.addHeader "Content-Type" "multipart/form-data")
+      (.setUrl url))
+    (doseq [part body-parts]
+      (.addBodyPart request-builder ^Part part))
+    (.build request-builder)))
+
 (defn- clear-ticket-stores! [^CasClient cas-client]
   (let [cls                   (.getClass cas-client)
         session-fetcher-field (.getDeclaredField cls "casSessionFetcher")
@@ -55,9 +66,8 @@
        (when-let [location (.get ^HttpHeaders headers "Location")]
          (str/starts-with? location (url-helper :cas.login.root)))))
 
-(defn- cas-http [^CasClient cas-client url-helper method url body]
-  (let [request  (json-request method url body)
-        execute! #(->> request
+(defn- execute-with-retry! [^CasClient cas-client url-helper ^Request request]
+  (let [execute! #(->> request
                        (.executeBlocking cas-client)
                        (process-response))
         response (execute!)]
@@ -77,14 +87,19 @@
       validation-response))
   (cas-authenticated-get [_ url]
     (try
-      (cas-http cas-client url-helper "GET" url nil)
+      (execute-with-retry! cas-client url-helper (json-request "GET" url nil))
       (catch Exception e
         (log/error e "cas-authenticated-get failed!"))))
   (cas-authenticated-post [_ url body]
     (try
-      (cas-http cas-client url-helper "POST" url body)
+      (execute-with-retry! cas-client url-helper (json-request "POST" url body))
+      (catch Exception e
+        (log/error e "cas-authenticated-post failed!"))))
+  (cas-authenticated-post-multipart-form-data [_ url body-parts]
+    (try
+      (execute-with-retry! cas-client url-helper (multipart-form-data-request "POST" url body-parts))
       (catch Exception e
-        (log/error e "cas-authenticated-post failed!")))))
+        (log/error e "cas-authenticated-post-multipart-form-data failed!")))))
 
 (defn create-cas-client [{:keys [username password]} url-helper service-url]
   (let [cas-config (CasConfig/SpringSessionCasConfig username password (url-helper :cas-client) service-url csrf-token caller-id)

src/yki/boundary/email.clj

@@ -2,8 +2,14 @@
   (:require
     [clojure.string :as str]
     [clojure.tools.logging :as log]
+    [integrant.core :as ig]
     [jsonista.core :as json]
-    [yki.util.http-util :as http-util]))
+    [yki.boundary.cas :as cas]
+    [yki.config :refer [oph-oid]]
+    [yki.util.http-util :as http-util])
+  (:import (org.asynchttpclient.request.body.multipart ByteArrayPart)))
+
+(def default-email-retention-period 180)
 
 (defn- log-disabled-email [recipients subject body attachments]
   (log/info
@@ -16,20 +22,108 @@
                     (str/join "," (map :name attachments))
                     "]")])))
 
-(defn send-email!
-  [url-helper {:keys [recipients subject body attachments]} disabled]
-  (let [url                (url-helper :ryhmasahkoposti-service)
-        wrapped-recipients (mapv (fn [rcp] {:email rcp}) recipients)]
-    (if disabled
-      (log-disabled-email recipients subject body attachments)
-      (let [email-data {:subject     subject
-                        :html        true
-                        :body        body
-                        :attachments attachments
-                        :charset     "UTF-8"}
-            response   (http-util/do-post url {:headers      {"content-type" "application/json; charset=UTF-8"}
-                                               :query-params {:sanitize "false"}
-                                               :body         (json/write-value-as-string {:email     email-data
-                                                                                          :recipient wrapped-recipients})})]
-        (when (not= 200 (:status response))
-          (throw (Exception. (str "Could not send email to " (str/join recipients)))))))))
+(defprotocol Email
+  (send-email! [_ email disabled?]))
+
+(defrecord OldEmailService [url-helper]
+  Email
+  (send-email! [_ email disabled?]
+    (let [{:keys [recipients subject body attachments]} email]
+      (if disabled?
+        (log-disabled-email recipients subject body attachments)
+        (let [email-data         {:subject     subject
+                                  :html        true
+                                  :body        body
+                                  :attachments attachments
+                                  :charset     "UTF-8"}
+              wrapped-recipients (mapv (fn [recipient]
+                                         ; Recipient can be
+                                         ; 1) a map containing keys :email and :name, OR
+                                         ; 2) the email as a string
+                                         (if (map? recipient)
+                                           {:email (:email recipient)}
+                                           {:email recipient})) recipients)
+              url                (url-helper :ryhmasahkoposti-service)
+              response           (http-util/do-post url {:headers      {"content-type" "application/json; charset=UTF-8"}
+                                                         :query-params {:sanitize "false"}
+                                                         :body         (json/write-value-as-string {:email     email-data
+                                                                                                    :recipient wrapped-recipients})})]
+          (when (not= 200 (:status response))
+            (throw (Exception. (str "Could not send email to " (str/join recipients))))))))))
+
+(defn- attachment->body-part [{:keys [name data contentType]}]
+  (ByteArrayPart. "liite" data contentType nil name))
+
+(defn- upload-attachment! [url-helper cas-client attachment]
+  (let [upload-attachment-url (url-helper :new-email-service.attachments)
+        body-parts            [(attachment->body-part attachment)]
+        {:keys [status body]} (cas/cas-authenticated-post-multipart-form-data cas-client upload-attachment-url body-parts)
+        response-body         (json/read-value body)]
+    (if (= 200 status)
+      (response-body "liiteTunniste")
+      (throw (ex-info "Error uploading attachment!" {:status     status
+                                                     :body       response-body
+                                                     :attachment (:name attachment)})))))
+
+(defn- email->message [{:keys [recipients subject body language metadata message-id retention-period]} attachment-ids]
+  {:otsikko                 subject
+   :sisalto                 body
+   :sisallonTyyppi          "html"
+   :kielet                  (some-> language (vector))
+   ; TODO Verify name and email for sender
+   ; TODO Possibly different reply-to address?
+   ; TODO Internationalization of name?
+   :lahettaja               {:nimi             "Yleiset kielitutkinnot / Opetushallitus"
+                             :sahkopostiOsoite "[email protected]"}
+   :vastaanottajat          (->> recipients
+                                 (mapv (fn [recipient]
+                                         ; Recipient can be
+                                         ; 1) a map containing keys :email and :name, OR
+                                         ; 2) the email as a string
+                                         (if (map? recipient)
+                                           {:sahkopostiOsoite (:email recipient)
+                                            :nimi             (:name recipient)}
+                                           {:sahkopostiOsoite recipient}))))
+   ; Enforce normal priority. If we were to use high priority, we'd need to throttle the rate of high priority messages ourselves.
+   :prioriteetti            "normaali"
+   :sailytysaika            (or retention-period default-email-retention-period)
+   :lahettavaPalvelu        "yki"
+   :metadata                metadata
+   :idempotencyKey          message-id
+   :kayttooikeusRajoitukset [{:oikeus       "APP_YKI_YLLAPITAJA"
+                              :organisaatio oph-oid}]
+   :liitteidenTunnisteet    attachment-ids
+   }
+  )
+
+(defrecord NewEmailService [url-helper cas-client]
+  Email
+  (send-email! [_ email disabled?]
+    (let [{:keys [recipients subject body attachments metadata message-id]} email]
+      (if disabled?
+        (log-disabled-email recipients subject body attachments)
+        ; TODO Consider failure modes!
+        ; TODO Eg. if uploading an attachment fails, retry whole operation?
+        ; TODO Not a problem right now, but what if there were multiple attachments and uploading only one of them would persistently fail?
+        ; TODO Should other uploaded attachments be explicitly removed? AFAIK, this is not even supported by the email service.
+        (let [send-msg-endpoint (url-helper :new-email-service.messages)
+              attachment-ids    (mapv #(upload-attachment! url-helper cas-client %) attachments)
+              {:keys [status body]} (cas/cas-authenticated-post cas-client send-msg-endpoint (email->message email attachment-ids))
+              response-body     (json/read-value body)]
+          (if (= 200 status)
+            ; TODO Store returned fields viestiTunniste and lahetysTunniste in DB?
+            response-body
+            (throw (ex-info "Error sending email!" {:status     status
+                                                    :body       response-body
+                                                    :message-id message-id
+                                                    ; TODO metadata isn't populated yet!
+                                                    :metadata   metadata}))))))))
+
+(defmethod ig/init-key ::email-client [_ {:keys [use-new-email-service? cas-client url-helper]}]
+  {:pre [(boolean? use-new-email-service?)
+         (or (not use-new-email-service?)
+             (some? cas-client))
+         (some? url-helper)]}
+  (if use-new-email-service?
+    (->NewEmailService url-helper (cas-client (url-helper :new-email-service.cas-login)))
+    (->OldEmailService url-helper)))

src/yki/config.clj

@@ -0,0 +1,3 @@
+(ns yki.config)
+
+(def oph-oid "1.2.246.562.10.00000000001")

src/yki/handler/exam_session.clj

@@ -10,7 +10,6 @@
     [ring.util.response :refer [bad-request not-found response]]
     [yki.boundary.exam-session-db :as exam-session-db]
     [yki.handler.routing :as routing]
-    [yki.middleware.auth :as auth]
     [yki.spec :as ys]
     [yki.util.audit-log :as audit-log]
     [yki.util.common :refer [string->date]]

src/yki/handler/login_link.clj

@@ -33,10 +33,14 @@
     (login-link-db/create-login-link! db (assoc login-link :code hashed))
     (log/info "Login link created for" email ". Adding to email queue")
     (pgq/put email-q
-             {:recipients [email]
-              :created    (System/currentTimeMillis)
-              :subject    (template-util/login-subject template-data)
-              :body       (template-util/render link-type lang template-data)})))
+             {:language         lang
+              :recipients       [{:email email}]
+              :created          (System/currentTimeMillis)
+              ; TODO Discuss if there is a need to retain login link messages for a longer or shorter period of time
+              :retention-period 7
+              :message-id       (random-uuid)
+              :subject          (template-util/login-subject template-data)
+              :body             (template-util/render link-type lang template-data)})))
 
 (defmethod ig/init-key :yki.handler/login-link [_ {:keys [db email-q url-helper access-log]}]
   {:pre [(some? db) (some? email-q) (some? url-helper) (some? access-log)]}

src/yki/job/scheduled_tasks.clj

@@ -34,8 +34,8 @@
                                        :interval  "1 DAY"})
 
 (defonce sync-onr-participant-data-handler-conf {:worker-id (str (UUID/randomUUID))
-                                                 :task "SYNC_ONR_PARTICIPANT_DATA_HANDLER"
-                                                 :interval "59 MINUTES"})
+                                                 :task      "SYNC_ONR_PARTICIPANT_DATA_HANDLER"
+                                                 :interval  "59 MINUTES"})
 
 (defn- take-with-error-handling
   "Takes message from queue and executes handler function with message.
@@ -91,14 +91,14 @@
        (log/error e "Participant sync handler failed"))))
 
 (defmethod ig/init-key ::email-queue-reader
-  [_ {:keys [email-q handle-at-once-at-most url-helper retry-duration-in-days disabled]}]
-  {:pre [(some? url-helper) (pos-int? handle-at-once-at-most) (some? email-q) (some? retry-duration-in-days)]}
+  [_ {:keys [email-q handle-at-once-at-most email-boundary retry-duration-in-days disabled]}]
+  {:pre [(some? email-boundary) (pos-int? handle-at-once-at-most) (some? email-q) (some? retry-duration-in-days)]}
   #(try
      (doseq [_ (range (min handle-at-once-at-most (pgq/count email-q)))]
        (take-with-error-handling email-q retry-duration-in-days
                                  (fn [email-req]
                                    (log/info "Email queue reader sending email to:" (:recipients email-req))
-                                   (email/send-email! url-helper email-req disabled))))
+                                   (email/send-email! email-boundary email-req disabled))))
      (catch Exception e
        (log/error e "Email queue reader failed"))))
 
@@ -120,6 +120,11 @@
          (doseq [exam-session exam-sessions-with-queue]
            (log/info "Exam session with queue and free space" exam-session)
            (try
+             ; TODO Instead of sending individual emails to each recipient in queue,
+             ; we could instead send out emails to multiple recipients at the same time.
+             ; This should work because there can be at most 50 people in queue for an individual exam session simultaneously,
+             ; while the (new) email service supports a maximum of 512 recipients per message.
+             ; Note that we'd still need to group the notifications by message language (fi, sv, en).
              (doseq [item (:queue exam-session)]
                (let [lang             (:lang item)
                      email            (:email item)
@@ -129,8 +134,10 @@
                      level            (template-util/get-level (:level_code exam-session) lang)]
                  (log/info "Sending notification to email" email)
                  (pgq/put email-q
-                          {:recipients [email]
+                          {:language   lang
+                           :recipients [{:email email}]
                            :created    (System/currentTimeMillis)
+                           :message-id (random-uuid)
                            :subject    (template-util/subject "queue" lang exam-session)
                            :body       (template-util/render
                                          "queue"
@@ -152,8 +159,8 @@
      (when (job-db/try-to-acquire-lock! db remove-old-data-handler-conf)
        (log/info "Old data removal started")
        (let [deleted-from-exam-session-queue (exam-session-db/remove-old-entries-from-exam-session-queue! db)
-             deleted-cas-tickets (cas-ticket-db/delete-old-tickets! db :virkailija)
-             deleted-cas-oppija-tickets (cas-ticket-db/delete-old-tickets! db :oppija)]
+             deleted-cas-tickets             (cas-ticket-db/delete-old-tickets! db :virkailija)
+             deleted-cas-oppija-tickets      (cas-ticket-db/delete-old-tickets! db :oppija)]
          (log/info "Removed old entries from exam-session-queue:" deleted-from-exam-session-queue)
          (log/info "Removed old CAS tickets:" deleted-cas-tickets)
          (log/info "Removed old CAS-oppija tickets:" deleted-cas-oppija-tickets)))

src/yki/middleware/auth.clj

@@ -9,12 +9,11 @@
     [ring.middleware.session :refer [wrap-session]]
     [ring.middleware.session.cookie :refer [cookie-store]]
     [ring.util.http-response :refer [found see-other]]
-    [yki.boundary.cas-ticket-db :as cas-ticket-db]))
+    [yki.boundary.cas-ticket-db :as cas-ticket-db]
+    [yki.config :refer [oph-oid]]))
 
 (def backend (session-backend))
 
-(def oph-oid "1.2.246.562.10.00000000001")
-
 (def admin-role "YLLAPITAJA")
 
 (def organizer-role "JARJESTAJA")