refactor part of count package to datastore, needs more work for other metrics besides pageviews

pkg/count/count.go

@@ -2,7 +2,7 @@ package count
 
 import (
 	"database/sql"
-	"log"
+	log "github.com/sirupsen/logrus"
 	"time"
 
 	"github.com/usefathom/fathom/pkg/datastore"
@@ -32,16 +32,19 @@ func getLastArchivedDate() string {
 
 // Archive aggregates data into daily totals
 func Archive() {
-	lastArchived := getLastArchivedDate()
+	start := time.Now()
 
-	CreateVisitorTotals(lastArchived)
+	lastArchived := getLastArchivedDate()
 	CreatePageviewTotals(lastArchived)
+	CreateVisitorTotals(lastArchived)
 	CreateScreenTotals(lastArchived)
 	CreateLanguageTotals(lastArchived)
 	CreateBrowserTotals(lastArchived)
 	CreateReferrerTotals(lastArchived)
-
 	datastore.SetOption("last_archived", time.Now().Format("2006-01-02"))
+
+	end := time.Now()
+	log.Infof("finished aggregating metrics. ran for %dms.", (end.UnixNano()-start.UnixNano())/1000000)
 }
 
 // Save the Total in the given database connection + table

pkg/count/pageviews.go

@@ -1,50 +1,21 @@
 package count
 
-import "github.com/usefathom/fathom/pkg/datastore"
+import (
+	"github.com/usefathom/fathom/pkg/datastore"
+	"log"
+	"time"
+)
 
 // CreatePageviewTotals aggregates pageview data for each page into daily totals
 func CreatePageviewTotals(since string) {
-	stmt, err := datastore.DB.Prepare(`SELECT
+	tomorrow := time.Now().AddDate(0, 0, 1).Format("2006-02-01")
-      pv.page_id,
+	totals, err := datastore.PageviewCountPerPageAndDay(tomorrow, since)
-      COUNT(*) AS count,
+	if err != nil {
-			COUNT(DISTINCT(pv.visitor_id)) AS count_unique,
+		log.Fatal(err)
-			DATE_FORMAT(pv.timestamp, "%Y-%m-%d") AS date_group
+	}
-    FROM pageviews pv
+
-    WHERE pv.timestamp > ?
+	err = datastore.SavePageviewTotals(totals)
-    GROUP BY pv.page_id, date_group`)
+	if err != nil {
-	checkError(err)
+		log.Fatal(err)
-	defer stmt.Close()
-
-	rows, err := stmt.Query(since)
-	checkError(err)
-	defer rows.Close()
-
-	datastore.DB.Begin()
-
-	datastore.DB.Exec("START TRANSACTION")
-	for rows.Next() {
-		var t Total
-		err = rows.Scan(&t.PageID, &t.Count, &t.CountUnique, &t.Date)
-		checkError(err)
-
-		stmt, err := datastore.DB.Prepare(`INSERT INTO total_pageviews(
-	    page_id,
-	    count,
-			count_unique,
-	    date
-	    ) VALUES( ?, ?, ?, ? ) ON DUPLICATE KEY UPDATE count = ?, count_unique = ?`)
-		checkError(err)
-		defer stmt.Close()
-
-		_, err = stmt.Exec(
-			t.PageID,
-			t.Count,
-			t.CountUnique,
-			t.Date,
-			t.Count,
-			t.CountUnique,
-		)
-		checkError(err)
 	}
-	datastore.DB.Exec("COMMIT")
 }

pkg/datastore/datastore.go

@@ -38,6 +38,11 @@ func New(driver string, dsn string) *sqlx.DB {
 	return dbx
 }
 
+// Get returns the underlying sqlx.DB instance. Use at your own risk.
+func Get() *sqlx.DB {
+	return dbx
+}
+
 func getDSN(driver string, host string, name string, user string, password string) string {
 	var dsn = fmt.Sprintf("%s:%s@%s/%s", user, password, host, name)
 

pkg/datastore/pageviews.go

@@ -42,3 +42,17 @@ func SavePageviews(pvs []*models.Pageview) error {
 	err = tx.Commit()
 	return err
 }
+
+func PageviewCountPerPageAndDay(before string, after string) ([]*models.Total, error) {
+	query := dbx.Rebind(`SELECT
+	    	pv.page_id,
+	    	COUNT(*) AS count,
+			COUNT(DISTINCT(pv.visitor_id)) AS count_unique,
+			DATE_FORMAT(pv.timestamp, "%Y-%m-%d") AS date_group
+	    FROM pageviews pv
+	    WHERE pv.timestamp < ? AND pv.timestamp > ?
+	    GROUP BY pv.page_id, date_group`)
+	var results []*models.Total
+	err := dbx.Select(&results, query, before, after)
+	return results, err
+}

pkg/datastore/total_pageviews.go

@@ -59,3 +59,30 @@ func TotalPageviewsPerPage(before int64, after int64, limit int64) ([]*models.Pa
 
 	return results, nil
 }
+
+func SavePageviewTotals(totals []*models.Total) error {
+	tx, err := dbx.Begin()
+	if err != nil {
+		return nil
+	}
+
+	query := dbx.Rebind(`INSERT INTO total_pageviews( page_id, count, count_unique, date ) VALUES( ?, ?, ?, ? ) ON DUPLICATE KEY UPDATE count = ?, count_unique = ?`)
+	stmt, err := tx.Prepare(query)
+	if err != nil {
+		return err
+	}
+
+	for _, t := range totals {
+		_, err = stmt.Exec(
+			t.PageID,
+			t.Count,
+			t.CountUnique,
+			t.Date,
+			t.Count,
+			t.CountUnique,
+		)
+	}
+
+	err = tx.Commit()
+	return err
+}

pkg/models/total.go

@@ -0,0 +1,11 @@
+package models
+
+// Total represents a daily aggregated total for a metric
+type Total struct {
+	ID          int64
+	PageID      int64 `db:"page_id"`
+	Value       string
+	Count       int64
+	CountUnique int64  `db:"count_unique"`
+	Date        string `db:"date_group"`
+}