groupme-lib/real_time.go

package groupme

import (
	"context"
	"errors"
	"gitea.watsonlabs.net/watsonb8/fayec"
	"gitea.watsonlabs.net/watsonb8/fayec/message"
	"gitea.watsonlabs.net/watsonb8/fayec/subscription"
	"log"
	"strings"
	"sync"
	"time"
)

const (
	PushServer   = "wss://push.groupme.com/faye"
	userChannel  = "/user/"
	groupChannel = "/group/"
	dmChannel    = "/direct_message/"
)

var (
	ErrHandlerNotFound    = errors.New("Handler not found")
	ErrListenerNotStarted = errors.New("GroupMe listener not started")
)

var concur = sync.Mutex{}
var token string

type HandlerAll interface {
	Handler

	//of self
	HandlerText
	HandlerLike
	HandlerMembership

	//of group
	HandleGroupTopic
	HandleGroupAvatar
	HandleGroupName
	HandleGroupLikeIcon

	//of group members
	HandleMemberNewNickname
	HandleMemberNewAvatar
	HandleMembers
}
type Handler interface {
	HandleError(error)
}
type HandlerText interface {
	HandleTextMessage(Message)
}
type HandlerLike interface {
	HandleLike(Message)
}
type HandlerMembership interface {
	HandleJoin(ID)
}

// Group Handlers
type HandleGroupTopic interface {
	HandleGroupTopic(group ID, newTopic string)
}

type HandleGroupName interface {
	HandleGroupName(group ID, newName string)
}
type HandleGroupAvatar interface {
	HandleGroupAvatar(group ID, newAvatar string)
}
type HandleGroupLikeIcon interface {
	HandleLikeIcon(group ID, PackID, PackIndex int, Type string)
}

// Group member handlers
type HandleMemberNewNickname interface {
	HandleNewNickname(group ID, user ID, newName string)
}

type HandleMemberNewAvatar interface {
	HandleNewAvatarInGroup(group ID, user ID, avatarURL string)
}
type HandleMembers interface {
	//HandleNewMembers returns only partial member with id and nickname; added is false if removing
	HandleMembers(group ID, members []Member, added bool)
}

// PushSubscription manages real time subscription
type PushSubscription struct {
	channel       chan message.Data
	client        *fayec.Client
	handlers      []Handler
	LastConnected int64
}

// NewPushSubscription creates and returns a push subscription object
func NewPushSubscription(context context.Context) PushSubscription {

	r := PushSubscription{
		channel: make(chan message.Data),
	}

	return r
}

func (r *PushSubscription) AddHandler(h Handler) {
	r.handlers = append(r.handlers, h)
}

// AddFullHandler is the same as AddHandler except it ensures the interface implements everything
func (r *PushSubscription) AddFullHandler(h HandlerAll) {
	r.handlers = append(r.handlers, h)
}

var RealTimeHandlers map[string]func(r *PushSubscription, channel string, data ...interface{})
var RealTimeSystemHandlers map[string]func(r *PushSubscription, channel string, id ID, rawData []byte)

// Listen connects to GroupMe. Runs in Goroutine.
func (r *PushSubscription) Connect(context context.Context, authorizationToken string) error {
	token = authorizationToken
	var authenticationExtension message.Extension = func(message *message.Message) {
		if message.Channel == "/meta/subscribe" {
			message.Ext = map[string]string{
				"access_token": authorizationToken,
				"timestamp":    string(time.Now().Unix()),
			}
		}
	}
	c, err := fayec.NewClient(PushServer, fayec.WithOutExtension(authenticationExtension))
	if err != nil {
		return err
	}
	r.client = c

	return nil
}

// SubscribeToUser to users
func (r *PushSubscription) SubscribeToUser(context context.Context, id ID) error {
	return r.subscribeWithPrefix(userChannel, context, id)
}

// SubscribeToGroup to groups for typing notification
func (r *PushSubscription) SubscribeToGroup(context context.Context, id ID) error {
	return r.subscribeWithPrefix(groupChannel, context, id)
}

// SubscribeToDM to users
func (r *PushSubscription) SubscribeToDM(context context.Context, id ID) error {
	id = ID(strings.Replace(id.String(), "+", "_", 1))
	return r.subscribeWithPrefix(dmChannel, context, id)
}

func (r *PushSubscription) subscribeWithPrefix(prefix string, context context.Context, groupID ID) error {
	concur.Lock()
	defer concur.Unlock()
	if r.client == nil {
		return ErrListenerNotStarted
	}

	var sub *subscription.Subscription
	sub, err := r.client.Subscribe(prefix + groupID.String())
	if err != nil {
		panic(err)
	}

	err = sub.OnMessage(func(channel string, data message.Data) {
		r.LastConnected = time.Now().Unix()
		dataMap := data.(map[string]interface{})
		content := dataMap["subject"]
		contentType := dataMap["type"].(string)

		handler, ok := RealTimeHandlers[contentType]
		if !ok {
			if contentType == "ping" ||
				len(contentType) == 0 ||
				content == "" {
				return
			}
			log.Println("Unable to handle GroupMe message type", contentType)
		}

		handler(r, channel, content)
	})

	return nil
}

// Connected check if connected
func (r *PushSubscription) Connected() bool {
	return r.LastConnected+30 >= time.Now().Unix()
}