Added topics, refactored recipient handling

5 files changed, 130 insertions, 69 deletions

diff --git a/src/datalayer.clj b/src/datalayer.clj
index 24c162a..ecdc6dc 100644
--- a/src/datalayer.clj
+++ b/src/datalayer.clj
@@ -1,4 +1,5 @@
 (ns datalayer
+  (:import java.util.Date)
   (:require redis
             tags)
   (:use clojure.contrib.sql
@@ -82,25 +83,6 @@ WHERE  u.nick = ANY(?)"
 (defn redis-popular-key [nick]
   (str "popular:" nick))
 
-(def popular-dumps-qry "
-select u.nick, u.avatar, r.key, m.message_id, m.content, m.created_on, count(*) as count,
-               array_agg(u2.nick) as user_nicks
-from users u, messages m, rooms r, tags t, users u2
-where lower(u.nick) = lower(?)
-and u.user_id = m.user_id and m.message_id = t.message_id
-and m.room_id = r.room_id and m.is_image = true and r.admin_only = false
-and t.tag = 'favorite'
-and t.user_id = u2.user_id
-group by u.nick, u.avatar, r.key, m.message_id, m.content, m.created_on
-order by count desc limit ? offset ?")
-
-(defn fetch-popular-dumps [nick viewer-nick]
-  (for [d (do-select [popular-dumps-qry nick 40 0])]
-    (let [favers (.getArray (:user_nicks d))]
-      (assoc d
-        :favers    favers
-        :favorited (some #(= % viewer-nick) favers)))))
-
 (defn fetch-popular-dumps-redis [nick viewer-nick]
   (let [msg-ids   (redis/with-server redis-server
                     (redis/zrevrange (redis-popular-key nick)
@@ -177,10 +159,14 @@ order by count desc limit ? offset ?")
 
 ;;;; Message insertion
 
+(defn direct-message-key [u-id]
+  (str "directmessage:" u-id))
+
 (def msg-insert-query
   "INSERT INTO messages (user_id, room_id, content, is_image, is_text)
    VALUES (?, ?, ?, ?, ?) RETURNING message_id, created_on")
 
+;; Note: direct-message recipients are inserted into postgres, but topics aren't.
 (defn insert-message-into-postgres! [author-id room-id content is-image is-text recips]
   (with-connection *db*
     (transaction
@@ -195,43 +181,81 @@ order by count desc limit ? offset ?")
           [msg-id author-id (:user_id r)]))
        [msg-id ts]))))
 
-(defn insert-recips-into-redis! [recips author-id ts content]
-  (let [dm-json (json-str {"author_id" author-id
-                           "recips" (map :nick recips)
-                           "content" content})]
+(defn insert-recips-into-redis! [recips author-id dt content]
+  (let [msg-json (json-str {"author_id" author-id
+                            "recips" (map :nick recips)
+                            "content" content})
+        ts       (.getTime dt)]
     (redis/with-server redis-server
       (redis/atomically
        (doseq [r recips]
-         (redis/zadd (str "directmessage:" (:user_id r))
-                     (.getTime ts)
-                     dm-json))))))
+         (redis/zadd (direct-message-key (:user_id r))
+                     ts
+                     msg-json))))))
 
-(defn insert-message! [author-id author-nick room-id content]
+(defn topic-key [topic]
+  (str "topic:" topic))
+
+(defn insert-topics-into-redis! [topics recips author-nick author-avatar dt msg-id content]
+  (let [ts         (.getTime dt)
+        msg-json   (json-str {"nick"          author-nick
+                              "avatar"        author-avatar
+                              "recips"        (map :recips recips)
+                              "content"       content
+                              "message_id"    msg-id
+                              "ts"            ts})]
+    (redis/with-server redis-server
+      (redis/atomically
+       (doseq [t topics]
+         (redis/lpush (topic-key t)
+                      msg-json))))))
+
+(defn insert-message! [author-id author-nick author-avatar room-id content]
   (let [msg-type    (classify-msg content)
         is-image    (boolean (#{:image :mixed} msg-type))
         is-text     (boolean (#{:mixed :text} msg-type))
         recips      (get-recips content)
-        [msg-id ts] (insert-message-into-postgres! author-id
+        topics      (get-topics content)
+        [msg-id dt] (insert-message-into-postgres! author-id
                                                    room-id
                                                    content
                                                    is-image
                                                    is-text
                                                    recips)]
     (if-not (empty? recips)
-      (insert-recips-into-redis! recips author-id ts content))
+      (insert-recips-into-redis! recips author-id dt content))
+    (if-not (empty? topics)
+      (insert-topics-into-redis! topics recips author-nick author-avatar dt msg-id content))
     {:author  author-nick
      :msg-id  msg-id
      :room    room-id
-     :db-ts   ts
+     :db-ts   dt
      :content content
-     :recips  (map (comp lower-case :nick) recips)}))
+     :recips  (map :nick recips)
+     :topics  topics
+     }))
 
-(defn fetch-private-messages [user-id]
+(defn fetch-direct-messages [user-id]
   (for [dm (redis/with-server redis-server
-             (redis/zrevrange (str "directmessage:" user-id) 0 40))]
+             (redis/zrevrange (direct-message-key user-id) 0 40))]
     (let [dm    (read-json dm)
           info  (fetch-user-id (get dm "author_id"))]
       {"nick"    (:nick info)
        "content" (get dm "content")
        "recips"  (get dm "recips" [])
        "avatar" (:avatar info)})))
+
+(def topic-fetch-num 40)
+
+(defn fetch-topic [viewer-id topic]
+  (let [redis-msgs (redis/with-server redis-server
+                     (redis/lrange (topic-key topic) 0 topic-fetch-num))
+        raw-msgs (for [m redis-msgs]
+                   (let [m (keywordify (read-json m))]
+                     (assoc m
+                       :created_on (Date. (:ts m)))))]
+    (if viewer-id
+      (tags/add-user-favs-to-msgs
+       raw-msgs
+       viewer-id)
+      raw-msgs)))
diff --git a/src/message.clj b/src/message.clj
index a8e0e9b..9f1380b 100644
--- a/src/message.clj
+++ b/src/message.clj
@@ -1,5 +1,6 @@
 (ns message
-  (:use user))
+  (:use user
+        utils))
 
 ;; Message parsing
 
@@ -26,7 +27,7 @@
 (defn get-recips [content]
   (filter
    boolean
-   (for [at-nick (re-seq recip-regex content)]
+   (for [at-nick (set (re-seq recip-regex content))]
      (fetch-nick (.substring (.trim at-nick) 1)))))
 
 (defn get-recips-from-msgs [msgs]
@@ -37,3 +38,10 @@
      boolean
      (for [r recips]
        (fetch-nick (.substring (.trim r) 1))))))
+
+(def topic-regex #"(?:^|\s)#\w+")
+
+(defn get-topics [content]
+  (set
+   (for [r (re-seq topic-regex content)]
+     (lower-case (.substring r 1)))))
diff --git a/src/site.clj b/src/site.clj
index 663a6df..6ebfb73 100644
--- a/src/site.clj
+++ b/src/site.clj
@@ -316,6 +316,9 @@ WHERE  user_id IN
                       (comp take-images :content)
                       dumps))))))
 
+(defn pull-recips [dumps]
+  (set (apply concat (map #(get % "recips") dumps))))
+
 (defn profile
   ([session profile-nick] (profile session profile-nick "profile"))
   ([session profile-nick template]
@@ -330,8 +333,8 @@ WHERE  user_id IN
                                    :user-tag-id (:user_id session)
 				   :nick profile-nick
 				   :limit 10)
-             dms           (fetch-private-messages (:user_id user-info))
-             recips        (set (apply concat (map #(get % "recips") dms)))
+             dms           (fetch-direct-messages (:user_id user-info))
+             recips        (pull-recips dms)
              imgs          (pull-random-dump-images dumps 5)]
          (do
            (.setAttribute st "is_home" is-home)
@@ -343,7 +346,7 @@ WHERE  user_id IN
 	   (.setAttribute st "score_ent" (score-to-entity score))
            (when-not (empty? dms)
              (.setAttribute st "dms" dms)
-             (.setAttribute st "recips" (json-str (map lower-case recips))))
+             (.setAttribute st "recips" (json-str recips)))
            (if (not (empty? imgs))
 	     (.setAttribute st "imgs" imgs))
 	   (.setAttribute st "debug_log_items" (logger))
@@ -421,6 +424,7 @@ WHERE  user_id IN
                        :msg-id msg-id
                        :date (if msg-id nil date)
                        :limit (inc *dumps-per-page*))
+        recips        (map :nick (get-recips-from-msgs raw-dumps))
         back-dumps    (if (or date msg-id)
                         (tags/fetch-dumps
                          :nick (:nick user-info)
@@ -432,8 +436,9 @@ WHERE  user_id IN
     (.setAttribute st "nick" (:nick user-info))
     (.setAttribute st "is_home" (= (:nick user-info) (:nick session)))
     (.setAttribute st "mini_profile" (build-mini-profile user-info))
-    (if (> (count dumps) 0)
+    (when (> (count dumps) 0)
       (.setAttribute st "dumps" dumps))
+    (.setAttribute st "recips" (json-str recips))
     (.setAttribute st "prev"
                    (if back-dumps
                      (cond
@@ -453,20 +458,18 @@ WHERE  user_id IN
 
 ;; Who faved me
 
-(def use-popular-redis true)
-
 (defn popular [session profile-nick]
   (if-let [user-info (fetch-nick profile-nick)]
     (let [st            (fetch-template "popular" session)
 	  profile-nick  (:nick user-info)
-	  raw-dumps     (if use-popular-redis
-                          (fetch-popular-dumps-redis profile-nick (:nick session))
-                          (fetch-popular-dumps       profile-nick (:nick session)))
+	  raw-dumps     (fetch-popular-dumps-redis profile-nick (:nick session))
           raw-dumps     (filter #(> (:count %) 0) raw-dumps)
+          recips        (get-recips-from-msgs raw-dumps)
 	  dumps         (map process-message-for-output raw-dumps)]
       (.setAttribute st "nick" profile-nick)
       (.setAttribute st "mini_profile" (build-mini-profile user-info))
       (.setAttribute st "dumps" dumps)
+      (.setAttribute st "recips" (json-str recips))
       (.toString st))
     (resp-error "NO_USER")))
 
@@ -538,6 +541,7 @@ WHERE  user_id IN
     (if (= nick-from-url (:nick message))
       (let [st (fetch-template "single_message" session)]
         (.setAttribute st "dump" (process-message-for-output message))
+        (.setAttribute st "recips" (json-str (map :nick (get-recips (:content message)))))
         (.toString st))
       (resp-error "NO_MESSAGE"))
     (resp-error "NO_MESSAGE")))
@@ -573,7 +577,7 @@ WHERE  user_id IN
       (doto st
 	(.setAttribute "users" (prepare-user-list room true))
 	(.setAttribute "messages" message-list)
-        (.setAttribute "recips" (json-str (map lower-case recips)))
+        (.setAttribute "recips" (json-str recips))
 	(.setAttribute "roomkey" (room :key))
 	(.setAttribute "isadminroom" (room :admin_only))
 	(.setAttribute "json_room_key" (json-str (room :key)))
@@ -605,7 +609,7 @@ WHERE  user_id IN
 						     :avatar    (session :avatar)}))
 	 (commute users assoc nick (user-struct-from-session session))))
      (resp-success (assoc (updates nick room old-ts)
-		          :timestamp now)))))
+                     :timestamp now)))))
 
 (defn validated-refresh [session params]
   (let [room-key (params :room)
@@ -638,16 +642,16 @@ WHERE  user_id IN
            mute                (resp-error (format-mute mute))
            :else
 	   (let [content    (validated-content content session)
-	         msg-info   (insert-message! user-id nick (:room_id room) content)
+	         msg-info   (insert-message! user-id nick (:avatar session) (:room_id room) content)
                  msg-id     (:msg-id msg-info)]
 	     (dosync
-              (if (not (contains? (ensure (room :users)) nick))
-                (login-user (user-struct-from-session session) room))
-              (add-message (build-msg nick content msg-id (:recips msg-info)) room))
+              (let [msg-struct (build-msg nick content msg-id (:recips msg-info))]
+                (if (not (contains? (ensure (room :users)) nick))
+                  (login-user (user-struct-from-session session) room))
+                (add-message msg-struct room)))
              (resp-success {:msgid msg-id
                             :recips (:recips msg-info)})))))
 
-
 (defn validated-msg [session params request]
   (cond
    (not (validate-room-access (params :room) session)) (resp-error "UNKNOWN_ROOM")
@@ -655,7 +659,6 @@ WHERE  user_id IN
 
 ;; Browser
 
-;; TODO: make work for all rooms
 (defn browser [session]
   (let [room  (lookup-room *default-room*)
         now   (System/currentTimeMillis)
@@ -675,6 +678,21 @@ WHERE  user_id IN
     (.toString st)))
 
 
+;; Topics
+
+(defn topic [session topic]
+  (let [topic  (lower-case topic)
+        msgs   (map
+                process-message-for-output
+                (fetch-topic (:user_id session) topic))
+        recips (pull-recips msgs)
+        st   (fetch-template "topic" session)]
+    (.setAttribute st "recips" (json-str recips))
+    (.setAttribute st "topic" topic)
+    (if-not (empty? msgs)
+      (.setAttribute st "dumps" msgs))
+    (.toString st)))
+
 ;; Chat Log
 
 (defn log [session room offset params]
@@ -690,7 +708,9 @@ WHERE  user_id IN
                             :image-only image-only
                             :amount (+ 1 *dumps-per-page*)
                             :offset dump-offset)
-        dumps       (map tags/add-favorited-flag (take *dumps-per-page* raw-dumps) (repeat session))
+        dumps       (map tags/add-favorited-flag
+                         (take *dumps-per-page* raw-dumps)
+                         (repeat session))
         ;;   json-tags (for [dump dumps :when (not (empty? (dump :tags)))] 
         ;;               (json-str {"id" (dump :message_id) "tags" (dump :tags) }))
         dumps       (map tags/remove-tags-for-output dumps)
@@ -734,8 +754,6 @@ WHERE  user_id IN
     (.setAttribute st "dumps" dumps)
     (.toString st)))
         
-
-
 ;; Altars
 
 ;; if :nick is in params, will fetch only altars by that nick
@@ -887,6 +905,7 @@ WHERE  user_id IN
                        :hide-vip (not (:is_admin session))
                        :date (if msg-id nil date)
                        :limit (inc *dumps-per-page*))
+        recips        (map :nick (get-recips-from-msgs raw-dumps))
         back-dumps    (if (or date msg-id)
                         (tags/fetch-tagged-dumps
                          :nick (:nick user-info)
@@ -906,6 +925,7 @@ WHERE  user_id IN
     (if (> (count raw-dumps) *dumps-per-page*)
       (.setAttribute st "next" (favorites-next-page-link (:nick user-info) (last raw-dumps))))
     (.setAttribute st "dumps" dumps)
+    (.setAttribute st "recips" (json-str recips))
     (.setAttribute st "infobar" (build-mini-profile user-info))
     (.setAttribute st "page_title" (format "%s'S FAVS" (:nick user-info)))
     (.setAttribute st "debug_log_items" (logger))
@@ -1069,11 +1089,17 @@ WHERE  user_id IN
           date       (today)
           dest       (open-file [*image-directory* date] filename)
           url        (image-url-from-file "images" date dest)
-          msg-info   (insert-message! (:user_id session) (:nick session)
-                                      (:room_id room) url)]
+          msg-info   (insert-message! (:user_id session)
+                                      (:nick session)
+                                      (:avatar session)
+                                      (:room_id room)
+                                      url)]
       (copy (:tempfile image) dest)
       (dosync
-       (let [msg (build-msg (:nick session) url (:msg-id msg-info) (:recips msg-info))]
+       (let [msg (build-msg (:nick session)
+                            url
+                            (:msg-id msg-info)
+                            (:recips msg-info))]
          (add-message msg room)))
       [200 "OK"])))
 
@@ -1180,6 +1206,9 @@ WHERE  user_id IN
   (GET "/r/:room/log/:offset"              (validated-log session (params :room) (params :offset) params))
 
   (GET "/favicon.ico" (serve-static "static" "favicon.ico"))
+
+  (GET "/t/:topic" (topic session (params :topic)))
+  
   (GET "/u/:nick"  (redirect-to (str "/" (params :nick))))
   (GET "/u/:nick/" (redirect-to (str "/" (params :nick))))
   (GET "/u/:nick/tag/:tag" (tagged-dumps-by-nick session params (request-url request)))
diff --git a/src/user.clj b/src/user.clj
index 7641bd8..8380bce 100644
--- a/src/user.clj
+++ b/src/user.clj
@@ -18,7 +18,7 @@
 
 ;;; User info cache
 
-(def user-cache-size 500)
+(def user-cache-size 99999)
 (def user-nick-cache (ref {}))
 (def user-id-cache (ref {}))
 
diff --git a/src/utils.clj b/src/utils.clj
index 8aaffba..ca527ac 100755
--- a/src/utils.clj
+++ b/src/utils.clj
@@ -27,7 +27,6 @@
       db-name "dumpfm"
       db-user "postgres"
       db-pass "root"]
-  ; TODO: use c3p0 for pooling?
   (def *db* {:datasource (doto (new PGPoolingDataSource)
                            (.setServerName   db-host)
                            (.setDatabaseName db-name)
@@ -48,9 +47,8 @@
       (throw (Exception. (str "Invalid url " u))))))
 
 (defn get-ip [request]
-  (let [ip (get (:headers request) "x-real-ip")  ; behind nginx
-        ip (if ip ip (:remote-addr request))] (str ip))  ; deployed locally
-)
+  (let [ip (get (:headers request) "x-real-ip")           ; behind nginx
+        ip (if ip ip (:remote-addr request))] (str ip)))  ; deployed locally
 
 (defn append [& seqs]
   (reduce into (map vector seqs)))
@@ -70,6 +68,9 @@
 (defn stringify-and-escape [m]
   (zipmap (map str* (keys m)) (map escape-html-deep (vals m))))
 
+(defn keywordify [m]
+  (zipmap (map keyword (keys m)) (vals m)))
+
 (defn nor [& args]
   (not-any? identity args))
 
@@ -371,16 +372,15 @@
 (.setRefreshInterval template-group 10)
 
 (defn initialize-template [st session]
+  (.setAttribute st "domain" config/*server-url*)
   (if (session :nick)
     (doto st
       (.setAttribute "user_email" (session :email))
       (.setAttribute "user_nick" (session :nick))
       (.setAttribute "user_avatar" (if (non-empty-string? (session :avatar))
-				     (session :avatar) nil))
-      (.setAttribute "isadmin" (session :is_admin))
-      (.setAttribute "domain" config/*server-url*))
-    (doto st
-      (.setAttribute "domain" config/*server-url*))))
+                                     (session :avatar)
+                                     nil))
+      (.setAttribute "isadmin" (session :is_admin)))))
 
 (defn fetch-template [template session]
   (try
@@ -446,7 +446,7 @@
             { :result result :time (System/currentTimeMillis)})
           result)))))
 
-;; Taken from Programming Clojure by Stuart Halloway
+;; From Programming Clojure by Stuart Halloway
 
 (defn index-filter [pred coll]
   (for [[idx elt] (indexed coll) :when (pred elt)] idx))