Notifications phase 1 (#1851)

Backend support for subscribing/unsubscribing to blocks, typically cards and boards. Notifies subscribers when changes are made to cards they are subscribed to.

server/ws/adapter.go

@@ -12,10 +12,12 @@ const (
 	websocketActionUnsubscribeBlocks    = "UNSUBSCRIBE_BLOCKS"
 	websocketActionUpdateBlock          = "UPDATE_BLOCK"
 	websocketActionUpdateConfig         = "UPDATE_CLIENT_CONFIG"
+	websocketActionUpdateSubscription   = "UPDATE_SUBSCRIPTION"
 )
 
 type Adapter interface {
 	BroadcastBlockChange(workspaceID string, block model.Block)
 	BroadcastBlockDelete(workspaceID, blockID, parentID string)
 	BroadcastConfigChange(clientConfig model.ClientConfig)
+	BroadcastSubscriptionChange(workspaceID string, subscription *model.Subscription)
 }

server/ws/common.go

@@ -10,6 +10,12 @@ type UpdateMsg struct {
 	Block  model.Block `json:"block"`
 }
 
+// UpdateSubscription is sent on subscription updates.
+type UpdateSubscription struct {
+	Action       string              `json:"action"`
+	Subscription *model.Subscription `json:"subscription"`
+}
+
 // WebsocketCommand is an incoming command from the client.
 type WebsocketCommand struct {
 	Action      string   `json:"action"`

server/ws/plugin_adapter.go

@@ -27,6 +27,7 @@ type PluginAdapterInterface interface {
 	BroadcastConfigChange(clientConfig model.ClientConfig)
 	BroadcastBlockChange(workspaceID string, block model.Block)
 	BroadcastBlockDelete(workspaceID, blockID, parentID string)
+	BroadcastSubscriptionChange(workspaceID string, subscription *model.Subscription)
 	HandleClusterEvent(ev mmModel.PluginClusterEvent)
 }
 
@@ -338,16 +339,16 @@ func (pa *PluginAdapter) BroadcastConfigChange(pluginConfig model.ClientConfig)
 
 // sendWorkspaceMessageSkipCluster sends a message to all the users
 // with a websocket client connected to.
-func (pa *PluginAdapter) sendWorkspaceMessageSkipCluster(workspaceID string, payload map[string]interface{}) {
+func (pa *PluginAdapter) sendWorkspaceMessageSkipCluster(event string, workspaceID string, payload map[string]interface{}) {
 	userIDs := pa.getUserIDsForWorkspace(workspaceID)
 	for _, userID := range userIDs {
-		pa.api.PublishWebSocketEvent(websocketActionUpdateBlock, payload, &mmModel.WebsocketBroadcast{UserId: userID})
+		pa.api.PublishWebSocketEvent(event, payload, &mmModel.WebsocketBroadcast{UserId: userID})
 	}
 }
 
 // sendWorkspaceMessage sends and propagates a message that is aimed
 // for all the users that are subscribed to a given workspace.
-func (pa *PluginAdapter) sendWorkspaceMessage(workspaceID string, payload map[string]interface{}) {
+func (pa *PluginAdapter) sendWorkspaceMessage(event string, workspaceID string, payload map[string]interface{}) {
 	go func() {
 		clusterMessage := &ClusterMessage{
 			WorkspaceID: workspaceID,
@@ -357,7 +358,7 @@ func (pa *PluginAdapter) sendWorkspaceMessage(workspaceID string, payload map[st
 		pa.sendMessageToCluster("websocket_message", clusterMessage)
 	}()
 
-	pa.sendWorkspaceMessageSkipCluster(workspaceID, payload)
+	pa.sendWorkspaceMessageSkipCluster(event, workspaceID, payload)
 }
 
 func (pa *PluginAdapter) BroadcastBlockChange(workspaceID string, block model.Block) {
@@ -371,7 +372,7 @@ func (pa *PluginAdapter) BroadcastBlockChange(workspaceID string, block model.Bl
 		Block:  block,
 	}
 
-	pa.sendWorkspaceMessage(workspaceID, utils.StructToMap(message))
+	pa.sendWorkspaceMessage(websocketActionUpdateBlock, workspaceID, utils.StructToMap(message))
 }
 
 func (pa *PluginAdapter) BroadcastBlockDelete(workspaceID, blockID, parentID string) {
@@ -385,3 +386,18 @@ func (pa *PluginAdapter) BroadcastBlockDelete(workspaceID, blockID, parentID str
 
 	pa.BroadcastBlockChange(workspaceID, block)
 }
+
+func (pa *PluginAdapter) BroadcastSubscriptionChange(workspaceID string, subscription *model.Subscription) {
+	pa.api.LogInfo("BroadcastingSubscriptionChange",
+		"workspaceID", workspaceID,
+		"blockID", subscription.BlockID,
+		"subscriberID", subscription.SubscriberID,
+	)
+
+	message := UpdateSubscription{
+		Action:       websocketActionUpdateSubscription,
+		Subscription: subscription,
+	}
+
+	pa.sendWorkspaceMessage(websocketActionUpdateSubscription, workspaceID, utils.StructToMap(message))
+}

server/ws/plugin_adapter_cluster.go

@@ -46,5 +46,20 @@ func (pa *PluginAdapter) HandleClusterEvent(ev mmModel.PluginClusterEvent) {
 		return
 	}
 
-	pa.sendWorkspaceMessageSkipCluster(clusterMessage.WorkspaceID, clusterMessage.Payload)
+	var action string
+	if actionRaw, ok := clusterMessage.Payload["action"]; ok {
+		if s, ok := actionRaw.(string); ok {
+			action = s
+		}
+	}
+	if action == "" {
+		// no action was specified in the event; assume block change and warn.
+		action = websocketActionUpdateBlock
+		pa.api.LogWarn("cannot determine action from cluster message data",
+			"id", ev.Id,
+			"payload", clusterMessage.Payload,
+		)
+	}
+
+	pa.sendWorkspaceMessageSkipCluster(action, clusterMessage.WorkspaceID, clusterMessage.Payload)
 }

server/ws/server.go

@@ -536,3 +536,7 @@ func (ws *Server) BroadcastConfigChange(clientConfig model.ClientConfig) {
 		}
 	}
 }
+
+func (ws *Server) BroadcastSubscriptionChange(workspaceID string, subscription *model.Subscription) {
+	// not implemented for standalone server.
+}