expanding plugin actions

API.md

@@ -91,15 +91,16 @@ The user agent `ua` is expected to follow [RFC 7231 section 5.5.3](http://tools.
 
 #### `{acc}`
 
-Message `{acc}` creats users or updates `tags` or authentication credentials `scheme` and `secret` of exiting users. To create a new user set `user` to the string `new` optionally followed by any character sequence, e.g. `newr15gsr`. Either authenticated or anonymous session can send an `{acc}` message to create a new user. To update tags or authentication credentials of the current user leave `user` unset.
+Message `{acc}` creates users or updates `tags` or authentication credentials `scheme` and `secret` of exiting users. To create a new user set `user` to the string `new` optionally followed by any character sequence, e.g. `newr15gsr`. Either authenticated or anonymous session can send an `{acc}` message to create a new user. To update tags or authentication credentials of the current user leave `user` unset.
 
 ```js
 acc: {
   id: "1a2b3", // string, client-provided message id, optional
   user: "new", // string, "new" to create a new user, default: current user, optional
   scheme: "basic", // authentication scheme for this account, required;
-               // "basic" and "anonymous" are currently supported for account creation. The current
-               // implementation of the basic scheme does not allow changes to username.
+               // "basic" and "anonymous" are currently supported for account creation. The 
+				// current implementation of the basic scheme does not allow changes to 
+				// username.
   secret: btoa("username:password"), // string, base64 encoded secret for the chosen
               // authentication scheme; to delete a scheme use a string with a single DEL
               // Unicode character "\u2421"; "token" and "basic" cannot be deleted

pbx/model.proto

@@ -10,10 +10,10 @@ service Node {
 // Plugin interface.
 service Plugin {
 	// This plugin method is called by Tinode server for every message received from the clients. The 
-	// method returns a ServerCtrl message. ServerCtrl.code is *not* 0 indicates that no further 
+	// method returns a ServerCtrl message. Non-zero ServerCtrl.code indicates that no further 
 	// processing is needed. The Tinode server will generate a {ctrl} message from the returned ServerCtrl 
 	// and forward it to the client session. 
-	// If ServerCtrl.code is 0, the server should continue with default processing of the client message.
+	// ServerCtrl.code equals to 0 instructs the server to continue with default processing of the client message.
   	rpc FireHose(ClientReq) returns (ServerResp) {}
 
 	// The following methods are for the Tinode server to report individual events.
@@ -439,4 +439,5 @@ message SubscriptionEvent {
 
 message MessageEvent {
 	Crud action = 1;
+	ServerData msg = 2;
 }

server/pbconverter.go

@@ -10,85 +10,106 @@ import (
 	"github.com/tinode/chat/pbx"
 )
 
+func pb_serv_ctrl_serialize(ctrl *MsgServerCtrl) *pbx.ServerMsg_Ctrl {
+	var params map[string][]byte
+	if ctrl.Params != nil {
+		if in, ok := ctrl.Params.(map[string]interface{}); ok {
+			params = interfaceMapToByteMap(in)
+		}
+	}
+
+	return &pbx.ServerMsg_Ctrl{Ctrl: &pbx.ServerCtrl{
+		Id:     ctrl.Id,
+		Topic:  ctrl.Topic,
+		Code:   int32(ctrl.Code),
+		Text:   ctrl.Text,
+		Params: params}}
+}
+
+func pb_serv_data_serialize(data *MsgServerData) *pbx.ServerMsg_Data {
+	return &pbx.ServerMsg_Data{Data: &pbx.ServerData{
+		Topic:      data.Topic,
+		FromUserId: data.From,
+		DeletedAt:  timeToInt64(data.DeletedAt),
+		SeqId:      int32(data.SeqId),
+		Head:       data.Head,
+		Content:    interfaceToBytes(data.Content)}}
+}
+
+func pb_serv_pres_serialize(pres *MsgServerPres) *pbx.ServerMsg_Pres {
+	var what pbx.ServerPres_What
+	switch pres.What {
+	case "on":
+		what = pbx.ServerPres_ON
+	case "off":
+		what = pbx.ServerPres_OFF
+	case "ua":
+		what = pbx.ServerPres_UA
+	case "upd":
+		what = pbx.ServerPres_UPD
+	case "gone":
+		what = pbx.ServerPres_GONE
+	case "acs":
+		what = pbx.ServerPres_ACS
+	case "term":
+		what = pbx.ServerPres_TERM
+	case "msg":
+		what = pbx.ServerPres_MSG
+	case "read":
+		what = pbx.ServerPres_READ
+	case "recv":
+		what = pbx.ServerPres_RECV
+	case "del":
+		what = pbx.ServerPres_DEL
+	default:
+		log.Fatal("Unknown pres.what value", pres.What)
+	}
+	return &pbx.ServerMsg_Pres{Pres: &pbx.ServerPres{
+		Topic:        pres.Topic,
+		Src:          pres.Src,
+		What:         what,
+		UserAgent:    pres.UserAgent,
+		SeqId:        int32(pres.SeqId),
+		DelId:        int32(pres.DelId),
+		DelSeq:       pb_DelQuery_serialize(pres.DelSeq),
+		TargetUserId: pres.AcsTarget,
+		ActorUserId:  pres.AcsActor,
+		Acs:          pb_AccessMode_serialize(pres.Acs)}}
+}
+
+func pb_serv_info_serialize(info *MsgServerInfo) *pbx.ServerMsg_Info {
+	return &pbx.ServerMsg_Info{Info: &pbx.ServerInfo{
+		Topic:      info.Topic,
+		FromUserId: info.From,
+		What:       pb_InfoNoteWhat_serialize(info.What),
+		SeqId:      int32(info.SeqId),
+	}}
+}
+
+func pb_serv_meta_serialize(meta *MsgServerMeta) *pbx.ServerMsg_Meta {
+	return &pbx.ServerMsg_Meta{Meta: &pbx.ServerMeta{
+		Id:    meta.Id,
+		Topic: meta.Topic,
+		Desc:  pb_TopicDesc_serialize(meta.Desc),
+		Sub:   pb_TopicSubSlice_serialize(meta.Sub),
+		Del:   pb_DelValues_serialize(meta.Del),
+	}}
+}
+
 // Convert ServerComMessage to pbx.ServerMsg
 func pb_serv_serialize(msg *ServerComMessage) *pbx.ServerMsg {
 	var pkt pbx.ServerMsg
 
 	if msg.Ctrl != nil {
-		var params map[string][]byte
-		if msg.Ctrl.Params != nil {
-			if in, ok := msg.Ctrl.Params.(map[string]interface{}); ok {
-				params = interfaceMapToByteMap(in)
-			}
-		}
-		pkt.Message = &pbx.ServerMsg_Ctrl{Ctrl: &pbx.ServerCtrl{
-			Id:     msg.Ctrl.Id,
-			Topic:  msg.Ctrl.Topic,
-			Code:   int32(msg.Ctrl.Code),
-			Text:   msg.Ctrl.Text,
-			Params: params}}
+		pkt.Message = pb_serv_ctrl_serialize(msg.Ctrl)
 	} else if msg.Data != nil {
-		pkt.Message = &pbx.ServerMsg_Data{Data: &pbx.ServerData{
-			Topic:      msg.Data.Topic,
-			FromUserId: msg.Data.From,
-			DeletedAt:  timeToInt64(msg.Data.DeletedAt),
-			SeqId:      int32(msg.Data.SeqId),
-			Head:       msg.Data.Head,
-			Content:    interfaceToBytes(msg.Data.Content)}}
+		pkt.Message = pb_serv_data_serialize(msg.Data)
 	} else if msg.Pres != nil {
-		var what pbx.ServerPres_What
-		switch msg.Pres.What {
-		case "on":
-			what = pbx.ServerPres_ON
-		case "off":
-			what = pbx.ServerPres_OFF
-		case "ua":
-			what = pbx.ServerPres_UA
-		case "upd":
-			what = pbx.ServerPres_UPD
-		case "gone":
-			what = pbx.ServerPres_GONE
-		case "acs":
-			what = pbx.ServerPres_ACS
-		case "term":
-			what = pbx.ServerPres_TERM
-		case "msg":
-			what = pbx.ServerPres_MSG
-		case "read":
-			what = pbx.ServerPres_READ
-		case "recv":
-			what = pbx.ServerPres_RECV
-		case "del":
-			what = pbx.ServerPres_DEL
-		default:
-			log.Fatal("Unknown pres.what value", msg.Pres.What)
-		}
-		pkt.Message = &pbx.ServerMsg_Pres{Pres: &pbx.ServerPres{
-			Topic:        msg.Pres.Topic,
-			Src:          msg.Pres.Src,
-			What:         what,
-			UserAgent:    msg.Pres.UserAgent,
-			SeqId:        int32(msg.Pres.SeqId),
-			DelId:        int32(msg.Pres.DelId),
-			DelSeq:       pb_DelQuery_serialize(msg.Pres.DelSeq),
-			TargetUserId: msg.Pres.AcsTarget,
-			ActorUserId:  msg.Pres.AcsActor,
-			Acs:          pb_AccessMode_serialize(msg.Pres.Acs)}}
+		pkt.Message = pb_serv_pres_serialize(msg.Pres)
 	} else if msg.Info != nil {
-		pkt.Message = &pbx.ServerMsg_Info{Info: &pbx.ServerInfo{
-			Topic:      msg.Info.Topic,
-			FromUserId: msg.Info.From,
-			What:       pb_InfoNoteWhat_serialize(msg.Info.What),
-			SeqId:      int32(msg.Info.SeqId),
-		}}
+		pkt.Message = pb_serv_info_serialize(msg.Info)
 	} else if msg.Meta != nil {
-		pkt.Message = &pbx.ServerMsg_Meta{Meta: &pbx.ServerMeta{
-			Id:    msg.Meta.Id,
-			Topic: msg.Meta.Topic,
-			Desc:  pb_TopicDesc_serialize(msg.Meta.Desc),
-			Sub:   pb_TopicSubSlice_serialize(msg.Meta.Sub),
-			Del:   pb_DelValues_serialize(msg.Meta.Del),
-		}}
+		pkt.Message = pb_serv_meta_serialize(msg.Meta)
 	}
 
 	return &pkt
@@ -631,6 +652,22 @@ func pb_TopicDesc_deserialize(desc *pbx.TopicDesc) *MsgTopicDesc {
 	}
 }
 
+func pb_Topic_serialize(topic *Topic) *pbx.TopicDesc {
+	if topic == nil {
+		return nil
+	}
+	return &pbx.TopicDesc{
+		CreatedAt: timeToInt64(&topic.created),
+		UpdatedAt: timeToInt64(&topic.updated),
+		Defacs: &pbx.DefaultAcsMode{
+			Auth: topic.accessAuth.String(),
+			Anon: topic.accessAnon.String()},
+		SeqId:  int32(topic.lastId),
+		DelId:  int32(topic.delId),
+		Public: interfaceToBytes(topic.public),
+	}
+}
+
 func pb_TopicSubSlice_serialize(subs []MsgTopicSub) []*pbx.TopicSub {
 	if subs == nil || len(subs) == 0 {
 		return nil

server/plugins.go

@@ -334,8 +334,7 @@ func pluginFireHose(sess *Session, msg *ClientComMessage) (*ClientComMessage, *S
 
 	var req *pbx.ClientReq
 
-	var id string
-	var topic string
+	id, topic := pluginIdAndTopic(msg)
 	ts := time.Now().UTC().Round(time.Millisecond)
 	for _, p := range plugins {
 		if !pluginDoFiltering(p.filterFireHose, msg) {
@@ -429,16 +428,77 @@ func pluginAccount(user *types.User, action int) {
 	}
 }
 
-func pluginTopic(msg *ServerComMessage) *ServerComMessage {
-	return nil
+func pluginTopic(topic *Topic, action int) {
+	if plugins == nil {
+		return
+	}
+
+	var event *pbx.TopicEvent
+	for _, p := range plugins {
+		if p.filterTopic == nil || p.filterTopic.byAction&action == 0 {
+			// Plugin is not interested in Message actions
+			continue
+		}
+
+		if event == nil {
+			event = &pbx.TopicEvent{
+				Action: pluginActionToCrud(action),
+				Name:   topic.name,
+				Desc:   pb_Topic_serialize(topic),
+			}
+		}
+
+		var ctx context.Context
+		var cancel context.CancelFunc
+		if p.timeout > 0 {
+			ctx, cancel = context.WithTimeout(context.Background(), p.timeout)
+			defer cancel()
+		} else {
+			ctx = context.Background()
+		}
+		if _, err := p.client.Topic(ctx, event); err != nil {
+			log.Println("plugins: Topic call failed", p.name, err)
+		}
+	}
+
+	return
 }
 
 func pluginSubscription(msg *ServerComMessage) *ServerComMessage {
 	return nil
 }
 
-func pluginMessage(msg *ServerComMessage) *ServerComMessage {
-	return nil
+func pluginMessage(data *MsgServerData, action int) {
+	if plugins == nil || action != plgActCreate {
+		return
+	}
+
+	var event *pbx.MessageEvent
+	for _, p := range plugins {
+		if p.filterMessage == nil || p.filterMessage.byAction&action == 0 {
+			// Plugin is not interested in Message actions
+			continue
+		}
+
+		if event == nil {
+			event = &pbx.MessageEvent{
+				Action: pluginActionToCrud(action),
+				Msg:    pb_serv_data_serialize(data).Data,
+			}
+		}
+
+		var ctx context.Context
+		var cancel context.CancelFunc
+		if p.timeout > 0 {
+			ctx, cancel = context.WithTimeout(context.Background(), p.timeout)
+			defer cancel()
+		} else {
+			ctx = context.Background()
+		}
+		if _, err := p.client.Message(ctx, event); err != nil {
+			log.Println("plugins: Message call failed", p.name, err)
+		}
+	}
 }
 
 // Returns false to skip, true to process
@@ -515,3 +575,37 @@ func pluginActionToCrud(action int) pbx.Crud {
 	}
 	panic("plugin: unknown action")
 }
+
+func pluginIdAndTopic(msg *ClientComMessage) (string, string) {
+	if msg.Hi != nil {
+		return msg.Hi.Id, ""
+	}
+	if msg.Acc != nil {
+		return msg.Acc.Id, ""
+	}
+	if msg.Login != nil {
+		return msg.Login.Id, ""
+	}
+	if msg.Sub != nil {
+		return msg.Sub.Id, msg.Sub.Topic
+	}
+	if msg.Leave != nil {
+		return msg.Leave.Id, msg.Leave.Topic
+	}
+	if msg.Pub != nil {
+		return msg.Pub.Id, msg.Pub.Topic
+	}
+	if msg.Get != nil {
+		return msg.Get.Id, msg.Get.Topic
+	}
+	if msg.Set != nil {
+		return msg.Set.Id, msg.Set.Topic
+	}
+	if msg.Del != nil {
+		return msg.Del.Id, msg.Del.Topic
+	}
+	if msg.Note != nil {
+		return "", msg.Note.Topic
+	}
+	panic("plugin: unknown message")
+}

server/session.go

@@ -674,6 +674,8 @@ func (s *Session) acc(msg *ClientComMessage) {
 
 		s.queueOut(NoErr(msg.Acc.Id, "", msg.timestamp))
 
+		// pluginAccount(&user, plgActCreate)
+
 	} else {
 		// session is not authenticated and this is not an attempt to create a new account
 		s.queueOut(ErrPermissionDenied(msg.Acc.Id, "", msg.timestamp))