Merge branch 'scheduled-check-0.28' into 'release/v0.28'

check unread messages on schedule

See merge request mashroom/backend/gmailparser!21

cmd/mashroom-gmailparser/main.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"os"
 	"regexp"
+	"time"
 
 	"cloud.google.com/go/pubsub"
 	"github.com/jmoiron/sqlx"
@@ -110,6 +111,13 @@ var (
 	}
 )
 
+var (
+	envScheduleInterval = env.ResolveDuration(
+		"SCHEDULE_INTERVAL",
+		time.Minute*5,
+	)
+)
+
 func main() {
 	defer tracer.Close()
 
@@ -211,6 +219,7 @@ func main() {
 			Mixpanel:     envMixpanelService,
 			Account:      envAccountService,
 		},
+		ScheduleInterval: envScheduleInterval,
 	})
 	if err != nil {
 		panic(err)

internal/parser/api.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"regexp"
 	"strings"
+	"time"
 
 	"cloud.google.com/go/pubsub"
 	account "gitlab.mashroom.com/mashroom/backend/account/api"
@@ -95,6 +96,7 @@ type Config struct {
 	GmailService     *gmail.Service
 	PubsubClient     *pubsub.Client
 	Connections      Connections
+	ScheduleInterval time.Duration
 }
 
 // Parser implementation

internal/parser/gmailparser.go

@@ -9,6 +9,7 @@ import (
 	"regexp"
 	"strings"
 	"sync"
+	"time"
 
 	"cloud.google.com/go/pubsub"
 	"github.com/nyaruka/phonenumbers"
@@ -123,6 +124,10 @@ func (p *Parser) processUnread(
 		return
 	}
 
+	if len(h.Messages) == 0 {
+		return
+	}
+
 	log.Infof("%d unread message(s) for '%s' label", len(h.Messages), v.labelID)
 
 	for _, m := range h.Messages {
@@ -353,6 +358,12 @@ func (p *Parser) handleSubscription(ctx context.Context, v *Subscription) error
 		}
 	}()
 
+	go func() {
+		for range time.Tick(p.ScheduleInterval) {
+			cond.Broadcast()
+		}
+	}()
+
 	go func() {
 		err = p.PubsubClient.Subscription(v.subscription).Receive(ctx, func(ctx context.Context, m *pubsub.Message) {
 			var payload notificationPayload