Add redis and MarkRead

cmd/server/main.go

@@ -24,16 +24,17 @@ import (
 	"log"
 	"net/http"
 	"strings"
+	"time"
 
-	//	"github.com/garyburd/redigo/redis"
+	"github.com/garyburd/redigo/redis"
 	"github.com/pstuifzand/microsub-server/microsub"
 	"willnorris.com/go/microformats"
 )
 
 var (
-	// pool        redis.Pool
+	pool        *redis.Pool
-	port int
+	port        int
-	// redisServer = flag.String("redis", "redis:6379", "")
+	redisServer = flag.String("redis", "redis:6379", "")
 )
 
 func init() {
@@ -63,9 +64,6 @@ func simplify(itemType string, item map[string][]interface{}) map[string]interfa
 				if text, e := content["value"]; e {
 					delete(content, "value")
 					content["text"] = text
-					// if _, e := content["html"]; !e {
-					// 	content["text"] = text
-					// }
 				}
 				feedItem[k] = content
 			}
@@ -197,9 +195,6 @@ func (h *microsubHandler) checkAuthToken(header string, token *TokenResponse) bo
 }
 
 func (h *microsubHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	//conn := pool.Get()
-	//defer conn.Close()
-
 	authorization := r.Header.Get("Authorization")
 
 	var token TokenResponse
@@ -310,13 +305,13 @@ func (h *microsubHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	return
 }
 
-// func newPool(addr string) *redis.Pool {
+func newPool(addr string) *redis.Pool {
-// 	return &redis.Pool{
+	return &redis.Pool{
-// 		MaxIdle:     3,
+		MaxIdle:     3,
-// 		IdleTimeout: 240 * time.Second,
+		IdleTimeout: 240 * time.Second,
-// 		Dial:        func() (redis.Conn, error) { return redis.Dial("tcp", addr) },
+		Dial:        func() (redis.Conn, error) { return redis.Dial("tcp", addr) },
-// 	}
+	}
-// }
+}
 
 func main() {
 	flag.Parse()
@@ -330,16 +325,19 @@ func main() {
 		}
 	}
 
+	pool = newPool(*redisServer)
+
+	conn := pool.Get()
+	defer conn.Close()
+
 	var backend microsub.Microsub
 
 	if createBackend {
 		backend = createMemoryBackend()
 	} else {
-		backend = loadMemoryBackend()
+		backend = loadMemoryBackend(conn)
 	}
 
-	//pool = newPool(*redisServer)
-
 	http.Handle("/microsub", &microsubHandler{backend})
 	log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", port), nil))
 }

cmd/server/memory.go

@@ -20,14 +20,17 @@ package main
 import (
 	"encoding/json"
 	"fmt"
+	"log"
 	"os"
 	"sort"
 	"strings"
 
+	"github.com/garyburd/redigo/redis"
 	"github.com/pstuifzand/microsub-server/microsub"
 )
 
 type memoryBackend struct {
+	Redis    redis.Conn
 	Channels map[string]microsub.Channel
 	Feeds    map[string][]microsub.Feed
 	NextUid  int
@@ -57,8 +60,9 @@ func (b *memoryBackend) save() {
 	jw.Encode(b)
 }
 
-func loadMemoryBackend() microsub.Microsub {
+func loadMemoryBackend(conn redis.Conn) microsub.Microsub {
 	backend := &memoryBackend{}
+	backend.Redis = conn
 	backend.load()
 
 	return backend
@@ -173,6 +177,10 @@ func (b *memoryBackend) TimelineGet(after, before, channel string) microsub.Time
 
 			// Filter items with "published" date
 			for _, r := range results {
+				if b.wasRead(channel, r) {
+					continue
+				}
+
 				if _, e := r["published"]; e {
 					items = append(items, r)
 				}
@@ -192,13 +200,34 @@ func (b *memoryBackend) TimelineGet(after, before, channel string) microsub.Time
 	}
 }
 
+func (b *memoryBackend) checkRead(channel string, uid string) bool {
+	args := redis.Args{}.Add(fmt.Sprintf("timeline:%s:read", channel)).Add(uid)
+	member, err := redis.Bool(b.Redis.Do("SISMEMBER", args...))
+	if err != nil {
+		log.Printf("Checking read for channel %s item %s has failed\n", channel, uid)
+	}
+	return member
+}
+
+func (b *memoryBackend) wasRead(channel string, item map[string]interface{}) bool {
+	if uid, e := item["uid"]; e {
+		return b.checkRead(channel, uid.(string))
+	}
+
+	if uid, e := item["url"]; e {
+		return b.checkRead(channel, uid.(string))
+	}
+
+	return false
+}
+
 func (b *memoryBackend) FollowGetList(uid string) []microsub.Feed {
 	return b.Feeds[uid]
 }
 
 func (b *memoryBackend) FollowURL(uid string, url string) microsub.Feed {
 	defer b.save()
-	feed := microsub.Feed{"feed", url}
+	feed := microsub.Feed{Type: "feed", URL: url}
 	b.Feeds[uid] = append(b.Feeds[uid], feed)
 	return feed
 }
@@ -218,15 +247,15 @@ func (b *memoryBackend) UnfollowURL(uid string, url string) {
 	}
 }
 
-// TODO: improve search for feeds
+// TODO: improve search for feeds, perhaps even with mf2 parser
 func (b *memoryBackend) Search(query string) []microsub.Feed {
 	return []microsub.Feed{
-		microsub.Feed{"feed", query},
+		microsub.Feed{Type: "feed", URL: query},
 	}
 }
 
-func (b *memoryBackend) PreviewURL(previewUrl string) microsub.Timeline {
+func (b *memoryBackend) PreviewURL(previewURL string) microsub.Timeline {
-	md, err := Fetch2(previewUrl)
+	md, err := Fetch2(previewURL)
 	if err != nil {
 		return microsub.Timeline{}
 	}
@@ -235,3 +264,10 @@ func (b *memoryBackend) PreviewURL(previewUrl string) microsub.Timeline {
 		Items: results,
 	}
 }
+
+func (b *memoryBackend) MarkRead(channel string, itemUids []string) {
+	args := redis.Args{}.Add(fmt.Sprintf("timeline:%s:read", channel)).AddFlat(itemUids)
+	if _, err := b.Redis.Do("SADD", args...); err != nil {
+		log.Printf("Marking read for channel %s has failed\n", channel)
+	}
+}

cmd/server/null.go

@@ -84,3 +84,6 @@ func (b *NullBackend) PreviewURL(url string) microsub.Timeline {
 		Items:  []map[string]interface{}{},
 	}
 }
+
+func (b *NullBackend) MarkRead(channel string, uids []string) {
+}

microsub/protocol.go

@@ -93,6 +93,8 @@ type Microsub interface {
 
 	TimelineGet(before, after, channel string) Timeline
 
+	MarkRead(channel string, entry []string)
+
 	FollowGetList(uid string) []Feed
 	FollowURL(uid string, url string) Feed