Name timeouts for the operations they limit

[onechat] / whatsapp_inbox.go

diff --git a/whatsapp_inbox.go b/whatsapp_inbox.go
index d76e43ca3b97ade93d2c2af60d0c421a8f31c12b..9d2183e7d5e282a96442f35ba13d0a7cf58ec1a1 100644 (file)
--- a/whatsapp_inbox.go
+++ b/whatsapp_inbox.go
@@ -2,7 +2,9 @@ package main
 
 import (
        "context"
 
 import (
        "context"

+       "crypto/sha256"

        "fmt"
        "fmt"

+       "log/slog"

        "slices"
        "strings"
        "sync"
        "slices"
        "strings"
        "sync"


@@ -14,41 +16,53 @@ import (
 )
 
 const (
 )
 
 const (

+       // whatsappContactLookupTimeout is the longest time of a read of names
+       // in the session.
+       whatsappContactLookupTimeout = 10 * time.Second
+
+       // whatsappImageDownloadTimeout is the longest time of an image
+       // download.
+       whatsappImageDownloadTimeout = 2 * time.Minute
+

        // whatsappSource is the source of the work events of the batches.
        whatsappSource = "whatsapp"
 
        // whatsappSource is the source of the work events of the batches.
        whatsappSource = "whatsapp"
 

-       // whatsappCatchupLimit is the longest wait for the end of the catch-up
-       // after a connection.
-       whatsappCatchupLimit = time.Minute
+       // whatsappCatchupTimeout is the longest wait for the end of the
+       // catch-up after a connection.
+       whatsappCatchupTimeout = time.Minute

 )
 
 // newWhatsAppInbox returns the inbox of the batches in state, with clock for
 )
 
 // newWhatsAppInbox returns the inbox of the batches in state, with clock for

-// the times of the batch texts and groups for the names of the groups.
-func newWhatsAppInbox(state *WhatsAppState, clock func() time.Time, groups *WhatsAppGroups) *WhatsAppInbox {
+// the times of the batch texts and groups for the names of the groups. images
+// holds shared images, and logger gets image failures.
+func newWhatsAppInbox(state *WhatsAppState, clock func() time.Time, groups *WhatsAppGroups, images *ImageStore, logger *slog.Logger) *WhatsAppInbox {

        return &WhatsAppInbox{
        return &WhatsAppInbox{

-               state:        state,
-               clock:        clock,
-               groups:       groups,
-               catchupLimit: whatsappCatchupLimit,
-               wake:         make(chan struct{}, 1),
-               failures:     make(chan error, 1),
+               state:    state,
+               clock:    clock,
+               groups:   groups,
+               images:   images,
+               logger:   logger,
+               wake:     make(chan struct{}, 1),
+               failures: make(chan error, 1),

        }
 }
 
        }
 }
 

-// WhatsAppInbox puts the text messages into batches, one chat at a time, as
+// WhatsAppInbox puts messages into batches, one chat at a time, as

 // WhatsAppState says. deliver logs each closed batch as a work message that
 // starts with [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
 // WhatsAppState says. deliver logs each closed batch as a work message that
 // starts with [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, 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.
+// whatsappCatchupTimeout passes, the closed batches wait in the state file. At
+// the end of 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


@@ -56,12 +70,13 @@ func newWhatsAppInbox(state *WhatsAppState, clock func() time.Time, groups *What
 // goroutines. wake tells Run that a batch can be due or that a catch-up ended,
 // and failures gives Run the error of a write of the state file.
 type WhatsAppInbox struct {
 // goroutines. wake tells Run that a batch can be due or that a catch-up ended,
 // and failures gives Run the error of a write of the state file.
 type WhatsAppInbox struct {

-       state        *WhatsAppState
-       clock        func() time.Time
-       groups       *WhatsAppGroups
-       catchupLimit time.Duration
-       wake         chan struct{}
-       failures     chan error
+       state    *WhatsAppState
+       clock    func() time.Time
+       groups   *WhatsAppGroups
+       images   *ImageStore
+       logger   *slog.Logger
+       wake     chan struct{}
+       failures chan error

 
        mutex        sync.Mutex
        catchingUp   bool
 
        mutex        sync.Mutex
        catchingUp   bool


@@ -87,14 +102,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.
 // 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()

+

        catchingUp := i.catchingUp
        caughtUp := i.caughtUp
        i.caughtUp = false
        i.mutex.Unlock()
 
        var err error
        catchingUp := i.catchingUp
        caughtUp := i.caughtUp
        i.caughtUp = false
        i.mutex.Unlock()
 
        var err error

+

        switch {
        case all || caughtUp:
                err = i.state.CloseAll()
        switch {
        case all || caughtUp:
                err = i.state.CloseAll()


@@ -108,13 +127,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)
                }


@@ -143,6 +188,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
 }
 


@@ -156,7 +208,7 @@ func (i *WhatsAppInbox) waitForCatchup() {
 }
 
 // startCatchup starts the catch-up of a new connection, which ends at
 }
 
 // startCatchup starts the catch-up of a new connection, which ends at

-// endCatchup or after catchupLimit.
+// endCatchup or after whatsappCatchupTimeout.

 func (i *WhatsAppInbox) startCatchup() {
        i.mutex.Lock()
        defer i.mutex.Unlock()
 func (i *WhatsAppInbox) startCatchup() {
        i.mutex.Lock()
        defer i.mutex.Unlock()


@@ -168,25 +220,32 @@ func (i *WhatsAppInbox) startCatchup() {
                i.catchupTimer.Stop()
        }
 
                i.catchupTimer.Stop()
        }
 

-       i.catchupTimer = time.AfterFunc(i.catchupLimit, func() {
+       i.catchupTimer = time.AfterFunc(whatsappCatchupTimeout, func() {

                i.caughtUpBy(catchups)
        })
 }
 
                i.caughtUpBy(catchups)
        })
 }
 

-// receive puts the text of event into its batch and wakes Run. link gives the
-// device and the own IDs. It returns false when the state file cannot hold the
+// receive stores an image before its batch and wakes Run. link gives the device
+// and the own IDs. It returns false when the state file cannot hold the

 // message, so that the library does not acknowledge it, and Run stops the
 // message, so that the library does not acknowledge it, and Run stops the

-// daemon. A message that is not text, or that the batches do not take, is
-// handled with nothing to do.
-func (i *WhatsAppInbox) receive(link WhatsAppLink, event *events.Message) bool {
-       ctx, cancel := context.WithTimeout(context.Background(), whatsappLookupTime)
+// daemon. An image that cannot be downloaded or stored gets a text marker. A
+// message that the batches do not take is handled with nothing to do.
+func (i *WhatsAppInbox) receive(ctx context.Context, link WhatsAppLink, event *events.Message) bool {
+       lookupCtx, cancel := context.WithTimeout(ctx, whatsappContactLookupTimeout)

        defer cancel()
 
        defer cancel()
 

-       chat, message, found := whatsappMessage(ctx, link.device, link.own, event)
+       chat, message, found := whatsappMessage(lookupCtx, link.device, link.own, event)

        if !found {
                return true
        }
 
        if !found {
                return true
        }
 

+       if event.Message.GetImageMessage() != nil {
+               ctx, cancel := context.WithTimeout(ctx, whatsappImageDownloadTimeout)
+               defer cancel()
+
+               message.Text = i.imageText(ctx, link, event, message.Text)
+       }
+

        _, err := i.state.Add(chat, message, time.Now())
        if err != nil {
                select {
        _, err := i.state.Add(chat, message, time.Now())
        if err != nil {
                select {


@@ -205,6 +264,28 @@ func (i *WhatsAppInbox) receive(link WhatsAppLink, event *events.Message) bool {
        return true
 }
 
        return true
 }
 

+// imageText stores the image of event before returning its marker and caption.
+// Failures keep the caption with a text marker. The daemon log identifies the
+// chat by a short hash, not its phone number.
+func (i *WhatsAppInbox) imageText(ctx context.Context, link WhatsAppLink, event *events.Message, caption string) string {
+       name, err := whatsappImage(ctx, link.client, i.images, event.Message.GetImageMessage())
+       if err != nil {
+               chat := event.Info.Chat.ToNonAD()
+               digest := sha256.Sum256([]byte(chat.String()))
+               redact := append(slices.Clone(link.own), chat.User)
+               i.logger.Error("image receive failed", "chat", fmt.Sprintf("%x", digest[:6]), "err", whatsappRedact(redact, err.Error()))
+
+               text := "(image not downloaded)"
+               if caption != "" {
+                       text += "\n" + caption
+               }
+
+               return text
+       }
+
+       return imageText([]string{name}, caption)
+}
+

 // caughtUpBy ends the catch-up of the connection with the number catchups, when
 // no later connection started a new one.
 func (i *WhatsAppInbox) caughtUpBy(catchups int) {
 // caughtUpBy ends the catch-up of the connection with the number catchups, when
 // no later connection started a new one.
 func (i *WhatsAppInbox) caughtUpBy(catchups int) {


@@ -264,10 +345,10 @@ func (i *WhatsAppInbox) close() {
 
 // whatsappMessage returns the chat and the batch message of event, or false
 // when the batches do not take it. device is the linked device, and own holds
 
 // whatsappMessage returns the chat and the batch message of event, or false
 // when the batches do not take it. device is the linked device, and own holds

-// the own IDs. The batches take the text of a private chat or a group, from
-// another person or from the user. They do not take media, reactions, edits,
-// deletions, view-once and disappearing messages, old history, status
-// broadcasts, channels, and the chat of the user with itself.
+// the own IDs. The batches take text, images, and media markers from a private
+// chat or a group, from another person or from the user. They do not take
+// reactions, edits, deletions, view-once and disappearing messages, old
+// history, status broadcasts, channels, and the chat of the user with itself.

 //
 // A mention of the user becomes @you, another mention @NAME when the name is
 // known, and the own number and LID become OWN, so that they never go to the
 //
 // A mention of the user becomes @you, another mention @NAME when the name is
 // known, and the own number and LID become OWN, so that they never go to the


@@ -277,7 +358,7 @@ func whatsappMessage(ctx context.Context, device *store.Device, own []string, ev
        chatJID := info.Chat.ToNonAD()
        switch {
        case device == nil:
        chatJID := info.Chat.ToNonAD()
        switch {
        case device == nil:

-       case event.IsEdit, info.Edit != "", event.IsEphemeral, event.IsViewOnce, event.SourceWebMsg != nil:
+       case event.IsEdit, info.Edit != "", event.IsEphemeral, event.IsViewOnce, event.Message.GetImageMessage().GetViewOnce(), event.Message.GetVideoMessage().GetViewOnce(), event.SourceWebMsg != nil:

        case chatJID.Server != types.DefaultUserServer && chatJID.Server != types.HiddenUserServer && chatJID.Server != types.GroupServer:
        case slices.Contains(own, chatJID.User):
        default:
        case chatJID.Server != types.DefaultUserServer && chatJID.Server != types.HiddenUserServer && chatJID.Server != types.GroupServer:
        case slices.Contains(own, chatJID.User):
        default:


@@ -288,7 +369,7 @@ func whatsappMessage(ctx context.Context, device *store.Device, own []string, ev
 }
 
 // whatsappText returns the chat and the batch message of event, a message that
 }
 
 // whatsappText returns the chat and the batch message of event, a message that

-// the batches take when it has text.
+// the batches take. An image returns its caption until receive stores it.

 func whatsappText(ctx context.Context, device *store.Device, own []string, event *events.Message) (WhatsAppChat, WhatsAppMessage, bool) {
        info := event.Info
        text := event.Message.GetConversation()
 func whatsappText(ctx context.Context, device *store.Device, own []string, event *events.Message) (WhatsAppChat, WhatsAppMessage, bool) {
        info := event.Info
        text := event.Message.GetConversation()


@@ -297,11 +378,33 @@ func whatsappText(ctx context.Context, device *store.Device, own []string, event
                text = extended.GetText()
        }
 
                text = extended.GetText()
        }
 

-       if strings.TrimSpace(text) == "" {
-               return WhatsAppChat{}, WhatsAppMessage{}, false
+       contextInfo := extended.GetContextInfo()
+       switch {
+       case event.Message.GetImageMessage() != nil:
+               image := event.Message.GetImageMessage()
+               text = image.GetCaption()
+               contextInfo = image.GetContextInfo()
+       case event.Message.GetStickerMessage() != nil:
+               text = "(sticker)"
+               contextInfo = event.Message.GetStickerMessage().GetContextInfo()
+       case event.Message.GetVideoMessage() != nil:
+               video := event.Message.GetVideoMessage()
+               text = "(video)"
+               if video.GetGifPlayback() {
+                       text = "(GIF)"
+               }
+
+               if video.GetCaption() != "" {
+                       text += "\n" + video.GetCaption()
+               }
+
+               contextInfo = video.GetContextInfo()
+       default:
+               if strings.TrimSpace(text) == "" {
+                       return WhatsAppChat{}, WhatsAppMessage{}, false
+               }

        }
 
        }
 

-       contextInfo := extended.GetContextInfo()

        message := WhatsAppMessage{
                ID:   info.ID,
                Time: info.Timestamp,
        message := WhatsAppMessage{
                ID:   info.ID,
                Time: info.Timestamp,