add dynamodb support

aws-sam/README.md

@@ -0,0 +1,3 @@
+# aws-sam
+
+This directory contains resources for deploying the tracker using the AWS Serverless Application Model (SAM). By deploying via serverless technologies, you can create a cost-effective (free or nearly free) scalable deployment.

aws-sam/gggtracker.cfn.yaml

@@ -0,0 +1,19 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Description: github.com/ccbrown/gggtracker
+Resources:
+  DynamoDBTable:
+    Type: AWS::DynamoDB::Table
+    Properties:
+      AttributeDefinitions:
+        - AttributeName: hk
+          AttributeType: B
+        - AttributeName: rk
+          AttributeType: B
+      KeySchema:
+        - AttributeName: hk
+          KeyType: HASH
+        - AttributeName: rk
+          KeyType: RANGE
+      ProvisionedThroughput:
+        ReadCapacityUnits: 25
+        WriteCapacityUnits: 25

main.go

@@ -4,7 +4,10 @@ import (
 	"fmt"
 	"net/http"
 	"path"
+	"strings"
 
+	"github.com/aws/aws-sdk-go-v2/aws/external"
+	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
 	"github.com/labstack/echo"
 	"github.com/labstack/echo/middleware"
 	log "github.com/sirupsen/logrus"
@@ -18,17 +21,29 @@ func main() {
 	pflag.IntP("port", "p", 8080, "the port to listen on")
 	pflag.String("staticdir", "./server/static", "the static files to serve")
 	pflag.String("ga", "", "a google analytics account")
-	pflag.String("db", "./gggtracker.db", "the database path")
+	pflag.String("db", "./gggtracker.db", "the database file path")
+	pflag.String("dynamodb-table", "", "if given, DynamoDB will be used instead of a database file")
 	pflag.String("forumsession", "", "the POESESSID cookie for a forum session")
 	viper.BindPFlags(pflag.CommandLine)
 	pflag.Parse()
 
 	viper.SetEnvPrefix("gggtracker")
+	viper.SetEnvKeyReplacer(strings.NewReplacer("-", "_"))
 	viper.AutomaticEnv()
 
 	e := echo.New()
 
-	db, err := server.OpenDatabase(viper.GetString("db"))
+	var db server.Database
+	var err error
+	if tableName := viper.GetString("dynamodb-table"); tableName != "" {
+		config, err := external.LoadDefaultAWSConfig()
+		if err != nil {
+			log.Fatal(err)
+		}
+		db, err = server.NewDynamoDBDatabase(dynamodb.New(config), tableName)
+	} else {
+		db, err = server.NewBoltDatabase(viper.GetString("db"))
+	}
 	if err != nil {
 		log.Fatal(err)
 	}

server/activity_handler.go

@@ -14,9 +14,12 @@ type jsonResponse struct {
 	Next     string                  `json:"next"`
 }
 
-func ActivityHandler(db *Database) echo.HandlerFunc {
+func ActivityHandler(db Database) echo.HandlerFunc {
 	return func(c echo.Context) error {
-		activity, next := db.Activity(c.QueryParam("next"), 50, LocaleForRequest(c.Request()).ActivityFilter)
+		activity, next, err := db.Activity(LocaleForRequest(c.Request()), c.QueryParam("next"), 50)
+		if err != nil {
+			return err
+		}
 		response := jsonResponse{
 			Next: next,
 		}

server/bolt_database.go

@@ -0,0 +1,84 @@
+package server
+
+import (
+	"encoding/base64"
+
+	"github.com/boltdb/bolt"
+)
+
+type BoltDatabase struct {
+	db *bolt.DB
+}
+
+func NewBoltDatabase(path string) (*BoltDatabase, error) {
+	db, err := bolt.Open(path, 0600, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	db.Update(func(tx *bolt.Tx) error {
+		_, err := tx.CreateBucketIfNotExists([]byte("activity"))
+		if err != nil {
+			return err
+		}
+		return nil
+	})
+
+	return &BoltDatabase{
+		db: db,
+	}, nil
+}
+
+func (db *BoltDatabase) AddActivity(activity []Activity) error {
+	return db.db.Update(func(tx *bolt.Tx) error {
+		b := tx.Bucket([]byte("activity"))
+		for _, a := range activity {
+			k, v, err := marshalActivity(a)
+			if err != nil {
+				return err
+			}
+			b.Put(k, v)
+		}
+		return nil
+	})
+}
+
+func (db *BoltDatabase) Activity(locale *Locale, start string, count int) ([]Activity, string, error) {
+	ret := []Activity(nil)
+	next := ""
+	if err := db.db.View(func(tx *bolt.Tx) error {
+		c := tx.Bucket([]byte("activity")).Cursor()
+		var k, v []byte
+		if start == "" {
+			k, v = c.Last()
+		} else {
+			s, err := base64.RawURLEncoding.DecodeString(start)
+			if err != nil {
+				k, v = c.Last()
+			} else {
+				k, v = c.Seek(s)
+				if k != nil {
+					k, v = c.Prev()
+				}
+			}
+		}
+		for len(ret) < count && k != nil {
+			activity, err := unmarshalActivity(k, v)
+			if err != nil {
+				return err
+			} else if locale.ActivityFilter(activity) {
+				ret = append(ret, activity)
+				next = base64.RawURLEncoding.EncodeToString(k)
+			}
+			k, v = c.Prev()
+		}
+		return nil
+	}); err != nil {
+		return nil, "", err
+	}
+	return ret, next, nil
+}
+
+func (db *BoltDatabase) Close() error {
+	return db.db.Close()
+}

server/bolt_database_test.go

@@ -0,0 +1,22 @@
+package server
+
+import (
+	"io/ioutil"
+	"os"
+	"path"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestBoltDatabase(t *testing.T) {
+	dir, err := ioutil.TempDir("testdata", "db")
+	require.NoError(t, err)
+	defer os.RemoveAll(dir)
+
+	db, err := NewBoltDatabase(path.Join(dir, "test.db"))
+	require.NoError(t, err)
+	defer db.Close()
+
+	testDatabase(t, db)
+}

server/database.go

@@ -1,133 +1,71 @@
 package server
 
 import (
-	"encoding/base64"
 	"encoding/binary"
-	"encoding/json"
+	"fmt"
 
-	"github.com/boltdb/bolt"
+	json "github.com/json-iterator/go"
 )
 
-type Database struct {
-	db *bolt.DB
-}
-
-func OpenDatabase(path string) (*Database, error) {
-	db, err := bolt.Open(path, 0600, nil)
-	if err != nil {
-		return nil, err
-	}
-
-	db.Update(func(tx *bolt.Tx) error {
-		_, err := tx.CreateBucketIfNotExists([]byte("activity"))
-		if err != nil {
-			return err
-		}
-		return nil
-	})
-
-	return &Database{
-		db: db,
-	}, nil
-}
-
-func (db *Database) Close() {
-	db.db.Close()
-}
-
 const (
 	ForumPostType = iota
 	RedditCommentType
 	RedditPostType
 )
 
-func (db *Database) AddActivity(activity []Activity) {
-	err := db.db.Update(func(tx *bolt.Tx) error {
-		b := tx.Bucket([]byte("activity"))
-		for _, a := range activity {
-			buf, err := json.Marshal(a)
-			if err != nil {
-				return err
-			}
-			k := make([]byte, 10)
-			binary.BigEndian.PutUint64(k, uint64(a.ActivityTime().Unix())<<24)
-			switch a.(type) {
-			case *ForumPost:
-				k[5] = ForumPostType
-			case *RedditComment:
-				k[5] = RedditCommentType
-			case *RedditPost:
-				k[5] = RedditPostType
-			}
-			binary.BigEndian.PutUint32(k[6:], a.ActivityKey())
-			b.Put(k, buf)
-		}
-		return nil
-	})
+type Database interface {
+	AddActivity(activity []Activity) error
+	Activity(locale *Locale, start string, count int) ([]Activity, string, error)
+	Close() error
+}
+
+func marshalActivity(a Activity) (key, value []byte, err error) {
+	buf, err := json.Marshal(a)
 	if err != nil {
-		panic(err)
+		return nil, nil, err
 	}
+	k := make([]byte, 10)
+	binary.BigEndian.PutUint64(k, uint64(a.ActivityTime().Unix())<<24)
+	switch a.(type) {
+	case *ForumPost:
+		k[5] = ForumPostType
+	case *RedditComment:
+		k[5] = RedditCommentType
+	case *RedditPost:
+		k[5] = RedditPostType
+	}
+	binary.BigEndian.PutUint32(k[6:], a.ActivityKey())
+	return k, buf, nil
 }
 
-func (db *Database) Activity(start string, count int, filter func(Activity) bool) ([]Activity, string) {
-	ret := []Activity(nil)
-	next := ""
-	err := db.db.View(func(tx *bolt.Tx) error {
-		c := tx.Bucket([]byte("activity")).Cursor()
-		var k, v []byte
-		if start == "" {
-			k, v = c.Last()
-		} else {
-			s, err := base64.RawURLEncoding.DecodeString(start)
-			if err != nil {
-				k, v = c.Last()
-			} else {
-				k, v = c.Seek(s)
-				if k != nil {
-					k, v = c.Prev()
-				}
-			}
+func unmarshalActivity(key, value []byte) (Activity, error) {
+	switch key[5] {
+	case ForumPostType:
+		post := &ForumPost{}
+		err := json.Unmarshal(value, post)
+		if err != nil {
+			return nil, err
 		}
-		for len(ret) < count && k != nil {
-			var activity Activity
-			switch k[5] {
-			case ForumPostType:
-				post := &ForumPost{}
-				err := json.Unmarshal(v, post)
-				if err != nil {
-					return err
-				}
-				if post.Host == "" {
-					post.Host = "www.pathofexile.com"
-				}
-				if post.Id != 0 {
-					activity = post
-				}
-			case RedditCommentType:
-				comment := &RedditComment{}
-				err := json.Unmarshal(v, comment)
-				if err != nil {
-					return err
-				}
-				activity = comment
-			case RedditPostType:
-				post := &RedditPost{}
-				err := json.Unmarshal(v, post)
-				if err != nil {
-					return err
-				}
-				activity = post
-			}
-			if activity != nil && (filter == nil || filter(activity)) {
-				ret = append(ret, activity)
-				next = base64.RawURLEncoding.EncodeToString(k)
-			}
-			k, v = c.Prev()
+		if post.Host == "" {
+			post.Host = "www.pathofexile.com"
 		}
-		return nil
-	})
-	if err != nil {
-		panic(err)
+		if post.Id != 0 {
+			return post, nil
+		}
+	case RedditCommentType:
+		comment := &RedditComment{}
+		err := json.Unmarshal(value, comment)
+		if err != nil {
+			return nil, err
+		}
+		return comment, nil
+	case RedditPostType:
+		post := &RedditPost{}
+		err := json.Unmarshal(value, post)
+		if err != nil {
+			return nil, err
+		}
+		return post, nil
 	}
-	return ret, next
+	return nil, fmt.Errorf("invalid activity key")
 }