From: Anderson A. Giacomolli Date: Sun, 8 Nov 2026 14:57:43 +0000 (-0300) Subject: Give the WhatsApp batches to the agent X-Git-Url: http://git.giacomolli.uk/onechat/commitdiff_plain/56d959f2d573eef1be5c04d0bf1659fa30d36e7c Give the WhatsApp batches to the agent The daemon reads the text messages of the private chats and the groups and gives them to the agent in batches. The messages of one chat form one batch while they are at most 20 seconds apart, for 2 minutes or 50 messages at the most, and a batch also closes after 20 quiet seconds in real time. The rule uses the WhatsApp times of the messages, so the catch-up after a stop gives the same batches as live messages. Each batch goes to the log as one work message that starts with [WhatsApp], with the chat, its ID for the later send tool, and the time, sender, and text of each message. It takes the path of the subagent reports: TurnLoop.Work logs it and, for a batch that needs the user, wakes the loop, so a batch that comes during a turn joins its next request. A batch starts a turn when it is private with a message of another person, or from a group with a message that mentions the user or replies to the user. Such a batch ends with a line that lets the agent end the turn with an empty reply. The chain already drops an empty reply, and Telegram drops the batches, so a batch that does not need the user gives no Telegram message. whatsapp/state.json keeps the batches that are not in the log. The library acknowledges a message only after the file holds it, and a batch leaves the file only after the log is durable, so a crash loses nothing and can repeat a batch once. A failed write stops the daemon. The own number never goes to the log: the messages of the user show as You, a mention of the user as @you, the number as OWN, and the chat of the user with itself is not read. The batches do not take media, reactions, edits, view-once and disappearing messages, or old history. Issue: 6f69ca1 --- diff --git a/README.rst b/README.rst index fd6cb95..b17cd64 100644 --- a/README.rst +++ b/README.rst @@ -208,6 +208,38 @@ shell of the agent runs as the user, so it can read the session. If the session leaks, remove the linked device in WhatsApp on the phone, and the session stops working. Without the file, WhatsApp is not linked. +The daemon reads the text messages of the private chats and the groups, +from other people and from the user. It does not read media, reactions, +edits, deletions, view-once and disappearing messages, or the history +from before the link. The messages of one chat form one batch until the +chat is quiet for 20 seconds, for 2 minutes or 50 messages at the most. +Each batch goes to the chat of the agent as one work message that starts +with ``[WhatsApp]``, with the name and the ID of the chat, and the time, +the sender, and the text of each message. The messages of the user show +as ``You``. ``whatsapp/state.json``, with mode 0600, keeps the batches +that are not in the chat yet, so they survive a restart. A crash just +after a batch goes to the chat can make the batch repeat once. + +A batch of a private chat with a message of another person starts a +turn, and so does a batch of a group with a message that mentions the +user or replies to a message of the user. The other batches go to the +chat with no turn, and the agent sees them at its next turn. A batch +that starts a turn ends with a line that tells the agent to end its turn +with an empty reply when nothing in the batch needs the user. No client +shows an empty reply, so such a batch gives no message in Telegram. The +terminal client shows each batch, and Telegram shows none. + +The phone number of the user never goes to the chat: a mention of the +user shows as ``@you``, the number shows as ``OWN``, and the chat of the +user with itself is not read. But the IDs of the other chats go to the +chat and to the model, because the agent needs them to send. The ID of a +private chat holds the phone number of the contact, such as +``15550002222@s.whatsapp.net``, and the ID of a group can hold the +number of the person who made it. + +Each batch of a group goes to the chat, also with no turn, and the +memory tree summarizes it with model calls, which use the quota. + When the session holds a linked device, the daemon connects at its start, and after a network failure it connects again by itself. When a disconnection lasts 10 minutes, all clients get one notice, and one more diff --git a/daemon.go b/daemon.go index 41b3ab6..991edfa 100644 --- a/daemon.go +++ b/daemon.go @@ -306,8 +306,9 @@ func daemonListenSocket(path string) (net.Listener, error) { // daemonServe runs the loop that start makes and serves the clients of listener // until ctx ends or the loop fails. It closes listener. It runs whatsapp, which -// logs its notes in the loop and tells all clients its notices. When bot is not nil, it runs the bot too, and a failure of the bot -// stops the agent. +// logs its batches and notes in the loop and tells all clients its notices, and +// a failure of whatsapp stops the agent. When bot is not nil, it runs the bot +// too, and a failure of the bot stops the agent. func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, whatsapp *WhatsApp, start func(notify func(Event)) (*TurnLoop, error)) error { ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -338,10 +339,12 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w hub.Accept(ctx, listener, l) }() + var whatsappErr error go func() { defer workers.Done() - whatsapp.Run(ctx, l, notify) + whatsappErr = whatsapp.Run(ctx, l, notify) + cancel() }() var botErr error @@ -365,7 +368,7 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w hub.CloseAll() workers.Wait() - return errors.Join(err, botErr) + return errors.Join(err, whatsappErr, botErr) } // daemonRunRequest runs one request of a client. The answers of a WhatsApp diff --git a/daemon_start_helper_test.go b/daemon_start_helper_test.go index b2c3787..f90d2d7 100644 --- a/daemon_start_helper_test.go +++ b/daemon_start_helper_test.go @@ -8,8 +8,8 @@ import ( "github.com/stretchr/testify/require" ) -// daemonStartTestWithBot is daemonStartTest with bot, which can be nil. WhatsApp -// is not linked. +// daemonStartTestWithBot is daemonStartTest with bot, which can be nil. +// WhatsApp is not linked. func daemonStartTestWithBot(t *testing.T, server *ModelFake, bot *TelegramBot) string { t.Helper() diff --git a/telegram.go b/telegram.go index 16eff80..814f454 100644 --- a/telegram.go +++ b/telegram.go @@ -160,10 +160,15 @@ type TelegramBot struct { } // Notify queues event for the chat. It never blocks, so a slow network does not -// stop the agent. +// stop the agent. The WhatsApp batches do not go to the chat: the reply of the +// agent tells the user what needs the user. func (b *TelegramBot) Notify(event Event) { switch event.Kind { - case "busy", "idle", "reply", "error", "notice", "subagent", "work": + case "busy", "idle", "reply", "error", "notice", "subagent": + case "work": + if event.Source == whatsappSource { + return + } case "user": if event.Source == telegramSource { return diff --git a/turn.go b/turn.go index 93052e2..d6a1c41 100644 --- a/turn.go +++ b/turn.go @@ -122,30 +122,45 @@ func (l *TurnLoop) Submit(source, text string) (Message, error) { // Submit does. A failed write stops the loop, because the log refuses all // writes after it. func (l *TurnLoop) Report(name, text, status string) { + _, err := l.Work(name, text, status, status != subagentStopped) + if err != nil { + l.fail(fmt.Errorf("%w: %w", errChainLogWrite, err)) + } +} + +// Work logs text as a work message of source, such as the name of a subagent, +// and tells the clients, with status. With wake, the message joins the waiting +// messages and wakes the loop, as a user message of Submit does: it joins the +// first request of a new turn, or the next request of a running turn after the +// tool results. +func (l *TurnLoop) Work(source, text, status string, wake bool) (Message, error) { message, err := l.memory.Append("work", text) - if err == nil { - l.notify(Event{ - Kind: "work", - ID: message.ID, - Text: text, - Source: name, - Status: status, - }) + if err != nil { + return Message{}, err } - l.mutex.Lock() - switch { - case err != nil: - l.failure = fmt.Errorf("%w: %w", errChainLogWrite, err) - case status != subagentStopped: - l.pending = append(l.pending, message) + l.notify(Event{ + Kind: "work", + ID: message.ID, + Text: text, + Source: source, + Status: status, + }) + + if !wake { + return message, nil } + + l.mutex.Lock() + l.pending = append(l.pending, message) l.mutex.Unlock() select { case l.wake <- struct{}{}: default: } + + return message, nil } // StopSubagent ends the run of the subagent name and tells if it ran. Its diff --git a/whatsapp.go b/whatsapp.go index 3c519e8..20143a9 100644 --- a/whatsapp.go +++ b/whatsapp.go @@ -8,6 +8,7 @@ import ( "net/url" "os" "path/filepath" + "slices" "strings" "sync" "time" @@ -15,6 +16,7 @@ import ( "go.mau.fi/whatsmeow" "go.mau.fi/whatsmeow/store" "go.mau.fi/whatsmeow/store/sqlstore" + "go.mau.fi/whatsmeow/types" "go.mau.fi/whatsmeow/types/events" waLog "go.mau.fi/whatsmeow/util/log" @@ -36,7 +38,8 @@ const ( // as "/whatsapp status". whatsappCommand = "/whatsapp" - // whatsappUsage is the answer to a WhatsApp command that does not exist. + // whatsappUsage is the answer to a WhatsApp command that does not + // exist. whatsappUsage = "Use /whatsapp pair PHONE, /whatsapp status, or /whatsapp unlink. PHONE is the number of your WhatsApp account with its country code, such as +15550001111." // whatsappPairSteps tells the user what to do with the pairing code, @@ -64,6 +67,20 @@ const ( // whatsappRequests is the number of pair and unlink commands that can // wait for Run. whatsappRequests = 8 + + // whatsappSource is the source of the work events of the batches. + whatsappSource = "whatsapp" + + // whatsappLookupTime is the longest time of a read of the names in the + // session database or of the groups from WhatsApp. + whatsappLookupTime = 10 * time.Second + + // whatsappGroupsRefresh is the shortest time between two reads of the + // groups from WhatsApp for a group with no name. + whatsappGroupsRefresh = time.Minute + + // whatsappUnknownGroup is the name of a group whose name is not known. + whatsappUnknownGroup = "A group with no known name" ) // WhatsAppClient is the part of the WhatsApp client that the daemon uses. The @@ -75,6 +92,7 @@ type WhatsAppClient interface { GetQRChannel(ctx context.Context) (<-chan whatsmeow.QRChannelItem, error) PairPhone(ctx context.Context, phone string, showPushNotification bool, clientType whatsmeow.PairClientType, clientDisplayName string) (string, error) Logout(ctx context.Context) error + GetJoinedGroups(ctx context.Context) ([]*types.GroupInfo, error) } // openWhatsApp opens the WhatsApp session in the data directory dir. Without @@ -91,6 +109,15 @@ func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logge noticeDelay: whatsappNoticeDelay, requests: make(chan WhatsAppRequest, whatsappRequests), logouts: make(chan struct{}, 1), + wake: make(chan struct{}, 1), + failures: make(chan error, 1), + groupNames: make(map[string]string), + } + + var err error + w.batches, err = openWhatsAppBatches(filepath.Join(w.dir, whatsappStateFileName), whatsappBatchQuiet, whatsappBatchWindow) + if err != nil { + return nil, err } info, err := os.Stat(w.sessionPath()) @@ -119,9 +146,9 @@ func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logge return w, nil } -// WhatsApp is the link of the daemon to the WhatsApp account of the user, as -// a linked device. The session database of the library in dir holds the keys -// of the device. Run keeps a linked device connected, and the library connects +// WhatsApp is the link of the daemon to the WhatsApp account of the user, as a +// linked device. The session database of the library in dir holds the keys of +// the device. Run keeps a linked device connected, and the library connects // again by itself after a network failure. // // The commands pair and unlink go to Run as requests, and Run runs one at a @@ -132,10 +159,21 @@ func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logge // client, and the state of the connection, which the event handler of the // library changes from its own goroutines. // +// The text messages go into batches, one chat at a time, as WhatsAppBatches +// says. Run logs each closed batch as a work message that starts with +// [WhatsApp], and a batch that needs the user wakes the turn loop. 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 library logs nothing, because its log lines hold the ID of the own // account, which is the phone number of the user. The own number and the own // LID never go to the log of the agent or to the log of the daemon: redact -// removes them from the errors. +// removes them from the errors, and a batch shows the messages of the user as +// You and a mention of the user as @you. The IDs of the other chats go to the +// log, and those of private chats hold phone numbers, because the agent needs +// them to send. type WhatsApp struct { dir string clock func() time.Time @@ -144,6 +182,9 @@ type WhatsApp struct { noticeDelay time.Duration requests chan WhatsAppRequest logouts chan struct{} + wake chan struct{} + failures chan error + batches *WhatsAppBatches mutex sync.Mutex container *sqlstore.Container @@ -159,13 +200,18 @@ type WhatsApp struct { noticed bool timer *time.Timer closed bool + groupNames map[string]string + groupsAt time.Time } -// Run connects a linked device, runs the pair and unlink commands, and handles -// a logout from the phone, until ctx ends. Its notes go to the log of l, and -// notify gets the notices for all clients. A failed first connection is -// logged, and the library tries again when the failure is one of the network. -func (w *WhatsApp) Run(ctx context.Context, l *TurnLoop, notify func(Event)) { +// Run connects a linked device, logs the closed batches in l, runs the pair and +// unlink commands, and handles a logout from the phone, until ctx ends. Its +// notes go to the log of l, and notify gets the notices for all clients. A +// failed first connection is logged, and the library tries again when the +// failure is one of the network. Run returns the error of a write of the log or +// of the state file, which stops the daemon, because then messages would be +// lost. +func (w *WhatsApp) Run(ctx context.Context, l *TurnLoop, notify func(Event)) error { w.mutex.Lock() w.notify = notify linked := w.linked @@ -184,11 +230,29 @@ func (w *WhatsApp) Run(ctx context.Context, l *TurnLoop, notify func(Event)) { }() for { + err := w.deliver(ctx, l) + if err != nil { + return err + } + + var due <-chan time.Time + wait, open := w.batches.Wait(time.Now()) + if open { + due = time.After(wait) + } + select { case <-ctx.Done(): - return + return nil + case err = <-w.failures: + return err + case <-w.wake: + case <-due: case <-w.logouts: - w.loggedOut(l) + err = w.loggedOut(ctx, l) + if err != nil { + return err + } case request := <-w.requests: cancel() pairing.Wait() @@ -196,7 +260,10 @@ func (w *WhatsApp) Run(ctx context.Context, l *TurnLoop, notify func(Event)) { switch { case request.unlink: - w.unlink(ctx, l, request.answer) + err = w.unlink(ctx, l, request.answer) + if err != nil { + return err + } case w.isLinked(): request.answer("WhatsApp is linked already. Run /whatsapp unlink first.") default: @@ -248,8 +315,8 @@ func (w *WhatsApp) Command(args string, answer func(text string)) { } // Status returns the state of the link: not linked, linked and connected, or -// linked and disconnected, with the times of the connection. It never holds -// the phone number. +// linked and disconnected, with the times of the connection. It never holds the +// phone number. func (w *WhatsApp) Status() string { w.mutex.Lock() defer w.mutex.Unlock() @@ -287,8 +354,8 @@ func (w *WhatsApp) sessionPath() string { return filepath.Join(w.dir, whatsappSessionFileName) } -// openSession opens the session database and its first device. The caller -// holds the mutex, or no other goroutine uses w yet. +// openSession opens the session database and its first device. The caller holds +// the mutex, or no other goroutine uses w yet. func (w *WhatsApp) openSession(ctx context.Context) error { address, err := whatsappSessionAddress(w.sessionPath()) if err != nil { @@ -331,8 +398,7 @@ func (w *WhatsApp) connect(ctx context.Context) { } } -// close ends the connection and the timer of the notice, and stops all -// notices. +// close ends the connection and the timer of the notice, and stops all notices. func (w *WhatsApp) close() { w.mutex.Lock() w.closed = true @@ -349,14 +415,19 @@ func (w *WhatsApp) close() { } } -// loggedOut deletes the session after a logout from the phone, notes it in the -// log of l, and tells all clients one time. -func (w *WhatsApp) loggedOut(l *TurnLoop) { +// loggedOut logs the open batches, deletes the session after a logout from the +// phone, notes it in the log of l, and tells all clients one time. +func (w *WhatsApp) loggedOut(ctx context.Context, l *TurnLoop) error { if !w.isLinked() { - return + return nil } - err := w.removeSession() + err := w.deliverAll(ctx, l) + if err != nil { + return err + } + + err = w.removeSession() if err != nil { w.logger.Printf("delete the session after the logout: %v", err) } @@ -367,13 +438,16 @@ func (w *WhatsApp) loggedOut(l *TurnLoop) { } w.broadcast("WhatsApp was unlinked from the phone. Run /whatsapp pair PHONE to link it again.") + + return nil } -// unlink logs out the linked device and deletes the session. When the logout -// call fails, as it does while WhatsApp is disconnected, the session is -// deleted all the same, and answer tells the user to remove the device in the -// phone. -func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text string)) { +// unlink logs the open batches, logs out the linked device, and deletes the +// session. When the logout call fails, as it does while WhatsApp is +// disconnected, the session is deleted all the same, and answer tells the user +// to remove the device in the phone. Its error is one of the log or the state +// file. +func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text string)) error { w.mutex.Lock() linked := w.linked client := w.client @@ -381,7 +455,12 @@ func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text str if !linked { answer("WhatsApp is not linked.") - return + return nil + } + + err := w.deliverAll(ctx, l) + if err != nil { + return err } var logoutErr error @@ -392,10 +471,10 @@ func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text str logoutErr = client.Logout(ctx) } - err := w.removeSession() + err = w.removeSession() if err != nil { answer("The unlink failed: " + w.redact(err.Error())) - return + return nil } err = l.note("WhatsApp is unlinked.") @@ -405,10 +484,105 @@ func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text str if client == nil || logoutErr != nil { answer("WhatsApp is unlinked here, but the phone did not get the logout. In WhatsApp on your phone, open Linked devices and remove the device.") - return + return nil } answer("WhatsApp is unlinked.") + + return nil +} + +// deliver logs the closed batches in l in order, each as a work message of +// whatsappSource that wakes the loop when the batch starts a turn. A batch +// leaves the state file after its message is durable in the log. A group with +// no name gets its name from WhatsApp first. +func (w *WhatsApp) deliver(ctx context.Context, l *TurnLoop) error { + _, err := w.batches.CloseDue(time.Now()) + if err != nil { + return err + } + + for _, batch := range w.batches.Closed() { + if batch.Chat.Group && batch.Chat.Name == "" { + batch.Chat.Name = w.groupName(ctx, batch.Chat.ID) + } + + _, err = l.Work(whatsappSource, batch.Text(w.clock().Location()), "", batch.Turn()) + if err != nil { + return fmt.Errorf("log a WhatsApp batch: %w", err) + } + + // A stop of the daemon during the flush is no failure: the + // batch stays in the state file and repeats once at the next + // start. + err = l.Flush(ctx) + if ctx.Err() != nil { + return nil + } + + if err != nil { + return fmt.Errorf("log a WhatsApp batch: %w", err) + } + + err = w.batches.Remove(batch.Number) + if err != nil { + return err + } + } + + return nil +} + +// deliverAll closes every open batch and logs all of them. +func (w *WhatsApp) deliverAll(ctx context.Context, l *TurnLoop) error { + err := w.batches.CloseAll() + if err != nil { + return err + } + + return w.deliver(ctx, l) +} + +// groupName returns the name of the group id, or whatsappUnknownGroup. A group +// that is not in the names of the groups reads the groups from WhatsApp again, +// at most once in whatsappGroupsRefresh. +func (w *WhatsApp) groupName(ctx context.Context, id string) string { + w.mutex.Lock() + name, found := w.groupNames[id] + client := w.client + fresh := time.Since(w.groupsAt) < whatsappGroupsRefresh + w.mutex.Unlock() + + if found { + return name + } + + if client == nil || fresh { + return whatsappUnknownGroup + } + + ctx, cancel := context.WithTimeout(ctx, whatsappLookupTime) + defer cancel() + + groups, err := client.GetJoinedGroups(ctx) + if err != nil { + w.logger.Printf("read the groups: %s", w.redact(err.Error())) + } + + w.mutex.Lock() + defer w.mutex.Unlock() + + w.groupsAt = time.Now() + for _, group := range groups { + w.groupNames[group.JID.String()] = group.Name + } + + name, found = w.groupNames[id] + if !found || name == "" { + return whatsappUnknownGroup + } + + return name } // removeSession ends the connection, closes the session database, and deletes @@ -449,11 +623,11 @@ func (w *WhatsApp) removeSession() error { return err } -// pair links a new device to the account of phone with a pairing code. It -// gives the steps and the code to answer, and then the end of the pairing: -// the linked number, an expired code, or a failure. A pairing that a new one -// ends gives no answer. A link notes "WhatsApp is linked." in the log of l, -// with no number. The client of a pairing that does not link is disconnected. +// pair links a new device to the account of phone with a pairing code. It gives +// the steps and the code to answer, and then the end of the pairing: the linked +// number, an expired code, or a failure. A pairing that a new one ends gives no +// answer. A link notes "WhatsApp is linked." in the log of l, with no number. +// The client of a pairing that does not link is disconnected. func (w *WhatsApp) pair(ctx context.Context, l *TurnLoop, phone string, answer func(text string)) { device, err := w.newDevice(ctx) if err != nil { @@ -583,15 +757,27 @@ func (w *WhatsApp) linkedBy(client WhatsAppClient) { w.linked = true } -// handle takes one event of the library and tells if it was handled. A +// handle takes one event of the library and tells if it was handled. A message +// goes to receive. The names of the groups follow the events of the groups. A // disconnection keeps the time of its start while the library tries again. // After noticeDelay of a disconnection, all clients get one notice, and they -// get one more when the connection comes back. A disconnection that the -// library does not try again gives a notice at once. +// get one more when the connection comes back. A disconnection that the library +// does not try again gives a notice at once. func (w *WhatsApp) handle(event any) bool { + message, isMessage := event.(*events.Message) + if isMessage { + return w.receive(message) + } + var notice string w.mutex.Lock() switch event := event.(type) { + case *events.GroupInfo: + if event.Name != nil { + w.groupNames[event.JID.String()] = event.Name.Name + } + case *events.JoinedGroup: + w.groupNames[event.JID.String()] = event.Name case *events.PairSuccess: w.own = whatsappOwnIDs(event.ID.User, event.LID.User) case *events.Connected: @@ -637,6 +823,129 @@ func (w *WhatsApp) handle(event any) bool { return true } +// receive puts the text of event into its batch and wakes Run. 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 (w *WhatsApp) receive(event *events.Message) bool { + ctx, cancel := context.WithTimeout(context.Background(), whatsappLookupTime) + defer cancel() + + chat, message, found := w.message(ctx, event) + if !found { + return true + } + + _, err := w.batches.Add(chat, message, time.Now()) + if err != nil { + select { + case w.failures <- err: + default: + } + + return false + } + + select { + case w.wake <- struct{}{}: + default: + } + + return true +} + +// message returns the chat and the batch message of event, or false when the +// batches do not take it. They 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. +// +// 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 +// log. +func (w *WhatsApp) message(ctx context.Context, event *events.Message) (WhatsAppChat, WhatsAppMessage, bool) { + w.mutex.Lock() + device := w.device + own := slices.Clone(w.own) + w.mutex.Unlock() + + info := event.Info + chatJID := info.Chat.ToNonAD() + switch { + case device == nil: + case event.IsEdit, info.Edit != "", event.IsEphemeral, event.IsViewOnce, event.SourceWebMsg != nil: + case chatJID.Server != types.DefaultUserServer && chatJID.Server != types.HiddenUserServer && chatJID.Server != types.GroupServer: + case slices.Contains(own, chatJID.User): + default: + return w.text(ctx, device, own, event) + } + + return WhatsAppChat{}, WhatsAppMessage{}, false +} + +// text returns the chat and the batch message of event, a message that the +// batches take when it has text. +func (w *WhatsApp) text(ctx context.Context, device *store.Device, own []string, event *events.Message) (WhatsAppChat, WhatsAppMessage, bool) { + info := event.Info + text := event.Message.GetConversation() + extended := event.Message.GetExtendedTextMessage() + if text == "" { + text = extended.GetText() + } + + if strings.TrimSpace(text) == "" { + return WhatsAppChat{}, WhatsAppMessage{}, false + } + + contextInfo := extended.GetContextInfo() + message := WhatsAppMessage{ + ID: info.ID, + Time: info.Timestamp, + Mine: info.IsFromMe, + } + + if !message.Mine { + message.Sender = whatsappContactName(ctx, device, info.Sender, info.PushName) + } + + for _, mentioned := range contextInfo.GetMentionedJID() { + jid, err := types.ParseJID(mentioned) + switch { + case err != nil: + case slices.Contains(own, jid.User): + message.Mention = true + text = strings.ReplaceAll(text, "@"+jid.User, "@you") + default: + text = strings.ReplaceAll(text, "@"+jid.User, "@"+whatsappContactName(ctx, device, jid, "")) + } + } + + participant, err := types.ParseJID(contextInfo.GetParticipant()) + message.Reply = err == nil && slices.Contains(own, participant.User) + + for _, id := range own { + text = strings.ReplaceAll(text, id, "OWN") + } + + message.Text = text + chatJID := info.Chat.ToNonAD() + chat := WhatsAppChat{ + ID: chatJID.String(), + Group: chatJID.Server == types.GroupServer, + } + + if !chat.Group { + push := info.PushName + if message.Mine { + push = "" + } + + chat.Name = whatsappContactName(ctx, device, chatJID, push) + } + + return chat, message, true +} + // disconnectedLong gives the notice of a long disconnection, unless the // disconnection with the number disconnections ended. func (w *WhatsApp) disconnectedLong(disconnections int) { @@ -690,9 +999,9 @@ type WhatsAppRequest struct { } // whatsappNewClient returns the client of the library for device, with handle -// as its event handler. The client tries again after a failed first -// connection, acknowledges a message only after handle returns true, and -// downloads no old history. The phone shows the device with the name OneChat. +// as its event handler. The client tries again after a failed first connection, +// acknowledges a message only after handle returns true, and downloads no old +// history. The phone shows the device with the name OneChat. func whatsappNewClient(device *store.Device, handle func(event any) bool) WhatsAppClient { store.SetOSInfo("OneChat", [3]uint32{0, 1, 0}) client := whatsmeow.NewClient(device, waLog.Noop) @@ -704,6 +1013,48 @@ func whatsappNewClient(device *store.Device, handle func(event any) bool) WhatsA return client } +// whatsappContactName returns the name of the person jid: the full name, the +// first name, the business name, or the push name of the contact in the session +// of device, then push, the push name of a message, then the phone number. A +// LID gets its phone number from the LID map of the session. +func whatsappContactName(ctx context.Context, device *store.Device, jid types.JID, push string) string { + jid = jid.ToNonAD() + candidates := []types.JID{jid} + phone := types.EmptyJID + switch jid.Server { + case types.DefaultUserServer: + phone = jid + case types.HiddenUserServer: + mapped, err := device.LIDs.GetPNForLID(ctx, jid) + if err == nil && !mapped.IsEmpty() { + phone = mapped + candidates = append(candidates, mapped) + } + } + + for _, candidate := range candidates { + contact, err := device.Contacts.GetContact(ctx, candidate) + if err != nil || !contact.Found { + continue + } + + for _, name := range []string{contact.FullName, contact.FirstName, contact.BusinessName, contact.PushName} { + if name != "" { + return name + } + } + } + + switch { + case push != "": + return push + case !phone.IsEmpty(): + return "+" + phone.User + default: + return jid.User + } +} + // whatsappOwnIDs returns the own number and the own LID that are not empty. func whatsappOwnIDs(ids ...string) []string { var own []string @@ -750,9 +1101,9 @@ func whatsappCommandArgs(line string) (string, bool) { return strings.TrimSpace(args), true } -// whatsappPhone returns the digits of phone, a number with its country code -// and an optional + and dashes, and tells if it has the form of a number: 7 to -// 15 digits, with no 0 first. +// whatsappPhone returns the digits of phone, a number with its country code and +// an optional + and dashes, and tells if it has the form of a number: 7 to 15 +// digits, with no 0 first. func whatsappPhone(phone string) (string, bool) { phone = strings.ReplaceAll(strings.TrimPrefix(phone, "+"), "-", "") valid := len(phone) >= 7 && len(phone) <= 15 && phone[0] != '0' && strings.Trim(phone, "0123456789") == "" diff --git a/whatsapp_batch.go b/whatsapp_batch.go new file mode 100644 index 0000000..37b7885 --- /dev/null +++ b/whatsapp_batch.go @@ -0,0 +1,332 @@ +package main + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "slices" + "strings" + "sync" + "time" +) + +const ( + // whatsappStateFileName is the state file in the WhatsApp directory: + // the batches that are not in the log yet. + whatsappStateFileName = "state.json" + + // whatsappBatchQuiet is how long a chat is quiet before its batch + // closes. + whatsappBatchQuiet = 20 * time.Second + + // whatsappBatchWindow is the longest time of a batch. + whatsappBatchWindow = 2 * time.Minute + + // whatsappBatchMessages is the most messages of a batch. + whatsappBatchMessages = 50 + + // whatsappBatchEmptyReply is the last line of a batch that starts a + // turn. + whatsappBatchEmptyReply = "If nothing in this batch needs the user, end your turn with an empty reply." + + // whatsappBatchDayLayout is the layout of the day in the first line of + // a batch. + whatsappBatchDayLayout = "Monday 2006-01-02" +) + +// openWhatsAppBatches reads the batches of the state file at path. A missing +// file holds no batches. A batch closes after quiet with no message, or after +// window at the latest. +func openWhatsAppBatches(path string, quiet, window time.Duration) (*WhatsAppBatches, error) { + b := &WhatsAppBatches{ + path: path, + quiet: quiet, + window: window, + } + + data, err := os.ReadFile(path) + if errors.Is(err, os.ErrNotExist) { + return b, nil + } + + if err != nil { + return nil, fmt.Errorf("read WhatsApp state: %w", err) + } + + err = json.Unmarshal(data, &b.state) + if err != nil { + return nil, fmt.Errorf("%s: %w", path, err) + } + + return b, nil +} + +// WhatsAppBatches holds the batches of the WhatsApp messages that are not in +// the log yet, open and closed, in the order of their first message. The +// messages of one chat form one batch while the times of the messages are at +// most quiet apart, at most window after the first, and the batch has fewer +// than whatsappBatchMessages messages. The times are the WhatsApp times, so a +// catch-up after a stop gives the same batches as live messages. A batch also +// closes when quiet passes after the arrival of its last message, or window +// after the arrival of its first message, in real time. +// +// Each change writes the state file whole, so the batches survive a restart. +// The mutex protects the batches, because the event handler of the library adds +// messages while Run closes and removes batches. +type WhatsAppBatches struct { + path string + quiet time.Duration + window time.Duration + + mutex sync.Mutex + state WhatsAppStateFile +} + +// Add puts message of chat into the open batch of chat, or into a new batch, +// with arrival as its time of arrival. A message that the batch holds already, +// which the servers sent again, changes nothing. It returns true when a batch +// closed. +func (b *WhatsAppBatches) Add(chat WhatsAppChat, message WhatsAppMessage, arrival time.Time) (bool, error) { + b.mutex.Lock() + defer b.mutex.Unlock() + + closed := false + index := b.open(chat.ID) + if index >= 0 { + batch := &b.state.Batches[index] + if slices.ContainsFunc(batch.Messages, func(held WhatsAppMessage) bool { + return held.ID == message.ID + }) { + return false, nil + } + + if message.Time.Sub(batch.Messages[len(batch.Messages)-1].Time) > b.quiet || message.Time.Sub(batch.Messages[0].Time) > b.window { + batch.Closed = true + closed = true + index = -1 + } + } + + if index < 0 { + b.state.Next++ + b.state.Batches = append(b.state.Batches, WhatsAppBatch{ + Number: b.state.Next, + Chat: chat, + First: arrival, + }) + index = len(b.state.Batches) - 1 + } + + batch := &b.state.Batches[index] + batch.Messages = append(batch.Messages, message) + batch.Last = arrival + if len(batch.Messages) >= whatsappBatchMessages { + batch.Closed = true + closed = true + } + + return closed, b.save() +} + +// CloseDue closes each open batch whose time is up at now and tells if one +// closed. +func (b *WhatsAppBatches) CloseDue(now time.Time) (bool, error) { + b.mutex.Lock() + defer b.mutex.Unlock() + + closed := false + for index := range b.state.Batches { + batch := &b.state.Batches[index] + if !batch.Closed && !now.Before(b.deadline(*batch)) { + batch.Closed = true + closed = true + } + } + + if !closed { + return false, nil + } + + return true, b.save() +} + +// CloseAll closes every open batch. +func (b *WhatsAppBatches) CloseAll() error { + b.mutex.Lock() + defer b.mutex.Unlock() + + for index := range b.state.Batches { + b.state.Batches[index].Closed = true + } + + return b.save() +} + +// Closed returns the closed batches in order. +func (b *WhatsAppBatches) Closed() []WhatsAppBatch { + b.mutex.Lock() + defer b.mutex.Unlock() + + var closed []WhatsAppBatch + for _, batch := range b.state.Batches { + if batch.Closed { + closed = append(closed, batch) + } + } + + return closed +} + +// Remove removes the batch with number, which is in the log. +func (b *WhatsAppBatches) Remove(number int) error { + b.mutex.Lock() + defer b.mutex.Unlock() + + b.state.Batches = slices.DeleteFunc(b.state.Batches, func(batch WhatsAppBatch) bool { + return batch.Number == number + }) + + return b.save() +} + +// Wait returns how long after now the first open batch closes, or false when no +// batch is open. +func (b *WhatsAppBatches) Wait(now time.Time) (time.Duration, bool) { + b.mutex.Lock() + defer b.mutex.Unlock() + + var first time.Time + for _, batch := range b.state.Batches { + deadline := b.deadline(batch) + if !batch.Closed && (first.IsZero() || deadline.Before(first)) { + first = deadline + } + } + + if first.IsZero() { + return 0, false + } + + return max(first.Sub(now), 0), true +} + +// open returns the index of the open batch of chat, or -1. The caller holds the +// mutex. +func (b *WhatsAppBatches) open(chat string) int { + return slices.IndexFunc(b.state.Batches, func(batch WhatsAppBatch) bool { + return !batch.Closed && batch.Chat.ID == chat + }) +} + +// deadline returns when batch closes in real time. +func (b *WhatsAppBatches) deadline(batch WhatsAppBatch) time.Time { + quiet := batch.Last.Add(b.quiet) + window := batch.First.Add(b.window) + if window.Before(quiet) { + return window + } + + return quiet +} + +// save writes the state file whole. The caller holds the mutex. +func (b *WhatsAppBatches) save() error { + data, err := json.Marshal(b.state) + if err != nil { + return fmt.Errorf("encode WhatsApp state: %w", err) + } + + err = os.MkdirAll(filepath.Dir(b.path), 0o700) + if err == nil { + err = writeFileAtomic(b.path, append(data, '\n')) + } + + if err != nil { + return fmt.Errorf("save WhatsApp state: %w", err) + } + + return nil +} + +// WhatsAppStateFile is the content of the state file: the number of the last +// batch and the batches that are not in the log yet. +type WhatsAppStateFile struct { + Next int `json:"next"` + Batches []WhatsAppBatch `json:"batches"` +} + +// WhatsAppBatch is one batch of the messages of one chat, with the arrival +// times of its first and its last message. +type WhatsAppBatch struct { + Number int `json:"number"` + Chat WhatsAppChat `json:"chat"` + Closed bool `json:"closed"` + First time.Time `json:"first"` + Last time.Time `json:"last"` + Messages []WhatsAppMessage `json:"messages"` +} + +// Turn tells if the batch starts a turn: a private batch with a message of +// another person, or a group batch with a message of another person that +// mentions the user or replies to a message of the user. +func (b WhatsAppBatch) Turn() bool { + return slices.ContainsFunc(b.Messages, func(message WhatsAppMessage) bool { + return !message.Mine && (!b.Chat.Group || message.Mention || message.Reply) + }) +} + +// Text returns the text of the batch for the log, with the times in location. +// The first line starts with [WhatsApp] and names the chat, its ID, and the day +// of the first message. Each message has a line with its time and its sender, +// and the messages of the user have the sender You. A batch that starts a turn +// ends with the line about the empty reply. +func (b WhatsAppBatch) Text(location *time.Location) string { + kind := "private chat" + if b.Chat.Group { + kind = "group" + } + + lines := []string{"[WhatsApp] " + b.Chat.Name + ", " + kind + " " + b.Chat.ID + ", " + b.Messages[0].Time.In(location).Format(whatsappBatchDayLayout)} + for _, message := range b.Messages { + sender := message.Sender + switch { + case message.Mine: + sender = "You" + case message.Mention: + sender += " (mentions you)" + case message.Reply: + sender += " (replies to you)" + } + + lines = append(lines, message.Time.In(location).Format("15:04")+" "+sender+": "+message.Text) + } + + if b.Turn() { + lines = append(lines, whatsappBatchEmptyReply) + } + + return strings.Join(lines, "\n") +} + +// WhatsAppChat is a chat: its ID, its name, and whether it is a group. +type WhatsAppChat struct { + ID string `json:"id"` + Name string `json:"name"` + Group bool `json:"group,omitempty"` +} + +// WhatsAppMessage is one text message of a batch: its ID, the name of its +// sender, its WhatsApp time, and its text. Mine marks a message of the user. +// Mention and Reply mark a message that mentions the user or replies to a +// message of the user. +type WhatsAppMessage struct { + ID string `json:"id"` + Sender string `json:"sender"` + Time time.Time `json:"time"` + Text string `json:"text"` + Mine bool `json:"mine,omitempty"` + Mention bool `json:"mention,omitempty"` + Reply bool `json:"reply,omitempty"` +} diff --git a/whatsapp_batch_test.go b/whatsapp_batch_test.go new file mode 100644 index 0000000..1a61440 --- /dev/null +++ b/whatsapp_batch_test.go @@ -0,0 +1,239 @@ +package main + +import ( + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestWhatsAppBatchesFollowTheTimesOfTheMessages(t *testing.T) { + testCases := []struct { + name string + times []time.Duration + sizes []int + }{ + { + name: "messages close together form one batch", + times: []time.Duration{0, 10 * time.Second, 30 * time.Second}, + sizes: []int{3}, + }, + { + name: "a quiet chat starts a new batch", + times: []time.Duration{0, 10 * time.Second, 31 * time.Second}, + sizes: []int{2, 1}, + }, + { + name: "a batch ends after two minutes", + times: []time.Duration{0, 20 * time.Second, 40 * time.Second, 60 * time.Second, 80 * time.Second, 100 * time.Second, 120 * time.Second, 140 * time.Second}, + sizes: []int{7, 1}, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + batches := whatsappTestBatches(t, filepath.Join(t.TempDir(), whatsappStateFileName)) + start := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + for index, offset := range tc.times { + _, err := batches.Add(whatsappTestChat, WhatsAppMessage{ + ID: string(rune('a' + index)), + Time: start.Add(offset), + Text: "hi", + }, time.Now()) + require.NoError(t, err) + } + + require.NoError(t, batches.CloseAll()) + var sizes []int + for _, batch := range batches.Closed() { + sizes = append(sizes, len(batch.Messages)) + } + + require.Equal(t, tc.sizes, sizes) + }) + } +} + +func TestWhatsAppBatchClosesAtFiftyMessages(t *testing.T) { + batches := whatsappTestBatches(t, filepath.Join(t.TempDir(), whatsappStateFileName)) + start := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + for index := range whatsappBatchMessages { + closed, err := batches.Add(whatsappTestChat, WhatsAppMessage{ + ID: time.Duration(index).String(), + Time: start, + Text: "hi", + }, time.Now()) + require.NoError(t, err) + require.Equal(t, index == whatsappBatchMessages-1, closed) + } + + require.Len(t, batches.Closed(), 1) +} + +func TestWhatsAppBatchClosesInRealTime(t *testing.T) { + batches := whatsappTestBatches(t, filepath.Join(t.TempDir(), whatsappStateFileName)) + arrival := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + for index, offset := range []time.Duration{0, 15 * time.Second, 30 * time.Second} { + _, err := batches.Add(whatsappTestChat, WhatsAppMessage{ + ID: string(rune('a' + index)), + Time: arrival, + Text: "hi", + }, arrival.Add(offset)) + require.NoError(t, err) + } + + wait, open := batches.Wait(arrival.Add(40 * time.Second)) + require.True(t, open) + require.Equal(t, 10*time.Second, wait) + + closed, err := batches.CloseDue(arrival.Add(49 * time.Second)) + require.NoError(t, err) + require.False(t, closed) + + closed, err = batches.CloseDue(arrival.Add(50 * time.Second)) + require.NoError(t, err) + require.True(t, closed) + + _, open = batches.Wait(arrival.Add(50 * time.Second)) + require.False(t, open) +} + +func TestWhatsAppBatchesSurviveARestart(t *testing.T) { + path := filepath.Join(t.TempDir(), whatsappDirName, whatsappStateFileName) + batches := whatsappTestBatches(t, path) + for _, id := range []string{"a", "a", "b"} { + _, err := batches.Add(whatsappTestChat, WhatsAppMessage{ + ID: id, + Time: time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC), + Text: "hi " + id, + }, time.Now()) + require.NoError(t, err) + } + + info, err := os.Stat(path) + require.NoError(t, err) + require.Equal(t, os.FileMode(0o600), info.Mode().Perm()) + + again := whatsappTestBatches(t, path) + require.NoError(t, again.CloseAll()) + closed := again.Closed() + require.Len(t, closed, 1) + require.Equal(t, []string{"hi a", "hi b"}, []string{closed[0].Messages[0].Text, closed[0].Messages[1].Text}) + + require.NoError(t, again.Remove(closed[0].Number)) + require.Empty(t, whatsappTestBatches(t, path).Closed()) +} + +func TestWhatsAppBatchesRefuseABrokenStateFile(t *testing.T) { + path := filepath.Join(t.TempDir(), whatsappStateFileName) + require.NoError(t, os.WriteFile(path, []byte("{"), 0o600)) + + _, err := openWhatsAppBatches(path, whatsappBatchQuiet, whatsappBatchWindow) + require.ErrorContains(t, err, path) +} + +func TestWhatsAppBatchText(t *testing.T) { + at := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + testCases := []struct { + name string + batch WhatsAppBatch + text string + }{ + { + name: "a private chat with a message of another person starts a turn", + batch: WhatsAppBatch{ + Chat: whatsappTestChat, + Messages: []WhatsAppMessage{ + { + Sender: "Ana Souza", + Time: at, + Text: "Are you at home?", + }, + { + Time: at.Add(3 * time.Minute), + Text: "Yes", + Mine: true, + }, + }, + }, + text: "[WhatsApp] Ana Souza, private chat 15550002222@s.whatsapp.net, Thursday 2026-01-01\n12:00 Ana Souza: Are you at home?\n12:03 You: Yes\n" + whatsappBatchEmptyReply, + }, + { + name: "a private chat with messages of the user only starts no turn", + batch: WhatsAppBatch{ + Chat: whatsappTestChat, + Messages: []WhatsAppMessage{ + { + Time: at, + Text: "On my way.", + Mine: true, + }, + }, + }, + text: "[WhatsApp] Ana Souza, private chat 15550002222@s.whatsapp.net, Thursday 2026-01-01\n12:00 You: On my way.", + }, + { + name: "a group with no mention of the user starts no turn", + batch: WhatsAppBatch{ + Chat: whatsappTestGroup, + Messages: []WhatsAppMessage{ + { + Sender: "Bruno Lima", + Time: at, + Text: "Who brings the cake?", + }, + }, + }, + text: "[WhatsApp] Family, group 120363000000000001@g.us, Thursday 2026-01-01\n12:00 Bruno Lima: Who brings the cake?", + }, + { + name: "a group with a mention of the user starts a turn", + batch: WhatsAppBatch{ + Chat: whatsappTestGroup, + Messages: []WhatsAppMessage{ + { + Sender: "Bruno Lima", + Time: at, + Text: "@you can you bring the cake?", + Mention: true, + }, + }, + }, + text: "[WhatsApp] Family, group 120363000000000001@g.us, Thursday 2026-01-01\n12:00 Bruno Lima (mentions you): @you can you bring the cake?\n" + whatsappBatchEmptyReply, + }, + { + name: "a group with a reply to the user starts a turn", + batch: WhatsAppBatch{ + Chat: whatsappTestGroup, + Messages: []WhatsAppMessage{ + { + Sender: "Bruno Lima", + Time: at, + Text: "Thanks!", + Reply: true, + }, + }, + }, + text: "[WhatsApp] Family, group 120363000000000001@g.us, Thursday 2026-01-01\n12:00 Bruno Lima (replies to you): Thanks!\n" + whatsappBatchEmptyReply, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.text, tc.batch.Text(time.UTC)) + }) + } +} + +// whatsappTestBatches opens the batches of the state file at path with the +// times of the plan. +func whatsappTestBatches(t *testing.T, path string) *WhatsAppBatches { + t.Helper() + + batches, err := openWhatsAppBatches(path, whatsappBatchQuiet, whatsappBatchWindow) + require.NoError(t, err) + + return batches +} diff --git a/whatsapp_fake_helper_test.go b/whatsapp_fake_helper_test.go index 1a7130c..7cb61b3 100644 --- a/whatsapp_fake_helper_test.go +++ b/whatsapp_fake_helper_test.go @@ -8,13 +8,16 @@ import ( "path/filepath" "sync" "testing" + "time" "github.com/stretchr/testify/require" "go.mau.fi/whatsmeow" waAdv "go.mau.fi/whatsmeow/proto/waAdv" + "go.mau.fi/whatsmeow/proto/waE2E" "go.mau.fi/whatsmeow/store" "go.mau.fi/whatsmeow/store/sqlstore" "go.mau.fi/whatsmeow/types" + "go.mau.fi/whatsmeow/types/events" waLog "go.mau.fi/whatsmeow/util/log" ) @@ -30,11 +33,39 @@ const ( whatsappTestCode = "ABCD-EFGH" ) +var ( + // whatsappTestOwn is the own account in the tests. + whatsappTestOwn = types.NewJID(whatsappTestNumber, types.DefaultUserServer) + + // whatsappTestAna is a contact in the tests. + whatsappTestAna = types.NewJID("15550002222", types.DefaultUserServer) + + // whatsappTestBruno is a member of the test group. + whatsappTestBruno = types.NewJID("15550003333", types.DefaultUserServer) + + // whatsappTestFamily is the test group. + whatsappTestFamily = types.NewJID("120363000000000001", types.GroupServer) + + // whatsappTestChat is the private chat with whatsappTestAna. + whatsappTestChat = WhatsAppChat{ + ID: whatsappTestAna.String(), + Name: "Ana Souza", + } + + // whatsappTestGroup is the chat of whatsappTestFamily. + whatsappTestGroup = WhatsAppChat{ + ID: whatsappTestFamily.String(), + Name: "Family", + Group: true, + } +) + // WhatsAppFake is a fake WhatsApp client. It keeps the event handler that the // daemon gives it, and Emit gives events to the handler, as the library does. // Each connection sends to connects, and each pairing sends its channel to -// pairings, so that the test gives the items of the pairing. A connection -// fails with connectErr, and a logout with logoutErr. +// pairings, so that the test gives the items of the pairing. A connection fails +// with connectErr, and a logout with logoutErr. groups are the groups of the +// account. type WhatsAppFake struct { connects chan struct{} pairings chan chan whatsmeow.QRChannelItem @@ -43,6 +74,7 @@ type WhatsAppFake struct { handle func(event any) bool connectErr error logoutErr error + groups []*types.GroupInfo phones []string logouts int } @@ -89,6 +121,14 @@ func (f *WhatsAppFake) Logout(ctx context.Context) error { return f.logoutErr } +// GetJoinedGroups returns groups. +func (f *WhatsAppFake) GetJoinedGroups(ctx context.Context) ([]*types.GroupInfo, error) { + f.mutex.Lock() + defer f.mutex.Unlock() + + return f.groups, nil +} + // Phones returns the phone numbers of the pairings. func (f *WhatsAppFake) Phones() []string { f.mutex.Lock() @@ -180,8 +220,27 @@ func whatsappTestSession(t *testing.T, dir string, linked bool) { require.NoError(t, container.Close()) } -// whatsappDiscardLog is a logger for a WhatsApp whose log a test does not -// read. +// whatsappDiscardLog is a logger for a WhatsApp whose log a test does not read. func whatsappDiscardLog() *log.Logger { return log.New(io.Discard, "", 0) } + +// whatsappTestMessage returns the event of the text message id of sender in +// chat at the time at. A sender with whatsappTestNumber is the user. +func whatsappTestMessage(chat, sender types.JID, id, text string, at time.Time) *events.Message { + return &events.Message{ + Info: types.MessageInfo{ + MessageSource: types.MessageSource{ + Chat: chat, + Sender: sender, + IsFromMe: sender.User == whatsappTestNumber, + IsGroup: chat.Server == types.GroupServer, + }, + ID: id, + Timestamp: at, + }, + Message: &waE2E.Message{ + Conversation: new(text), + }, + } +} diff --git a/whatsapp_test.go b/whatsapp_test.go index e0018e5..38a7b8d 100644 --- a/whatsapp_test.go +++ b/whatsapp_test.go @@ -2,6 +2,7 @@ package main import ( "errors" + "net/http" "os" "path/filepath" "strings" @@ -10,6 +11,8 @@ import ( "github.com/stretchr/testify/require" "go.mau.fi/whatsmeow" + "go.mau.fi/whatsmeow/proto/waE2E" + "go.mau.fi/whatsmeow/proto/waWeb" "go.mau.fi/whatsmeow/types" "go.mau.fi/whatsmeow/types/events" ) @@ -467,3 +470,203 @@ func whatsappTestLogTexts(t *testing.T, memory *Memory) []string { return texts } + +func TestWhatsAppPrivateBatchStartsATurnWithAnEmptyReply(t *testing.T) { + whatsapp, fake := whatsappStartTestLinked(t) + require.NoError(t, whatsapp.device.Contacts.PutContactName(t.Context(), whatsappTestAna, "Ana", "Ana Souza")) + + server := newModelFake(t) + server.reply = func(w http.ResponseWriter, call ModelFakeCall) { + modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5)) + } + + telegram := newTelegramFake(t) + bot, _ := telegramStartTestBot(t, telegram) + socket, memory := daemonStartTestWithWhatsApp(t, server, bot, whatsapp) + client := terminalStartTestClient(t, socket) + client.WaitFor(t, "[connected to OneChat]") + <-fake.connects + + at := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + require.True(t, fake.Emit(whatsappTestMessage(whatsappTestAna, whatsappTestAna, "m1", "Are you at home?", at))) + require.True(t, fake.Emit(whatsappTestMessage(whatsappTestAna, whatsappTestOwn, "m2", "Yes", at))) + + text := "[WhatsApp] Ana Souza, private chat 15550002222@s.whatsapp.net, Thursday 2026-01-01\n12:00 Ana Souza: Are you at home?\n12:00 You: Yes\n" + whatsappBatchEmptyReply + client.WaitFor(t, text+"\n[working]\n[done]\n") + require.Equal(t, 1, server.Calls()) + require.Contains(t, string(server.Call(0).raw), "Are you at home?") + require.Equal(t, []string{text}, whatsappTestLogTexts(t, memory)) + require.Empty(t, telegram.Sent()) + require.Eventually(t, func() bool { + return len(whatsapp.batches.Closed()) == 0 + }, 10*time.Second, time.Millisecond) +} + +func TestWhatsAppGroupTurns(t *testing.T) { + testCases := []struct { + name string + message func(event *events.Message) + line string + turn bool + }{ + { + name: "plain", + message: func(event *events.Message) { + }, + line: "12:00 Bruno Lima: Who brings the cake?", + }, + { + name: "mention", + message: func(event *events.Message) { + event.Message = &waE2E.Message{ + ExtendedTextMessage: &waE2E.ExtendedTextMessage{ + Text: new("@" + whatsappTestLID + " can you bring the cake? Call " + whatsappTestNumber + "."), + ContextInfo: &waE2E.ContextInfo{ + MentionedJID: []string{whatsappTestLID + "@lid"}, + }, + }, + } + }, + line: "12:00 Bruno Lima (mentions you): @you can you bring the cake? Call OWN.", + turn: true, + }, + { + name: "reply", + message: func(event *events.Message) { + event.Message = &waE2E.Message{ + ExtendedTextMessage: &waE2E.ExtendedTextMessage{ + Text: new("Who brings the cake?"), + ContextInfo: &waE2E.ContextInfo{ + Participant: new(whatsappTestOwn.String()), + }, + }, + } + }, + line: "12:00 Bruno Lima (replies to you): Who brings the cake?", + turn: true, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + whatsapp, fake := whatsappStartTestLinked(t) + fake.groups = []*types.GroupInfo{ + { + JID: whatsappTestFamily, + GroupName: types.GroupName{ + Name: "Family", + }, + }, + } + + server := newModelFake(t) + socket, memory := daemonStartTestWithWhatsApp(t, server, nil, whatsapp) + client := terminalStartTestClient(t, socket) + client.WaitFor(t, "[connected to OneChat]") + <-fake.connects + + event := whatsappTestMessage(whatsappTestFamily, whatsappTestBruno, "m1", "Who brings the cake?", time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)) + event.Info.PushName = "Bruno Lima" + tc.message(event) + require.True(t, fake.Emit(event)) + + text := "[WhatsApp] Family, group 120363000000000001@g.us, Thursday 2026-01-01\n" + tc.line + if tc.turn { + text += "\n" + whatsappBatchEmptyReply + client.WaitFor(t, "[done]") + require.Equal(t, 1, server.Calls()) + } else { + client.WaitFor(t, text) + time.Sleep(100 * time.Millisecond) + require.Zero(t, server.Calls()) + } + + require.Equal(t, text, whatsappTestLogTexts(t, memory)[0]) + }) + } +} + +func TestWhatsAppTakesOnlyText(t *testing.T) { + at := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + testCases := []struct { + name string + change func(event *events.Message) + }{ + { + name: "an edit", + change: func(event *events.Message) { + event.IsEdit = true + }, + }, + { + name: "old history", + change: func(event *events.Message) { + event.SourceWebMsg = &waWeb.WebMessageInfo{} + }, + }, + { + name: "a view-once message", + change: func(event *events.Message) { + event.IsViewOnce = true + }, + }, + { + name: "a disappearing message", + change: func(event *events.Message) { + event.IsEphemeral = true + }, + }, + { + name: "the chat of the user with itself", + change: func(event *events.Message) { + event.Info.Chat = whatsappTestOwn + }, + }, + { + name: "a status broadcast", + change: func(event *events.Message) { + event.Info.Chat = types.StatusBroadcastJID + }, + }, + { + name: "a photo with a caption", + change: func(event *events.Message) { + event.Message = &waE2E.Message{ + ImageMessage: &waE2E.ImageMessage{ + Caption: new("look"), + }, + } + }, + }, + { + name: "a text of spaces", + change: func(event *events.Message) { + event.Message.Conversation = new(" ") + }, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + whatsapp, _ := whatsappStartTestLinked(t) + event := whatsappTestMessage(whatsappTestAna, whatsappTestAna, "m1", "hi", at) + tc.change(event) + require.True(t, whatsapp.handle(event)) + require.NoError(t, whatsapp.batches.CloseAll()) + require.Empty(t, whatsapp.batches.Closed()) + }) + } +} + +// whatsappStartTestLinked opens a WhatsApp with a linked device in a temporary +// directory, with a quiet time of 50 ms for its batches. +func whatsappStartTestLinked(t *testing.T) (*WhatsApp, *WhatsAppFake) { + t.Helper() + + dir := t.TempDir() + whatsappTestSession(t, dir, true) + whatsapp, fake, _ := whatsappStartTest(t, dir, newTestClock()) + whatsapp.batches.quiet = 50 * time.Millisecond + + return whatsapp, fake +}