import (
"context"
+ "crypto/sha256"
"fmt"
+ "log/slog"
"slices"
"strings"
"sync"
)
// 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{
state: state,
clock: clock,
groups: groups,
+ images: images,
+ logger: logger,
+ lookupTime: whatsappLookupTime,
+ imageTime: whatsappImageTime,
catchupLimit: whatsappCatchupLimit,
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
// 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
state *WhatsAppState
clock func() time.Time
groups *WhatsAppGroups
+ images *ImageStore
+ logger *slog.Logger
+ lookupTime time.Duration
+ imageTime time.Duration
catchupLimit time.Duration
wake chan struct{}
failures chan error
// 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()
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)
}
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
}
})
}
-// 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
-// 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, i.lookupTime)
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 event.Message.GetImageMessage() != nil {
+ ctx, cancel := context.WithTimeout(ctx, i.imageTime)
+ defer cancel()
+
+ message.Text = i.imageText(ctx, link, event, message.Text)
+ }
+
_, err := i.state.Add(chat, message, time.Now())
if err != nil {
select {
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) {
// 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
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:
}
// 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()
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,
projects / onechat / blobdiff