X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/56d959f2d573eef1be5c04d0bf1659fa30d36e7c..0da1ef837bbd190cc5787214187d1b29d62b9410:/whatsapp.go?ds=sidebyside

diff --git a/whatsapp.go b/whatsapp.go
index 20143a9..8306734 100644
--- a/whatsapp.go
+++ b/whatsapp.go
@@ -81,6 +81,10 @@ const (
 
 	// whatsappUnknownGroup is the name of a group whose name is not known.
 	whatsappUnknownGroup = "A group with no known name"
+
+	// whatsappCatchupLimit is the longest wait for the end of the catch-up
+	// after a connection.
+	whatsappCatchupLimit = time.Minute
 )
 
 // WhatsAppClient is the part of the WhatsApp client that the daemon uses. The
@@ -102,16 +106,17 @@ type WhatsAppClient interface {
 // the client of a device with its event handler.
 func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logger *log.Logger, newClient func(device *store.Device, handle func(event any) bool) WhatsAppClient) (*WhatsApp, error) {
 	w := &WhatsApp{
-		dir:         filepath.Join(dir, whatsappDirName),
-		clock:       clock,
-		logger:      logger,
-		newClient:   newClient,
-		noticeDelay: whatsappNoticeDelay,
-		requests:    make(chan WhatsAppRequest, whatsappRequests),
-		logouts:     make(chan struct{}, 1),
-		wake:        make(chan struct{}, 1),
-		failures:    make(chan error, 1),
-		groupNames:  make(map[string]string),
+		dir:          filepath.Join(dir, whatsappDirName),
+		clock:        clock,
+		logger:       logger,
+		newClient:    newClient,
+		noticeDelay:  whatsappNoticeDelay,
+		catchupLimit: whatsappCatchupLimit,
+		requests:     make(chan WhatsAppRequest, whatsappRequests),
+		logouts:      make(chan struct{}, 1),
+		wake:         make(chan struct{}, 1),
+		failures:     make(chan error, 1),
+		groupNames:   make(map[string]string),
 	}
 
 	var err error
@@ -140,6 +145,7 @@ func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logge
 
 	if w.device.ID != nil {
 		w.linked = true
+		w.catchingUp = true
 		w.own = whatsappOwnIDs(w.device.ID.User, w.device.LID.User)
 	}
 
@@ -161,11 +167,18 @@ func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logge
 //
 // The text messages go into batches, one chat at a time, as WhatsAppBatches
 // says. Run logs each closed batch as a work message that starts with
-// [WhatsApp], and a batch that needs the user wakes the turn loop. The event
-// handler acknowledges a message only after the state file holds it, so the
-// servers send again a message that a crash lost. A batch leaves the state file
-// only after its work message is durable in the log. A crash between the two
-// makes the batch repeat once.
+// [WhatsApp], and a batch that needs the user wakes the turn loop.
+//
+// After each connection, the servers send the messages that came while the
+// daemon was away, and then OfflineSyncCompleted. Until then, or until
+// catchupLimit passes, the closed batches wait in the state file. At the end of
+// the catch-up, Run closes all batches, logs them, and wakes the turn loop
+// once, so the whole catch-up gives one turn and not one turn for each chat.
+// The batches of the state file at the start wait for the first catch-up in the
+// same way. The event handler acknowledges a message only after the state file
+// holds it, so the servers send again a message that a crash lost. A batch
+// leaves the state file only after its work message is durable in the log. A
+// crash between the two makes the batch repeat once.
 //
 // The library logs nothing, because its log lines hold the ID of the own
 // account, which is the phone number of the user. The own number and the own
@@ -175,16 +188,17 @@ func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logge
 // log, and those of private chats hold phone numbers, because the agent needs
 // them to send.
 type WhatsApp struct {
-	dir         string
-	clock       func() time.Time
-	logger      *log.Logger
-	newClient   func(device *store.Device, handle func(event any) bool) WhatsAppClient
-	noticeDelay time.Duration
-	requests    chan WhatsAppRequest
-	logouts     chan struct{}
-	wake        chan struct{}
-	failures    chan error
-	batches     *WhatsAppBatches
+	dir          string
+	clock        func() time.Time
+	logger       *log.Logger
+	newClient    func(device *store.Device, handle func(event any) bool) WhatsAppClient
+	noticeDelay  time.Duration
+	catchupLimit time.Duration
+	requests     chan WhatsAppRequest
+	logouts      chan struct{}
+	wake         chan struct{}
+	failures     chan error
+	batches      *WhatsAppBatches
 
 	mutex          sync.Mutex
 	container      *sqlstore.Container
@@ -202,6 +216,10 @@ type WhatsApp struct {
 	closed         bool
 	groupNames     map[string]string
 	groupsAt       time.Time
+	catchingUp     bool
+	caughtUp       bool
+	catchups       int
+	catchupTimer   *time.Timer
 }
 
 // Run connects a linked device, logs the closed batches in l, runs the pair and
@@ -230,7 +248,7 @@ func (w *WhatsApp) Run(ctx context.Context, l *TurnLoop, notify func(Event)) err
 	}()
 
 	for {
-		err := w.deliver(ctx, l)
+		err := w.deliver(ctx, l, false)
 		if err != nil {
 			return err
 		}
@@ -406,6 +424,10 @@ func (w *WhatsApp) close() {
 		w.timer.Stop()
 	}
 
+	if w.catchupTimer != nil {
+		w.catchupTimer.Stop()
+	}
+
 	client := w.client
 	w.client = nil
 	w.mutex.Unlock()
@@ -422,7 +444,7 @@ func (w *WhatsApp) loggedOut(ctx context.Context, l *TurnLoop) error {
 		return nil
 	}
 
-	err := w.deliverAll(ctx, l)
+	err := w.deliver(ctx, l, true)
 	if err != nil {
 		return err
 	}
@@ -458,7 +480,7 @@ func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text str
 		return nil
 	}
 
-	err := w.deliverAll(ctx, l)
+	err := w.deliver(ctx, l, true)
 	if err != nil {
 		return err
 	}
@@ -493,21 +515,39 @@ func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text str
 }
 
 // deliver logs the closed batches in l in order, each as a work message of
-// whatsappSource that wakes the loop when the batch starts a turn. A batch
-// leaves the state file after its message is durable in the log. A group with
-// no name gets its name from WhatsApp first.
-func (w *WhatsApp) deliver(ctx context.Context, l *TurnLoop) error {
-	_, err := w.batches.CloseDue(time.Now())
+// whatsappSource, and wakes the loop once with the batches that start a turn. A
+// batch leaves the state file after its message is durable in the log. A group
+// with no name gets its name from WhatsApp first. During a catch-up it logs
+// nothing, and at its end it closes all batches first. With all, it closes and
+// logs all batches also during a catch-up.
+func (w *WhatsApp) deliver(ctx context.Context, l *TurnLoop, all bool) error {
+	w.mutex.Lock()
+	catchingUp := w.catchingUp
+	caughtUp := w.caughtUp
+	w.caughtUp = false
+	w.mutex.Unlock()
+
+	var err error
+	switch {
+	case all || caughtUp:
+		err = w.batches.CloseAll()
+	case catchingUp:
+		return nil
+	default:
+		_, err = w.batches.CloseDue(time.Now())
+	}
+
 	if err != nil {
 		return err
 	}
 
+	var turns []Message
 	for _, batch := range w.batches.Closed() {
 		if batch.Chat.Group && batch.Chat.Name == "" {
 			batch.Chat.Name = w.groupName(ctx, batch.Chat.ID)
 		}
 
-		_, err = l.Work(whatsappSource, batch.Text(w.clock().Location()), "", batch.Turn())
+		message, err := l.Work(whatsappSource, batch.Text(w.clock().Location()), "")
 		if err != nil {
 			return fmt.Errorf("log a WhatsApp batch: %w", err)
 		}
@@ -528,19 +568,15 @@ func (w *WhatsApp) deliver(ctx context.Context, l *TurnLoop) error {
 		if err != nil {
 			return err
 		}
-	}
-
-	return nil
-}
 
-// deliverAll closes every open batch and logs all of them.
-func (w *WhatsApp) deliverAll(ctx context.Context, l *TurnLoop) error {
-	err := w.batches.CloseAll()
-	if err != nil {
-		return err
+		if batch.Turn() {
+			turns = append(turns, message)
+		}
 	}
 
-	return w.deliver(ctx, l)
+	l.Wake(turns...)
+
+	return nil
 }
 
 // groupName returns the name of the group id, or whatsappUnknownGroup. A group
@@ -594,6 +630,7 @@ func (w *WhatsApp) removeSession() error {
 	w.linked = false
 	w.own = nil
 	w.connected = false
+	w.catchingUp = false
 	if w.timer != nil {
 		w.timer.Stop()
 	}
@@ -759,10 +796,11 @@ func (w *WhatsApp) linkedBy(client WhatsAppClient) {
 
 // handle takes one event of the library and tells if it was handled. A message
 // goes to receive. The names of the groups follow the events of the groups. A
-// disconnection keeps the time of its start while the library tries again.
-// After noticeDelay of a disconnection, all clients get one notice, and they
-// get one more when the connection comes back. A disconnection that the library
-// does not try again gives a notice at once.
+// connection starts a catch-up, which ends at OfflineSyncCompleted or after
+// catchupLimit. A disconnection keeps the time of its start while the library
+// tries again. After noticeDelay of a disconnection, all clients get one
+// notice, and they get one more when the connection comes back. A disconnection
+// that the library does not try again gives a notice at once.
 func (w *WhatsApp) handle(event any) bool {
 	message, isMessage := event.(*events.Message)
 	if isMessage {
@@ -783,6 +821,16 @@ func (w *WhatsApp) handle(event any) bool {
 	case *events.Connected:
 		w.connected = true
 		w.connectedAt = w.clock()
+		w.catchingUp = true
+		w.catchups++
+		catchups := w.catchups
+		if w.catchupTimer != nil {
+			w.catchupTimer.Stop()
+		}
+
+		w.catchupTimer = time.AfterFunc(w.catchupLimit, func() {
+			w.caughtUpBy(catchups)
+		})
 		if w.timer != nil {
 			w.timer.Stop()
 		}
@@ -802,6 +850,8 @@ func (w *WhatsApp) handle(event any) bool {
 		}
 
 		w.connected = false
+	case *events.OfflineSyncCompleted:
+		w.endCatchup()
 	case *events.LoggedOut:
 		select {
 		case w.logouts <- struct{}{}:
@@ -946,6 +996,36 @@ func (w *WhatsApp) text(ctx context.Context, device *store.Device, own []string,
 	return chat, message, true
 }
 
+// caughtUpBy ends the catch-up of the connection with the number catchups, when
+// no later connection started a new one.
+func (w *WhatsApp) caughtUpBy(catchups int) {
+	w.mutex.Lock()
+	defer w.mutex.Unlock()
+
+	if w.catchups == catchups {
+		w.endCatchup()
+	}
+}
+
+// endCatchup ends a running catch-up and wakes Run, which logs all batches. The
+// caller holds the mutex.
+func (w *WhatsApp) endCatchup() {
+	if !w.catchingUp {
+		return
+	}
+
+	w.catchingUp = false
+	w.caughtUp = true
+	if w.catchupTimer != nil {
+		w.catchupTimer.Stop()
+	}
+
+	select {
+	case w.wake <- struct{}{}:
+	default:
+	}
+}
+
 // disconnectedLong gives the notice of a long disconnection, unless the
 // disconnection with the number disconnections ended.
 func (w *WhatsApp) disconnectedLong(disconnections int) {