X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/0b97bb2cc4bc0b8f8fe27cc7aa3f9d1d89ed28c8..a6a82d96b4906e9e1f09f45f69f48124f11ae1bb:/whatsapp.go?ds=sidebyside

diff --git a/whatsapp.go b/whatsapp.go
index 44b9dfa..c5af946 100644
--- a/whatsapp.go
+++ b/whatsapp.go
@@ -8,7 +8,6 @@ import (
 	"errors"
 	"fmt"
 	"log"
-	"maps"
 	"net/url"
 	"os"
 	"path/filepath"
@@ -77,28 +76,10 @@ const (
 	// 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"
-
-	// whatsappCatchupLimit is the longest wait for the end of the catch-up
-	// after a connection.
-	whatsappCatchupLimit = time.Minute
-
-	// whatsappGroupsAge is how long whatsapp_find uses the groups that it
-	// read from WhatsApp.
-	whatsappGroupsAge = time.Hour
-
 	// whatsappFindMatches is the most matches of whatsapp_find.
 	whatsappFindMatches = 20
 
@@ -145,17 +126,14 @@ type WhatsAppClient interface {
 // 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),
-		groupNames:   make(map[string]string),
+		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
@@ -164,6 +142,8 @@ func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logge
 		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
@@ -184,7 +164,7 @@ func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logge
 
 	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)
 	}
 
@@ -204,20 +184,9 @@ 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 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
@@ -227,17 +196,16 @@ func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logge
 // 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
+	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
@@ -253,13 +221,6 @@ type WhatsApp struct {
 	noticed        bool
 	timer          *time.Timer
 	closed         bool
-	groupNames     map[string]string
-	groupsAt       time.Time
-	groupsRead     time.Time
-	catchingUp     bool
-	caughtUp       bool
-	catchups       int
-	catchupTimer   *time.Timer
 }
 
 // Run connects a linked device, logs the closed batches in l, runs the pair and
@@ -288,24 +249,18 @@ func (w *WhatsApp) Run(ctx context.Context, l *TurnLoop, notify func(Event)) err
 	}()
 
 	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 {
@@ -443,6 +398,21 @@ func (w *WhatsApp) isLinked() bool {
 	return w.linked
 }
 
+// link returns a copy of the link of the device: whether it is linked, the
+// device, the client, whether it is connected, and the own IDs.
+func (w *WhatsApp) link() WhatsAppLink {
+	w.mutex.Lock()
+	defer w.mutex.Unlock()
+
+	return WhatsAppLink{
+		linked:    w.linked,
+		device:    w.device,
+		client:    w.client,
+		connected: w.connected,
+		own:       slices.Clone(w.own),
+	}
+}
+
 // connect makes the client of the linked device and connects it.
 func (w *WhatsApp) connect(ctx context.Context) {
 	w.mutex.Lock()
@@ -464,14 +434,12 @@ func (w *WhatsApp) close() {
 		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()
 	}
@@ -484,7 +452,7 @@ func (w *WhatsApp) loggedOut(ctx context.Context, l *TurnLoop) error {
 		return nil
 	}
 
-	err := w.deliver(ctx, l, true)
+	err := w.inbox.deliver(ctx, l, w.link(), true)
 	if err != nil {
 		return err
 	}
@@ -520,7 +488,7 @@ func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text str
 		return nil
 	}
 
-	err := w.deliver(ctx, l, true)
+	err := w.inbox.deliver(ctx, l, w.link(), true)
 	if err != nil {
 		return err
 	}
@@ -554,130 +522,6 @@ func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text str
 	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.groupName(ctx, 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
-}
-
-// 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]
-	fresh := time.Since(w.groupsAt) < whatsappGroupsRefresh
-	w.mutex.Unlock()
-
-	if !found && !fresh {
-		w.readGroups(ctx)
-
-		w.mutex.Lock()
-		name, found = w.groupNames[id]
-		w.mutex.Unlock()
-	}
-
-	if !found || name == "" {
-		return whatsappUnknownGroup
-	}
-
-	return name
-}
-
-// readGroups reads the names of the groups of the account from WhatsApp, and
-// tells if it could. They stay in memory only.
-func (w *WhatsApp) readGroups(ctx context.Context) bool {
-	w.mutex.Lock()
-	client := w.client
-	w.mutex.Unlock()
-
-	if client == nil {
-		return false
-	}
-
-	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()
-	if err != nil {
-		return false
-	}
-
-	w.groupsRead = w.groupsAt
-	for _, group := range groups {
-		w.groupNames[group.JID.String()] = group.Name
-	}
-
-	return true
-}
-
 // 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
@@ -690,9 +534,10 @@ func (w *WhatsApp) Find(ctx context.Context, name string) (string, error) {
 	linked := w.linked
 	device := w.device
 	own := slices.Clone(w.own)
-	stale := time.Since(w.groupsRead) >= whatsappGroupsAge
 	w.mutex.Unlock()
 
+	stale := w.groups.stale()
+
 	if !linked {
 		return whatsappNotLinked, nil
 	}
@@ -707,10 +552,8 @@ func (w *WhatsApp) Find(ctx context.Context, name string) (string, error) {
 		return "", fmt.Errorf("read the contacts: %s", w.redact(err.Error()))
 	}
 
-	read := !stale || w.readGroups(ctx)
-	w.mutex.Lock()
-	groups := maps.Clone(w.groupNames)
-	w.mutex.Unlock()
+	read := !stale || w.groups.read(ctx, w.link())
+	groups := w.groups.all()
 
 	if len(contacts) == 0 && len(groups) == 0 {
 		if !read {
@@ -762,7 +605,6 @@ func (w *WhatsApp) removeSession() error {
 	w.linked = false
 	w.own = nil
 	w.connected = false
-	w.catchingUp = false
 	if w.timer != nil {
 		w.timer.Stop()
 	}
@@ -773,6 +615,8 @@ func (w *WhatsApp) removeSession() error {
 	w.device = nil
 	w.mutex.Unlock()
 
+	w.inbox.stopCatchup()
+
 	if client != nil {
 		client.Disconnect()
 	}
@@ -933,42 +777,40 @@ func (w *WhatsApp) linkedBy(client WhatsAppClient) {
 }
 
 // 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 {
-	message, isMessage := event.(*events.Message)
-	if isMessage {
-		return w.receive(message)
-	}
-
-	var notice string
-	w.mutex.Lock()
 	switch event := event.(type) {
+	case *events.Message:
+		return w.inbox.receive(w.link(), event)
 	case *events.GroupInfo:
 		if event.Name != nil {
-			w.groupNames[event.JID.String()] = event.Name.Name
+			w.groups.set(event.JID.String(), event.Name.Name)
 		}
+
+		return true
 	case *events.JoinedGroup:
-		w.groupNames[event.JID.String()] = event.Name
+		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
+	w.mutex.Lock()
+	switch event := event.(type) {
 	case *events.PairSuccess:
 		w.own = whatsappOwnIDs(event.ID.User, event.LID.User)
 	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()
 		}
@@ -988,8 +830,6 @@ func (w *WhatsApp) handle(event any) bool {
 		}
 
 		w.connected = false
-	case *events.OfflineSyncCompleted:
-		w.endCatchup()
 	case *events.LoggedOut:
 		select {
 		case w.logouts <- struct{}{}:
@@ -1011,72 +851,6 @@ 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()
-
-	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) {
@@ -1183,7 +957,7 @@ func (w *WhatsApp) Send(ctx context.Context, id, chat, text string) (string, err
 		record.State = whatsappFailed
 		record.Answer = "The send failed: " + w.redact(err.Error()) + "."
 	} else {
-		name := w.groupName(ctx, jid.String())
+		name := w.groups.name(ctx, w.link(), jid.String())
 		if jid.Server != types.GroupServer {
 			name = whatsappContactName(ctx, device, jid, "")
 		}
@@ -1206,6 +980,16 @@ type WhatsAppMatch struct {
 	kind string
 }
 
+// WhatsAppLink is a copy of the link of the device at one time, for the parts
+// of WhatsApp that do not own it: the groups, the batches, find, and send.
+type WhatsAppLink struct {
+	linked    bool
+	device    *store.Device
+	client    WhatsAppClient
+	connected bool
+	own       []string
+}
+
 // WhatsAppRequest is a pair or unlink command for Run, with the function that
 // gives its answers to the client that asked. A pairing has the phone number.
 type WhatsAppRequest struct {
@@ -1229,90 +1013,6 @@ func whatsappNewClient(device *store.Device, handle func(event any) bool) WhatsA
 	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