Lafargue: Add XMPP notification capability.

6 files changed, 137 insertions, 9 deletions

diff --git a/config.sexp.sample b/config.sexp.sample
index 37bcfcf..f314815 100644
--- a/config.sexp.sample
+++ b/config.sexp.sample
@@ -8,4 +8,13 @@
  :websocket-base "ws://localhost:3001"
  :base-uri       "http://localhost:3001"
  :tag-base       "example.com"
- :web-port       3001}
+ :web-port       3001
+ :mode           :production   ;or :dev
+ :xmpp           {:user         "benki"
+                  :service-name "example.com"
+                  :server       "jabber.example.com"
+                  :port         5222
+                  :password     ""
+                  :tls          :required}  ;or one of: true, false
+ }
+
diff --git a/project.clj b/project.clj
index 4d8e35e..d6940c3 100644
--- a/project.clj
+++ b/project.clj
@@ -44,6 +44,8 @@
                  [org.apache.ws.commons.axiom/axiom-api "1.2.12"]
                  [clj-apache-http "2.3.2"]
                  [org.pegdown/pegdown "1.1.0"]
+                 [jivesoftware/smack "3.1.0"]
+                 [jivesoftware/smackx "3.1.0"]
                  ]
   :plugins [[lein-swank "1.4.3"]]
   :exclusions [org.clojure/clojure-contrib]  ;you know, the old pre-1.3.0 versions
diff --git a/schema.sql b/schema.sql
index c12f73e..f1f07fc 100644
--- a/schema.sql
+++ b/schema.sql
@@ -34,9 +34,16 @@ CREATE TABLE user_nicknames(
   PRIMARY KEY(nickname),
   FOREIGN KEY("user") REFERENCES users
 );
-
 CREATE INDEX user_nicknames_user ON user_nicknames ("user");
 
+CREATE TABLE user_jids(
+  "user" INTEGER NOT NULL,
+  jid    VARCHAR NOT NULL,
+  PRIMARY KEY("user", jid),
+  FOREIGN KEY("user") REFERENCES users
+);
+CREATE INDEX user_jids_user ON user_jids ("user");
+
 CREATE TABLE page_keys(
   "user" INTEGER NOT NULL,
   page   VARCHAR NOT NULL,
diff --git a/src/mulk/benki/lazychat.clj b/src/mulk/benki/lazychat.clj
index e2caaf8..ecc4ac1 100644
--- a/src/mulk/benki/lazychat.clj
+++ b/src/mulk/benki/lazychat.clj
@@ -21,7 +21,8 @@
             [lamina.core          :as lamina]
             [aleph.http           :as ahttp]
             [aleph.formats        :as aformats]
-            [clojure.data.json    :as json])
+            [clojure.data.json    :as json]
+            [mulk.benki.xmpp      :as xmpp])
   (:import [org.apache.abdera Abdera]))
 
 
@@ -29,6 +30,29 @@
 (defonce lafargue-events (channel))
 
 
+(defmethod xmpp/format-message ::lafargue-message
+  [message]
+  (fmt nil "<~A>\n\n~A" (:first_name message) (:content message)))
+
+(defn determine-targets [message]
+  (letfn [(protected-targets []
+            (with-dbt
+              (map :id (query "SELECT id FROM users WHERE status IN ('admin', 'approved')"))))]
+    (into #{}
+          (concat (:targets message)
+                  (case (keyword (:visibility message))
+                    :personal  nil
+                    :protected (protected-targets)
+                    :public    (cons nil (protected-targets)))))))
+
+(defn push-message-to-xmpp [msg]
+  (let [targets (filter integer? (determine-targets msg))]
+    (enqueue xmpp/messages {:message msg,
+                            :targets targets})))
+
+(defn start-xmpp-pump []
+  (receive-all lafargue-events push-message-to-xmpp))
+
 (defn create-lazychat-message! [{content  :content,  visibility :visibility
                                  format   :format,   targets    :targets,
                                  referees :referees, id         :id}]
@@ -53,10 +77,12 @@
                              [:message :target]
                              [id (int target)]))
         (enqueue lafargue-events
-                 {:content  content,  :visibility visibility,
-                  :format   format,   :targets    targets,
-                  :referees referees, :id         id,
-                  :author   *user*,   :date       (java.util.Date.)})))))
+                 (with-meta
+                   {:content  content,  :visibility visibility,
+                    :format   format,   :targets    targets,
+                    :referees referees, :id         id,
+                    :author   *user*,   :date       (java.util.Date.)}
+                   {:type ::lafargue-message}))))))
 
 (defn select-message [id]
   (let [message  (query1 "SELECT author, content, format, visibility, date
@@ -231,3 +257,7 @@
   (with-auth
     (response/json
       (with-dbt (query1 "SELECT NEXTVAL('lazychat_messages_id_seq')")))))
+
+(defn init-lazychat! []
+  (future
+    (receive-all lafargue-events push-message-to-xmpp)))
diff --git a/src/mulk/benki/main.clj b/src/mulk/benki/main.clj
index 8a7f094..031a500 100644
--- a/src/mulk/benki/main.clj
+++ b/src/mulk/benki/main.clj
@@ -5,7 +5,7 @@
         [hiccup core     page-helpers]
         [mulk.benki      util config db])
   (:require [noir server options]
-            [mulk.benki wiki auth book_marx id lazychat]
+            [mulk.benki wiki auth book_marx id lazychat xmpp]
             [ring.middleware.file]
             [noir.session      :as session]
             [noir.request      :as request]
@@ -92,7 +92,10 @@
                               :websocket true})))
 
 (defonce server
-  (future (run-server)))
+  (do
+    (future (mulk.benki.xmpp/init-xmpp!))
+    (future (mulk.benki.lazychat/init-lazychat!))
+    (future (run-server))))
 
 (defn -main [& args]
   (loop []
diff --git a/src/mulk/benki/xmpp.clj b/src/mulk/benki/xmpp.clj
new file mode 100644
index 0000000..4f967ba
--- /dev/null
+++ b/src/mulk/benki/xmpp.clj
@@ -0,0 +1,77 @@
+(ns mulk.benki.xmpp
+  (:refer-clojure)
+  (:use [clojure     repl]
+        [noir        core]
+        [noir-async  core]
+        [mulk.benki  auth config db util webutil]
+        ;;
+        [clojure.core.match :only [match]]
+        [lamina.core        :only [channel enqueue enqueue-and-close receive-all
+                                   map* filter*]])
+  (:require [clojure.algo.monads  :as m]
+            [clojure.java.jdbc    :as sql]
+            [clojure.string       :as string]
+            [lamina.core          :as lamina]
+            [clojure.data.json    :as json])
+  (:import [org.jivesoftware.smack ConnectionConfiguration
+                                   ConnectionConfiguration$SecurityMode
+                                   XMPPConnection
+                                   MessageListener]))
+
+
+(defonce xmpp     (atom nil))
+(defonce messages (channel))
+
+
+(defn- connect []
+  (let [xmpp-config       (:xmpp @benki-config)
+        connection-config (doto (ConnectionConfiguration. (:server       xmpp-config)
+                                                          (:port         xmpp-config)
+                                                          (:service-name xmpp-config))
+                            (.setSecurityMode (case (:tls xmpp-config)
+                                                true      ConnectionConfiguration$SecurityMode/enabled
+                                                false     ConnectionConfiguration$SecurityMode/disabled
+                                                :required ConnectionConfiguration$SecurityMode/required))
+                            (.setVerifyRootCAEnabled true)
+                            (.setVerifyChainEnabled  true)
+                            ;;(.setCompressionEnabled  true)
+                            (.setSASLAuthenticationEnabled true))]
+    (doto (XMPPConnection. connection-config)
+      (.connect)
+      (.login (:user xmpp-config) (:password xmpp-config) (:resource xmpp-config)))))
+
+(defn reconnect! []
+  (swap! xmpp #(do (when %
+                     (.disconnect %))
+                   (connect))))
+
+(defmulti format-message type)
+
+(defn- ->pgarray [coll]
+  (fmt nil "{~{~A~^,~}}" coll))
+
+(defn- push-message [targets message]
+  (let [chat-manager (.getChatManager @xmpp)
+        notification (format-message message)
+        recipients   (with-dbt
+                       (map :jid
+                            (query "SELECT jid FROM user_jids j INNER JOIN unnest(?::INTEGER[]) t ON j.user = t"
+                                   (->pgarray targets))))]
+    (doseq [recipient recipients]
+      (future
+        (let [chat (.createChat chat-manager
+                                recipient
+                                (reify MessageListener
+                                  (processMessage [self chat message]
+                                    nil)))]
+          (.sendMessage chat notification))))))
+
+(defn- startup-client []
+  (receive-all messages
+               (fn [{targets :targets, msg :message}]
+                 (push-message targets msg))))
+
+(defn init-xmpp! []
+  (future
+    (reconnect!)
+    (startup-client)))