// 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
- // whatsappCatchupLimit is the longest wait for the end of the catch-up
- // after a connection.
- whatsappCatchupLimit = time.Minute
-
// whatsappFindMatches is the most matches of whatsapp_find.
whatsappFindMatches = 20
// the client of a device with its event handler.
func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logger *log.Logger, newClient func(device *store.Device, handle func(event any) bool) WhatsAppClient) (*WhatsApp, error) {
w := &WhatsApp{
- dir: filepath.Join(dir, whatsappDirName),
- clock: clock,
- logger: logger,
- newClient: newClient,
- noticeDelay: whatsappNoticeDelay,
- catchupLimit: whatsappCatchupLimit,
- requests: make(chan WhatsAppRequest, whatsappRequests),
- logouts: make(chan struct{}, 1),
- wake: make(chan struct{}, 1),
- failures: make(chan error, 1),
- groups: newWhatsAppGroups(logger),
+ dir: filepath.Join(dir, whatsappDirName),
+ clock: clock,
+ logger: logger,
+ newClient: newClient,
+ noticeDelay: whatsappNoticeDelay,
+ requests: make(chan WhatsAppRequest, whatsappRequests),
+ logouts: make(chan struct{}, 1),
+ groups: newWhatsAppGroups(logger),
}
var err error
return nil, err
}
+ w.inbox = newWhatsAppInbox(w.state, clock, w.groups)
+
info, err := os.Stat(w.sessionPath())
if errors.Is(err, os.ErrNotExist) {
return w, nil
if w.device.ID != nil {
w.linked = true
- w.catchingUp = true
+ w.inbox.waitForCatchup()
w.own = whatsappOwnIDs(w.device.ID.User, w.device.LID.User)
}
// 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 WhatsAppState 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.
-//
-// 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, Run 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 inbox puts the text messages into batches and logs them, and groups
+// holds the names of the groups. Both get a WhatsAppLink from link for each
+// call, so they use the current device and client.
//
// 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
// 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
- logger *log.Logger
- newClient func(device *store.Device, handle func(event any) bool) WhatsAppClient
- noticeDelay time.Duration
- catchupLimit time.Duration
- requests chan WhatsAppRequest
- logouts chan struct{}
- wake chan struct{}
- failures chan error
- state *WhatsAppState
- groups *WhatsAppGroups
+ dir string
+ clock func() time.Time
+ logger *log.Logger
+ newClient func(device *store.Device, handle func(event any) bool) WhatsAppClient
+ noticeDelay time.Duration
+ requests chan WhatsAppRequest
+ logouts chan struct{}
+ state *WhatsAppState
+ groups *WhatsAppGroups
+ inbox *WhatsAppInbox
mutex sync.Mutex
container *sqlstore.Container
noticed bool
timer *time.Timer
closed bool
- catchingUp bool
- caughtUp bool
- catchups int
- catchupTimer *time.Timer
}
// Run connects a linked device, logs the closed batches in l, runs the pair and
}()
for {
- err := w.deliver(ctx, l, false)
+ err := w.inbox.deliver(ctx, l, w.link(), false)
if err != nil {
return err
}
- var due <-chan time.Time
- wait, open := w.state.Wait(time.Now())
- if open {
- due = time.After(wait)
- }
-
select {
case <-ctx.Done():
return nil
- case err = <-w.failures:
+ case err = <-w.inbox.failures:
return err
- case <-w.wake:
- case <-due:
+ case <-w.inbox.wake:
+ case <-w.inbox.due():
case <-w.logouts:
err = w.loggedOut(ctx, l)
if err != nil {
w.timer.Stop()
}
- if w.catchupTimer != nil {
- w.catchupTimer.Stop()
- }
-
client := w.client
w.client = nil
w.mutex.Unlock()
+ w.inbox.close()
+
if client != nil {
client.Disconnect()
}
return nil
}
- err := w.deliver(ctx, l, true)
+ err := w.inbox.deliver(ctx, l, w.link(), true)
if err != nil {
return err
}
return nil
}
- err := w.deliver(ctx, l, true)
+ err := w.inbox.deliver(ctx, l, w.link(), true)
if err != nil {
return err
}
return nil
}
-// deliver logs the closed batches in l in order, each as a work message of
-// whatsappSource, and wakes the loop once with the batches that start 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. 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.
-func (w *WhatsApp) deliver(ctx context.Context, l *TurnLoop, all bool) error {
- w.mutex.Lock()
- catchingUp := w.catchingUp
- caughtUp := w.caughtUp
- w.caughtUp = false
- w.mutex.Unlock()
-
- var err error
- switch {
- case all || caughtUp:
- err = w.state.CloseAll()
- case catchingUp:
- return nil
- default:
- _, err = w.state.CloseDue(time.Now())
- }
-
- if err != nil {
- return err
- }
-
- var turns []Message
- for _, batch := range w.state.Closed() {
- if batch.Chat.Group && batch.Chat.Name == "" {
- batch.Chat.Name = w.groups.name(ctx, w.link(), batch.Chat.ID)
- }
-
- message, err := l.Work(whatsappSource, batch.Text(w.clock().Location()), "")
- 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.state.Remove(batch.Number)
- if err != nil {
- return err
- }
-
- if batch.Turn() {
- turns = append(turns, message)
- }
- }
-
- l.Wake(turns...)
-
- return nil
-}
-
// Find returns the contacts and the groups whose name contains name, with no
// regard to case and accents, as lines "ID | name | contact" or "ID | name |
// group", sorted by name. It gives at most whatsappFindMatches lines and the
w.linked = false
w.own = nil
w.connected = false
- w.catchingUp = false
if w.timer != nil {
w.timer.Stop()
}
w.device = nil
w.mutex.Unlock()
+ w.inbox.stopCatchup()
+
if client != nil {
client.Disconnect()
}
}
// 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
-// connection starts a catch-up, which ends at OfflineSyncCompleted or after
-// catchupLimit. 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.
+// goes to the inbox. The names of the groups follow the events of the groups. A
+// connection starts a catch-up of the inbox, which ends at OfflineSyncCompleted
+// or after its catchupLimit. 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.
func (w *WhatsApp) handle(event any) bool {
switch event := event.(type) {
case *events.Message:
- return w.receive(event)
+ return w.inbox.receive(w.link(), event)
case *events.GroupInfo:
if event.Name != nil {
w.groups.set(event.JID.String(), event.Name.Name)
case *events.JoinedGroup:
w.groups.set(event.JID.String(), event.Name)
return true
+ case *events.OfflineSyncCompleted:
+ w.inbox.endCatchup()
+ return true
+ case *events.Connected:
+ w.inbox.startCatchup()
}
var notice string
case *events.Connected:
w.connected = true
w.connectedAt = w.clock()
- w.catchingUp = true
- w.catchups++
- catchups := w.catchups
- if w.catchupTimer != nil {
- w.catchupTimer.Stop()
- }
-
- w.catchupTimer = time.AfterFunc(w.catchupLimit, func() {
- w.caughtUpBy(catchups)
- })
if w.timer != nil {
w.timer.Stop()
}
}
w.connected = false
- case *events.OfflineSyncCompleted:
- w.endCatchup()
case *events.LoggedOut:
select {
case w.logouts <- struct{}{}:
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()
-
- w.mutex.Lock()
- device := w.device
- own := slices.Clone(w.own)
- w.mutex.Unlock()
-
- chat, message, found := whatsappMessage(ctx, device, own, event)
- if !found {
- return true
- }
-
- _, err := w.state.Add(chat, message, time.Now())
- if err != nil {
- select {
- case w.failures <- err:
- default:
- }
-
- return false
- }
-
- select {
- case w.wake <- struct{}{}:
- default:
- }
-
- return true
-}
-
-// caughtUpBy ends the catch-up of the connection with the number catchups, when
-// no later connection started a new one.
-func (w *WhatsApp) caughtUpBy(catchups int) {
- w.mutex.Lock()
- defer w.mutex.Unlock()
-
- if w.catchups == catchups {
- w.endCatchup()
- }
-}
-
-// endCatchup ends a running catch-up and wakes Run, which logs all batches. The
-// caller holds the mutex.
-func (w *WhatsApp) endCatchup() {
- if !w.catchingUp {
- return
- }
-
- w.catchingUp = false
- w.caughtUp = true
- if w.catchupTimer != nil {
- w.catchupTimer.Stop()
- }
-
- select {
- case w.wake <- struct{}{}:
- default:
- }
-}
-
// disconnectedLong gives the notice of a long disconnection, unless the
// disconnection with the number disconnections ended.
func (w *WhatsApp) disconnectedLong(disconnections int) {
return client
}
-// 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.
-//
-// 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 whatsappMessage(ctx context.Context, device *store.Device, own []string, event *events.Message) (WhatsAppChat, WhatsAppMessage, bool) {
- 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 whatsappText(ctx, device, own, event)
- }
-
- return WhatsAppChat{}, WhatsAppMessage{}, false
-}
-
-// whatsappText returns the chat and the batch message of event, a message that
-// the batches take when it has text.
-func whatsappText(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)
-
- message.Text = whatsappRedact(own, 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
-}
-
// 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
projects / onechat / blobdiff