Merge pull request nyaruka#39 from tybritten/master

Allow for Configurable Archive Period

archiver.go

@@ -105,7 +105,7 @@ WHERE o.is_active = TRUE order by o.id
 `
 
 // GetActiveOrgs returns the active organizations sorted by id
-func GetActiveOrgs(ctx context.Context, db *sqlx.DB) ([]Org, error) {
+func GetActiveOrgs(ctx context.Context, db *sqlx.DB, conf *Config) ([]Org, error) {
 	ctx, cancel := context.WithTimeout(ctx, time.Minute)
 	defer cancel()
 
@@ -117,7 +117,7 @@ func GetActiveOrgs(ctx context.Context, db *sqlx.DB) ([]Org, error) {
 
 	orgs := make([]Org, 0, 10)
 	for rows.Next() {
-		org := Org{ActiveDays: 90}
+		org := Org{ActiveDays: conf.ArchiveLength}
 		err = rows.StructScan(&org)
 		if err != nil {
 			return nil, errors.Wrapf(err, "error scanning active org")

archiver_test.go

@@ -34,7 +34,8 @@ func TestGetMissingDayArchives(t *testing.T) {
 
 	// get the tasks for our org
 	ctx := context.Background()
-	orgs, err := GetActiveOrgs(ctx, db)
+	config := NewConfig()
+	orgs, err := GetActiveOrgs(ctx, db, config)
 	assert.NoError(t, err)
 
 	now := time.Date(2018, 1, 8, 12, 30, 0, 0, time.UTC)
@@ -58,14 +59,31 @@ func TestGetMissingDayArchives(t *testing.T) {
 	assert.Equal(t, time.Date(2017, 8, 11, 0, 0, 0, 0, time.UTC), tasks[0].StartDate)
 	assert.Equal(t, time.Date(2017, 10, 1, 0, 0, 0, 0, time.UTC), tasks[21].StartDate)
 	assert.Equal(t, time.Date(2017, 10, 10, 0, 0, 0, 0, time.UTC), tasks[30].StartDate)
+
+	// org 3 again, but changing the archive period so we have no tasks
+	orgs[2].ActiveDays = 200
+	tasks, err = GetMissingDailyArchives(ctx, db, now, orgs[2], MessageType)
+	assert.NoError(t, err)
+	assert.Equal(t, 0, len(tasks))
+
+	// org 1 again, but lowering the archive period so we have tasks
+	orgs[0].ActiveDays = 2
+	tasks, err = GetMissingDailyArchives(ctx, db, now, orgs[0], MessageType)
+	assert.NoError(t, err)
+	assert.Equal(t, 58, len(tasks))
+	assert.Equal(t, time.Date(2017, 11, 10, 0, 0, 0, 0, time.UTC), tasks[0].StartDate)
+	assert.Equal(t, time.Date(2017, 12, 1, 0, 0, 0, 0, time.UTC), tasks[21].StartDate)
+	assert.Equal(t, time.Date(2017, 12, 10, 0, 0, 0, 0, time.UTC), tasks[30].StartDate)
+
 }
 
 func TestGetMissingMonthArchives(t *testing.T) {
 	db := setup(t)
 
 	// get the tasks for our org
 	ctx := context.Background()
-	orgs, err := GetActiveOrgs(ctx, db)
+	config := NewConfig()
+	orgs, err := GetActiveOrgs(ctx, db, config)
 	assert.NoError(t, err)
 
 	now := time.Date(2018, 1, 8, 12, 30, 0, 0, time.UTC)
@@ -87,6 +105,7 @@ func TestGetMissingMonthArchives(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, 1, len(tasks))
 	assert.Equal(t, time.Date(2017, 8, 1, 0, 0, 0, 0, time.UTC), tasks[0].StartDate)
+
 }
 
 func TestCreateMsgArchive(t *testing.T) {
@@ -96,7 +115,8 @@ func TestCreateMsgArchive(t *testing.T) {
 	err := EnsureTempArchiveDirectory("/tmp")
 	assert.NoError(t, err)
 
-	orgs, err := GetActiveOrgs(ctx, db)
+	config := NewConfig()
+	orgs, err := GetActiveOrgs(ctx, db, config)
 	assert.NoError(t, err)
 	now := time.Date(2018, 1, 8, 12, 30, 0, 0, time.UTC)
 
@@ -172,7 +192,8 @@ func TestCreateRunArchive(t *testing.T) {
 	err := EnsureTempArchiveDirectory("/tmp")
 	assert.NoError(t, err)
 
-	orgs, err := GetActiveOrgs(ctx, db)
+	config := NewConfig()
+	orgs, err := GetActiveOrgs(ctx, db, config)
 	assert.NoError(t, err)
 	now := time.Date(2018, 1, 8, 12, 30, 0, 0, time.UTC)
 
@@ -228,7 +249,8 @@ func TestWriteArchiveToDB(t *testing.T) {
 	db := setup(t)
 	ctx := context.Background()
 
-	orgs, err := GetActiveOrgs(ctx, db)
+	config := NewConfig()
+	orgs, err := GetActiveOrgs(ctx, db, config)
 	assert.NoError(t, err)
 	now := time.Date(2018, 1, 8, 12, 30, 0, 0, time.UTC)
 
@@ -281,11 +303,11 @@ func TestArchiveOrgMessages(t *testing.T) {
 	ctx := context.Background()
 	deleteTransactionSize = 1
 
-	orgs, err := GetActiveOrgs(ctx, db)
+	config := NewConfig()
+	orgs, err := GetActiveOrgs(ctx, db, config)
 	assert.NoError(t, err)
 	now := time.Date(2018, 1, 8, 12, 30, 0, 0, time.UTC)
 
-	config := NewConfig()
 	os.Args = []string{"rp-archiver"}
 
 	loader := ezconf.NewLoader(&config, "archiver", "Archives RapidPro runs and msgs to S3", nil)
@@ -422,11 +444,11 @@ func TestArchiveOrgRuns(t *testing.T) {
 	db := setup(t)
 	ctx := context.Background()
 
-	orgs, err := GetActiveOrgs(ctx, db)
+	config := NewConfig()
+	orgs, err := GetActiveOrgs(ctx, db, config)
 	assert.NoError(t, err)
 	now := time.Date(2018, 1, 8, 12, 30, 0, 0, time.UTC)
 
-	config := NewConfig()
 	os.Args = []string{"rp-archiver"}
 
 	loader := ezconf.NewLoader(&config, "archiver", "Archives RapidPro runs and msgs to S3", nil)

cmd/rp-archiver/main.go

@@ -84,7 +84,7 @@ func main() {
 
 		// get our active orgs
 		ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
-		orgs, err := archiver.GetActiveOrgs(ctx, db)
+		orgs, err := archiver.GetActiveOrgs(ctx, db, config)
 		cancel()
 
 		if err != nil {

config.go

@@ -21,6 +21,7 @@ type Config struct {
 
 	ArchiveMessages bool `help:"whether we should archive messages"`
 	ArchiveRuns     bool `help:"whether we should archive runs"`
+	ArchiveLength   int  `help:"how many days back to archive"`
 	Delete          bool `help:"whether to delete messages and runs from the db after archival (default false)"`
 }
 
@@ -45,6 +46,7 @@ func NewConfig() *Config {
 
 		ArchiveMessages: true,
 		ArchiveRuns:     true,
+		ArchiveLength:   90,
 		Delete:          false,
 	}