Extend event service & event queue item to support subscription types

On top of 'channel-type' subscriptions we had, we now support 'user-type' subscriptions to handle
subscription requests (ie: we want this specific user to be subscribed to that specific channel)

sam/repository/channel.go

@@ -157,6 +157,7 @@ func (r *channel) FindMembers(userID uint64) (types.ChannelMemberSet, error) {
 func (r *channel) CreateChannel(mod *types.Channel) (*types.Channel, error) {
 	mod.ID = factory.Sonyflake.NextID()
 	mod.CreatedAt = time.Now()
+	mod.UpdatedAt = nil
 	mod.Meta = coalesceJson(mod.Meta, []byte("{}"))
 
 	if mod.Type == "" {

sam/service/events.go

@@ -35,8 +35,9 @@ func (svc *event) With(ctx context.Context) EventService {
 	}
 }
 
+// Message sends message events to subscribers
 func (svc *event) Message(m *types.Message) error {
-	return svc.push(payload.Message(m), m.ChannelID)
+	return svc.push(payload.Message(m), types.EventQueueItemSubTypeChannel, m.ChannelID)
 }
 
 // Channel notifies subscribers about channel change
@@ -48,10 +49,18 @@ func (svc *event) Channel(ch *types.Channel) error {
 		sub = 0
 	}
 
-	return svc.push(payload.Channel(ch), sub)
+	return svc.push(payload.Channel(ch), types.EventQueueItemSubTypeChannel, sub)
 }
 
-func (svc *event) push(m outgoing.MessageEncoder, sub uint64) error {
+// Join sends force-channel-join event to all matching subscriptions
+//
+// Subscription will match when session's user ID is the same as sub
+// We pack channel ID (the id to subscribe to) as payload
+func (svc *event) Join(userID, channelID uint64) error {
+	return svc.push(payload.Channel(&types.Channel{ID: channelID}), types.EventQueueItemSubTypeUser, userID)
+}
+
+func (svc *event) push(m outgoing.MessageEncoder, subType types.EventQueueItemSubType, sub uint64) error {
 	var enc, err = m.EncodeMessage()
 	if err != nil {
 		return err
@@ -61,6 +70,7 @@ func (svc *event) push(m outgoing.MessageEncoder, sub uint64) error {
 
 	if sub > 0 {
 		item.Subscriber = payload.Uint64toa(sub)
+		item.SubType = subType
 	}
 
 	return svc.events.Push(svc.ctx, item)

sam/types/event_queue.go

@@ -6,9 +6,17 @@ import (
 
 type (
 	EventQueueItem struct {
-		ID         uint64          `db:"id"`
-		Origin     uint64          `db:"origin"`
-		Subscriber string          `db:"subscriber"`
-		Payload    json.RawMessage `db:"payload"`
+		ID         uint64                `db:"id"`
+		Origin     uint64                `db:"origin"`
+		SubType    EventQueueItemSubType `db:"subtype"`
+		Subscriber string                `db:"subscriber"`
+		Payload    json.RawMessage       `db:"payload"`
 	}
+
+	EventQueueItemSubType string
+)
+
+const (
+	EventQueueItemSubTypeUser    = "user"
+	EventQueueItemSubTypeChannel = "channel"
 )

sam/websocket/event_queue.go

@@ -2,9 +2,12 @@ package websocket
 
 import (
 	"context"
+	"encoding/json"
 
 	"github.com/titpetric/factory"
 
+	"github.com/crusttech/crust/internal/payload"
+	"github.com/crusttech/crust/internal/payload/outgoing"
 	"github.com/crusttech/crust/sam/repository"
 	"github.com/crusttech/crust/sam/types"
 )
@@ -63,12 +66,30 @@ func (eq *eventQueue) feedSessions(ctx context.Context, config *repository.Flags
 				s.sendBytes(item.Payload)
 			})
 		} else {
-			// Distribute payload to specific subscribers
-			store.Walk(func(s *Session) {
-				if s.subs.Get(item.Subscriber) != nil {
-					s.sendBytes(item.Payload)
+			switch item.SubType {
+			case types.EventQueueItemSubTypeUser:
+				// Holds channel info (in fact, ID only) when payload is unmarshaled
+				var join *outgoing.Channel
+				if err := json.Unmarshal(item.Payload, join); err == nil {
+					return err
 				}
-			})
+
+				// This store.Walk handler does send to subscribed sessions but
+				// subscribes all sessions that belong to the same user
+				store.Walk(func(s *Session) {
+					if payload.Uint64toa(s.user.Identity()) == item.Subscriber {
+						s.subs.Add(join.ID)
+					}
+				})
+			case types.EventQueueItemSubTypeChannel:
+				// Distribute payload to specific subscribers
+				store.Walk(func(s *Session) {
+					if s.subs.Get(item.Subscriber) != nil {
+						s.sendBytes(item.Payload)
+					}
+				})
+			}
+
 		}
 	}
 }

sam/websocket/session_incoming_channel.go

@@ -77,21 +77,7 @@ func (s *Session) channelCreate(ctx context.Context, p *incoming.ChannelCreate)
 		return err
 	}
 
-	// Explicitly subscribe to newly created channel
-	s.subs.Add(payload.Uint64toa(ch.ID))
-
-	// @todo this should go over all user's sessions and subscribe there as well
-
-	// @todo load channel member count
-
-	pl := payload.Channel(ch)
-
-	if ch.Type == types.ChannelTypePublic {
-		return s.sendToAll(pl)
-	}
-
-	// By default, just send reply to user
-	return s.sendReply(pl)
+	return nil
 }
 
 func (s *Session) channelDelete(ctx context.Context, p *incoming.ChannelDelete) (err error) {
@@ -116,12 +102,6 @@ func (s *Session) channelUpdate(ctx context.Context, p *incoming.ChannelUpdate)
 		ch.Type = types.ChannelType(*p.Type)
 	}
 
-	ch, err = s.svc.ch.With(ctx).Update(ch)
-	if err != nil {
-		return err
-	}
-
-	// @todo load channel member count
-
-	return s.sendToAllSubscribers(payload.Channel(ch), p.ID)
+	_, err = s.svc.ch.With(ctx).Update(ch)
+	return err
 }