From d453db89a7c1a3a67c8b3473b6ac93b5f1b90b29 Mon Sep 17 00:00:00 2001
From: Philipp Heckel <pheckel@datto.com>
Date: Mon, 8 Nov 2021 09:46:31 -0500
Subject: [PATCH] Add since=all; make poll=1 default to since=all

---
 server/cache.go         |  2 +-
 server/cache_mem.go     |  4 ++--
 server/cache_sqlite.go  |  4 ++--
 server/server.go        | 45 +++++++++++++++++++++++++++++++++--------
 server/static/js/app.js |  2 +-
 5 files changed, 43 insertions(+), 14 deletions(-)

diff --git a/server/cache.go b/server/cache.go
index df4c2c9e..fb13095c 100644
--- a/server/cache.go
+++ b/server/cache.go
@@ -7,7 +7,7 @@ import (
 
 type cache interface {
 	AddMessage(m *message) error
-	Messages(topic string, since time.Time) ([]*message, error)
+	Messages(topic string, since sinceTime) ([]*message, error)
 	MessageCount(topic string) (int, error)
 	Topics() (map[string]*topic, error)
 	Prune(keep time.Duration) error
diff --git a/server/cache_mem.go b/server/cache_mem.go
index 6c3a94f1..83b0f36d 100644
--- a/server/cache_mem.go
+++ b/server/cache_mem.go
@@ -29,7 +29,7 @@ func (s *memCache) AddMessage(m *message) error {
 	return nil
 }
 
-func (s *memCache) Messages(topic string, since time.Time) ([]*message, error) {
+func (s *memCache) Messages(topic string, since sinceTime) ([]*message, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if _, ok := s.messages[topic]; !ok {
@@ -38,7 +38,7 @@ func (s *memCache) Messages(topic string, since time.Time) ([]*message, error) {
 	messages := make([]*message, 0) // copy!
 	for _, m := range s.messages[topic] {
 		msgTime := time.Unix(m.Time, 0)
-		if msgTime == since || msgTime.After(since) {
+		if msgTime == since.Time() || msgTime.After(since.Time()) {
 			messages = append(messages, m)
 		}
 	}
diff --git a/server/cache_sqlite.go b/server/cache_sqlite.go
index 2a07f741..89e932e8 100644
--- a/server/cache_sqlite.go
+++ b/server/cache_sqlite.go
@@ -55,8 +55,8 @@ func (c *sqliteCache) AddMessage(m *message) error {
 	return err
 }
 
-func (c *sqliteCache) Messages(topic string, since time.Time) ([]*message, error) {
-	rows, err := c.db.Query(selectMessagesSinceTimeQuery, topic, since.Unix())
+func (c *sqliteCache) Messages(topic string, since sinceTime) ([]*message, error) {
+	rows, err := c.db.Query(selectMessagesSinceTimeQuery, topic, since.Time().Unix())
 	if err != nil {
 		return nil, err
 	}
diff --git a/server/server.go b/server/server.go
index 27f0eb85..1cb185cb 100644
--- a/server/server.go
+++ b/server/server.go
@@ -53,6 +53,25 @@ type indexPage struct {
 	CacheDuration string
 }
 
+type sinceTime time.Time
+
+func (t sinceTime) IsAll() bool {
+	return t == sinceAllMessages
+}
+
+func (t sinceTime) IsNone() bool {
+	return t == sinceNoMessages
+}
+
+func (t sinceTime) Time() time.Time {
+	return time.Time(t)
+}
+
+var (
+	sinceAllMessages = sinceTime(time.Unix(0, 0))
+	sinceNoMessages  = sinceTime(time.Unix(1, 0))
+)
+
 const (
 	messageLimit = 512
 )
@@ -318,8 +337,8 @@ func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, v *visi
 	}
 }
 
-func (s *Server) sendOldMessages(t *topic, since time.Time, sub subscriber) error {
-	if since.IsZero() {
+func (s *Server) sendOldMessages(t *topic, since sinceTime, sub subscriber) error {
+	if since.IsNone() {
 		return nil
 	}
 	messages, err := s.cache.Messages(t.id, since)
@@ -334,17 +353,27 @@ func (s *Server) sendOldMessages(t *topic, since time.Time, sub subscriber) erro
 	return nil
 }
 
-func parseSince(r *http.Request) (time.Time, error) {
+// parseSince returns a timestamp identifying the time span from which cached messages should be received.
+//
+// Values in the "since=..." parameter can be either a unix timestamp or a duration (e.g. 12h), or
+// "all" for all messages.
+func parseSince(r *http.Request) (sinceTime, error) {
 	if !r.URL.Query().Has("since") {
-		return time.Time{}, nil
+		if r.URL.Query().Has("poll") {
+			return sinceAllMessages, nil
+		}
+		return sinceNoMessages, nil
 	}
-	if since, err := strconv.ParseInt(r.URL.Query().Get("since"), 10, 64); err == nil {
-		return time.Unix(since, 0), nil
+	if r.URL.Query().Get("since") == "all" {
+		return sinceAllMessages, nil
+	}
+	if s, err := strconv.ParseInt(r.URL.Query().Get("since"), 10, 64); err == nil {
+		return sinceTime(time.Unix(s, 0)), nil
 	}
 	if d, err := time.ParseDuration(r.URL.Query().Get("since")); err == nil {
-		return time.Now().Add(-1 * d), nil
+		return sinceTime(time.Now().Add(-1 * d)), nil
 	}
-	return time.Time{}, errHTTPBadRequest
+	return sinceNoMessages, errHTTPBadRequest
 }
 
 func (s *Server) handleOptions(w http.ResponseWriter, r *http.Request) error {
diff --git a/server/static/js/app.js b/server/static/js/app.js
index d4de8de3..ea9e335e 100644
--- a/server/static/js/app.js
+++ b/server/static/js/app.js
@@ -118,7 +118,7 @@ const test = (topic) => {
 };
 
 const fetchCachedMessages = async (topic) => {
-    const topicJsonUrl = `/${topic}/json?poll=1&since=12h`; // Poll!
+    const topicJsonUrl = `/${topic}/json?poll=1`; // Poll!
     for await (let line of makeTextFileLineIterator(topicJsonUrl)) {
         const message = JSON.parse(line);
         topics[topic]['messages'].push(message);