implement results type container

pkg/aggregator/aggregator.go

@@ -1,4 +1,4 @@
-package counter
+package aggregator
 
 import (
 	"time"
@@ -9,39 +9,61 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-func Aggregate() error {
+func Run() {
 	// Get unprocessed pageviews
 	pageviews, err := datastore.GetProcessablePageviews()
 	if err != nil && err != datastore.ErrNoResults {
 		log.Error(err)
-		return err
+		return
 	}
 
 	//  Do we have anything to process?
 	if len(pageviews) == 0 {
-		return nil
+		return
 	}
 
-	sites := map[string]*models.SiteStats{}
-	pages := map[string]*models.PageStats{}
-	referrers := map[string]*models.ReferrerStats{}
+	results := Process(pageviews)
+
+	// update stats
+	for _, site := range results.Sites {
+		err = datastore.UpdateSiteStats(site)
+		if err != nil {
+			log.Error(err)
+		}
+	}
+
+	for _, pageStats := range results.Pages {
+		err = datastore.UpdatePageStats(pageStats)
+		if err != nil {
+			log.Error(err)
+		}
+	}
+
+	for _, referrerStats := range results.Referrers {
+		err = datastore.UpdateReferrerStats(referrerStats)
+		if err != nil {
+			log.Error(err)
+		}
+	}
+
+	// finally, remove pageviews that we just processed
+	err = datastore.DeletePageviews(pageviews)
+	if err != nil {
+		log.Error(err)
+	}
+}
+
+func Process(pageviews []*models.Pageview) *Results {
+	log.Debugf("processing %d pageviews", len(pageviews))
+	results := NewResults()
 
 	for _, p := range pageviews {
-		date := p.Timestamp.Format("2006-01-02")
-
-		var site *models.SiteStats
-		var ok bool
-		if site, ok = sites[date]; !ok {
-			site, err = getSiteStats(p.Timestamp)
-			if err != nil {
-				log.Error(err)
-				continue // TODO: Pageview should not be deleted if this happens
-			}
-
-			sites[date] = site
+		site, err := results.GetSiteStats(p.Timestamp)
+		if err != nil {
+			log.Error(err)
+			continue
 		}
 
-		// site stats
 		site.Pageviews += 1
 
 		// TODO: Weight isn't right here because we need the number of pageview with a known time of page, not all pageviews
@@ -63,15 +85,10 @@ func Aggregate() error {
 			}
 		}
 
-		// page stats
-		var pageStats *models.PageStats
-		if pageStats, ok = pages[date+p.Pathname]; !ok {
-			pageStats, err = getPageStats(p.Timestamp, p.Hostname, p.Pathname)
-			if err != nil {
-				log.Error(err)
-				continue
-			}
-			pages[date+p.Hostname+p.Pathname] = pageStats
+		pageStats, err := results.GetPageStats(p.Timestamp, p.Hostname, p.Pathname)
+		if err != nil {
+			log.Error(err)
+			continue
 		}
 
 		pageStats.Pageviews += 1
@@ -95,15 +112,10 @@ func Aggregate() error {
 
 		// referrer stats
 		if p.Referrer != "" {
-			var referrerStats *models.ReferrerStats
-			var ok bool
-			if referrerStats, ok = referrers[date+p.Referrer]; !ok {
-				referrerStats, err = getReferrerStats(p.Timestamp, p.Referrer)
-				if err != nil {
-					log.Error(err)
-					continue
-				}
-				referrers[date+p.Referrer] = referrerStats
+			referrerStats, err := results.GetReferrerStats(p.Timestamp, p.Referrer)
+			if err != nil {
+				log.Error(err)
+				continue
 			}
 
 			referrerStats.Pageviews += 1
@@ -126,41 +138,11 @@ func Aggregate() error {
 
 	}
 
-	// update stats
-	for _, site := range sites {
-		err = datastore.UpdateSiteStats(site)
-		if err != nil {
-			log.Error(err)
-		}
-	}
-
-	for _, pageStats := range pages {
-		err = datastore.UpdatePageStats(pageStats)
-		if err != nil {
-			log.Error(err)
-			return err
-		}
-	}
-
-	for _, referrerStats := range referrers {
-		err = datastore.UpdateReferrerStats(referrerStats)
-		if err != nil {
-			log.Error(err)
-			return err
-		}
-	}
-
-	// finally, remove pageviews that we just processed
-	err = datastore.DeletePageviews(pageviews)
-	if err != nil {
-		log.Error(err)
-	}
-
-	return nil
+	return results
 }
 
-func getSiteStats(date time.Time) (*models.SiteStats, error) {
-	stats, err := datastore.GetSiteStats(date)
+func getSiteStats(t time.Time) (*models.SiteStats, error) {
+	stats, err := datastore.GetSiteStats(t)
 	if err != nil && err != datastore.ErrNoResults {
 		return nil, err
 	}
@@ -170,7 +152,7 @@ func getSiteStats(date time.Time) (*models.SiteStats, error) {
 	}
 
 	stats = &models.SiteStats{
-		Date: date,
+		Date: t,
 	}
 	err = datastore.InsertSiteStats(stats)
 	return stats, err

pkg/aggregator/aggregator_test.go

@@ -0,0 +1,9 @@
+package aggregator
+
+import (
+	"testing"
+)
+
+func TestProcess(t *testing.T) {
+
+}

pkg/aggregator/result.go

@@ -0,0 +1,70 @@
+package aggregator
+
+import (
+	"time"
+
+	"github.com/usefathom/fathom/pkg/models"
+)
+
+type Results struct {
+	Sites     map[string]*models.SiteStats
+	Pages     map[string]*models.PageStats
+	Referrers map[string]*models.ReferrerStats
+}
+
+func NewResults() *Results {
+	return &Results{
+		Sites:     map[string]*models.SiteStats{},
+		Pages:     map[string]*models.PageStats{},
+		Referrers: map[string]*models.ReferrerStats{},
+	}
+}
+
+func (r *Results) GetSiteStats(t time.Time) (*models.SiteStats, error) {
+	var stats *models.SiteStats
+	var ok bool
+	var err error
+
+	date := t.Format("2006-01-02")
+	if stats, ok = r.Sites[date]; !ok {
+		stats, err = getSiteStats(t)
+		if err != nil {
+			return nil, err
+		}
+		r.Sites[date] = stats
+	}
+	return stats, nil
+}
+
+func (r *Results) GetPageStats(t time.Time, hostname string, pathname string) (*models.PageStats, error) {
+	var stats *models.PageStats
+	var ok bool
+	var err error
+
+	date := t.Format("2006-01-02")
+	if stats, ok = r.Pages[date+hostname+pathname]; !ok {
+		stats, err = getPageStats(t, hostname, pathname)
+		if err != nil {
+			return nil, err
+		}
+		r.Pages[date+hostname+pathname] = stats
+	}
+	return stats, nil
+}
+
+func (r *Results) GetReferrerStats(t time.Time, referrer string) (*models.ReferrerStats, error) {
+	var stats *models.ReferrerStats
+	var ok bool
+	var err error
+
+	date := t.Format("2006-01-02")
+	if stats, ok = r.Referrers[date+referrer]; !ok {
+		stats, err = getReferrerStats(t, referrer)
+		if err != nil {
+			return nil, err
+		}
+		r.Referrers[date+referrer] = stats
+	}
+
+	return stats, nil
+}

pkg/api/collect.go

@@ -8,7 +8,7 @@ import (
 	"time"
 
 	"github.com/mssola/user_agent"
-	"github.com/usefathom/fathom/pkg/counter"
+	"github.com/usefathom/fathom/pkg/aggregator"
 	"github.com/usefathom/fathom/pkg/datastore"
 	"github.com/usefathom/fathom/pkg/models"
 )
@@ -98,13 +98,13 @@ func NewCollectHandler() http.Handler {
 
 // runs the aggregate func every minute
 func aggregate() {
-	counter.Aggregate()
+	aggregator.Run()
 	timeout := 1 * time.Minute
 
 	for {
 		select {
 		case <-time.After(timeout):
-			counter.Aggregate()
+			aggregator.Run()
 		}
 	}
 }