Import WhatsApp history into a child

[onechat] / whatsapp_inbox.go

diff --git a/whatsapp_inbox.go b/whatsapp_inbox.go
index d76e43ca3b97ade93d2c2af60d0c421a8f31c12b..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
-// 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,14 +89,18 @@ 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()
+
        catchingUp := i.catchingUp
        caughtUp := i.caughtUp
        i.caughtUp = false
        i.mutex.Unlock()
 
        var err error
+
        switch {
        case all || caughtUp:
                err = i.state.CloseAll()
@@ -108,13 +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)
                }
@@ -143,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
 }