the api should work with unsaved pulses now

.clj-kondo/config.edn

@@ -162,7 +162,8 @@
     metabase.audit                 #{metabase.audit}
     metabase.bootstrap             #{metabase.bootstrap}
     metabase.cmd                   #{}                                                              ; there are no namespaces here since you shouldn't be using this in any other module.
-    metabase.channel               #{metabase.channel.core}
+    metabase.channel               #{metabase.channel.core
+                                     metabase.channel.shared}
     metabase.config                #{metabase.config}
     metabase.core                  #{metabase.core.initialization-status}                           ; TODO -- only namespace used outside of EE, this probably belongs in `metabase.server` anyway since that's the only place it's used.
     metabase.db                    #{metabase.db

src/metabase/api/pulse.clj

@@ -8,6 +8,7 @@
    [metabase.api.alert :as api.alert]
    [metabase.api.common :as api]
    [metabase.api.common.validation :as validation]
+   [metabase.channel.shared :as channel.shared]
    [metabase.config :as config]
    [metabase.email :as email]
    [metabase.events :as events]
@@ -275,7 +276,7 @@
                [:body {:style "margin: 0;"}
                 (binding [render/*include-title*   true
                           render/*include-buttons* true]
-                  (render/render-pulse-card-for-display (metabase.pulse/defaulted-timezone card) card result))]])}))
+                  (render/render-pulse-card-for-display (channel.shared/defaulted-timezone card) card result))]])}))
 
 (api/defendpoint GET "/preview_dashboard/:id"
   "Get HTML rendering of a Dashboard with `id`.
@@ -307,7 +308,7 @@
         data      (:data result)
         card-type (render/detect-pulse-chart-type card nil data)
         card-html (html (binding [render/*include-title* true]
-                          (render/render-pulse-card-for-display (metabase.pulse/defaulted-timezone card) card result)))]
+                          (render/render-pulse-card-for-display (channel.shared/defaulted-timezone card) card result)))]
     {:id              id
      :pulse_card_type card-type
      :pulse_card_html card-html
@@ -325,7 +326,7 @@
   (let [card   (api/read-check Card id)
         result (pulse-card-query-results card)
         ba     (binding [render/*include-title* true]
-                 (render/render-pulse-card-to-png (metabase.pulse/defaulted-timezone card) card result preview-card-width))]
+                 (render/render-pulse-card-to-png (channel.shared/defaulted-timezone card) card result preview-card-width))]
     {:status 200, :headers {"Content-Type" "image/png"}, :body (ByteArrayInputStream. ba)}))
 
 (api/defendpoint POST "/test"

src/metabase/channel/core.clj

@@ -21,7 +21,7 @@
   {:added    "0.51.0"
    :arglists '([channel-type notification-content recipients])}
   (fn [channel-type notification-content _recipients]
-    [channel-type (:payload-type notification-content)]))
+    [channel-type (:notification_type notification-content)]))
 
 (defmulti send!
   "Send a message to a channel."

src/metabase/channel/email.clj

@@ -55,8 +55,11 @@
                                  :bcc?         (email/bcc-enabled?)}))
 
 (mu/defmethod channel/render-notification [:channel/email :notification/alert] :- [:sequential EmailMessage]
-  [_channel-type {:keys [card pulse payload channel]} recipients]
-  (let [condition-kwd             (messages/pulse->alert-condition-kwd pulse)
+  [_channel-type {:keys [payload context]} recipients]
+  (let [{:keys [card
+                pulse
+                trigger]}         context
+        condition-kwd             (messages/pulse->alert-condition-kwd pulse)
         email-subject             (case condition-kwd
                                     :meets (trs "Alert: {0} has reached its goal" (:name card))
                                     :below (trs "Alert: {0} has gone below its goal" (:name card))
@@ -68,14 +71,14 @@
         email-to-users            (when (> (count user-emails) 0)
                                     (construct-pulse-email
                                      email-subject user-emails
-                                     (messages/render-alert-email timezone pulse channel
+                                     (messages/render-alert-email timezone pulse trigger
                                                                   [payload]
                                                                   goal
                                                                   nil)))
         email-to-nonusers         (for [non-user-email non-user-emails]
                                     (construct-pulse-email
                                      email-subject [non-user-email]
-                                     (messages/render-alert-email timezone pulse channel
+                                     (messages/render-alert-email timezone pulse trigger
                                                                   [payload]
                                                                   goal
                                                                   non-user-email)))]
@@ -86,8 +89,10 @@
 ;; ------------------------------------------------------------------------------------------------;;
 
 (mu/defmethod channel/render-notification [:channel/email :notification/dashboard-subscription] :- [:sequential EmailMessage]
-  [_channel-details {:keys [dashboard payload pulse]} recipients]
-  (let [{:keys [user-emails
+  [_channel-details {:keys [payload context]} recipients]
+  (let [{dashsub   :pulse
+         dashboard :dashboard}       context
+        {:keys [user-emails
                 non-user-emails]} (recipients->emails recipients)
         timezone                  (some->> payload (some :card) channel.shared/defaulted-timezone)
         email-subject             (:name dashboard)
@@ -95,10 +100,10 @@
                                     (construct-pulse-email
                                      email-subject
                                      user-emails
-                                     (messages/render-pulse-email timezone pulse dashboard payload nil)))
+                                     (messages/render-pulse-email timezone dashsub dashboard payload nil)))
         email-to-nonusers         (for [non-user-email non-user-emails]
                                     (construct-pulse-email
                                      email-subject
                                      [non-user-email]
-                                     (messages/render-pulse-email timezone pulse dashboard payload non-user-email)))]
+                                     (messages/render-pulse-email timezone dashsub dashboard payload non-user-email)))]
     (filter some? (conj email-to-nonusers email-to-users))))

src/metabase/channel/slack.clj

@@ -96,10 +96,10 @@
 ;; ------------------------------------------------------------------------------------------------;;
 
 (mu/defmethod channel/render-notification [:channel/slack :notification/alert] :- [:sequential SlackMessage]
-  [_channel-details {:keys [payload card]} channel-ids]
+  [_channel-details {:keys [payload context]} channel-ids]
   (let [attachments [{:blocks [{:type "header"
                                 :text {:type "plain_text"
-                                       :text (str "🔔 " (:name card))
+                                       :text (str "🔔 " (get-in context [:card :name]))
                                        :emoji true}}]}
                      (payload->attachment-data payload (slack/files-channel))]]
     (for [channel-id channel-ids]
@@ -158,10 +158,12 @@
       attachment)))
 
 (mu/defmethod channel/render-notification [:channel/slack :notification/dashboard-subscription] :- [:sequential SlackMessage]
-  [_channel-type {:keys [payload dashboard pulse]} channel-ids]
-  (for [channel-id channel-ids]
-    {:channel-id  channel-id
-     :attachments (remove nil?
-                          (flatten [(slack-dashboard-header pulse dashboard)
-                                    (create-slack-attachment-data payload)
-                                    (slack-dashboard-footer pulse dashboard)]))}))
+  [_channel-type {:keys [payload context]} channel-ids]
+  (let [{dashsub :pulse
+         dashboard :dashboard}           context]
+    (for [channel-id channel-ids]
+      {:channel-id  channel-id
+       :attachments (remove nil?
+                            (flatten [(slack-dashboard-header dashsub dashboard)
+                                      (create-slack-attachment-data payload)
+                                      (slack-dashboard-footer dashsub dashboard)]))})))

src/metabase/notification/core.clj

@@ -0,0 +1,277 @@
+(ns metabase.notification.core
+  "Notifications do:
+  - execute notificatino based on its type
+  - deliver the message to all of its channel"
+  (:require
+   [metabase.events :as events]
+   [metabase.channel.core :as channel]
+   [metabase.notification.execute :as noti.execute]
+   [metabase.util.i18n :refer [tru]]
+   [metabase.util.log :as log]
+   [metabase.util.malli :as mu]
+   [metabase.util.malli.schema :as ms]
+   [metabase.util.ui-logic :as ui-logic]
+   [metabase.util.retry :as retry]
+   [toucan2.core :as t2]
+   [metabase.notification.core :as noti]))
+
+(def ^:private payload-types #{:notification/alert :notification/dashboard-subscription})
+
+(def ^:private Notification
+  "Saved notification"
+  [:map
+   [:notification_type (into [:enum ] payload-types)]
+   [:notification_id   pos-int?]
+   [:creator_id   pos-int?]])
+
+(def ^:private HydratedNotification
+  [:map
+   [:notification_type             :keyword]
+   [:pulse                         :map]
+   [:creator_id                    pos-int?]
+   [:notification_id   {:optional true} [:maybe pos-int?]]
+   ;; alert only
+   [:card_id      {:optional true} pos-int?]
+   [:trigger      {:optional true} :map]
+   ;; dashboard subscription only
+   [:dashboard_id {:optional true} pos-int?]])
+
+(def ^:private NotificationRecipients
+  [:multi {:dispatch :channel_type}
+   [:channel/email [:map [:recipients [:sequential [:multi {:dispatch :kind}
+                                                     [:user           :map]
+                                                     [:external-email ms/Email]]]]]]
+   [:channel/slack [:map [:recipients [:sequential :string]]]]
+   [:else :any]])
+
+(mu/defn ^:private notification->channel+recipients :- [:sequential NotificationRecipients]
+  [notification pc-ids]
+  (let [pcs (if (some? pc-ids)
+              (t2/select :model/PulseChannel :pulse_id (:notification_id notification)
+                         :enabled true
+                         :id [:in pc-ids])
+              (t2/select :model/PulseChannel :pulse_id (:notification_id notification) :enabled true))]
+    (for [pc pcs]
+      (let [channel-type (keyword "channel" (name (:channel_type pc)))]
+        {:channel_type  channel-type
+         :recipients    (if (= :channel/email channel-type)
+                          (concat (map (fn [user]
+                                         {:kind :user
+                                          :user user})
+                                       ;; TODO: n+1 here
+                                       (t2/select :model/User
+                                                  {:left-join [[:pulse_channel_recipient :pcr] [:= :core_user.id :pcr.user_id]
+                                                               [:pulse_channel :pc] [:= :pc.id :pcr.pulse_channel_id]
+                                                               [:pulse :p] [:= :p.id :pc.pulse_id]]
+                                                   :where     [:and
+                                                               [:= :p.id (:notification_id notification)]
+                                                               [:= :core_user.is_active true]]
+                                                   :order-by [[:core_user.id :asc]]}))
+                                  ;; non-user-email
+                                  (map (fn [email] {:recipient email
+                                                    :kind      :external-email}) (get-in pc [:recipients :emails])))
+                          [(get-in pc [:details :channel])])}))))
+
+
+;; ------------------------------------------------------------------------------------------------;;
+;;                                        Multimethods                                             ;;
+;; ------------------------------------------------------------------------------------------------;;
+
+(defmulti ^:private execute-notification
+  :notification_type)
+
+(defmulti ^:private should-send-notification?
+  "Returns true if given the pulse type and resultset a new notification (pulse or alert) should be sent"
+  :notification_type)
+
+;; ------------------------------------------------------------------------------------------------;;
+;;                                           Alerts                                                ;;
+;; ------------------------------------------------------------------------------------------------;;
+
+(mu/defn ^:private hydrate-notification :- HydratedNotification
+  [notification :- [:or Notification HydratedNotification]]
+  (let [pulse (or (:pulse notification) (t2/select-one :model/Pulse (:notification_id notification)))]
+    (cond-> notification
+      (not (contains? notification :pulse))
+      (assoc :pulse pulse)
+
+      (and (= :notification/alert (:notification_type notification))
+           (not (contains? notification :card)))
+      ;; there should be only one pulse card for alert
+      (assoc :card_id (t2/select-one-fn :id :model/PulseCard :pulse_id (:id pulse)))
+
+      (and (= :notification/dashboard-subscription (:notification_type notification))
+           (not (contains? notification :dashboard)))
+      (assoc :dashboard_id (:dashboard_id pulse))
+
+      (and (= :notification/alert (:notification_type notification))
+           (not (contains? notification :trigger)))
+      (assoc :trigger (t2/select-one-fn
+                       ;; we use these for alert email. see [[metabase.email.messages/alert-context]]
+                       ;; currently notification's triggere is tied to pulseChannel, but should it'll be 2 different
+                       ;; concept and these info will come directly from the trigger (scheduler) so we don't have to query
+                       ;; it here
+                       #(select-keys % [:schedule_type :schedule_hour :schedule_day :schedule_frame])
+                       :model/PulseChannel :pulse_id  (:id pulse) :enabled true)))))
+
+(def ^:private AlertPayload
+  [:map {:closed true}
+   [:notification_type [:= :notification/alert]]
+   [:context           [:map
+                        [:card    :map]
+                        [:trigger :map]
+                        [:pulse   :map]]]
+   [:payload           [:map
+                        [:type [:= :card]]
+                        [:card :map]
+                        [:result :map]]]])
+
+(mu/defmethod execute-notification :notification/alert :- AlertPayload
+  [{:keys [creator_id] :as hydated-noti} :- HydratedNotification]
+  (let [card (t2/select-one :model/Card (:card_id hydated-noti))]
+   {:notification_type :notification/alert
+    :payload      (noti.execute/execute-card creator_id card)
+    :context      {:pulse   (:pulse hydated-noti)
+                   :trigger (:trigger hydated-noti)
+                   :card    card}}))
+
+(defn- is-card-empty?
+  "Check if the part is empty"
+  [part]
+  (if-let [result (:result part)]
+    (or (zero? (-> result :row_count))
+        ;; Many aggregations result in [[nil]] if there are no rows to aggregate after filters
+        (= [[nil]]
+           (-> result :data :rows)))
+    ;; Text cards have no result; treat as empty
+    true))
+
+(defn- goal-met? [{:keys [alert_above_goal], :as pulse} [first-result]]
+  (let [goal-comparison      (if alert_above_goal >= <)
+        goal-val             (ui-logic/find-goal-value first-result)
+        comparison-col-rowfn (ui-logic/make-goal-comparison-rowfn (:card first-result)
+                                                            (get-in first-result [:result :data]))]
+
+    (when-not (and goal-val comparison-col-rowfn)
+      (throw (ex-info (tru "Unable to compare results to goal for alert.")
+                      {:pulse  pulse
+                       :result first-result})))
+    (boolean
+     (some (fn [row]
+             (goal-comparison (comparison-col-rowfn row) goal-val))
+           (get-in first-result [:result :data :rows])))))
+
+(defmethod should-send-notification? :notification/alert
+  [payload]
+  (let [alert           (get-in payload [:context :pulse])
+        alert-condition (:alert_condition alert)]
+    (cond
+      (= "rows" alert-condition)
+      (not (is-card-empty? (:payload payload)))
+
+      (= "goal" alert-condition)
+      (goal-met? alert (:payload payload))
+
+      :else
+      (let [^String error-text (tru "Unrecognized alert with condition ''{0}''" alert-condition)]
+        (throw (IllegalArgumentException. error-text))))))
+
+;; ------------------------------------------------------------------------------------------------;;
+;;                                    Dashboard Subscriptions                                      ;;
+;; ------------------------------------------------------------------------------------------------;;
+
+(def ^:private DashboardSubscriptionPayload
+  [:map {:closed true}
+   [:notification_type [:= :notification/dashboard-subscription]]
+   [:context      [:map {:closed true}
+                   [:pulse :map]
+                   [:dashboard :map]]]
+   [:payload      [:sequential :any]]])
+
+(mu/defmethod execute-notification :notification/dashboard-subscription :- DashboardSubscriptionPayload
+  [hydrated-noti :- HydratedNotification]
+  (let [dashboard (t2/select-one :model/Dashboard (:dashboard_id hydrated-noti))
+        payload   (noti.execute/execute-dashboard (:pulse hydrated-noti) dashboard)]
+    {:notification_type :notification/dashboard-subscription
+     :payload      payload
+     :context      {:pulse     (:pulse hydrated-noti)
+                    :dashboard dashboard}}))
+
+(defn- are-all-parts-empty?
+  "Do none of the cards have any results?"
+  [results]
+  (every? is-card-empty? results))
+
+(defmethod should-send-notification? :notification/dashboard-subscription
+  [payload]
+  (cond
+   (get-in payload [:context :dashboard :archived]) false
+   (get-in payload [:context :pulse :skip_if_empty]) (not (are-all-parts-empty? (:payload payload)))
+   :else true))
+
+;; ------------------------------------------------------------------------------------------------;;
+;;                                           Public APIs                                           ;;
+;; ------------------------------------------------------------------------------------------------;;
+
+(def ^:private payload-type->event-type
+  {:notification/alert                  :event/alert-send
+   :notification/dashboard-subscription :event/subscription-send})
+
+(defn- channel-send!
+  [& args]
+  (try
+    (apply channel/send! args)
+    (catch Exception e
+      ;; Token errors have already been logged and we should not retry.
+      (when-not (and (= :channel/slack (first args))
+                     (contains? (:errors (ex-data e)) :slack-token))
+        (throw e)))))
+
+(defn- send-retrying!
+  [& args]
+  (try
+    (apply (retry/decorate channel-send!) args)
+    (catch Throwable e
+      (log/error e "Error sending notification!"))))
+
+(mu/defn send-notification!
+  "Send the notification."
+  ([notification         :- [:or Notification HydratedNotification]
+    &
+    pc-ids-or-recipients #_:- #_[:maybe [:sequential [:or pos-int? NotificationRecipients]]]]
+   (let [notification           (hydrate-notification notification)
+         channel+recipients     (if (and (seq pc-ids-or-recipients) (map? (first pc-ids-or-recipients)))
+                                  pc-ids-or-recipients
+                                  (notification->channel+recipients notification pc-ids-or-recipients))
+         payload                (execute-notification notification)
+         payload-type           (:notification_type payload)
+         notification-info      (if (= :notification/alert payload-type)
+                                  (get-in payload [:context :alert])
+                                  (get-in payload [:context :dashboard-subscription]))]
+     (if (should-send-notification? payload)
+       (do
+        (events/publish-event! (get payload-type->event-type payload-type)
+                               {:id      (:notification_id notification)
+                                :user-id (:creator_id notification)
+                                :object  {:recipients (map :recipients channel+recipients)
+                                          :filters    (:parameters notification-info)}})
+        (doseq [channel channel+recipients]
+          (try
+            (log/infof "Sending notification %d to channel: %s" (:notification_id notification) channel)
+            (doseq [message (channel/render-notification (:channel_type channel)
+                                                         payload
+                                                         (:recipients channel))]
+              (send-retrying! (:channel_type channel) message))
+            (catch Exception e
+              (log/errorf e "Error sending %s %d to channel %s"
+                          (:notification_type notification) (:notification_id notification) (:channel_type channel)))))
+        (when (and (= :notification/alert payload-type)
+                   (:alert_first_only notification-info))
+          (t2/delete! :model/Pulse (:id notification-info))))
+       (log/infof "Skipping notification %s" (select-keys notification [:notification_id :notification_type]))))))
+
+(comment
+ (ngoc/with-tc
+   (send-notification! {:notification_type :notification/dashboard-subscription
+                        :notification_id   2
+                        :creator_id 2})))

src/metabase/notification/execute.clj

@@ -0,0 +1,239 @@
+(ns metabase.notification.execute
+  (:require
+   [metabase.api.common :as api]
+   [metabase.models.dashboard :as dashboard]
+   [metabase.models.dashboard-card :as dashboard-card]
+   [metabase.models.interface :as mi]
+   [metabase.models.serialization :as serdes]
+   [metabase.pulse.parameters :as pulse-params]
+   [metabase.query-processor :as qp]
+   [metabase.query-processor.dashboard :as qp.dashboard]
+   [metabase.query-processor.middleware.permissions :as qp.perms]
+   [metabase.server.middleware.session :as mw.session]
+   [metabase.shared.parameters.parameters :as shared.params]
+   [metabase.util :as u]
+   [metabase.util.log :as log]
+   [metabase.util.urls :as urls]
+   [toucan2.core :as t2]))
+
+(defn execute-card
+  "Execute the query for a single Card with [[user-perm-id]]'s permissions."
+  [user-perm-id card]
+  ;; The Card must either be executed in the context of a User
+  {:pre [(integer? user-perm-id)]}
+  (let [card-id (:id card)]
+    (try
+      (when-let [{query     :dataset_query
+                  metadata  :result_metadata
+                  card-type :type
+                  :as       card} (t2/select-one :model/Card :id card-id, :archived false)]
+        (let [query         (assoc query :async? false)
+              process-query (fn []
+                              (binding [qp.perms/*card-id* card-id]
+                                (qp/process-query
+                                 (qp/userland-query-with-default-constraints
+                                  (assoc query :middleware {:skip-results-metadata? true
+                                                            :process-viz-settings?  true
+                                                            :js-int-to-string?      false})
+                                  (cond-> {:executed-by user-perm-id
+                                                  :context     :pulse
+                                                  :card-id     card-id}
+                                          (= card-type :model)
+                                          (assoc :metadata/model-metadata metadata))))))
+              result        (if user-perm-id
+                              (mw.session/with-current-user user-perm-id
+                                (process-query))
+                              (process-query))]
+          {:card   card
+           :result result
+           :type   :card}))
+      (catch Throwable e
+        (log/warnf e "Error running query for Card %s" card-id)))))
+
+(defn- is-card-empty?
+  "Check if the card is empty"
+  [card]
+  (if-let [result (:result card)]
+    (or (zero? (-> result :row_count))
+        ;; Many aggregations result in [[nil]] if there are no rows to aggregate after filters
+        (= [[nil]]
+           (-> result :data :rows)))
+    ;; Text cards have no result; treat as empty
+    true))
+
+(defn execute-dashboard-subscription-card
+  "Returns subscription result for a card.
+
+  This function should be executed under pulse's creator permissions."
+  [dashcard parameters]
+  (try
+    (let [{card-id      :card_id
+           dashboard-id :dashboard_id} dashcard
+          card                         (t2/select-one :model/Card :id card-id)
+          multi-cards                  (dashboard-card/dashcard->multi-cards dashcard)
+          result-fn                    (fn [card-id]
+                                         {:card     (if (= card-id (:id card))
+                                                      card
+                                                      (t2/select-one :model/Card :id card-id))
+                                          :dashcard dashcard
+                                          :type     :card
+                                          :result   (qp.dashboard/process-query-for-dashcard
+                                                      :dashboard-id  dashboard-id
+                                                      :card-id       card-id
+                                                      :dashcard-id   (u/the-id dashcard)
+                                                      :context       :dashboard-subscription
+                                                      :export-format :api
+                                                      :parameters    parameters
+                                                      :middleware    {:process-viz-settings? true
+                                                                      :js-int-to-string?     false}
+                                                      :make-run      (fn make-run [qp _export-format]
+                                                                       (^:once fn* [query info]
+                                                                               (qp
+                                                                                 (qp/userland-query-with-default-constraints query info)
+                                                                                 nil))))})
+          result                       (result-fn card-id)
+          series-results               (map (comp result-fn :id) multi-cards)]
+      (when-not (and (get-in dashcard [:visualization_settings :card.hide_empty])
+                     (is-card-empty? (assoc card :result (:result result))))
+        (update result :dashcard assoc :series-results series-results)))
+    (catch Throwable e
+      (log/warnf e "Error running query for Card %s" (:card_id dashcard)))))
+
+
+(defn- merge-default-values
+  "For the specific case of Dashboard Subscriptions we should use `:default` parameter values as the actual `:value` for
+  the parameter if none is specified. Normally the FE client will take `:default` and pass it in as `:value` if it
+  wants to use it (see #20503 for more details) but this obviously isn't an option for Dashboard Subscriptions... so
+  go thru `parameters` and change `:default` to `:value` unless a `:value` is explicitly specified."
+  [parameters]
+  (for [{default-value :default, :as parameter} parameters]
+    (merge
+     (when default-value
+       {:value default-value})
+     (dissoc parameter :default))))
+
+(defn virtual-card-of-type?
+  "Check if dashcard is a virtual with type `ttype`, if `true` returns the dashcard, else returns `nil`.
+
+  There are currently 4 types of virtual card: \"text\", \"action\", \"link\", \"placeholder\"."
+  [dashcard ttype]
+  (when (= ttype (get-in dashcard [:visualization_settings :virtual_card :display]))
+    dashcard))
+
+(defn- link-card-entity->url
+  [{:keys [db_id id model] :as _entity}]
+  (case model
+    "card"       (urls/card-url id)
+    "dataset"    (urls/card-url id)
+    "collection" (urls/collection-url id)
+    "dashboard"  (urls/dashboard-url id)
+    "database"   (urls/database-url id)
+    "table"      (urls/table-url db_id id)))
+
+(defn- link-card->text-part
+  [{:keys [entity url] :as _link-card}]
+  (let [url-link-card? (some? url)]
+    {:text (str (format
+                  "### [%s](%s)"
+                  (if url-link-card? url (:name entity))
+                  (if url-link-card? url (link-card-entity->url entity)))
+                (when-let [description (if url-link-card? nil (:description entity))]
+                  (format "\n%s" description)))
+     :type :text}))
+
+(defn- dashcard-link-card->part
+  "Convert a dashcard that is a link card to pulse part.
+
+  This function should be executed under pulse's creator permissions."
+  [dashcard]
+  (assert api/*current-user-id* "Makes sure you wrapped this with a `with-current-user`.")
+  (let [link-card (get-in dashcard [:visualization_settings :link])]
+    (cond
+      (some? (:url link-card))
+      (link-card->text-part link-card)
+
+      ;; if link card link to an entity, update the setting because
+      ;; the info in viz-settings might be out-of-date
+      (some? (:entity link-card))
+      (let [{:keys [model id]} (:entity link-card)
+            instance           (t2/select-one
+                                 (serdes/link-card-model->toucan-model model)
+                                 (dashboard-card/link-card-info-query-for-model model id))]
+        (when (mi/can-read? instance)
+          (link-card->text-part (assoc link-card :entity instance)))))))
+
+(defn- escape-heading-markdown
+  [dashcard]
+  (if (= "heading" (get-in dashcard [:visualization_settings :virtual_card :display]))
+    ;; If there's no heading text, the heading is empty, so we return nil.
+    (when (get-in dashcard [:visualization_settings :text])
+      (update-in dashcard [:visualization_settings :text]
+                 #(str "## " (shared.params/escape-chars % shared.params/escaped-chars-regex))))
+    dashcard))
+
+(defn- dashcard->part
+  "Given a dashcard returns its part based on its type.
+
+  The result will follow the pulse's creator permissions."
+  [dashcard pulse dashboard]
+  (assert api/*current-user-id* "Makes sure you wrapped this with a `with-current-user`.")
+  (cond
+    (:card_id dashcard)
+    (let [parameters (merge-default-values (pulse-params/parameters pulse dashboard))]
+      (execute-dashboard-subscription-card dashcard parameters))
+
+    ;; actions
+    (virtual-card-of-type? dashcard "action")
+    nil
+
+    ;; link cards
+    (virtual-card-of-type? dashcard "link")
+    (dashcard-link-card->part dashcard)
+
+    ;; placeholder cards aren't displayed
+    (virtual-card-of-type? dashcard "placeholder")
+    nil
+
+    ;; text cards have existed for a while and I'm not sure if all existing text cards
+    ;; will have virtual_card.display = "text", so assume everything else is a text card
+    :else
+    (let [parameters (merge-default-values (pulse-params/parameters pulse dashboard))]
+      (some-> dashcard
+              (pulse-params/process-virtual-dashcard parameters)
+              escape-heading-markdown
+              :visualization_settings
+              (assoc :type :text)))))
+
+(defn- dashcards->part
+  [dashcards pulse dashboard]
+  (let [ordered-dashcards (sort dashboard-card/dashcard-comparator dashcards)]
+    (doall (for [dashcard ordered-dashcards
+                 :let     [part (dashcard->part dashcard pulse dashboard)]
+                 :when    (some? part)]
+             part))))
+
+(defn- tab->part
+  [{:keys [name]}]
+  {:text name
+   :type :tab-title})
+
+(defn execute-dashboard
+  "Fetch all the dashcards in a dashboard for a Pulse, and execute non-text cards.
+
+  The generated parts will follow the pulse's creator permissions."
+  [dashsub dashboard]
+  (let [dashboard-id (u/the-id dashboard)]
+    (mw.session/with-current-user (:creator_id dashsub)
+      (let [parts (if (dashboard/has-tabs? dashboard)
+                    (let [tabs-with-cards (t2/hydrate (t2/select :model/DashboardTab :dashboard_id dashboard-id) :tab-cards)]
+                      (doall (flatten (for [{:keys [cards] :as tab} tabs-with-cards]
+                                        (concat [(tab->part tab)] (dashcards->part cards dashsub dashboard))))))
+                    (dashcards->part (t2/select :model/DashboardCard :dashboard_id dashboard-id) dashsub dashboard))]
+        (if (:skip_if_empty dashsub)
+          ;; Remove cards that have no results when empty results aren't wanted
+          (remove (fn [{part-type :type :as part}]
+                    (and
+                      (= part-type :card)
+                      (zero? (get-in part [:result :row_count] 0))))
+                  parts)
+          parts)))))

src/metabase/pulse.clj

@@ -1,321 +1,10 @@
 (ns metabase.pulse
   "Public API for sending Pulses."
   (:require
-   [metabase.api.common :as api]
-   [metabase.channel.core :as channel]
-   [metabase.events :as events]
-   [metabase.models.dashboard :as dashboard :refer [Dashboard]]
-   [metabase.models.dashboard-card :as dashboard-card]
-   [metabase.models.database :refer [Database]]
-   [metabase.models.interface :as mi]
-   [metabase.models.pulse :as pulse :refer [Pulse]]
-   [metabase.models.serialization :as serdes]
-   [metabase.pulse.parameters :as pulse-params]
-   [metabase.pulse.util :as pu]
-   [metabase.query-processor.timezone :as qp.timezone]
-   [metabase.server.middleware.session :as mw.session]
-   [metabase.shared.parameters.parameters :as shared.params]
-   [metabase.util :as u]
-   [metabase.util.i18n :refer [tru]]
-   [metabase.util.log :as log]
-   [metabase.util.malli :as mu]
-   [metabase.util.malli.schema :as ms]
-   [metabase.util.retry :as retry]
-   [metabase.util.ui-logic :as ui-logic]
-   [metabase.util.urls :as urls]
-   [toucan2.core :as t2]))
+   [metabase.notification.core :as noti]))
 
 (set! *warn-on-reflection* true)
 
-;;; ------------------------------------------------- PULSE SENDING --------------------------------------------------
-
-(defn- merge-default-values
-  "For the specific case of Dashboard Subscriptions we should use `:default` parameter values as the actual `:value` for
-  the parameter if none is specified. Normally the FE client will take `:default` and pass it in as `:value` if it
-  wants to use it (see #20503 for more details) but this obviously isn't an option for Dashboard Subscriptions... so
-  go thru `parameters` and change `:default` to `:value` unless a `:value` is explicitly specified."
-  [parameters]
-  (for [{default-value :default, :as parameter} parameters]
-    (merge
-     (when default-value
-       {:value default-value})
-     (dissoc parameter :default))))
-
-
-(defn virtual-card-of-type?
-  "Check if dashcard is a virtual with type `ttype`, if `true` returns the dashcard, else returns `nil`.
-
-  There are currently 4 types of virtual card: \"text\", \"action\", \"link\", \"placeholder\"."
-  [dashcard ttype]
-  (when (= ttype (get-in dashcard [:visualization_settings :virtual_card :display]))
-    dashcard))
-
-(defn- link-card-entity->url
-  [{:keys [db_id id model] :as _entity}]
-  (case model
-    "card"       (urls/card-url id)
-    "dataset"    (urls/card-url id)
-    "collection" (urls/collection-url id)
-    "dashboard"  (urls/dashboard-url id)
-    "database"   (urls/database-url id)
-    "table"      (urls/table-url db_id id)))
-
-(defn- link-card->text-part
-  [{:keys [entity url] :as _link-card}]
-  (let [url-link-card? (some? url)]
-    {:text (str (format
-                  "### [%s](%s)"
-                  (if url-link-card? url (:name entity))
-                  (if url-link-card? url (link-card-entity->url entity)))
-                (when-let [description (if url-link-card? nil (:description entity))]
-                  (format "\n%s" description)))
-     :type :text}))
-
-(defn- dashcard-link-card->part
-  "Convert a dashcard that is a link card to pulse part.
-
-  This function should be executed under pulse's creator permissions."
-  [dashcard]
-  (assert api/*current-user-id* "Makes sure you wrapped this with a `with-current-user`.")
-  (let [link-card (get-in dashcard [:visualization_settings :link])]
-    (cond
-      (some? (:url link-card))
-      (link-card->text-part link-card)
-
-      ;; if link card link to an entity, update the setting because
-      ;; the info in viz-settings might be out-of-date
-      (some? (:entity link-card))
-      (let [{:keys [model id]} (:entity link-card)
-            instance           (t2/select-one
-                                 (serdes/link-card-model->toucan-model model)
-                                 (dashboard-card/link-card-info-query-for-model model id))]
-        (when (mi/can-read? instance)
-          (link-card->text-part (assoc link-card :entity instance)))))))
-
-(defn- escape-heading-markdown
-  [dashcard]
-  (if (= "heading" (get-in dashcard [:visualization_settings :virtual_card :display]))
-    ;; If there's no heading text, the heading is empty, so we return nil.
-    (when (get-in dashcard [:visualization_settings :text])
-      (update-in dashcard [:visualization_settings :text]
-                 #(str "## " (shared.params/escape-chars % shared.params/escaped-chars-regex))))
-    dashcard))
-
-(defn- dashcard->part
-  "Given a dashcard returns its part based on its type.
-
-  The result will follow the pulse's creator permissions."
-  [dashcard pulse dashboard]
-  (assert api/*current-user-id* "Makes sure you wrapped this with a `with-current-user`.")
-  (cond
-    (:card_id dashcard)
-    (let [parameters (merge-default-values (pulse-params/parameters pulse dashboard))]
-      (pu/execute-dashboard-subscription-card dashcard parameters))
-
-    ;; actions
-    (virtual-card-of-type? dashcard "action")
-    nil
-
-    ;; link cards
-    (virtual-card-of-type? dashcard "link")
-    (dashcard-link-card->part dashcard)
-
-    ;; placeholder cards aren't displayed
-    (virtual-card-of-type? dashcard "placeholder")
-    nil
-
-    ;; text cards have existed for a while and I'm not sure if all existing text cards
-    ;; will have virtual_card.display = "text", so assume everything else is a text card
-    :else
-    (let [parameters (merge-default-values (pulse-params/parameters pulse dashboard))]
-      (some-> dashcard
-              (pulse-params/process-virtual-dashcard parameters)
-              escape-heading-markdown
-              :visualization_settings
-              (assoc :type :text)))))
-
-(defn- dashcards->part
-  [dashcards pulse dashboard]
-  (let [ordered-dashcards (sort dashboard-card/dashcard-comparator dashcards)]
-    (doall (for [dashcard ordered-dashcards
-                 :let     [part (dashcard->part dashcard pulse dashboard)]
-                 :when    (some? part)]
-             part))))
-
-(defn- tab->part
-  [{:keys [name]}]
-  {:text name
-   :type :tab-title})
-
-(defn- execute-dashboard
-  "Fetch all the dashcards in a dashboard for a Pulse, and execute non-text cards.
-
-  The generated parts will follow the pulse's creator permissions."
-  [{:keys [skip_if_empty] pulse-creator-id :creator_id :as pulse} dashboard & {:as _options}]
-  (let [dashboard-id (u/the-id dashboard)]
-    (mw.session/with-current-user pulse-creator-id
-      (let [parts (if (dashboard/has-tabs? dashboard)
-                    (let [tabs-with-cards (t2/hydrate (t2/select :model/DashboardTab :dashboard_id dashboard-id) :tab-cards)]
-                      (doall (flatten (for [{:keys [cards] :as tab} tabs-with-cards]
-                                        (concat [(tab->part tab)] (dashcards->part cards pulse dashboard))))))
-                    (dashcards->part (t2/select :model/DashboardCard :dashboard_id dashboard-id) pulse dashboard))]
-        (if skip_if_empty
-          ;; Remove cards that have no results when empty results aren't wanted
-          (remove (fn [{part-type :type :as part}]
-                    (and
-                      (= part-type :card)
-                      (zero? (get-in part [:result :row_count] 0))))
-                  parts)
-          parts)))))
-
-(defn- database-id [card]
-  (or (:database_id card)
-      (get-in card [:dataset_query :database])))
-
-(mu/defn defaulted-timezone :- :string
-  "Returns the timezone ID for the given `card`. Either the report timezone (if applicable) or the JVM timezone."
-  [card :- (ms/InstanceOf :model/Card)]
-  (or (some->> card database-id (t2/select-one Database :id) qp.timezone/results-timezone-id)
-      (qp.timezone/system-timezone-id)))
-
-(defn- are-all-parts-empty?
-  "Do none of the cards have any results?"
-  [results]
-  (every? pu/is-card-empty? results))
-
-(defn- goal-met? [{:keys [alert_above_goal], :as pulse} [first-result]]
-  (let [goal-comparison      (if alert_above_goal >= <)
-        goal-val             (ui-logic/find-goal-value first-result)
-        comparison-col-rowfn (ui-logic/make-goal-comparison-rowfn (:card first-result)
-                                                            (get-in first-result [:result :data]))]
-
-    (when-not (and goal-val comparison-col-rowfn)
-      (throw (ex-info (tru "Unable to compare results to goal for alert.")
-                      {:pulse  pulse
-                       :result first-result})))
-    (boolean
-     (some (fn [row]
-             (goal-comparison (comparison-col-rowfn row) goal-val))
-           (get-in first-result [:result :data :rows])))))
-
-
-;;; +----------------------------------------------------------------------------------------------------------------+
-;;; |                                         Creating Notifications To Send                                         |
-;;; +----------------------------------------------------------------------------------------------------------------+
-
-(defn- alert-or-pulse [pulse]
-  (if (:dashboard_id pulse)
-    :pulse
-    :alert))
-
-(defmulti ^:private should-send-notification?
-  "Returns true if given the pulse type and resultset a new notification (pulse or alert) should be sent"
-  (fn [pulse _parts] (alert-or-pulse pulse)))
-
-(defmethod should-send-notification? :alert
-  [{:keys [alert_condition] :as alert} parts]
-  (cond
-    (= "rows" alert_condition)
-    (not (are-all-parts-empty? parts))
-
-    (= "goal" alert_condition)
-    (goal-met? alert parts)
-
-    :else
-    (let [^String error-text (tru "Unrecognized alert with condition ''{0}''" alert_condition)]
-      (throw (IllegalArgumentException. error-text)))))
-
-(defmethod should-send-notification? :pulse
-  [pulse parts]
-  (if (:skip_if_empty pulse)
-    (not (are-all-parts-empty? parts))
-    true))
-
-(defn- get-notification-info
-  [pulse parts channel]
-  (let [alert? (nil? (:dashboard_id pulse))]
-    (merge {:payload-type (if alert?
-                            :notification/alert
-                            :notification/dashboard-subscription)
-            :payload      (if alert? (first parts) parts)
-            :pulse        pulse
-            :channel      channel}
-           (if alert?
-             {:card  (t2/select-one :model/Card (-> parts first :card :id))}
-             {:dashboard (t2/select-one :model/Dashboard (:dashboard_id pulse))}))))
-
-(defn- channels-to-channel-recipients
-  [channel]
-  (if (= :slack (keyword (:channel_type channel)))
-    [(get-in channel [:details :channel])]
-    (for [recipient (:recipients channel)]
-      (if-not (:id recipient)
-        {:kind :external-email
-         :email (:email recipient)}
-        {:kind :user
-         :user recipient}))))
-
-(defn- channel-send!
-  [& args]
-  (try
-    (apply channel/send! args)
-    (catch Exception e
-      ;; Token errors have already been logged and we should not retry.
-      (when-not (and (= :channel/slack (first args))
-                     (contains? (:errors (ex-data e)) :slack-token))
-        (throw e)))))
-
-(defn- send-retrying!
-  [& args]
-  (try
-    (apply (retry/decorate channel-send!) args)
-    (catch Throwable e
-      (log/error e "Error sending notification!"))))
-
-(defn- execute-pulse
-  [{:keys [cards] pulse-id :id :as pulse} dashboard]
-  (if dashboard
-    ;; send the dashboard
-    (execute-dashboard pulse dashboard)
-    ;; send the cards instead
-    (for [card cards
-          ;; Pulse ID may be `nil` if the Pulse isn't saved yet
-          :let [part (pu/execute-card pulse (u/the-id card) :pulse-id pulse-id)]
-          ;; some cards may return empty part, e.g. if the card has been archived
-          :when part]
-      part)))
-
-(defn- send-pulse!*
-  [{:keys [channels channel-ids] pulse-id :id :as pulse} dashboard]
-  (let [parts                  (execute-pulse pulse dashboard)
-        ;; `channel-ids` is the set of channels to send to now, so only send to those. Note the whole set of channels
-        channels               (if (seq channel-ids)
-                                 (filter #((set channel-ids) (:id %)) channels)
-                                 channels)]
-    (if (should-send-notification? pulse parts)
-      (let [event-type (if (= :pulse (alert-or-pulse pulse))
-                         :event/subscription-send
-                         :event/alert-send)]
-        (events/publish-event! event-type {:id      (:id pulse)
-                                           :user-id (:creator_id pulse)
-                                           :object  {:recipients (map :recipients (:channels pulse))
-                                                     :filters    (:parameters pulse)}})
-        (u/prog1 (doseq [channel channels]
-                   (try
-                     (let [channel-type (if (= :email (keyword (:channel_type channel)))
-                                          :channel/email
-                                          :channel/slack)
-                           messages     (channel/render-notification channel-type
-                                                                     (get-notification-info pulse parts channel)
-                                                                     (channels-to-channel-recipients channel))]
-                       (doseq [message messages]
-                         (send-retrying! channel-type message)))
-                     (catch Exception e
-                       (log/errorf e "Error sending %s %d to channel %s" (alert-or-pulse pulse) (:id pulse) (:channel_type channel)))))
-          (when (:alert_first_only pulse)
-            (t2/delete! Pulse :id pulse-id))))
-      (log/infof "Skipping sending %s %d" (alert-or-pulse pulse) (:id pulse)))))
-
 ;;; +----------------------------------------------------------------------------------------------------------------+
 ;;; |                                             Sending Notifications                                              |
 ;;; +----------------------------------------------------------------------------------------------------------------+
@@ -330,14 +19,10 @@
    Example:
 
     (send-pulse! pulse)                    ; Send to all Channels
-    (send-pulse! pulse :channel-ids [312]) ; Send only to Channel with :id = 312"
-  [{:keys [dashboard_id], :as pulse} & {:keys [channel-ids]}]
+    (send-pulse! pulse  [312]) ; Send only to Channel with :id = 312"
+  [pulse & {:keys [channel-ids]}]
   {:pre [(map? pulse) (integer? (:creator_id pulse))]}
-  (let [dashboard (t2/select-one Dashboard :id dashboard_id)
-        pulse     (-> (mi/instance Pulse pulse)
-                      ;; This is usually already done by this step, in the `send-pulses` task which uses `retrieve-pulse`
-                      ;; to fetch the Pulse.
-                      pulse/hydrate-notification
-                      (merge (when channel-ids {:channel-ids channel-ids})))]
-    (when (not (:archived dashboard))
-      (send-pulse!* pulse dashboard))))
+  (noti/send-notification! {:payload_type (if (:dashboard_id pulse)
+                                            :notification/dashboard-subscription
+                                           :notificaiton/alert)
+                            :payload_id    (:id pulse)}pulse channel-ids))

src/metabase/pulse/preview.clj

@@ -8,8 +8,9 @@
    [hickory.core :as hik]
    [hickory.render :as hik.r]
    [hickory.zip :as hik.z]
+   [metabase.channel.shared :as channel.shared]
    [metabase.email.messages :as messages]
-   [metabase.pulse :as pulse]
+   [metabase.notification.execute :as notification.execute]
    [metabase.pulse.markdown :as markdown]
    [metabase.pulse.render :as render]
    [metabase.pulse.render.image-bundle :as img]
@@ -55,7 +56,7 @@
             [:td {:style (style/style (merge table-style-map {:max-width "400px"}))}
              content])]
     (if card
-      (let [base-render (render/render-pulse-card :inline (pulse/defaulted-timezone card) card dashcard result)
+      (let [base-render (render/render-pulse-card :inline (channel.shared/defaulted-timezone card) card dashcard result)
             html-src    (-> base-render :content)
             img-src     (-> base-render
                             (png/render-html-to-png 1200)
@@ -78,7 +79,7 @@
          (markdown/process-markdown (:text dashboard-result) :html)])
        (cellfn nil)])))
 
-(def ^:private execute-dashboard #'pulse/execute-dashboard)
+(def ^:private execute-dashboard #'notification.execute/execute-dashboard)
 
 (defn render-dashboard-to-hiccup
   "Given a dashboard ID, renders all of the dashcards to hiccup datastructure."