X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/522d7fd117285e4e4ce3ea5ec9b5a1722521b681..b2fc2d6a737dd5bab25e57ba822dd07258de93e6:/whatsapp_inbox.go?ds=inline

diff --git a/whatsapp_inbox.go b/whatsapp_inbox.go
index c7fbf51..adff79e 100644
--- a/whatsapp_inbox.go
+++ b/whatsapp_inbox.go
@@ -42,13 +42,15 @@ func newWhatsAppInbox(state *WhatsAppState, clock func() time.Time, groups *What
 // 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, deliver 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 catch-up, deliver closes all batches and logs them. It wakes the root
+// once for its batches, so their catch-up gives one turn, not one 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
+// or agent queue. A crash between the two can repeat an unrouted batch once. A
+// routed batch enters the durable agent queue, which prevents a second log
+// entry and wakes only the child. A missing child uses the root.
 //
 // The state file is the one of WhatsApp, which keeps the send records in it
 // too. groups gives the names of the groups. The mutex protects the state of
@@ -87,6 +89,8 @@ func (i *WhatsAppInbox) due() <-chan time.Time {
 // with no name gets its name from WhatsApp first, with the client of link.
 // 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.
+// Routed batches enter the durable child queue instead of the root log and wake
+// their child even when the batch does not need a root turn.
 func (i *WhatsAppInbox) deliver(ctx context.Context, l *TurnLoop, link WhatsAppLink, all bool) error {
 	i.mutex.Lock()
 
@@ -110,14 +114,39 @@ func (i *WhatsAppInbox) deliver(ctx context.Context, l *TurnLoop, link WhatsAppL
 		return err
 	}
 
+	batches := i.state.Closed()
+	if l.messages != nil {
+		err = l.messages.PruneWhatsApp(batches)
+		if err != nil {
+			return fmt.Errorf("prune WhatsApp receipts: %w", err)
+		}
+	}
+
 	var turns []Message
 
-	for _, batch := range i.state.Closed() {
+	for _, batch := range batches {
 		if batch.Chat.Group && batch.Chat.Name == "" {
 			batch.Chat.Name = i.groups.name(ctx, link, batch.Chat.ID)
 		}
 
-		message, err := l.Work(whatsappSource, batch.Text(i.clock().Location()), "")
+		text := batch.Text(i.clock().Location())
+		if l.messages != nil {
+			routed, err := l.messages.RouteWhatsApp(batch, text)
+			if err != nil {
+				return fmt.Errorf("route a WhatsApp batch: %w", err)
+			}
+
+			if routed {
+				err = i.state.Remove(batch.Number)
+				if err != nil {
+					return fmt.Errorf("remove routed WhatsApp batch: %w", err)
+				}
+
+				continue
+			}
+		}
+
+		message, err := l.Work(whatsappSource, text, "")
 		if err != nil {
 			return fmt.Errorf("log a WhatsApp batch: %w", err)
 		}
@@ -146,6 +175,13 @@ func (i *WhatsAppInbox) deliver(ctx context.Context, l *TurnLoop, link WhatsAppL
 
 	l.Wake(turns...)
 
+	if l.messages != nil {
+		err = l.messages.PruneWhatsApp(i.state.Closed())
+		if err != nil {
+			return fmt.Errorf("prune WhatsApp receipts: %w", err)
+		}
+	}
+
 	return nil
 }