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
  93         catchingUp := i.catchingUp
  94         caughtUp := i.caughtUp
  95         i.caughtUp = false
  96         i.mutex.Unlock()
  97
  98         var err error
  99
 100         switch {
 101         case all || caughtUp:
 102                 err = i.state.CloseAll()
 103         case catchingUp:
 104                 return nil
 105         default:
 106                 _, err = i.state.CloseDue(time.Now())
 107         }
 108
 109         if err != nil {
 110                 return err
 111         }
 112
 113         var turns []Message
 114
 115         for _, batch := range i.state.Closed() {
 116                 if batch.Chat.Group && batch.Chat.Name == "" {
 117                         batch.Chat.Name = i.groups.name(ctx, link, batch.Chat.ID)
 118                 }
 119
 120                 message, err := l.Work(whatsappSource, batch.Text(i.clock().Location()), "")
 121                 if err != nil {
 122                         return fmt.Errorf("log a WhatsApp batch: %w", err)
 123                 }
 124
 125                 // A stop of the daemon during the flush is no failure: the
 126                 // batch stays in the state file and repeats once at the next
 127                 // start.
 128                 err = l.Flush(ctx)
 129                 if ctx.Err() != nil {
 130                         return nil
 131                 }
 132
 133                 if err != nil {
 134                         return fmt.Errorf("log a WhatsApp batch: %w", err)
 135                 }
 136
 137                 err = i.state.Remove(batch.Number)
 138                 if err != nil {
 139                         return err
 140                 }
 141
 142                 if batch.Turn() {
 143                         turns = append(turns, message)
 144                 }
 145         }
 146
 147         l.Wake(turns...)
 148
 149         return nil
 150 }
 151
 152 // waitForCatchup makes the batches wait for a catch-up, as for the batches of
 153 // the state file at the start, before the first connection.
 154 func (i *WhatsAppInbox) waitForCatchup() {
 155         i.mutex.Lock()
 156         defer i.mutex.Unlock()
 157
 158         i.catchingUp = true
 159 }
 160
 161 // startCatchup starts the catch-up of a new connection, which ends at
 162 // endCatchup or after catchupLimit.
 163 func (i *WhatsAppInbox) startCatchup() {
 164         i.mutex.Lock()
 165         defer i.mutex.Unlock()
 166
 167         i.catchingUp = true
 168         i.catchups++
 169         catchups := i.catchups
 170         if i.catchupTimer != nil {
 171                 i.catchupTimer.Stop()
 172         }
 173
 174         i.catchupTimer = time.AfterFunc(i.catchupLimit, func() {
 175                 i.caughtUpBy(catchups)
 176         })
 177 }
 178
 179 // receive puts the text of event into its batch and wakes Run. link gives the
 180 // device and the own IDs. It returns false when the state file cannot hold the
 181 // message, so that the library does not acknowledge it, and Run stops the
 182 // daemon. A message that is not text, or that the batches do not take, is
 183 // handled with nothing to do.
 184 func (i *WhatsAppInbox) receive(link WhatsAppLink, event *events.Message) bool {
 185         ctx, cancel := context.WithTimeout(context.Background(), whatsappLookupTime)
 186         defer cancel()
 187
 188         chat, message, found := whatsappMessage(ctx, link.device, link.own, event)
 189         if !found {
 190                 return true
 191         }
 192
 193         _, err := i.state.Add(chat, message, time.Now())
 194         if err != nil {
 195                 select {
 196                 case i.failures <- err:
 197                 default:
 198                 }
 199
 200                 return false
 201         }
 202
 203         select {
 204         case i.wake <- struct{}{}:
 205         default:
 206         }
 207
 208         return true
 209 }
 210
 211 // caughtUpBy ends the catch-up of the connection with the number catchups, when
 212 // no later connection started a new one.
 213 func (i *WhatsAppInbox) caughtUpBy(catchups int) {
 214         i.mutex.Lock()
 215         defer i.mutex.Unlock()
 216
 217         if i.catchups == catchups {
 218                 i.finishCatchup()
 219         }
 220 }
 221
 222 // finishCatchup ends a running catch-up and wakes Run, which logs all batches.
 223 // The caller holds the mutex.
 224 func (i *WhatsAppInbox) finishCatchup() {
 225         if !i.catchingUp {
 226                 return
 227         }
 228
 229         i.catchingUp = false
 230         i.caughtUp = true
 231         if i.catchupTimer != nil {
 232                 i.catchupTimer.Stop()
 233         }
 234
 235         select {
 236         case i.wake <- struct{}{}:
 237         default:
 238         }
 239 }
 240
 241 // endCatchup ends a running catch-up at OfflineSyncCompleted.
 242 func (i *WhatsAppInbox) endCatchup() {
 243         i.mutex.Lock()
 244         defer i.mutex.Unlock()
 245
 246         i.finishCatchup()
 247 }
 248
 249 // stopCatchup ends a running catch-up with no wake, after the session is
 250 // removed.
 251 func (i *WhatsAppInbox) stopCatchup() {
 252         i.mutex.Lock()
 253         defer i.mutex.Unlock()
 254
 255         i.catchingUp = false
 256 }
 257
 258 // close stops the timer of the catch-up.
 259 func (i *WhatsAppInbox) close() {
 260         i.mutex.Lock()
 261         defer i.mutex.Unlock()
 262
 263         if i.catchupTimer != nil {
 264                 i.catchupTimer.Stop()
 265         }
 266 }
 267
 268 // whatsappMessage returns the chat and the batch message of event, or false
 269 // when the batches do not take it. device is the linked device, and own holds
 270 // the own IDs. The batches take the text of a private chat or a group, from
 271 // another person or from the user. They do not take media, reactions, edits,
 272 // deletions, view-once and disappearing messages, old history, status
 273 // broadcasts, channels, and the chat of the user with itself.
 274 //
 275 // A mention of the user becomes @you, another mention @NAME when the name is
 276 // known, and the own number and LID become OWN, so that they never go to the
 277 // log.
 278 func whatsappMessage(ctx context.Context, device *store.Device, own []string, event *events.Message) (WhatsAppChat, WhatsAppMessage, bool) {
 279         info := event.Info
 280         chatJID := info.Chat.ToNonAD()
 281         switch {
 282         case device == nil:
 283         case event.IsEdit, info.Edit != "", event.IsEphemeral, event.IsViewOnce, event.SourceWebMsg != nil:
 284         case chatJID.Server != types.DefaultUserServer && chatJID.Server != types.HiddenUserServer && chatJID.Server != types.GroupServer:
 285         case slices.Contains(own, chatJID.User):
 286         default:
 287                 return whatsappText(ctx, device, own, event)
 288         }
 289
 290         return WhatsAppChat{}, WhatsAppMessage{}, false
 291 }
 292
 293 // whatsappText returns the chat and the batch message of event, a message that
 294 // the batches take when it has text.
 295 func whatsappText(ctx context.Context, device *store.Device, own []string, event *events.Message) (WhatsAppChat, WhatsAppMessage, bool) {
 296         info := event.Info
 297         text := event.Message.GetConversation()
 298         extended := event.Message.GetExtendedTextMessage()
 299         if text == "" {
 300                 text = extended.GetText()
 301         }
 302
 303         if strings.TrimSpace(text) == "" {
 304                 return WhatsAppChat{}, WhatsAppMessage{}, false
 305         }
 306
 307         contextInfo := extended.GetContextInfo()
 308         message := WhatsAppMessage{
 309                 ID:   info.ID,
 310                 Time: info.Timestamp,
 311                 Mine: info.IsFromMe,
 312         }
 313
 314         if !message.Mine {
 315                 message.Sender = whatsappContactName(ctx, device, info.Sender, info.PushName)
 316         }
 317
 318         for _, mentioned := range contextInfo.GetMentionedJID() {
 319                 jid, err := types.ParseJID(mentioned)
 320                 switch {
 321                 case err != nil:
 322                 case slices.Contains(own, jid.User):
 323                         message.Mention = true
 324                         text = strings.ReplaceAll(text, "@"+jid.User, "@you")
 325                 default:
 326                         text = strings.ReplaceAll(text, "@"+jid.User, "@"+whatsappContactName(ctx, device, jid, ""))
 327                 }
 328         }
 329
 330         participant, err := types.ParseJID(contextInfo.GetParticipant())
 331         message.Reply = err == nil && slices.Contains(own, participant.User)
 332
 333         message.Text = whatsappRedact(own, text)
 334         chatJID := info.Chat.ToNonAD()
 335         chat := WhatsAppChat{
 336                 ID:    chatJID.String(),
 337                 Group: chatJID.Server == types.GroupServer,
 338         }
 339
 340         if !chat.Group {
 341                 push := info.PushName
 342                 if message.Mine {
 343                         push = ""
 344                 }
 345
 346                 chat.Name = whatsappContactName(ctx, device, chatJID, push)
 347         }
 348
 349         return chat, message, true
 350 }