mautrix-whatsapp/user.go

// mautrix-whatsapp - A Matrix-WhatsApp puppeting bridge.
// Copyright (C) 2020 Tulir Asokan
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.

package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/skip2/go-qrcode"
	log "maunium.net/go/maulogger/v2"

	"github.com/Rhymen/go-whatsapp"
	waBinary "github.com/Rhymen/go-whatsapp/binary"
	waProto "github.com/Rhymen/go-whatsapp/binary/proto"

	"maunium.net/go/mautrix"
	"maunium.net/go/mautrix/event"
	"maunium.net/go/mautrix/format"
	"maunium.net/go/mautrix/id"

	"maunium.net/go/mautrix-whatsapp/database"
)

type User struct {
	*database.User
	Conn *whatsapp.Conn

	bridge *Bridge
	log    log.Logger

	Admin               bool
	Whitelisted         bool
	RelaybotWhitelisted bool

	IsRelaybot bool

	ConnectionErrors int
	CommunityID      string

	cleanDisconnection  bool
	batteryWarningsSent int
	lastReconnection    int64
	pushName            string

	chatListReceived chan struct{}
	syncPortalsDone  chan struct{}

	messageInput  chan PortalMessage
	messageOutput chan PortalMessage

	syncStart chan struct{}
	syncWait  sync.WaitGroup

	mgmtCreateLock  sync.Mutex
	connLock        sync.Mutex
	cancelReconnect func()

	prevBridgeStatus *AsmuxPong
}

func (bridge *Bridge) GetUserByMXID(userID id.UserID) *User {
	_, isPuppet := bridge.ParsePuppetMXID(userID)
	if isPuppet || userID == bridge.Bot.UserID {
		return nil
	}
	bridge.usersLock.Lock()
	defer bridge.usersLock.Unlock()
	user, ok := bridge.usersByMXID[userID]
	if !ok {
		return bridge.loadDBUser(bridge.DB.User.GetByMXID(userID), &userID)
	}
	return user
}

func (bridge *Bridge) GetUserByJID(userID whatsapp.JID) *User {
	bridge.usersLock.Lock()
	defer bridge.usersLock.Unlock()
	user, ok := bridge.usersByJID[userID]
	if !ok {
		return bridge.loadDBUser(bridge.DB.User.GetByJID(userID), nil)
	}
	return user
}

func (user *User) addToJIDMap() {
	user.bridge.usersLock.Lock()
	user.bridge.usersByJID[user.JID] = user
	user.bridge.usersLock.Unlock()
}

func (user *User) removeFromJIDMap() {
	user.bridge.usersLock.Lock()
	jidUser, ok := user.bridge.usersByJID[user.JID]
	if ok && user == jidUser {
		delete(user.bridge.usersByJID, user.JID)
	}
	user.bridge.usersLock.Unlock()
	user.bridge.Metrics.TrackLoginState(user.JID, false)
	user.sendBridgeStatus(AsmuxPong{Error: AsmuxWANotLoggedIn})
}

func (bridge *Bridge) GetAllUsers() []*User {
	bridge.usersLock.Lock()
	defer bridge.usersLock.Unlock()
	dbUsers := bridge.DB.User.GetAll()
	output := make([]*User, len(dbUsers))
	for index, dbUser := range dbUsers {
		user, ok := bridge.usersByMXID[dbUser.MXID]
		if !ok {
			user = bridge.loadDBUser(dbUser, nil)
		}
		output[index] = user
	}
	return output
}

func (bridge *Bridge) loadDBUser(dbUser *database.User, mxid *id.UserID) *User {
	if dbUser == nil {
		if mxid == nil {
			return nil
		}
		dbUser = bridge.DB.User.New()
		dbUser.MXID = *mxid
		dbUser.Insert()
	}
	user := bridge.NewUser(dbUser)
	bridge.usersByMXID[user.MXID] = user
	if len(user.JID) > 0 {
		bridge.usersByJID[user.JID] = user
	}
	if len(user.ManagementRoom) > 0 {
		bridge.managementRooms[user.ManagementRoom] = user
	}
	return user
}

func (user *User) GetPortals() []*Portal {
	keys := user.User.GetPortalKeys()
	portals := make([]*Portal, len(keys))

	user.bridge.portalsLock.Lock()
	for i, key := range keys {
		portal, ok := user.bridge.portalsByJID[key]
		if !ok {
			portal = user.bridge.loadDBPortal(user.bridge.DB.Portal.GetByJID(key), &key)
		}
		portals[i] = portal
	}
	user.bridge.portalsLock.Unlock()
	return portals
}

func (bridge *Bridge) NewUser(dbUser *database.User) *User {
	user := &User{
		User:   dbUser,
		bridge: bridge,
		log:    bridge.Log.Sub("User").Sub(string(dbUser.MXID)),

		IsRelaybot: false,

		chatListReceived: make(chan struct{}, 1),
		syncPortalsDone:  make(chan struct{}, 1),
		syncStart:        make(chan struct{}, 1),
		messageInput:     make(chan PortalMessage),
		messageOutput:    make(chan PortalMessage, bridge.Config.Bridge.UserMessageBuffer),
	}
	user.RelaybotWhitelisted = user.bridge.Config.Bridge.Permissions.IsRelaybotWhitelisted(user.MXID)
	user.Whitelisted = user.bridge.Config.Bridge.Permissions.IsWhitelisted(user.MXID)
	user.Admin = user.bridge.Config.Bridge.Permissions.IsAdmin(user.MXID)
	go user.handleMessageLoop()
	go user.runMessageRingBuffer()
	return user
}

func (user *User) GetManagementRoom() id.RoomID {
	if len(user.ManagementRoom) == 0 {
		user.mgmtCreateLock.Lock()
		defer user.mgmtCreateLock.Unlock()
		if len(user.ManagementRoom) > 0 {
			return user.ManagementRoom
		}
		resp, err := user.bridge.Bot.CreateRoom(&mautrix.ReqCreateRoom{
			Topic:    "WhatsApp bridge notices",
			IsDirect: true,
		})
		if err != nil {
			user.log.Errorln("Failed to auto-create management room:", err)
		} else {
			user.SetManagementRoom(resp.RoomID)
		}
	}
	return user.ManagementRoom
}

func (user *User) SetManagementRoom(roomID id.RoomID) {
	existingUser, ok := user.bridge.managementRooms[roomID]
	if ok {
		existingUser.ManagementRoom = ""
		existingUser.Update()
	}

	user.ManagementRoom = roomID
	user.bridge.managementRooms[user.ManagementRoom] = user
	user.Update()
}

func (user *User) SetSession(session *whatsapp.Session) {
	if session == nil {
		user.Session = nil
		user.LastConnection = 0
	} else if len(session.Wid) > 0 {
		user.Session = session
	} else {
		return
	}
	user.Update()
}

func (user *User) Connect(evenIfNoSession bool) bool {
	user.connLock.Lock()
	if user.Conn != nil {
		user.connLock.Unlock()
		if user.Conn.IsConnected() {
			return true
		} else {
			return user.RestoreSession()
		}
	} else if !evenIfNoSession && user.Session == nil {
		user.connLock.Unlock()
		return false
	}
	user.log.Debugln("Connecting to WhatsApp")
	if user.Session != nil {
		user.sendBridgeStatus(AsmuxPong{Error: AsmuxWAConnecting})
	}
	timeout := time.Duration(user.bridge.Config.Bridge.ConnectionTimeout)
	if timeout == 0 {
		timeout = 20
	}
	user.Conn = whatsapp.NewConn(&whatsapp.Options{
		Timeout:         timeout * time.Second,
		LongClientName:  user.bridge.Config.WhatsApp.OSName,
		ShortClientName: user.bridge.Config.WhatsApp.BrowserName,
		ClientVersion:   WAVersion,
		Log:             user.log.Sub("Conn"),
		Handler:         []whatsapp.Handler{user},
	})
	user.setupAdminTestHooks()
	user.connLock.Unlock()
	return user.RestoreSession()
}

func (user *User) DeleteConnection() {
	user.connLock.Lock()
	if user.Conn == nil {
		user.connLock.Unlock()
		return
	}
	err := user.Conn.Disconnect()
	if err != nil && err != whatsapp.ErrNotConnected {
		user.log.Warnln("Error disconnecting: %v", err)
	}
	user.Conn.RemoveHandlers()
	user.Conn = nil
	user.bridge.Metrics.TrackConnectionState(user.JID, false)
	user.sendBridgeStatus(AsmuxPong{Error: AsmuxWANotConnected})
	user.connLock.Unlock()
}

func (user *User) RestoreSession() bool {
	if user.Session != nil {
		user.Conn.SetSession(*user.Session)
		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
		defer cancel()
		err := user.Conn.Restore(true, ctx)
		if err == whatsapp.ErrAlreadyLoggedIn {
			return true
		} else if err != nil {
			user.log.Errorln("Failed to restore session:", err)
			if errors.Is(err, whatsapp.ErrUnpaired) {
				user.sendMarkdownBridgeAlert("\u26a0 Failed to connect to WhatsApp: unpaired from phone. " +
					"To re-pair your phone, log in again.")
				user.removeFromJIDMap()
				//user.JID = ""
				user.SetSession(nil)
				user.DeleteConnection()
				return false
			} else {
				user.sendBridgeStatus(AsmuxPong{Error: AsmuxWANotConnected})
				user.sendMarkdownBridgeAlert("\u26a0 Failed to connect to WhatsApp. Make sure WhatsApp " +
					"on your phone is reachable and use `reconnect` to try connecting again.")
			}
			user.log.Debugln("Disconnecting due to failed session restore...")
			err = user.Conn.Disconnect()
			if err != nil {
				user.log.Errorln("Failed to disconnect after failed session restore:", err)
			}
			return false
		}
		user.ConnectionErrors = 0
		user.log.Debugln("Session restored successfully")
		user.PostLogin()
	}
	return true
}

func (user *User) HasSession() bool {
	return user.Session != nil
}

func (user *User) IsConnected() bool {
	return user.Conn != nil && user.Conn.IsConnected() && user.Conn.IsLoggedIn()
}

func (user *User) IsLoginInProgress() bool {
	return user.Conn != nil && user.Conn.IsLoginInProgress()
}

func (user *User) loginQrChannel(ce *CommandEvent, qrChan <-chan string, eventIDChan chan<- id.EventID) {
	var qrEventID id.EventID
	for code := range qrChan {
		if code == "stop" {
			return
		}
		qrCode, err := qrcode.Encode(code, qrcode.Low, 256)
		if err != nil {
			user.log.Errorln("Failed to encode QR code:", err)
			ce.Reply("Failed to encode QR code: %v", err)
			return
		}

		bot := user.bridge.AS.BotClient()

		resp, err := bot.UploadBytes(qrCode, "image/png")
		if err != nil {
			user.log.Errorln("Failed to upload QR code:", err)
			ce.Reply("Failed to upload QR code: %v", err)
			return
		}

		if qrEventID == "" {
			sendResp, err := bot.SendImage(ce.RoomID, code, resp.ContentURI)
			if err != nil {
				user.log.Errorln("Failed to send QR code to user:", err)
				return
			}
			qrEventID = sendResp.EventID
			eventIDChan <- qrEventID
		} else {
			_, err = bot.SendMessageEvent(ce.RoomID, event.EventMessage, &event.MessageEventContent{
				MsgType: event.MsgImage,
				Body:    code,
				URL:     resp.ContentURI.CUString(),
				NewContent: &event.MessageEventContent{
					MsgType: event.MsgImage,
					Body:    code,
					URL:     resp.ContentURI.CUString(),
				},
				RelatesTo: &event.RelatesTo{
					Type:    event.RelReplace,
					EventID: qrEventID,
				},
			})
			if err != nil {
				user.log.Errorln("Failed to send edited QR code to user:", err)
			}
		}
	}
}

func (user *User) Login(ce *CommandEvent) {
	qrChan := make(chan string, 3)
	eventIDChan := make(chan id.EventID, 1)
	go user.loginQrChannel(ce, qrChan, eventIDChan)
	session, jid, err := user.Conn.Login(qrChan, nil, user.bridge.Config.Bridge.LoginQRRegenCount)
	qrChan <- "stop"
	if err != nil {
		var eventID id.EventID
		select {
		case eventID = <-eventIDChan:
		default:
		}
		reply := event.MessageEventContent{
			MsgType: event.MsgText,
		}
		if err == whatsapp.ErrAlreadyLoggedIn {
			reply.Body = "You're already logged in"
		} else if err == whatsapp.ErrLoginInProgress {
			reply.Body = "You have a login in progress already."
		} else if err == whatsapp.ErrLoginTimedOut {
			reply.Body = "QR code scan timed out. Please try again."
		} else {
			user.log.Warnln("Failed to log in:", err)
			reply.Body = fmt.Sprintf("Unknown error while logging in: %v", err)
		}
		msg := reply
		if eventID != "" {
			msg.NewContent = &reply
			msg.RelatesTo = &event.RelatesTo{
				Type:    event.RelReplace,
				EventID: eventID,
			}
		}
		_, _ = ce.Bot.SendMessageEvent(ce.RoomID, event.EventMessage, &msg)
		return
	}
	// TODO there's a bit of duplication between this and the provisioning API login method
	//      Also between the two logout methods (commands.go and provisioning.go)
	user.log.Debugln("Successful login as", jid, "via command")
	user.ConnectionErrors = 0
	user.JID = strings.Replace(jid, whatsapp.OldUserSuffix, whatsapp.NewUserSuffix, 1)
	user.addToJIDMap()
	user.SetSession(&session)
	ce.Reply("Successfully logged in, synchronizing chats...")
	user.PostLogin()
}

type Chat struct {
	Portal          *Portal
	LastMessageTime uint64
	MarkAsRead      bool
	Contact         whatsapp.Contact
}

type ChatList []Chat

func (cl ChatList) Len() int {
	return len(cl)
}

func (cl ChatList) Less(i, j int) bool {
	return cl[i].LastMessageTime > cl[j].LastMessageTime
}

func (cl ChatList) Swap(i, j int) {
	cl[i], cl[j] = cl[j], cl[i]
}

func (user *User) PostLogin() {
	user.sendBridgeStatus(AsmuxPong{OK: true})
	user.bridge.Metrics.TrackConnectionState(user.JID, true)
	user.bridge.Metrics.TrackLoginState(user.JID, true)
	user.bridge.Metrics.TrackBufferLength(user.MXID, 0)
	user.log.Debugln("Locking processing of incoming messages and starting post-login sync")
	// TODO can an old sync still be ongoing when PostLogin is called again?
	// TODO 2: can the new sync happen before this happens?
	user.chatListReceived = make(chan struct{}, 1)
	user.syncPortalsDone = make(chan struct{}, 1)
	user.syncWait.Add(1)
	user.syncStart <- struct{}{}
	go user.intPostLogin()
}

func (user *User) tryAutomaticDoublePuppeting() {
	if !user.bridge.Config.CanDoublePuppet(user.MXID) {
		return
	}
	user.log.Debugln("Checking if double puppeting needs to be enabled")
	puppet := user.bridge.GetPuppetByJID(user.JID)
	if len(puppet.CustomMXID) > 0 {
		user.log.Debugln("User already has double-puppeting enabled")
		// Custom puppet already enabled
		return
	}
	accessToken, err := puppet.loginWithSharedSecret(user.MXID)
	if err != nil {
		user.log.Warnln("Failed to login with shared secret:", err)
		return
	}
	err = puppet.SwitchCustomMXID(accessToken, user.MXID)
	if err != nil {
		puppet.log.Warnln("Failed to switch to auto-logined custom puppet:", err)
		return
	}
	user.log.Infoln("Successfully automatically enabled custom puppet")
}

func (user *User) sendBridgeNotice(formatString string, args ...interface{}) {
	notice := fmt.Sprintf(formatString, args...)
	_, err := user.bridge.Bot.SendNotice(user.GetManagementRoom(), notice)
	if err != nil {
		user.log.Warnf("Failed to send bridge notice \"%s\": %v", notice, err)
	}
}

func (user *User) sendMarkdownBridgeAlert(formatString string, args ...interface{}) {
	notice := fmt.Sprintf(formatString, args...)
	content := format.RenderMarkdown(notice, true, false)
	_, err := user.bridge.Bot.SendMessageEvent(user.GetManagementRoom(), event.EventMessage, content)
	if err != nil {
		user.log.Warnf("Failed to send bridge alert \"%s\": %v", notice, err)
	}
}

func (user *User) postConnPing() bool {
	user.log.Debugln("Making post-connection ping")
	var err error
	for i := 0; ; i++ {
		err = user.Conn.AdminTest()
		if err == nil {
			user.log.Debugln("Post-connection ping OK")
			return true
		} else if errors.Is(err, whatsapp.ErrConnectionTimeout) && i < 5 {
			user.log.Warnfln("Post-connection ping timed out, sending new one")
		} else {
			break
		}
	}
	user.log.Errorfln("Post-connection ping failed: %v. Disconnecting and then reconnecting after a second", err)
	disconnectErr := user.Conn.Disconnect()
	if disconnectErr != nil {
		user.log.Warnln("Error while disconnecting after failed post-connection ping:", disconnectErr)
	}
	user.sendBridgeStatus(AsmuxPong{Error: AsmuxWANotConnected})
	user.bridge.Metrics.TrackDisconnection(user.MXID)
	go func() {
		time.Sleep(1 * time.Second)
		user.tryReconnect(fmt.Sprintf("Post-connection ping failed: %v", err))
	}()
	return false
}

func (user *User) intPostLogin() {
	defer user.syncWait.Done()
	user.lastReconnection = time.Now().Unix()
	user.createCommunity()
	user.tryAutomaticDoublePuppeting()

	user.log.Debugln("Waiting for chat list receive confirmation")
	select {
	case <-user.chatListReceived:
		user.log.Debugln("Chat list receive confirmation received in PostLogin")
	case <-time.After(time.Duration(user.bridge.Config.Bridge.ChatListWait) * time.Second):
		user.log.Warnln("Timed out waiting for chat list to arrive!")
		user.postConnPing()
		return
	}

	if !user.postConnPing() {
		user.log.Debugln("Post-connection ping failed, unlocking processing of incoming messages.")
		return
	}

	user.log.Debugln("Waiting for portal sync complete confirmation")
	select {
	case <-user.syncPortalsDone:
		user.log.Debugln("Post-connection portal sync complete, unlocking processing of incoming messages.")
	// TODO this is too short, maybe a per-portal duration?
	case <-time.After(time.Duration(user.bridge.Config.Bridge.PortalSyncWait) * time.Second):
		user.log.Warnln("Timed out waiting for portal sync to complete! Unlocking processing of incoming messages.")
	}
}

type NormalMessage interface {
	GetInfo() whatsapp.MessageInfo
}

func (user *User) HandleEvent(event interface{}) {
	switch v := event.(type) {
	case NormalMessage:
		info := v.GetInfo()
		user.messageInput <- PortalMessage{info.RemoteJid, user, v, info.Timestamp}
	case whatsapp.MessageRevocation:
		user.messageInput <- PortalMessage{v.RemoteJid, user, v, 0}
	case whatsapp.StreamEvent:
		user.HandleStreamEvent(v)
	case []whatsapp.Chat:
		user.HandleChatList(v)
	case []whatsapp.Contact:
		user.HandleContactList(v)
	case error:
		user.HandleError(v)
	case whatsapp.Contact:
		go user.HandleNewContact(v)
	case whatsapp.BatteryMessage:
		user.HandleBatteryMessage(v)
	case whatsapp.CallInfo:
		user.HandleCallInfo(v)
	case whatsapp.PresenceEvent:
		go user.HandlePresence(v)
	case whatsapp.JSONMsgInfo:
		go user.HandleMsgInfo(v)
	case whatsapp.ReceivedMessage:
		user.HandleReceivedMessage(v)
	case whatsapp.ReadMessage:
		user.HandleReadMessage(v)
	case whatsapp.JSONCommand:
		user.HandleCommand(v)
	case whatsapp.ChatUpdate:
		user.HandleChatUpdate(v)
	case whatsapp.ConnInfo:
		user.HandleConnInfo(v)
	case json.RawMessage:
		user.HandleJSONMessage(v)
	case *waProto.WebMessageInfo:
		user.updateLastConnectionIfNecessary()
		// TODO trace log
		//user.log.Debugfln("WebMessageInfo: %+v", v)
	case *waBinary.Node:
		user.log.Debugfln("Unknown binary message: %+v", v)
	default:
		user.log.Debugfln("Unknown type of event in HandleEvent: %T", v)
	}
}

func (user *User) HandleStreamEvent(evt whatsapp.StreamEvent) {
	if evt.Type == whatsapp.StreamSleep {
		if user.lastReconnection+60 > time.Now().Unix() {
			user.lastReconnection = 0
			user.log.Infoln("Stream went to sleep soon after reconnection, making new post-connection ping in 20 seconds")
			go func() {
				time.Sleep(20 * time.Second)
				// TODO if this happens during the post-login sync, it can get stuck forever
				// TODO check if the above is still true
				user.postConnPing()
			}()
		}
	} else {
		user.log.Infofln("Stream event: %+v", evt)
	}
}

func (user *User) HandleChatList(chats []whatsapp.Chat) {
	user.log.Infoln("Chat list received")
	chatMap := make(map[string]whatsapp.Chat)
	for _, chat := range user.Conn.Store.Chats {
		chatMap[chat.JID] = chat
	}
	for _, chat := range chats {
		chatMap[chat.JID] = chat
	}
	select {
	case user.chatListReceived <- struct{}{}:
	default:
	}
	go user.syncPortals(chatMap, false)
}

func (user *User) syncPortals(chatMap map[string]whatsapp.Chat, createAll bool) {
	// TODO use contexts instead of checking if user.Conn is the same?
	connAtStart := user.Conn
	if chatMap == nil {
		chatMap = user.Conn.Store.Chats
	}
	user.log.Infoln("Reading chat list")
	chats := make(ChatList, 0, len(chatMap))
	existingKeys := user.GetInCommunityMap()
	portalKeys := make([]database.PortalKeyWithMeta, 0, len(chatMap))
	for _, chat := range chatMap {
		ts, err := strconv.ParseUint(chat.LastMessageTime, 10, 64)
		if err != nil {
			user.log.Warnfln("Non-integer last message time in %s: %s", chat.JID, chat.LastMessageTime)
			continue
		}
		portal := user.GetPortalByJID(chat.JID)

		chats = append(chats, Chat{
			Portal:          portal,
			Contact:         user.Conn.Store.Contacts[chat.JID],
			LastMessageTime: ts,
			MarkAsRead:      chat.Unread == "0",
		})
		var inCommunity, ok bool
		if inCommunity, ok = existingKeys[portal.Key]; !ok || !inCommunity {
			inCommunity = user.addPortalToCommunity(portal)
			if portal.IsPrivateChat() {
				puppet := user.bridge.GetPuppetByJID(portal.Key.JID)
				user.addPuppetToCommunity(puppet)
			}
		}
		portalKeys = append(portalKeys, database.PortalKeyWithMeta{PortalKey: portal.Key, InCommunity: inCommunity})
	}
	user.log.Infoln("Read chat list, updating user-portal mapping")
	err := user.SetPortalKeys(portalKeys)
	if err != nil {
		user.log.Warnln("Failed to update user-portal mapping:", err)
	}
	sort.Sort(chats)
	limit := user.bridge.Config.Bridge.InitialChatSync
	if limit < 0 {
		limit = len(chats)
	}
	if user.Conn != connAtStart {
		user.log.Debugln("Connection seems to have changed before sync, cancelling")
		return
	}
	now := uint64(time.Now().Unix())
	user.log.Infoln("Syncing portals")
	doublePuppet := user.bridge.GetPuppetByCustomMXID(user.MXID)
	for i, chat := range chats {
		if chat.LastMessageTime+user.bridge.Config.Bridge.SyncChatMaxAge < now {
			break
		}
		create := (chat.LastMessageTime >= user.LastConnection && user.LastConnection > 0) || i < limit
		if len(chat.Portal.MXID) > 0 || create || createAll {
			// Don't sync unless chat meta sync is enabled or portal doesn't exist
			if user.bridge.Config.Bridge.ChatMetaSync || len(chat.Portal.MXID) == 0 {
				chat.Portal.Sync(user, chat.Contact)
			}
			err = chat.Portal.BackfillHistory(user, chat.LastMessageTime)
			if err != nil {
				chat.Portal.log.Errorln("Error backfilling history:", err)
			}
			if chat.MarkAsRead && doublePuppet != nil && doublePuppet.CustomIntent() != nil {
				lastMessage := user.bridge.DB.Message.GetLastInChat(chat.Portal.Key)
				if lastMessage != nil {
					err = doublePuppet.CustomIntent().MarkRead(chat.Portal.MXID, lastMessage.MXID)
					if err != nil {
						user.log.Warnln("Failed to mark %s in %s as read after backfill: %v", lastMessage.MXID, chat.Portal.MXID, err)
					}
				}
			}
		}
	}
	if user.Conn != connAtStart {
		user.log.Debugln("Connection seems to have changed during sync, cancelling")
		return
	}
	user.UpdateDirectChats(nil)
	user.log.Infoln("Finished syncing portals")
	select {
	case user.syncPortalsDone <- struct{}{}:
	default:
	}
}

func (user *User) getDirectChats() map[id.UserID][]id.RoomID {
	res := make(map[id.UserID][]id.RoomID)
	privateChats := user.bridge.DB.Portal.FindPrivateChats(user.JID)
	for _, portal := range privateChats {
		if len(portal.MXID) > 0 {
			res[user.bridge.FormatPuppetMXID(portal.Key.JID)] = []id.RoomID{portal.MXID}
		}
	}
	return res
}

func (user *User) UpdateDirectChats(chats map[id.UserID][]id.RoomID) {
	if !user.bridge.Config.Bridge.SyncDirectChatList {
		return
	}
	puppet := user.bridge.GetPuppetByCustomMXID(user.MXID)
	if puppet == nil || puppet.CustomIntent() == nil {
		return
	}
	intent := puppet.CustomIntent()
	method := http.MethodPatch
	if chats == nil {
		chats = user.getDirectChats()
		method = http.MethodPut
	}
	user.log.Debugln("Updating m.direct list on homeserver")
	var err error
	if user.bridge.Config.Homeserver.Asmux {
		urlPath := intent.BuildBaseURL("_matrix", "client", "unstable", "com.beeper.asmux", "dms")
		_, err = intent.MakeFullRequest(mautrix.FullRequest{
			Method:      method,
			URL:         urlPath,
			Headers:     http.Header{"X-Asmux-Auth": {user.bridge.AS.Registration.AppToken}},
			RequestJSON: chats,
		})
	} else {
		existingChats := make(map[id.UserID][]id.RoomID)
		err = intent.GetAccountData(event.AccountDataDirectChats.Type, &existingChats)
		if err != nil {
			user.log.Warnln("Failed to get m.direct list to update it:", err)
			return
		}
		for userID, rooms := range existingChats {
			if _, ok := user.bridge.ParsePuppetMXID(userID); !ok {
				// This is not a ghost user, include it in the new list
				chats[userID] = rooms
			} else if _, ok := chats[userID]; !ok && method == http.MethodPatch {
				// This is a ghost user, but we're not replacing the whole list, so include it too
				chats[userID] = rooms
			}
		}
		err = intent.SetAccountData(event.AccountDataDirectChats.Type, &chats)
	}
	if err != nil {
		user.log.Warnln("Failed to update m.direct list:", err)
	}
}

func (user *User) HandleContactList(contacts []whatsapp.Contact) {
	contactMap := make(map[whatsapp.JID]whatsapp.Contact)
	for _, contact := range contacts {
		contactMap[contact.JID] = contact
	}
	go user.syncPuppets(contactMap)
}

func (user *User) syncPuppets(contacts map[whatsapp.JID]whatsapp.Contact) {
	if contacts == nil {
		contacts = user.Conn.Store.Contacts
	}

	_, hasSelf := contacts[user.JID]
	if !hasSelf {
		contacts[user.JID] = whatsapp.Contact{
			Name:   user.pushName,
			Notify: user.pushName,
			JID:    user.JID,
		}
	}

	user.log.Infoln("Syncing puppet info from contacts")
	for jid, contact := range contacts {
		if strings.HasSuffix(jid, whatsapp.NewUserSuffix) {
			puppet := user.bridge.GetPuppetByJID(contact.JID)
			puppet.Sync(user, contact)
		} else if strings.HasSuffix(jid, whatsapp.BroadcastSuffix) {
			portal := user.GetPortalByJID(contact.JID)
			portal.Sync(user, contact)
		}
	}
	user.log.Infoln("Finished syncing puppet info from contacts")
}

func (user *User) updateLastConnectionIfNecessary() {
	if user.LastConnection+60 < uint64(time.Now().Unix()) {
		user.UpdateLastConnection()
	}
}

func (user *User) HandleError(err error) {
	if !errors.Is(err, whatsapp.ErrInvalidWsData) {
		user.log.Errorfln("WhatsApp error: %v", err)
	}
	if closed, ok := err.(*whatsapp.ErrConnectionClosed); ok {
		user.bridge.Metrics.TrackDisconnection(user.MXID)
		if closed.Code == 1000 && user.cleanDisconnection {
			user.cleanDisconnection = false
			if !user.bridge.Config.Bridge.AggressiveReconnect {
				user.sendBridgeStatus(AsmuxPong{Error: AsmuxWANotConnected})
				user.bridge.Metrics.TrackConnectionState(user.JID, false)
				user.log.Infoln("Clean disconnection by server")
				return
			} else {
				user.log.Debugln("Clean disconnection by server, but aggressive reconnection is enabled")
			}
		}
		go user.tryReconnect(fmt.Sprintf("Your WhatsApp connection was closed with websocket status code %d", closed.Code))
	} else if failed, ok := err.(*whatsapp.ErrConnectionFailed); ok {
		disconnectErr := user.Conn.Disconnect()
		if disconnectErr != nil {
			user.log.Warnln("Failed to disconnect after connection fail:", disconnectErr)
		}
		user.bridge.Metrics.TrackDisconnection(user.MXID)
		user.ConnectionErrors++
		go user.tryReconnect(fmt.Sprintf("Your WhatsApp connection failed: %v", failed.Err))
	} else if err == whatsapp.ErrPingFalse {
		disconnectErr := user.Conn.Disconnect()
		if disconnectErr != nil {
			user.log.Warnln("Failed to disconnect after failed ping:", disconnectErr)
		}
		user.bridge.Metrics.TrackDisconnection(user.MXID)
		user.ConnectionErrors++
		go user.tryReconnect(fmt.Sprintf("Your WhatsApp connection failed: %v", err))
	}
	// Otherwise unknown error, probably mostly harmless
}

func (user *User) tryReconnect(msg string) {
	user.bridge.Metrics.TrackConnectionState(user.JID, false)
	if user.ConnectionErrors > user.bridge.Config.Bridge.MaxConnectionAttempts {
		user.sendMarkdownBridgeAlert("%s. Use the `reconnect` command to reconnect.", msg)
		user.sendBridgeStatus(AsmuxPong{Error: AsmuxWANotConnected})
		return
	}
	if user.bridge.Config.Bridge.ReportConnectionRetry {
		user.sendBridgeNotice("%s. Reconnecting...", msg)
		// Don't want the same error to be repeated
		msg = ""
	}
	var tries uint
	var exponentialBackoff bool
	baseDelay := time.Duration(user.bridge.Config.Bridge.ConnectionRetryDelay)
	if baseDelay < 0 {
		exponentialBackoff = true
		baseDelay = -baseDelay + 1
	}
	delay := baseDelay
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	user.cancelReconnect = cancel
	for user.ConnectionErrors <= user.bridge.Config.Bridge.MaxConnectionAttempts {
		select {
		case <-ctx.Done():
			user.log.Debugln("tryReconnect context cancelled, aborting reconnection attempts")
			return
		default:
		}
		user.sendBridgeStatus(AsmuxPong{Error: AsmuxWAConnecting})
		err := user.Conn.Restore(true, ctx)
		if err == nil {
			user.ConnectionErrors = 0
			if user.bridge.Config.Bridge.ReportConnectionRetry {
				user.sendBridgeNotice("Reconnected successfully")
			}
			user.PostLogin()
			return
		} else if errors.Is(err, whatsapp.ErrBadRequest) {
			user.log.Warnln("Got init 400 error when trying to reconnect, resetting connection...")
			err = user.Conn.Disconnect()
			if err != nil {
				user.log.Debugln("Error while disconnecting for connection reset:", err)
			}
		} else if errors.Is(err, whatsapp.ErrUnpaired) {
			user.log.Errorln("Got init 401 (unpaired) error when trying to reconnect, not retrying")
			user.removeFromJIDMap()
			//user.JID = ""
			user.SetSession(nil)
			user.DeleteConnection()
			user.sendMarkdownBridgeAlert("\u26a0 Failed to reconnect to WhatsApp: unpaired from phone. " +
				"To re-pair your phone, log in again.")
			user.sendBridgeStatus(AsmuxPong{Error: AsmuxWANotLoggedIn})
			return
		} else if errors.Is(err, whatsapp.ErrAlreadyLoggedIn) {
			user.log.Warnln("Reconnection said we're already logged in, not trying anymore")
			return
		} else {
			user.log.Errorln("Error while trying to reconnect after disconnection:", err)
		}
		tries++
		user.ConnectionErrors++
		if user.ConnectionErrors <= user.bridge.Config.Bridge.MaxConnectionAttempts {
			if exponentialBackoff {
				delay = (1 << tries) + baseDelay
			}
			if user.bridge.Config.Bridge.ReportConnectionRetry {
				user.sendBridgeNotice("Reconnection attempt failed: %v. Retrying in %d seconds...", err, delay)
			}
			time.Sleep(delay * time.Second)
		}
	}

	user.sendBridgeStatus(AsmuxPong{Error: AsmuxWANotConnected})
	if user.bridge.Config.Bridge.ReportConnectionRetry {
		user.sendMarkdownBridgeAlert("%d reconnection attempts failed. Use the `reconnect` command to try to reconnect manually.", tries)
	} else {
		user.sendMarkdownBridgeAlert("\u26a0 %s. Additionally, %d reconnection attempts failed. Use the `reconnect` command to try to reconnect.", msg, tries)
	}
}

func (user *User) PortalKey(jid whatsapp.JID) database.PortalKey {
	return database.NewPortalKey(jid, user.JID)
}

func (user *User) GetPortalByJID(jid whatsapp.JID) *Portal {
	return user.bridge.GetPortalByJID(user.PortalKey(jid))
}

func (user *User) runMessageRingBuffer() {
	for msg := range user.messageInput {
		select {
		case user.messageOutput <- msg:
			user.bridge.Metrics.TrackBufferLength(user.MXID, len(user.messageOutput))
		default:
			dropped := <-user.messageOutput
			user.log.Warnln("Buffer is full, dropping message in", dropped.chat)
			user.messageOutput <- msg
		}
	}
}

func (user *User) handleMessageLoop() {
	for {
		select {
		case msg := <-user.messageOutput:
			user.bridge.Metrics.TrackBufferLength(user.MXID, len(user.messageOutput))
			user.GetPortalByJID(msg.chat).messages <- msg
		case <-user.syncStart:
			user.log.Debugln("Processing of incoming messages is locked")
			user.bridge.Metrics.TrackSyncLock(user.JID, true)
			user.syncWait.Wait()
			user.bridge.Metrics.TrackSyncLock(user.JID, false)
			user.log.Debugln("Processing of incoming messages unlocked")
		}
	}
}

func (user *User) HandleNewContact(contact whatsapp.Contact) {
	user.log.Debugfln("Contact message: %+v", contact)
	if strings.HasSuffix(contact.JID, whatsapp.OldUserSuffix) {
		contact.JID = strings.Replace(contact.JID, whatsapp.OldUserSuffix, whatsapp.NewUserSuffix, -1)
	}
	if strings.HasSuffix(contact.JID, whatsapp.NewUserSuffix) {
		puppet := user.bridge.GetPuppetByJID(contact.JID)
		puppet.UpdateName(user, contact)
	} else if strings.HasSuffix(contact.JID, whatsapp.BroadcastSuffix) {
		portal := user.GetPortalByJID(contact.JID)
		portal.UpdateName(contact.Name, "", nil, true)
	}
}

func (user *User) HandleBatteryMessage(battery whatsapp.BatteryMessage) {
	user.log.Debugfln("Battery message: %+v", battery)
	var notice string
	if !battery.Plugged && battery.Percentage < 15 && user.batteryWarningsSent < 1 {
		notice = fmt.Sprintf("Phone battery low (%d %% remaining)", battery.Percentage)
		user.batteryWarningsSent = 1
	} else if !battery.Plugged && battery.Percentage < 5 && user.batteryWarningsSent < 2 {
		notice = fmt.Sprintf("Phone battery very low (%d %% remaining)", battery.Percentage)
		user.batteryWarningsSent = 2
	} else if battery.Percentage > 15 || battery.Plugged {
		user.batteryWarningsSent = 0
	}
	if notice != "" {
		go user.sendBridgeNotice("%s", notice)
	}
}

type FakeMessage struct {
	Text  string
	ID    string
	Alert bool
}

func (user *User) HandleCallInfo(info whatsapp.CallInfo) {
	if info.Data != nil {
		return
	}
	data := FakeMessage{
		ID: info.ID,
	}
	switch info.Type {
	case whatsapp.CallOffer:
		if !user.bridge.Config.Bridge.CallNotices.Start {
			return
		}
		data.Text = "Incoming call"
		data.Alert = true
	case whatsapp.CallOfferVideo:
		if !user.bridge.Config.Bridge.CallNotices.Start {
			return
		}
		data.Text = "Incoming video call"
		data.Alert = true
	case whatsapp.CallTerminate:
		if !user.bridge.Config.Bridge.CallNotices.End {
			return
		}
		data.Text = "Call ended"
		data.ID += "E"
	default:
		return
	}
	portal := user.GetPortalByJID(info.From)
	if portal != nil {
		portal.messages <- PortalMessage{info.From, user, data, 0}
	}
}

func (user *User) HandlePresence(info whatsapp.PresenceEvent) {
	puppet := user.bridge.GetPuppetByJID(info.SenderJID)
	switch info.Status {
	case whatsapp.PresenceUnavailable:
		_ = puppet.DefaultIntent().SetPresence("offline")
	case whatsapp.PresenceAvailable:
		if len(puppet.typingIn) > 0 && puppet.typingAt+15 > time.Now().Unix() {
			portal := user.bridge.GetPortalByMXID(puppet.typingIn)
			_, _ = puppet.IntentFor(portal).UserTyping(puppet.typingIn, false, 0)
			puppet.typingIn = ""
			puppet.typingAt = 0
		} else {
			_ = puppet.DefaultIntent().SetPresence("online")
		}
	case whatsapp.PresenceComposing:
		portal := user.GetPortalByJID(info.JID)
		if len(puppet.typingIn) > 0 && puppet.typingAt+15 > time.Now().Unix() {
			if puppet.typingIn == portal.MXID {
				return
			}
			_, _ = puppet.IntentFor(portal).UserTyping(puppet.typingIn, false, 0)
		}
		puppet.typingIn = portal.MXID
		puppet.typingAt = time.Now().Unix()
		_, _ = puppet.IntentFor(portal).UserTyping(portal.MXID, true, 15*1000)
	}
}

func (user *User) HandleMsgInfo(info whatsapp.JSONMsgInfo) {
	if (info.Command == whatsapp.MsgInfoCommandAck || info.Command == whatsapp.MsgInfoCommandAcks) && info.Acknowledgement == whatsapp.AckMessageRead {
		portal := user.GetPortalByJID(info.ToJID)
		if len(portal.MXID) == 0 {
			return
		}

		intent := user.bridge.GetPuppetByJID(info.SenderJID).IntentFor(portal)
		for _, msgID := range info.IDs {
			msg := user.bridge.DB.Message.GetByJID(portal.Key, msgID)
			if msg == nil || msg.IsFakeMXID() {
				continue
			}

			err := intent.MarkRead(portal.MXID, msg.MXID)
			if err != nil {
				user.log.Warnln("Failed to mark message %s as read by %s: %v", msg.MXID, info.SenderJID, err)
			}
		}
	}
}

func (user *User) HandleReceivedMessage(received whatsapp.ReceivedMessage) {
	if received.Type == "read" {
		go user.markSelfRead(received.Jid, received.Index)
	} else {
		user.log.Debugfln("Unknown received message type: %+v", received)
	}
}

func (user *User) HandleReadMessage(read whatsapp.ReadMessage) {
	user.log.Debugfln("Received chat read message: %+v", read)
	go user.markSelfRead(read.Jid, "")
}

func (user *User) markSelfRead(jid, messageID string) {
	if strings.HasSuffix(jid, whatsapp.OldUserSuffix) {
		jid = strings.Replace(jid, whatsapp.OldUserSuffix, whatsapp.NewUserSuffix, -1)
	}
	puppet := user.bridge.GetPuppetByJID(user.JID)
	if puppet == nil {
		return
	}
	intent := puppet.CustomIntent()
	if intent == nil {
		return
	}
	portal := user.GetPortalByJID(jid)
	if portal == nil {
		return
	}
	var message *database.Message
	if messageID == "" {
		message = user.bridge.DB.Message.GetLastInChat(portal.Key)
		if message == nil {
			return
		}
		user.log.Debugfln("User read chat %s/%s in WhatsApp mobile (last known event: %s/%s)", portal.Key.JID, portal.MXID, message.JID, message.MXID)
	} else {
		message = user.bridge.DB.Message.GetByJID(portal.Key, messageID)
		if message == nil || message.IsFakeMXID() {
			return
		}
		user.log.Debugfln("User read message %s/%s in %s/%s in WhatsApp mobile", message.JID, message.MXID, portal.Key.JID, portal.MXID)
	}
	err := intent.MarkRead(portal.MXID, message.MXID)
	if err != nil {
		user.log.Warnfln("Failed to bridge own read receipt in %s: %v", jid, err)
	}
}

func (user *User) HandleCommand(cmd whatsapp.JSONCommand) {
	switch cmd.Type {
	case whatsapp.CommandPicture:
		if strings.HasSuffix(cmd.JID, whatsapp.NewUserSuffix) {
			puppet := user.bridge.GetPuppetByJID(cmd.JID)
			go puppet.UpdateAvatar(user, cmd.ProfilePicInfo)
		} else if user.bridge.Config.Bridge.ChatMetaSync {
			portal := user.GetPortalByJID(cmd.JID)
			go portal.UpdateAvatar(user, cmd.ProfilePicInfo, true)
		}
	case whatsapp.CommandDisconnect:
		if cmd.Kind == "replaced" {
			user.cleanDisconnection = true
			go user.sendMarkdownBridgeAlert("\u26a0 Your WhatsApp connection was closed by the server because you opened another WhatsApp Web client.\n\n" +
				"Use the `reconnect` command to disconnect the other client and resume bridging.")
		} else {
			user.log.Warnln("Unknown kind of disconnect:", string(cmd.Raw))
			go user.sendMarkdownBridgeAlert("\u26a0 Your WhatsApp connection was closed by the server (reason code: %s).\n\n"+
				"Use the `reconnect` command to reconnect.", cmd.Kind)
		}
	}
}

func (user *User) HandleChatUpdate(cmd whatsapp.ChatUpdate) {
	if cmd.Command != whatsapp.ChatUpdateCommandAction {
		return
	}

	portal := user.GetPortalByJID(cmd.JID)
	if len(portal.MXID) == 0 {
		if cmd.Data.Action == whatsapp.ChatActionIntroduce || cmd.Data.Action == whatsapp.ChatActionCreate {
			go func() {
				err := portal.CreateMatrixRoom(user)
				if err != nil {
					user.log.Errorln("Failed to create portal room after receiving join event:", err)
				}
			}()
		}
		return
	}

	// These don't come down the message history :(
	switch cmd.Data.Action {
	case whatsapp.ChatActionAddTopic:
		go portal.UpdateTopic(cmd.Data.AddTopic.Topic, cmd.Data.SenderJID, nil, true)
	case whatsapp.ChatActionRemoveTopic:
		go portal.UpdateTopic("", cmd.Data.SenderJID, nil, true)
	case whatsapp.ChatActionRemove:
		// We ignore leaving groups in the message history to avoid accidentally leaving rejoined groups,
		// but if we get a real-time command that says we left, it should be safe to bridge it.
		if !user.bridge.Config.Bridge.ChatMetaSync {
			for _, jid := range cmd.Data.UserChange.JIDs {
				if jid == user.JID {
					go portal.HandleWhatsAppKick(nil, cmd.Data.SenderJID, cmd.Data.UserChange.JIDs)
					break
				}
			}
		}
	}

	if !user.bridge.Config.Bridge.ChatMetaSync {
		// Ignore chat update commands, we're relying on the message history.
		return
	}

	switch cmd.Data.Action {
	case whatsapp.ChatActionNameChange:
		go portal.UpdateName(cmd.Data.NameChange.Name, cmd.Data.SenderJID, nil, true)
	case whatsapp.ChatActionPromote:
		go portal.ChangeAdminStatus(cmd.Data.UserChange.JIDs, true)
	case whatsapp.ChatActionDemote:
		go portal.ChangeAdminStatus(cmd.Data.UserChange.JIDs, false)
	case whatsapp.ChatActionAnnounce:
		go portal.RestrictMessageSending(cmd.Data.Announce)
	case whatsapp.ChatActionRestrict:
		go portal.RestrictMetadataChanges(cmd.Data.Restrict)
	case whatsapp.ChatActionRemove:
		go portal.HandleWhatsAppKick(nil, cmd.Data.SenderJID, cmd.Data.UserChange.JIDs)
	case whatsapp.ChatActionAdd:
		go portal.HandleWhatsAppInvite(cmd.Data.SenderJID, nil, cmd.Data.UserChange.JIDs)
	case whatsapp.ChatActionIntroduce:
		if cmd.Data.SenderJID != "unknown" {
			go portal.Sync(user, whatsapp.Contact{JID: portal.Key.JID})
		}
	}
}

func (user *User) HandleConnInfo(info whatsapp.ConnInfo) {
	if user.Session != nil && info.Connected && len(info.ClientToken) > 0 {
		user.log.Debugln("Received new tokens")
		user.Session.ClientToken = info.ClientToken
		user.Session.ServerToken = info.ServerToken
		user.Session.Wid = info.WID
		user.Update()
	}
	if len(info.PushName) > 0 {
		user.pushName = info.PushName
	}
}

func (user *User) HandleJSONMessage(message json.RawMessage) {
	if !json.Valid(message) {
		return
	}
	user.log.Debugfln("JSON message: %s", message)
	user.updateLastConnectionIfNecessary()
}

func (user *User) NeedsRelaybot(portal *Portal) bool {
	return !user.HasSession() || !user.IsInPortal(portal.Key)
}