path: root/connector/xmpp/xmpp.go

package xmpp

import (
	"crypto/tls"
	"fmt"
	"strings"
	"sync"
	"time"

	gxmpp "github.com/matterbridge/go-xmpp"
	"github.com/rs/xid"
	log "github.com/sirupsen/logrus"

	. "git.deuxfleurs.fr/Deuxfleurs/easybridge/connector"
)

// User id format: username@server (= JID)
// OR: nickname@room_name@muc_server

// Room id format: room_name@muc_server (= MUC ID)

type XMPP struct {
	handler Handler

	connectorLoopNum int
	connected        bool
	timeout          int

	server        string
	port          int
	ssl           bool
	jid           string
	jid_localpart string
	password      string
	nickname      string

	conn *gxmpp.Client

	stateLock sync.Mutex
	muc       map[RoomID]*mucInfo
}

type mucInfo struct {
	joined        bool
	pendingJoins  map[UserID]string
	pendingLeaves map[UserID]struct{}
}

func (xm *XMPP) SetHandler(h Handler) {
	xm.handler = h
}

func (xm *XMPP) Protocol() string {
	return XMPP_PROTOCOL
}

func (xm *XMPP) Configure(c Configuration) error {
	if xm.conn != nil {
		xm.Close()
	}

	// Parse and validate configuration
	var err error

	xm.port, err = c.GetInt("port", 5222)
	if err != nil {
		return err
	}

	xm.ssl, err = c.GetBool("ssl", true)
	if err != nil {
		return err
	}

	xm.jid, err = c.GetString("jid")
	if err != nil {
		return err
	}
	jid_parts := strings.Split(xm.jid, "@")
	if len(jid_parts) != 2 {
		return fmt.Errorf("Invalid JID: %s", xm.jid)
	}
	xm.server = jid_parts[1]
	xm.jid_localpart = jid_parts[0]

	xm.nickname, _ = c.GetString("nickname", xm.jid_localpart)

	xm.password, err = c.GetString("password")
	if err != nil {
		return err
	}

	// Try to connect
	xm.muc = make(map[RoomID]*mucInfo)

	xm.connectorLoopNum += 1
	go xm.connectLoop(xm.connectorLoopNum)

	for i := 0; i < 42; i++ {
		time.Sleep(time.Duration(1) * time.Second)
		if xm.connected {
			return nil
		}
	}
	return fmt.Errorf("Failed to connect after 42s attempting")
}

func (xm *XMPP) connectLoop(num int) {
	xm.timeout = 10
	for {
		if xm.connectorLoopNum != num {
			return
		}
		tc := &tls.Config{
			ServerName:         xm.server,
			InsecureSkipVerify: true,
		}
		options := gxmpp.Options{
			Host:      xm.server,
			User:      xm.jid,
			Password:  xm.password,
			NoTLS:     true,
			StartTLS:  xm.ssl,
			Session:   true,
			TLSConfig: tc,
		}
		var err error
		xm.conn, err = options.NewClient()
		if err != nil {
			xm.connected = false
			xm.handler.SystemMessage(fmt.Sprintf("XMPP failed to connect (%s). Retrying in %ds", err, xm.timeout))
			time.Sleep(time.Duration(xm.timeout) * time.Second)
			xm.timeout *= 2
			if xm.timeout > 600 {
				xm.timeout = 600
			}
		} else {
			xm.connected = true
			xm.timeout = 10

			for muc, mucInfo := range xm.muc {
				if mucInfo.joined {
					_, err := xm.conn.JoinMUCNoHistory(string(muc), xm.nickname)
					if err != nil {
						xm.handler.SystemMessage(fmt.Sprintf("Could not rejoin MUC %s after reconnection: %s", muc, err))
						xm.handler.Left(RoomID(muc))
						mucInfo.joined = false
					}
				}
			}

			err = xm.handleXMPP()

			xm.connected = false
			xm.handler.SystemMessage(fmt.Sprintf("XMPP disconnected (%s), reconnecting)", err))
		}
	}
}

func (xm *XMPP) xmppKeepAlive() chan bool {
	done := make(chan bool)
	go func() {
		ticker := time.NewTicker(90 * time.Second)
		defer ticker.Stop()
		for {
			select {
			case <-ticker.C:
				if err := xm.conn.PingC2S("", ""); err != nil {
					log.Debugf("PING failed %#v\n", err)
				}
			case <-done:
				return
			}
		}
	}()
	return done
}

func (xm *XMPP) handleXMPP() error {
	done := xm.xmppKeepAlive()
	defer close(done)

	for {
		m, err := xm.conn.Recv()
		if err != nil {
			return err
		}

		log.Tracef("XMPP: %#v\n", m)

		xm.handleXMPPStance(m)
	}
}

func (xm *XMPP) handleXMPPStance(m interface{}) {
	xm.stateLock.Lock()
	defer xm.stateLock.Unlock()

	switch v := m.(type) {
	case gxmpp.Chat:
		remote_sp := strings.Split(v.Remote, "/")

		// Skip self-sent events
		if v.Remote == xm.jid || (v.Type == "groupchat" && len(remote_sp) == 2 && remote_sp[1] == xm.nickname) {
			return
		}

		// If empty text, make sure we joined the room
		// We would do this at every incoming message if it were not so costly
		if v.Text == "" && v.Type == "groupchat" {
			xm.handler.Joined(RoomID(remote_sp[0]))
		}

		// Handle subject change in group chats
		if v.Subject != "" && v.Type == "groupchat" {
			author := UserID("")
			if len(remote_sp) == 2 {
				if remote_sp[1] == xm.nickname {
					author = xm.User()
				} else {
					author = UserID(remote_sp[1] + "@" + remote_sp[0])
				}
			}
			xm.handler.RoomInfoUpdated(RoomID(remote_sp[0]), author, &RoomInfo{
				Topic: v.Subject,
			})
		}

		// Handle text message
		if v.Text != "" {
			event := &Event{
				Type: EVENT_MESSAGE,
				Text: v.Text,
			}

			if strings.HasPrefix(event.Text, "/me ") {
				event.Type = EVENT_ACTION
				event.Text = strings.Replace(event.Text, "/me ", "", 1)
			}

			if v.Type == "chat" {
				event.Author = UserID(remote_sp[0])
				xm.handler.Event(event)
			}
			if v.Type == "groupchat" && len(remote_sp) == 2 {
				// First flush pending leaves and joins
				room_id := RoomID(remote_sp[0])
				if muc, ok := xm.muc[room_id]; ok {
					muc.flushLeavesJoins(room_id, xm.handler)
				}

				// Now send event
				event.Room = room_id
				event.Author = UserID(remote_sp[1] + "@" + remote_sp[0])
				event.Id = v.ID
				xm.handler.Event(event)
			}
		}
	case gxmpp.Presence:
		remote := strings.Split(v.From, "/")
		room := RoomID(remote[0])
		if mucInfo, ok := xm.muc[room]; ok {
			// skip presence with no user and self-presence
			if len(remote) < 2 || remote[1] == xm.nickname {
				return
			}

			user := UserID(remote[1] + "@" + remote[0])
			if v.Type != "unavailable" {
				if _, ok := mucInfo.pendingLeaves[user]; ok {
					delete(mucInfo.pendingLeaves, user)
				} else {
					mucInfo.pendingJoins[user] = remote[1]
				}
			} else {
				if _, ok := mucInfo.pendingJoins[user]; ok {
					delete(mucInfo.pendingJoins, user)
				} else {
					mucInfo.pendingLeaves[user] = struct{}{}
				}
			}
		}
	}
}

func (muc *mucInfo) flushLeavesJoins(room RoomID, handler Handler) {
	for user, display_name := range muc.pendingJoins {
		handler.Event(&Event{
			Type:   EVENT_JOIN,
			Room:   room,
			Author: user,
		})
		handler.UserInfoUpdated(user, &UserInfo{
			DisplayName: display_name,
		})
	}
	for user, _ := range muc.pendingLeaves {
		handler.Event(&Event{
			Type:   EVENT_LEAVE,
			Room:   room,
			Author: user,
		})
	}
	muc.pendingJoins = make(map[UserID]string)
	muc.pendingLeaves = make(map[UserID]struct{})
}

func (xm *XMPP) User() UserID {
	return UserID(xm.jid)
}

func (xm *XMPP) SetUserInfo(info *UserInfo) error {
	return fmt.Errorf("Not implemented")
}

func (xm *XMPP) SetRoomInfo(roomId RoomID, info *RoomInfo) error {
	if info.Topic != "" {
		_, err := xm.conn.Send(gxmpp.Chat{
			Type:    "groupchat",
			Remote:  string(roomId),
			Subject: info.Topic,
		})
		if err != nil {
			return err
		}
	}

	if info.Picture.MediaObject != nil {
		// TODO
		return fmt.Errorf("Room picture change not implemented on xmpp")
	}

	if info.Name != "" && info.Name != string(roomId) {
		// TODO
		return fmt.Errorf("Room name change not implemented on xmpp")
	}
	return nil
}

func (xm *XMPP) Join(roomId RoomID) error {
	xm.stateLock.Lock()
	defer xm.stateLock.Unlock()

	xm.muc[roomId] = &mucInfo{
		pendingJoins:  make(map[UserID]string),
		pendingLeaves: make(map[UserID]struct{}),
	}

	log.Tracef("Join %s with nick %s\n", roomId, xm.nickname)
	_, err := xm.conn.JoinMUCNoHistory(string(roomId), xm.nickname)

	if err == nil {
		xm.muc[roomId].joined = true
	}

	return err
}

func (xm *XMPP) Invite(userId UserID, roomId RoomID) error {
	if roomId == "" {
		xm.conn.RequestSubscription(string(userId))
		xm.conn.ApproveSubscription(string(userId))
		return nil
	}
	// TODO
	return fmt.Errorf("Not implemented")
}

func (xm *XMPP) Leave(roomId RoomID) {
	xm.stateLock.Lock()
	defer xm.stateLock.Unlock()

	xm.conn.LeaveMUC(string(roomId))

	if muc, ok := xm.muc[roomId]; ok {
		muc.joined = false
	}
}

func (xm *XMPP) SearchForUsers(query string) ([]UserSearchResult, error) {
	// TODO: search roster
	return nil, fmt.Errorf("Not implemented")
}

func (xm *XMPP) Send(event *Event) (string, error) {
	xm.stateLock.Lock()
	defer xm.stateLock.Unlock()

	if event.Attachments != nil && len(event.Attachments) > 0 {
		for _, at := range event.Attachments {
			url := at.URL()
			if url == "" {
				// TODO find a way to send them using some hosing of some kind
				return "", fmt.Errorf("Attachment without URL sent to XMPP")
			} else {
				event.Text += fmt.Sprintf("\n%s (%s, %dkb)",
					url, at.Mimetype(), at.Size()/1024)
			}
		}
	}

	if event.Id == "" {
		event.Id = xid.New().String()
	}

	log.Tracef("xm *XMPP Send %#v\n", event)
	if len(event.Recipient) > 0 {
		_, err := xm.conn.Send(gxmpp.Chat{
			Type:   "chat",
			Remote: string(event.Recipient),
			Text:   event.Text,
		})
		return event.Id, err
	} else if len(event.Room) > 0 {
		if muc, ok := xm.muc[event.Room]; ok {
			muc.flushLeavesJoins(event.Room, xm.handler)
		}
		_, err := xm.conn.Send(gxmpp.Chat{
			Type:   "groupchat",
			Remote: string(event.Room),
			Text:   event.Text,
			ID:     event.Id,
		})
		return event.Id, err
	} else {
		return "", fmt.Errorf("Invalid event")
	}
}

func (xm *XMPP) UserCommand(cmd string) {
	xm.handler.SystemMessage("Command not supported.")
}

func (xm *XMPP) Close() {
	xm.stateLock.Lock()
	defer xm.stateLock.Unlock()

	if xm.conn != nil {
		xm.conn.Close()
	}
	xm.conn = nil
	xm.connectorLoopNum += 1
	xm.connected = false
}