X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/522d7fd117285e4e4ce3ea5ec9b5a1722521b681..3b2bf881d484ba1a33ff77a342564b8195a75c80:/whatsapp_inbox.go diff --git a/whatsapp_inbox.go b/whatsapp_inbox.go index c7fbf51..06f1c55 100644 --- a/whatsapp_inbox.go +++ b/whatsapp_inbox.go @@ -2,7 +2,9 @@ package main import ( "context" + "crypto/sha256" "fmt" + "log/slog" "slices" "strings" "sync" @@ -23,32 +25,39 @@ const ( ) // 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 @@ -59,6 +68,10 @@ type WhatsAppInbox struct { 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 @@ -87,6 +100,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 +125,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 +186,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 } @@ -176,20 +223,27 @@ func (i *WhatsAppInbox) startCatchup() { }) } -// 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 { @@ -208,6 +262,28 @@ func (i *WhatsAppInbox) receive(link WhatsAppLink, event *events.Message) bool { 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) { @@ -267,10 +343,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 -// 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 @@ -280,7 +356,7 @@ func whatsappMessage(ctx context.Context, device *store.Device, own []string, ev 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: @@ -291,7 +367,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 -// 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() @@ -300,11 +376,33 @@ func whatsappText(ctx context.Context, device *store.Device, own []string, event 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,