Deliver routed WhatsApp batches to children

[onechat] / whatsapp_inbox.go

diff --git a/whatsapp_inbox.go b/whatsapp_inbox.go
index c7fbf51ff4ece235dd6b39de109818b6b4ce90e4..adff79e698f22d28bb1ce388298c90839d263cb7 100644 (file)
--- 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
 // 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
 //
 // 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.
 // 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()
 
 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
        }
 
                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
 
        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)
                }
 
                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)
                }
                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...)
 
 
        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
 }
 
        return nil
 }