get the management of the PulseChannels to work and be connected through the api.

src/metabase/api/pulse.clj

@@ -15,7 +15,7 @@
   "Fetch all `Pulses`"
   []
   (-> (db/sel :many Pulse (order :name :ASC))
-      (hydrate :creator :cards)))
+      (hydrate :creator :cards :channels)))
 
 
 (defendpoint POST "/"
@@ -24,17 +24,16 @@
   {name     [Required NonEmptyString]
    cards    [Required ArrayOfMaps]
    channels [Required ArrayOfMaps]}
-  (clojure.pprint/pprint body)
   (let-500 [card-ids (filter identity (map :id cards))
-            pulse (pulse/create-pulse name *current-user-id* card-ids channels)]
-    (hydrate pulse :cards)))
+            pulse    (pulse/create-pulse name *current-user-id* card-ids channels)]
+    (hydrate pulse :cards :channels)))
 
 
 (defendpoint GET "/:id"
   "Fetch `Pulse` with ID."
   [id]
   (->404 (db/sel :one Pulse :id id)
-         (hydrate :creator :cards)))
+         (hydrate :creator :cards :channels)))
 
 
 (defendpoint PUT "/:id"
@@ -43,13 +42,13 @@
   {name     [Required NonEmptyString]
    cards    [Required ArrayOfMaps]
    channels [Required ArrayOfMaps]}
-  (let-404 [pulse (db/sel :one Pulse :id id)]
-    (->500
-      (pulse/update-pulse {:id       id
-                           :name     name
-                           :cards    (filter identity (into [] (map :id cards)))
-                           :channels []})
-      (hydrate :creator :cards))))
+  (check-404 (db/exists? Pulse :id id))
+  (let-500 [result (pulse/update-pulse {:id       id
+                                        :name     name
+                                        :cards    (filter identity (into [] (map :id cards)))
+                                        :channels channels})]
+    (-> (db/sel :one Pulse :id id)
+        (hydrate :creator :cards :channels))))
 
 
 (defendpoint DELETE "/:id"

src/metabase/models/pulse.clj

 (ns metabase.models.pulse
-  (:require [korma.core :as k]
+  (:require [clojure.tools.logging :as log]
+            [korma.core :as k]
             [korma.db :as kdb]
             [metabase.db :as db]
             (metabase.models [card :refer [Card]]
                              [common :refer [perms-readwrite]]
                              [interface :refer :all]
                              [pulse-card :refer [PulseCard]]
-                             [pulse-channel :refer [PulseChannel]]
+                             [pulse-channel :refer [PulseChannel] :as pulse-channel]
                              [user :refer [User]])
             [metabase.util :as u]))
 
@@ -53,49 +54,83 @@
    *  If an existing `PulseCard` has no corresponding ID in CARD-IDs, it will be deleted.
    *  All cards will be updated with a `position` according to their place in the collection of CARD-IDS"
   {:arglists '([pulse card-ids])}
-  [{:keys [id] :as pulse} card-ids]
+  [{:keys [id]} card-ids]
   {:pre [(integer? id)
          (coll? card-ids)
          (every? integer? card-ids)]}
-  ;; IMPORTANT!  This is done in a transaction so that we can have the delete & insert be atomic
-  (kdb/transaction
-    ;; first off, just delete any cards associated with this pulse (we add them again below)
-    (k/delete PulseCard (k/where {:pulse_id id}))
-    ;; now just insert all of the cards that were given to us
-    (let [cards (map-indexed (fn [idx itm] {:pulse_id id :card_id itm :position idx}) card-ids)]
-      (k/insert PulseCard (k/values cards))))
-  pulse)
+  ;; first off, just delete any cards associated with this pulse (we add them again below)
+  (k/delete PulseCard (k/where {:pulse_id id}))
+  ;; now just insert all of the cards that were given to us
+  (let [cards (map-indexed (fn [idx itm] {:pulse_id id :card_id itm :position idx}) card-ids)]
+    (k/insert PulseCard (k/values cards))))
+
+(defn- update-pulse-channel
+  "Utility function which determines how to properly update a single pulse channel given "
+  [pulse-id new-channel existing-channel]
+  ;; NOTE that we force the :id of the channel being updated to the :id we *know* from our
+  ;;      existing list of `PulseChannels` pulled from the db to ensure we affect the right record
+  (let [channel (when new-channel (assoc new-channel :pulse_id pulse-id :id (:id existing-channel)))]
+    (cond
+      ;; 1. in channels, NOT in db-channels = CREATE
+      (and channel (not existing-channel))  (pulse-channel/create-pulse-channel channel)
+      ;; 2. NOT in channels, in db-channels = DELETE
+      (and (nil? channel) existing-channel) (db/cascade-delete PulseChannel :id (:id existing-channel))
+      ;; 3. in channels, in db-channels = UPDATE
+      (and channel existing-channel)        (pulse-channel/update-pulse-channel channel)
+      ;; 4. NOT in channels, NOT in db-channels = NO-OP
+      :else nil)))
+
+(defn update-pulse-channels
+  "Update the `PulseChannels` for a given PULSE.
+   CHANNELS should be a definitive collection of *all* of the channels for the the pulse.
+
+   * If a channel in the list has no existing `PulseChannel` object, one will be created.
+   * If an existing `PulseChannel` has no corresponding entry in CHANNELS, it will be deleted.
+   * All previously existing channels will be updated with their most recent information."
+  {:arglists '([pulse channels])}
+  [{:keys [id]} channels]
+  {:pre [(integer? id)
+         (coll? channels)
+         (every? map? channels)]}
+  (let [new-channels   (group-by #(keyword (:channel_type %)) channels)
+        old-channels   (group-by #(keyword (:channel_type %)) (db/sel :many PulseChannel :pulse_id id))
+        handle-channel #(update-pulse-channel id (first (get new-channels %)) (first (get old-channels %)))]
+    (assert (= 0 (count (get new-channels nil))) "Cannot have channels without a :channel_type attribute")
+    (dorun (map handle-channel (vec (keys pulse-channel/channel-types))))))
 
 (defn update-pulse
   "Update an existing `Pulse`"
   [{:keys [id name cards channels] :as pulse}]
-  {:pre [(string? name)
+  {:pre [(integer? id)
+         (string? name)
          (coll? cards)
          (> (count cards) 0)
          (every? integer? cards)
-         ;(coll? channels)
-         ;(every? map? channels)
-         ]}
+         (coll? channels)
+         (every? map? channels)]}
   ;; TODO: ideally this would all be in a transaction
   (db/upd Pulse id :name name)
   (update-pulse-cards pulse cards)
+  (update-pulse-channels pulse channels)
   (db/sel :one Pulse :id id))
 
 (defn create-pulse
   "Create a new `Pulse` by inserting it into the database along with all associated pieces of data such as:
-  `PulseCards`, `PulseChannels`, and `PulseChannelRecipients`.
-
-   This entire operation is atomic and happens within a transaction, so any failure will void the entire creation."
+  `PulseCards`, `PulseChannels`, and `PulseChannelRecipients`."
   [name creator-id cards channels]
   {:pre [(string? name)
          (integer? creator-id)
          (coll? cards)
          (> (count cards) 0)
          (every? integer? cards)
-         ;(coll? channels)
-         ;(every? map? channels)
-         ]}
-  (let [pulse (db/ins Pulse
-                :creator_id creator-id
-                :name name)]
-    (update-pulse-cards pulse cards)))
+         (coll? channels)
+         (every? map? channels)]}
+  ;; TODO: ideally this would all be in a transaction
+  (let [{:keys [id] :as pulse} (db/ins Pulse
+                                 :creator_id creator-id
+                                 :name name)]
+    ;; add cards to the Pulse
+    (update-pulse-cards pulse cards)
+    ;; add channels to the Pulse
+    (update-pulse-channels pulse channels)
+    (db/sel :one Pulse :id id)))

src/metabase/models/pulse_channel.clj

@@ -2,7 +2,7 @@
   (:require [clojure.set :as set]
             [korma.core :as k]
             [metabase.api.common :refer [check]]
-            [metabase.db :refer :all]
+            [metabase.db :as db]
             (metabase.models [pulse-channel-recipient :refer [PulseChannelRecipient]]
                              [interface :refer :all]
                              [user :refer [User]])
@@ -15,36 +15,36 @@
   "Map which contains the definitions for each type of pulse channel we allow.  Each key is a channel type with a map
    which contains any other relevant information for defining the channel.  E.g.
 
-   {:email {:name \"Email\"}
-    :slack {:name \"Slack\"}}"
-  {:email {}
-   :slack {}})
-
-(defn pulse-channel?
-  "Predicate function which returns `true` if the given channel is a valid option as a pulse channel type, `false` otherwise."
+   {:email {:name \"Email\", :recipients? true}
+    :slack {:name \"Slack\", :recipients? false}}"
+  {:email {:recipients? true}
+   :slack {:recipients? false}})
+
+(defn channel-type?
+  "Predicate function which returns `true` if the given argument is a valid value as a channel-type, `false` otherwise."
+  [channel-type]
+  (contains? (set (keys channel-types)) (keyword channel-type)))
+
+(def schedule-types
+  "Map which contains the definitions for each type of pulse schedule type we allow.  Each key is a schedule-type with
+   a map which contains any other relevant information related to the defined schedule-type.  E.g.
+
+   {:hourly {:name \"Hourly\"}
+    :dailye {:name \"Daily\"}}"
+  {:hourly {}
+   :daily  {}
+   :weekly {}})
+
+(defn schedule-type?
+  "Predicate function which returns `true` if the given argument is a valid value as a schedule-type, `false` otherwise."
+  [schedule-type]
+  (contains? (set (keys schedule-types)) (keyword schedule-type)))
+
+(defn supports-recipients?
+  "Predicate function which returns `true` if the given channel type supports a list of recipients, `false` otherwise."
   [channel]
-  (contains? (set (keys channel-types)) (keyword channel)))
-
-(def modes
-  {:active   {:id 1
-              :name "Active"}
-   :disabled {:id 2
-              :name "Disabled"}})
-
-(def mode-kws
-  (set (keys modes)))
-
-(defn mode->id [mode]
-  {:pre [(contains? mode-kws mode)]}
-  (:id (modes mode)))
+  (boolean (:recipients? (get channel-types channel))))
 
-(defn mode->name [mode]
-  {:pre [(contains? mode-kws mode)]}
-  (:name (modes mode)))
-
-(def modes-input
-  [{:id (mode->id :active),   :name (mode->name :active)}
-   {:id (mode->id :disabled), :name (mode->name :disabled)}])
 
 (def days-of-week
   "Simple `vector` of the days in the week used for reference and lookups.
@@ -87,22 +87,22 @@
 (defentity PulseChannel
   [(k/table :pulse_channel)
    (hydration-keys pulse_channel)
-   (types :schedule_details :json)
+   (types :details :json, :schedule_details :json)
    timestamped]
 
   (post-select [_ {:keys [id creator_id] :as pulse-channel}]
     (map->PulseChannelInstance
       (u/assoc* pulse-channel
-                :recipients   (delay (sel :many User
+                :recipients   (delay (db/sel :many User
                                           (k/where {:id [in (k/subselect PulseChannelRecipient (k/fields :user_id) (k/where {:pulse_channel_id id}))]}))))))
 
   (pre-cascade-delete [_ {:keys [id]}]
-    (cascade-delete PulseChannelRecipient :pulse_channel_id id)))
+    (db/cascade-delete PulseChannelRecipient :pulse_channel_id id)))
 
 (extend-ICanReadWrite PulseChannelEntity :read :always, :write :superuser)
 
 
-;; ## Related Functions
+;; ## Helper Functions
 
 (defn update-recipients
   "Update the `PulseChannelRecipients` for PULSE.
@@ -115,7 +115,7 @@
   {:pre [(integer? id)
          (coll? user-ids)
          (every? integer? user-ids)]}
-  (let [recipients-old (set (sel :many :field [PulseChannelRecipient :user_id] :pulse_channel_id id))
+  (let [recipients-old (set (db/sel :many :field [PulseChannelRecipient :user_id] :pulse_channel_id id))
         recipients-new (set user-ids)
         recipients+    (set/difference recipients-new recipients-old)
         recipients-    (set/difference recipients-old recipients-new)]
@@ -128,3 +128,37 @@
       (k/delete PulseChannelRecipient
               (k/where {:pulse_channel_id id
                       :user_id [in recipients-]})))))
+
+(defn update-pulse-channel
+  "Updates an existing `PulseChannel` along with all related data associated with the channel such as `PulseChannelRecipients`."
+  [{:keys [id channel_type details recipients schedule_details schedule_type]
+    :or   {details          {}
+           recipients       []
+           schedule_details {}}}]
+  {:pre [(integer? id)
+         (channel-type? channel_type)
+         (schedule-type? schedule_type)]}
+  (db/upd PulseChannel id
+    :details          details
+    :schedule_type    schedule_type
+    :schedule_details schedule_details)
+  (when (and (supports-recipients? channel_type) (seq recipients))
+    "do recipients"))
+
+(defn create-pulse-channel
+  "Create a new `PulseChannel` along with all related data associated with the channel such as `PulseChannelRecipients`."
+  [{:keys [channel_type details pulse_id recipients schedule_details schedule_type]
+    :or   {details          {}
+           recipients       []
+           schedule_details {}}}]
+  {:pre [(channel-type? channel_type)
+         (integer? pulse_id)
+         (schedule-type? schedule_type)]}
+  (let [{:keys [id]} (db/ins PulseChannel
+                       :pulse_id         pulse_id
+                       :channel_type     channel_type
+                       :details          details
+                       :schedule_type    schedule_type
+                       :schedule_details schedule_details)]
+    (when (and (supports-recipients? channel_type) (seq recipients))
+      "do recipients")))

src/metabase/models/pulse_channel_recipient.clj

 (ns metabase.models.pulse-channel-recipient
-  (:require [korma.core :as k]))
+  (:require [korma.core :as k]
+            [metabase.models.interface :refer :all]))
 
-(k/defentity PulseChannelRecipient
-  (k/table :pulse_channel_recipient))
+(defentity PulseChannelRecipient
+  [(k/table :pulse_channel_recipient)])

test/metabase/api/pulse_test.clj

@@ -93,6 +93,17 @@
   (-> (select-keys card [:id :name :description])
       (assoc :display (name (:display card)))))
 
+(defn pulse-channel-details [channel]
+  (match-$ channel
+    {:id               $
+     :pulse_id         $
+     :channel_type     $
+     :details          $
+     :schedule_type    $
+     :schedule_details $
+     :created_at       $
+     :updated_at       $}))
+
 (defn pulse-details [pulse]
   (match-$ pulse
     {:id           $
@@ -102,8 +113,8 @@
      :updated_at   $
      :creator_id   $
      :creator      (user-details @(:creator pulse))
-     :cards        [(map pulse-card-details @(:cards pulse))]
-     :channels     []}))
+     :cards        (mapv pulse-card-details @(:cards pulse))
+     :channels     (mapv pulse-channel-details @(:channels pulse))}))
 
 (defn pulse-response [{:keys [created_at updated_at] :as pulse}]
   (-> pulse
@@ -124,9 +135,9 @@
   ((user->client :rasta) :post 400 "pulse" {:name  "abc"
                                             :cards "foobar"}))
 
-;(expect {:errors {:cards "Invalid value 'abc' for 'cards': array value must be a map."}}
-;  ((user->client :rasta) :post 400 "pulse" {:name  "abc"
-;                                            :cards ["abc"]}))
+(expect {:errors {:cards "Invalid value 'abc' for 'cards': array value must be a map."}}
+  ((user->client :rasta) :post 400 "pulse" {:name  "abc"
+                                            :cards ["abc"]}))
 
 (expect {:errors {:channels "field is a required param."}}
   ((user->client :rasta) :post 400 "pulse" {:name "abc"
@@ -134,13 +145,13 @@
 
 (expect {:errors {:channels "Invalid value 'foobar' for 'channels': value must be an array."}}
   ((user->client :rasta) :post 400 "pulse" {:name    "abc"
-                                            :cards   [1 2 3]
+                                            :cards   [{:id 100} {:id 200}]
                                             :channels "foobar"}))
 
-;(expect {:errors {:channels "Invalid value 'abc' for 'channels': array value must be a map."}}
-;  ((user->client :rasta) :post 400 "pulse" {:name     "abc"
-;                                            :cards    [{:id 100} {:id 200}]
-;                                            :channels ["abc"]}))
+(expect {:errors {:channels "Invalid value 'abc' for 'channels': array value must be a map."}}
+  ((user->client :rasta) :post 400 "pulse" {:name     "abc"
+                                            :cards    [{:id 100} {:id 200}]
+                                            :channels ["abc"]}))
 
 (expect-let [card1 (new-card)
              card2 (new-card)]