member_groups.clj

(defn denorm-groups-to-member-groups-s
  "denorms from groups to the group side of member-groups"
  [cassandra org-id]
  (ddo [g-s (m/select-buffered
             cassandra
             g.ent/RawGroups
             [:org_id]
             [org-id]
             {:from :groups_full_by_id
              :order-by [[:id :asc]]})

        mg-s (m/select-buffered
              cassandra
              g.ent/MemberGroups
              [:org_id]
              [org-id]
              {:from :group_members_by_group_id
               :order-by [[:group_id :asc]]})

        change-s
        (cross/inner-join-streams
         {:key-compare-fn cass.uuid/-compare
          :skey-streams {:group g-ss
                         :member-group mg-ss}
          :finish-merge-fn (fn [{g :group mg :member-group}]
                             [mg
                              (when (and g (not (:disabled g)))
                                (denorm/denormalize-fields-to-target
                                 g.ent/RawGroups
                                 g.ent/MemberGroups
                                 g
                                 mg))])})

        change-r-s (m/change-buffered
                    cassandra
                    g.ent/MemberGroups
                    change-s)]
    (stream/reduce (fn [c _] (inc c)) 0 change-r-s)))