spaceapi/xmpp/xmpp.go

package xmpp

import (
	"crypto/tls"
	"fmt"
	"log"
	"net"
	"strings"
	"time"

	"git.hacknology.de/projekte/spaceapi"
	"github.com/mattn/go-xmpp"
)

const (
	commandHelp  = "!help"
	commandState = "!state"
	commandClose = "!closed"

	helpMessage = `Spacebot hat folgende Kommandos:

!help   - Diese Nachricht.
!state  - Ist der Space offen?
!closed - Space ist zu.

Gern geschehen.`

	errorNonPublic = "Bitte mach das öffentlich, damit jeder weiß von wem die Information ist."

	stateUnknown = "Ich weiß leider nicht ob der Space offen ist."
	stateOpen    = "Space ist OFFEN."
	stateClosed  = "Space ist zu."
)

// Listener returns a storage listener that broadcasts changes in the state to an XMPP MUC.
func Listener(storage *spaceapi.Storage, jid, password, target, handle string) (spaceapi.Listener, error) {
	xmppHost, err := lookupHost(jid)
	if err != nil {
		return nil, fmt.Errorf("can not find SRV record: %s", err)
	}

	clientOpts := xmpp.Options{
		Host:     xmppHost,
		User:     jid,
		Password: password,
		NoTLS:    true,
		Debug:    false,
		Session:  false,
		TLSConfig: &tls.Config{
			InsecureSkipVerify: true,
		},
	}

	sendChan := make(chan string, 1)
	go clientLoop(storage, sendChan, clientOpts, target, handle)

	return func(old, new spaceapi.SpaceStatus) {
		if new.State.Open == nil {
			return
		}

		if old.State.Open == new.State.Open {
			return
		}

		msg := "Space ist jetzt OFFEN!"
		if !*new.State.Open {
			msg = "Space ist jetzt ZU!"
		}

		sendChan <- msg
	}, nil
}

const (
	defaultClientPort = 5222
	backOffFactor     = 1.2
	maxReconnectWait  = 5 * time.Minute
)

func clientLoop(storage *spaceapi.Storage, sendChan <-chan string, clientOpts xmpp.Options, target, handle string) {
	wait := 1 * time.Second
	for {
		client, err := clientConnect(clientOpts, target, handle)
		if err != nil {
			log.Printf("Error connecting to XMPP server: %s", err)

			wait = time.Duration(float64(wait) * backOffFactor)
			if wait > maxReconnectWait {
				wait = maxReconnectWait
			}

			log.Printf("Retrying in %s...", wait)
			time.Sleep(wait)
			continue
		}

		receiveChan, errorChan := receiveLoop(storage, client, target)

	receiveLoop:
		for {
			select {
			case err := <-errorChan:
				log.Printf("Error receiving messages: %s", err)
				break receiveLoop
			case msg := <-sendChan:
				if err := sendGroupMessage(client, target, msg); err != nil {
					log.Printf("Error sending message: %s", err)
				}
			case chat := <-receiveChan:
				if err := handleMessage(storage, client, target, chat); err != nil {
					log.Printf("Error handling message: %s", err)
				}
			}
		}

		log.Printf("Closing XMPP connection.")
		client.Close()
	}
}

func handleMessage(storage *spaceapi.Storage, client *xmpp.Client, target string, chat xmpp.Chat) error {
	switch chat.Text {
	case commandHelp:
		if err := sendMessage(client, chat.Remote, helpMessage); err != nil {
			return fmt.Errorf("error sending reply: %s", err)
		}
	case commandState:
		state := storage.Status().State
		msg := stateUnknown

		if state.Open != nil {
			switch *state.Open {
			case true:
				msg = stateOpen
			case false:
				msg = stateClosed
			}
		}

		sendFunc := sendGroupMessage
		replyTarget := target
		if chat.Type != "groupchat" {
			sendFunc = sendMessage
			replyTarget = chat.Remote
		}

		if err := sendFunc(client, replyTarget, msg); err != nil {
			return fmt.Errorf("error sending reply: %s", err)
		}
	case commandClose:
		if chat.Type != "groupchat" {
			if err := sendMessage(client, chat.Remote, errorNonPublic); err != nil {
				return fmt.Errorf("error sending reply: %s", err)
			}

			return nil
		}

		storage.Modify(func(status *spaceapi.SpaceStatus) {
			open := false
			status.State.Open = &open
		})
	}

	return nil
}

func clientConnect(clientOpts xmpp.Options, target, handle string) (*xmpp.Client, error) {
	log.Printf("Connecting to %s as %s", clientOpts.Host, clientOpts.User)
	client, err := clientOpts.NewClient()
	if err != nil {
		return nil, fmt.Errorf("can not connect to server: %s", err)
	}

	if _, err := client.JoinMUCNoHistory(target, handle); err != nil {
		return nil, fmt.Errorf("can not join room: %s", err)
	}

	return client, nil
}

func receiveLoop(storage *spaceapi.Storage, client *xmpp.Client, target string) (<-chan xmpp.Chat, <-chan error) {
	receiveChan := make(chan xmpp.Chat)
	errorChan := make(chan error)

	go func() {
		defer close(errorChan)
		defer close(receiveChan)

		for {
			chat, err := client.Recv()
			if err != nil {
				errorChan <- err
				return
			}

			switch v := chat.(type) {
			case xmpp.Chat:
				receiveChan <- v
			}
		}
	}()

	return receiveChan, errorChan
}

func lookupHost(jid string) (string, error) {
	if !strings.Contains(jid, "@") {
		return "", fmt.Errorf("not a valid JID: %s", jid)
	}

	parts := strings.SplitN(jid, "@", 2)

	_, addrs, err := net.LookupSRV("xmpp-client", "tcp", parts[1])
	if err != nil {
		return "", fmt.Errorf("%s", err)
	}

	if len(addrs) == 0 {
		return fmt.Sprintf("%s:%d", parts[1], defaultClientPort), nil
	}

	return fmt.Sprintf("%s:%d", addrs[0].Target, addrs[0].Port), nil
}

func sendMessage(client *xmpp.Client, target, msg string) error {
	chat := xmpp.Chat{
		Remote: target,
		Type:   "chat",
		Text:   msg,
	}

	if _, err := client.Send(chat); err != nil {
		return err
	}

	return nil
}

func sendGroupMessage(client *xmpp.Client, target, msg string) error {
	chat := xmpp.Chat{
		Remote: target,
		Type:   "groupchat",
		Text:   msg,
	}

	if _, err := client.Send(chat); err != nil {
		return err
	}

	return nil
}