fold filtering into db layer

Author: abr-egn

server/activity_handler.go

@@ -2,7 +2,6 @@ package server
 
 import (
 	"github.com/labstack/echo"
-	log "github.com/sirupsen/logrus"
 )
 
 type jsonResponseActivity struct {
@@ -17,7 +16,21 @@ type jsonResponse struct {
 
 func ActivityHandler(db Database) echo.HandlerFunc {
 	return func(c echo.Context) error {
-		activity, next, err := fetchActivity(db, LocaleForRequest(c.Request()), c.QueryParam("next"), c.QueryParam("nohelp") == "true")
+		locale := LocaleForRequest(c.Request())
+		filter := func(a Activity) bool {
+			return true
+		}
+		if c.QueryParam("nohelp") == "true" {
+			filter = func(a Activity) bool {
+				if fp, ok := a.(*ForumPost); ok {
+					if fp.ForumId == locale.HelpForumId {
+						return false
+					}
+				}
+				return true
+			}
+		}
+		activity, next, err := db.Activity(locale, c.QueryParam("next"), 50, filter)
 		if err != nil {
 			return err
 		}
@@ -42,51 +55,3 @@ func ActivityHandler(db Database) echo.HandlerFunc {
 		return c.JSON(200, response)
 	}
 }
-
-const MinPageSize = 50
-const DbRequestSize = 50
-const MaxDbRequests = 10
-
-func fetchActivity(db Database, locale *Locale, start string, nohelp bool) ([]Activity, string, error) {
-	activity := []Activity{}
-	next := start
-	pred := func(a Activity) bool {
-		return true
-	}
-	if nohelp {
-		pred = func(a Activity) bool {
-			if fp, ok := a.(*ForumPost); ok {
-				if fp.ForumId == locale.HelpForumId {
-					return false
-				}
-			}
-			return true
-		}
-	}
-	for i := 0; i < MaxDbRequests && len(activity) < MinPageSize; i++ {
-		as, n, err := db.Activity(locale, next, DbRequestSize)
-		if err != nil {
-			return nil, "", err
-		}
-		next = n
-		if len(as) == 0 && n == "" {
-			log.Debug("end of activity db")
-			break
-		}
-		filtered := 0
-		for _, a := range as {
-			if pred(a) {
-				activity = append(activity, a)
-			} else {
-				filtered++
-			}
-		}
-		log.WithFields(log.Fields{
-			"count":    len(as),
-			"buffered": len(activity),
-			"filtered": filtered,
-			"next":     next,
-		}).Debug("processed activity batch")
-	}
-	return activity, next, nil
-}

server/bolt_database.go

@@ -43,7 +43,7 @@ func (db *BoltDatabase) AddActivity(activity []Activity) error {
 	})
 }
 
-func (db *BoltDatabase) Activity(locale *Locale, start string, count int) ([]Activity, string, error) {
+func (db *BoltDatabase) Activity(locale *Locale, start string, count int, filter func(a Activity) bool) ([]Activity, string, error) {
 	ret := []Activity(nil)
 	next := ""
 	if err := db.db.View(func(tx *bolt.Tx) error {
@@ -66,7 +66,7 @@ func (db *BoltDatabase) Activity(locale *Locale, start string, count int) ([]Act
 			activity, err := unmarshalActivity(k, v)
 			if err != nil {
 				return err
-			} else if activity != nil && locale.ActivityFilter(activity) {
+			} else if activity != nil && locale.ActivityFilter(activity) && filter(activity) {
 				ret = append(ret, activity)
 				next = base64.RawURLEncoding.EncodeToString(k)
 			}

server/database.go

@@ -17,7 +17,7 @@ const (
 
 type Database interface {
 	AddActivity(activity []Activity) error
-	Activity(locale *Locale, start string, count int) ([]Activity, string, error)
+	Activity(locale *Locale, start string, count int, filter func(a Activity) bool) ([]Activity, string, error)
 	Close() error
 }
 

server/database_test.go

@@ -31,21 +31,23 @@ func testDatabase_ForumPosts(t *testing.T, db Database) {
 
 	db.AddActivity([]Activity{post1, post2})
 
-	posts, next, err := db.Activity(locale, "", 1)
+	all := func(a Activity) bool { return true }
+
+	posts, next, err := db.Activity(locale, "", 1, all)
 	require.NoError(t, err)
 	require.Equal(t, 1, len(posts))
 	assert.Equal(t, post1.Id, posts[0].(*ForumPost).Id)
 	assert.Equal(t, post1.Poster, posts[0].(*ForumPost).Poster)
 	assert.Equal(t, post1.Time.Unix(), posts[0].(*ForumPost).Time.Unix())
 
-	posts, next, err = db.Activity(locale, next, 1)
+	posts, next, err = db.Activity(locale, next, 1, all)
 	require.NoError(t, err)
 	require.Equal(t, 1, len(posts))
 	assert.Equal(t, post2.Id, posts[0].(*ForumPost).Id)
 	assert.Equal(t, post2.Poster, posts[0].(*ForumPost).Poster)
 	assert.Equal(t, post2.Time.Unix(), posts[0].(*ForumPost).Time.Unix())
 
-	posts, _, err = db.Activity(locale, next, 1)
+	posts, _, err = db.Activity(locale, next, 1, all)
 	require.NoError(t, err)
 	require.Equal(t, 0, len(posts))
 }

server/dynamodb_database.go

@@ -82,7 +82,7 @@ func (db *DynamoDBDatabase) AddActivity(activity []Activity) error {
 	return nil
 }
 
-func (db *DynamoDBDatabase) Activity(locale *Locale, start string, count int) ([]Activity, string, error) {
+func (db *DynamoDBDatabase) Activity(locale *Locale, start string, count int, filter func(a Activity) bool) ([]Activity, string, error) {
 	var activity []Activity
 
 	var startKey map[string]dynamodb.AttributeValue
@@ -120,7 +120,7 @@ func (db *DynamoDBDatabase) Activity(locale *Locale, start string, count int) ([
 		for _, item := range result.Items {
 			if a, err := unmarshalActivity(item["rk"].B, item["v"].B); err != nil {
 				return nil, "", err
-			} else if a != nil {
+			} else if a != nil && filter(a) {
 				activity = append(activity, a)
 			}
 		}

server/rss_handler.go

@@ -43,7 +43,7 @@ type rssResponse struct {
 
 func RSSHandler(db Database) echo.HandlerFunc {
 	return func(c echo.Context) error {
-		activity, _, err := db.Activity(LocaleForRequest(c.Request()), c.QueryParam("next"), 50)
+		activity, _, err := db.Activity(LocaleForRequest(c.Request()), c.QueryParam("next"), 50, func(a Activity) bool { return true })
 		if err != nil {
 			return err
 		}