                  (let [ks' (cond-> ks
                              many? (conj id))
                        xch (chan 1 (delta/nest ks'))]
-                   (vswap! subs assoc ks'
-                           [xch (compscribe* xch
-                                             subs-fn
-                                             unsubs-fn
-                                             (get direct-hooks ks)
-                                             id)])
+                   (subscribe compscribe-service
+                              [(get direct-hooks ks) id]
+                              xch)
+                   (vswap! subs assoc ks' xch)
                    ;; xch will be closed by the subscribed-composition
                    (mix-in target-mix xch)))
-        ch-in (subs-fn endpoint id)]
+        ch-in (chan)]
+    (subscribe source-service [endpoint id] ch-in)
     (go-loop []
       (if-some [delta (<! ch-in)]
         (let [[{:keys [subs-one subs-many unsubs]} delta]
               (extract-hooks conformed-spec delta)
 
               subs-by-pks (group-by-pks @subs)]
-          (doseq [[ks [xch unsubs-fn]]
+          #_(println delta conformed-spec [subs-one subs-many unsubs])
+          (doseq [[ks xch]
                   (->> unsubs
                        (mapcat (fn [ks]
                                  (concat (->> (get subs-by-pks ks)

    ch."))

(defn- compscribe*
  [outer-target subs-fn unsubs-fn
   [endpoint direct-hooks deep-hooks :as conformed-spec] id]
  (let [;; Once intercepted, events need to go through target-mix so
        ;; that mix-ins and mix-outs have synchronous effects

  as ids and are transformed to assert a map {(id streamed-val)+} at
  key.

  Subscriptions are made at service, which has to implement IStream,
  with [endpoint id] as identifier argument.

  id is used to make an initial subscription at the root endpoint.

