git.giacomolli.uk - onechat/blob - whatsapp_inbox.go

   1 package main
   2
   3 import (
   4         "context"
   5         "fmt"
   6         "slices"
   7         "strings"
   8         "sync"
   9         "time"
  10
  11         "go.mau.fi/whatsmeow/store"
  12         "go.mau.fi/whatsmeow/types"
  13         "go.mau.fi/whatsmeow/types/events"
  14 )
  15
  16 const (
  17         // whatsappSource is the source of the work events of the batches.
  18         whatsappSource = "whatsapp"
  19
  20         // whatsappCatchupLimit is the longest wait for the end of the catch-up
  21         // after a connection.
  22         whatsappCatchupLimit = time.Minute
  23 )
  24
  25 // newWhatsAppInbox returns the inbox of the batches in state, with clock for
  26 // the times of the batch texts and groups for the names of the groups.
  27 func newWhatsAppInbox(state *WhatsAppState, clock func() time.Time, groups *WhatsAppGroups) *WhatsAppInbox {
  28         return &WhatsAppInbox{
  29                 state:        state,
  30                 clock:        clock,
  31                 groups:       groups,
  32                 catchupLimit: whatsappCatchupLimit,
  33                 wake:         make(chan struct{}, 1),
  34                 failures:     make(chan error, 1),
  35         }
  36 }
  37
  38 // WhatsAppInbox puts the text messages into batches, one chat at a time, as
  39 // WhatsAppState says. deliver logs each closed batch as a work message that
  40 // starts with [WhatsApp], and a batch that needs the user wakes the turn loop.
  41 //
  42 // After each connection, the servers send the messages that came while the
  43 // daemon was away, and then OfflineSyncCompleted. Until then, or until
  44 // catchupLimit passes, the closed batches wait in the state file. At the end of
  45 // the catch-up, deliver closes all batches, logs them, and wakes the turn loop
  46 // once, so the whole catch-up gives one turn and not one turn for each chat.
  47 // The batches of the state file at the start wait for the first catch-up in the
  48 // same way. The event handler acknowledges a message only after the state file
  49 // holds it, so the servers send again a message that a crash lost. A batch
  50 // leaves the state file only after its work message is durable in the log. A
  51 // crash between the two makes the batch repeat once.
  52 //
  53 // The state file is the one of WhatsApp, which keeps the send records in it
  54 // too. groups gives the names of the groups. The mutex protects the state of
  55 // the catch-up, which the event handler of the library changes from its own
  56 // goroutines. wake tells Run that a batch can be due or that a catch-up ended,
  57 // and failures gives Run the error of a write of the state file.
  58 type WhatsAppInbox struct {
  59         state        *WhatsAppState
  60         clock        func() time.Time
  61         groups       *WhatsAppGroups
  62         catchupLimit time.Duration
  63         wake         chan struct{}
  64         failures     chan error
  65
  66         mutex        sync.Mutex
  67         catchingUp   bool
  68         caughtUp     bool
  69         catchups     int
  70         catchupTimer *time.Timer
  71 }
  72
  73 // due returns a channel that gets a value when the next open batch is due, or
  74 // nil when no batch is open.
  75 func (i *WhatsAppInbox) due() <-chan time.Time {
  76         wait, open := i.state.Wait(time.Now())
  77         if !open {
  78                 return nil
  79         }
  80
  81         return time.After(wait)
  82 }
  83
  84 // deliver logs the closed batches in l in order, each as a work message of
  85 // whatsappSource, and wakes the loop once with the batches that start a turn. A
  86 // batch leaves the state file after its message is durable in the log. A group
  87 // with no name gets its name from WhatsApp first, with the client of link.
  88 // During a catch-up it logs nothing, and at its end it closes all batches
  89 // first. With all, it closes and logs all batches also during a catch-up.
  90 func (i *WhatsAppInbox) deliver(ctx context.Context, l *TurnLoop, link WhatsAppLink, all bool) error {
  91         i.mutex.Lock()
  92         catchingUp := i.catchingUp
  93         caughtUp := i.caughtUp
  94         i.caughtUp = false
  95         i.mutex.Unlock()
  96
  97         var err error
  98         switch {
  99         case all || caughtUp:
 100                 err = i.state.CloseAll()
 101         case catchingUp:
 102                 return nil
 103         default:
 104                 _, err = i.state.CloseDue(time.Now())
 105         }
 106
 107         if err != nil {
 108                 return err
 109         }
 110
 111         var turns []Message
 112         for _, batch := range i.state.Closed() {
 113                 if batch.Chat.Group && batch.Chat.Name == "" {
 114                         batch.Chat.Name = i.groups.name(ctx, link, batch.Chat.ID)
 115                 }
 116
 117                 message, err := l.Work(whatsappSource, batch.Text(i.clock().Location()), "")
 118                 if err != nil {
 119                         return fmt.Errorf("log a WhatsApp batch: %w", err)
 120                 }
 121
 122                 // A stop of the daemon during the flush is no failure: the
 123                 // batch stays in the state file and repeats once at the next
 124                 // start.
 125                 err = l.Flush(ctx)
 126                 if ctx.Err() != nil {
 127                         return nil
 128                 }
 129
 130                 if err != nil {
 131                         return fmt.Errorf("log a WhatsApp batch: %w", err)
 132                 }
 133
 134                 err = i.state.Remove(batch.Number)
 135                 if err != nil {
 136                         return err
 137                 }
 138
 139                 if batch.Turn() {
 140                         turns = append(turns, message)
 141                 }
 142         }
 143
 144         l.Wake(turns...)
 145
 146         return nil
 147 }
 148
 149 // waitForCatchup makes the batches wait for a catch-up, as for the batches of
 150 // the state file at the start, before the first connection.
 151 func (i *WhatsAppInbox) waitForCatchup() {
 152         i.mutex.Lock()
 153         defer i.mutex.Unlock()
 154
 155         i.catchingUp = true
 156 }
 157
 158 // startCatchup starts the catch-up of a new connection, which ends at
 159 // endCatchup or after catchupLimit.
 160 func (i *WhatsAppInbox) startCatchup() {
 161         i.mutex.Lock()
 162         defer i.mutex.Unlock()
 163
 164         i.catchingUp = true
 165         i.catchups++
 166         catchups := i.catchups
 167         if i.catchupTimer != nil {
 168                 i.catchupTimer.Stop()
 169         }
 170
 171         i.catchupTimer = time.AfterFunc(i.catchupLimit, func() {
 172                 i.caughtUpBy(catchups)
 173         })
 174 }
 175
 176 // receive puts the text of event into its batch and wakes Run. link gives the
 177 // device and the own IDs. It returns false when the state file cannot hold the
 178 // message, so that the library does not acknowledge it, and Run stops the
 179 // daemon. A message that is not text, or that the batches do not take, is
 180 // handled with nothing to do.
 181 func (i *WhatsAppInbox) receive(link WhatsAppLink, event *events.Message) bool {
 182         ctx, cancel := context.WithTimeout(context.Background(), whatsappLookupTime)
 183         defer cancel()
 184
 185         chat, message, found := whatsappMessage(ctx, link.device, link.own, event)
 186         if !found {
 187                 return true
 188         }
 189
 190         _, err := i.state.Add(chat, message, time.Now())
 191         if err != nil {
 192                 select {
 193                 case i.failures <- err:
 194                 default:
 195                 }
 196
 197                 return false
 198         }
 199
 200         select {
 201         case i.wake <- struct{}{}:
 202         default:
 203         }
 204
 205         return true
 206 }
 207
 208 // caughtUpBy ends the catch-up of the connection with the number catchups, when
 209 // no later connection started a new one.
 210 func (i *WhatsAppInbox) caughtUpBy(catchups int) {
 211         i.mutex.Lock()
 212         defer i.mutex.Unlock()
 213
 214         if i.catchups == catchups {
 215                 i.finishCatchup()
 216         }
 217 }
 218
 219 // finishCatchup ends a running catch-up and wakes Run, which logs all batches.
 220 // The caller holds the mutex.
 221 func (i *WhatsAppInbox) finishCatchup() {
 222         if !i.catchingUp {
 223                 return
 224         }
 225
 226         i.catchingUp = false
 227         i.caughtUp = true
 228         if i.catchupTimer != nil {
 229                 i.catchupTimer.Stop()
 230         }
 231
 232         select {
 233         case i.wake <- struct{}{}:
 234         default:
 235         }
 236 }
 237
 238 // endCatchup ends a running catch-up at OfflineSyncCompleted.
 239 func (i *WhatsAppInbox) endCatchup() {
 240         i.mutex.Lock()
 241         defer i.mutex.Unlock()
 242
 243         i.finishCatchup()
 244 }
 245
 246 // stopCatchup ends a running catch-up with no wake, after the session is
 247 // removed.
 248 func (i *WhatsAppInbox) stopCatchup() {
 249         i.mutex.Lock()
 250         defer i.mutex.Unlock()
 251
 252         i.catchingUp = false
 253 }
 254
 255 // close stops the timer of the catch-up.
 256 func (i *WhatsAppInbox) close() {
 257         i.mutex.Lock()
 258         defer i.mutex.Unlock()
 259
 260         if i.catchupTimer != nil {
 261                 i.catchupTimer.Stop()
 262         }
 263 }
 264
 265 // whatsappMessage returns the chat and the batch message of event, or false
 266 // when the batches do not take it. device is the linked device, and own holds
 267 // the own IDs. The batches take the text of a private chat or a group, from
 268 // another person or from the user. They do not take media, reactions, edits,
 269 // deletions, view-once and disappearing messages, old history, status
 270 // broadcasts, channels, and the chat of the user with itself.
 271 //
 272 // A mention of the user becomes @you, another mention @NAME when the name is
 273 // known, and the own number and LID become OWN, so that they never go to the
 274 // log.
 275 func whatsappMessage(ctx context.Context, device *store.Device, own []string, event *events.Message) (WhatsAppChat, WhatsAppMessage, bool) {
 276         info := event.Info
 277         chatJID := info.Chat.ToNonAD()
 278         switch {
 279         case device == nil:
 280         case event.IsEdit, info.Edit != "", event.IsEphemeral, event.IsViewOnce, event.SourceWebMsg != nil:
 281         case chatJID.Server != types.DefaultUserServer && chatJID.Server != types.HiddenUserServer && chatJID.Server != types.GroupServer:
 282         case slices.Contains(own, chatJID.User):
 283         default:
 284                 return whatsappText(ctx, device, own, event)
 285         }
 286
 287         return WhatsAppChat{}, WhatsAppMessage{}, false
 288 }
 289
 290 // whatsappText returns the chat and the batch message of event, a message that
 291 // the batches take when it has text.
 292 func whatsappText(ctx context.Context, device *store.Device, own []string, event *events.Message) (WhatsAppChat, WhatsAppMessage, bool) {
 293         info := event.Info
 294         text := event.Message.GetConversation()
 295         extended := event.Message.GetExtendedTextMessage()
 296         if text == "" {
 297                 text = extended.GetText()
 298         }
 299
 300         if strings.TrimSpace(text) == "" {
 301                 return WhatsAppChat{}, WhatsAppMessage{}, false
 302         }
 303
 304         contextInfo := extended.GetContextInfo()
 305         message := WhatsAppMessage{
 306                 ID:   info.ID,
 307                 Time: info.Timestamp,
 308                 Mine: info.IsFromMe,
 309         }
 310
 311         if !message.Mine {
 312                 message.Sender = whatsappContactName(ctx, device, info.Sender, info.PushName)
 313         }
 314
 315         for _, mentioned := range contextInfo.GetMentionedJID() {
 316                 jid, err := types.ParseJID(mentioned)
 317                 switch {
 318                 case err != nil:
 319                 case slices.Contains(own, jid.User):
 320                         message.Mention = true
 321                         text = strings.ReplaceAll(text, "@"+jid.User, "@you")
 322                 default:
 323                         text = strings.ReplaceAll(text, "@"+jid.User, "@"+whatsappContactName(ctx, device, jid, ""))
 324                 }
 325         }
 326
 327         participant, err := types.ParseJID(contextInfo.GetParticipant())
 328         message.Reply = err == nil && slices.Contains(own, participant.User)
 329
 330         message.Text = whatsappRedact(own, text)
 331         chatJID := info.Chat.ToNonAD()
 332         chat := WhatsAppChat{
 333                 ID:    chatJID.String(),
 334                 Group: chatJID.Server == types.GroupServer,
 335         }
 336
 337         if !chat.Group {
 338                 push := info.PushName
 339                 if message.Mine {
 340                         push = ""
 341                 }
 342
 343                 chat.Name = whatsappContactName(ctx, device, chatJID, push)
 344         }
 345
 346         return chat, message, true
 347 }