send gcal reminders in parallel (#339)

Author: zoom-ua

gcalbot/gcalbot/reminderscheduler/send.go

@@ -1,10 +1,13 @@
 package reminderscheduler
 
 import (
+	"context"
 	"fmt"
 	"time"
 
+	"github.com/keybase/go-keybase-chat-bot/kbchat/types/chat1"
 	"github.com/keybase/managed-bots/gcalbot/gcalbot"
+	"github.com/keybase/pipeliner"
 )
 
 func (r *ReminderScheduler) sendReminderLoop(shutdownCh chan struct{}) error {
@@ -37,6 +40,10 @@ func (r *ReminderScheduler) sendReminderLoop(shutdownCh chan struct{}) error {
 
 func (r *ReminderScheduler) sendReminders(sendMinute time.Time) {
 	timestamp := getReminderTimestamp(sendMinute, 0)
+	var sendTasks []struct {
+		convID  chat1.ConvIDStr
+		message string
+	}
 	r.minuteReminders.ForEachReminderMessageInMinute(timestamp, func(msg *ReminderMessage) {
 		for duration := range msg.MinuteReminders {
 			msgTimestamp := getReminderTimestamp(msg.StartTime, duration)
@@ -48,12 +55,17 @@ func (r *ReminderScheduler) sendReminders(sendMinute time.Time) {
 				} else {
 					eventSummary = "An event"
 				}
+				var message string
 				if minutesBefore == 0 {
-					r.ChatEcho(msg.KeybaseConvID, "%s is starting now: %s", eventSummary, msg.MsgContent)
+					message = fmt.Sprintf("%s is starting now: %s", eventSummary, msg.MsgContent)
 				} else {
-					r.ChatEcho(msg.KeybaseConvID, "%s is starting in %s: %s",
+					message = fmt.Sprintf("%s is starting in %s: %s",
 						eventSummary, gcalbot.MinutesBeforeString(minutesBefore), msg.MsgContent)
 				}
+				sendTasks = append(sendTasks, struct {
+					convID  chat1.ConvIDStr
+					message string
+				}{msg.KeybaseConvID, message})
 				delete(msg.MinuteReminders, duration)
 				r.stats.Count("sendReminders - reminder")
 			}
@@ -65,9 +77,26 @@ func (r *ReminderScheduler) sendReminders(sendMinute time.Time) {
 		}
 	})
 	r.minuteReminders.RemoveMinute(timestamp)
+
+	const sendWindow = 10
+	ctx := context.Background()
+	pipe := pipeliner.NewPipeliner(sendWindow)
+	worker := func(_ context.Context, i int) error { // nolint:unparam
+		t := sendTasks[i]
+		r.ChatEcho(t.convID, "%s", t.message)
+		return nil
+	}
+	for i := range sendTasks {
+		if err := pipe.WaitForRoom(ctx); err != nil {
+			break
+		}
+		go func() { pipe.CompleteOne(worker(ctx, i)) }()
+	}
+	_ = pipe.Flush(ctx)
+
 	sendDuration := time.Since(sendMinute)
 	if sendDuration.Seconds() > 15 {
-		r.Errorf("sending reminders took %s", sendDuration.String())
+		r.Errorf("sending %d reminders took %s", len(sendTasks), sendDuration.String())
 	}
 	r.stats.Value("sendReminders - duration - seconds", sendDuration.Seconds())
 }

gcalbot/gcalbot/schedulescheduler/send.go

@@ -10,6 +10,7 @@ import (
 	"golang.org/x/oauth2"
 
 	"google.golang.org/api/calendar/v3"
+	"google.golang.org/api/googleapi"
 
 	"github.com/keybase/managed-bots/gcalbot/gcalbot"
 )
@@ -131,7 +132,12 @@ func (s *ScheduleScheduler) SendDailyScheduleMessage(sendMinute time.Time, subsc
 	for index, calendarID := range subscription.CalendarIDs {
 		cal, err := srv.Calendars.Get(calendarID).Fields("summary").Do()
 		if err != nil {
-			s.Errorf("error getting calendar summary from API: %s", err)
+			if gerr, ok := err.(*googleapi.Error); ok && gerr.Code == 404 {
+				// Calendar was deleted or user lost access; use ID as display name
+				s.Debug("calendar no longer accessible (404): %s", calendarID)
+			} else {
+				s.Errorf("error getting calendar summary from API: %s", err)
+			}
 			calendarSummaries[index] = calendarID // use the cal id if there is an error
 		} else {
 			calendarSummaries[index] = cal.Summary

go.mod

@@ -16,6 +16,7 @@ require (
 	github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51
 	github.com/keybase/go-codec v0.0.0-20180928230036-164397562123
 	github.com/keybase/go-keybase-chat-bot v0.0.0-20260127182354-7367dd3315a3
+	github.com/keybase/pipeliner v0.0.0-20260107202117-7e0ff0869cf1
 	github.com/opensearch-project/opensearch-go/v4 v4.5.0
 	github.com/stathat/go v1.0.0
 	github.com/stretchr/testify v1.11.1

go.sum

@@ -100,6 +100,8 @@ github.com/keybase/go-codec v0.0.0-20180928230036-164397562123 h1:yg56lYPqh9suJe
 github.com/keybase/go-codec v0.0.0-20180928230036-164397562123/go.mod h1:r/eVVWCngg6TsFV/3HuS9sWhDkAzGG8mXhiuYA+Z/20=
 github.com/keybase/go-keybase-chat-bot v0.0.0-20260127182354-7367dd3315a3 h1:0uwBoNFUfi+vqQplGh2md1bak0WYelaoqkjn6hBSwJM=
 github.com/keybase/go-keybase-chat-bot v0.0.0-20260127182354-7367dd3315a3/go.mod h1:wl5lBoVNkepL8Hzs7jyqg3GS6U+by4yQeNr7oT0Evt0=
+github.com/keybase/pipeliner v0.0.0-20260107202117-7e0ff0869cf1 h1:aqymqcmuSOLTlGuWnfii7odizltdPaLi2hhDCpRwPqs=
+github.com/keybase/pipeliner v0.0.0-20260107202117-7e0ff0869cf1/go.mod h1:YT8b13LS9QDe45OyX1BFgXD0BRANHm0xauB+LNfV6WM=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=