From 2d1135ca46038f9615464f2544ad4e5980e9d77c Mon Sep 17 00:00:00 2001 From: Doug Lauder Date: Fri, 22 Dec 2023 17:00:27 -0500 Subject: [PATCH] Shared channels plugin APIs for MS Teams plugin (#25805) New plugin APIs and hooks for accessing Shared Channels service via plugin. - RegisterPluginForSharedChannels(opts model.RegisterPluginOpts) (remoteID string, err error) - UnregisterPluginForSharedChannels(pluginID string) error - ShareChannel(sc *model.SharedChannel) (*model.SharedChannel, error) - UpdateSharedChannel(sc *model.SharedChannel) (*model.SharedChannel, error) - UnshareChannel(channelID string) (unshared bool, err error) - UpdateSharedChannelCursor(channelID, remoteID string, cusror model.GetPostsSinceForSyncCursor) error - SyncSharedChannel(channelID string) error - InviteRemoteToChannel(channelID string, remoteID string, userID string) error - UninviteRemoteFromChannel(channelID string, remoteID string) error Hooks - OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) - OnSharedChannelsPing(rc *model.RemoteCluster) bool --- server/channels/api4/shared_channel_test.go | 4 +- server/channels/app/app_iface.go | 22 +- server/channels/app/channel.go | 2 +- server/channels/app/helper_test.go | 2 +- .../app/opentracing/opentracing_layer.go | 259 ++++++++++--- server/channels/app/plugin_api.go | 36 ++ server/channels/app/remote_cluster.go | 52 +++ server/channels/app/server.go | 5 +- server/channels/app/shared_channel.go | 181 +++++++++- server/channels/app/shared_channel_test.go | 2 +- .../app/slashcommands/command_share.go | 54 +-- .../app/slashcommands/command_share_test.go | 22 +- .../channels/app/slashcommands/helper_test.go | 2 +- .../opentracinglayer/opentracinglayer.go | 18 + .../channels/store/retrylayer/retrylayer.go | 21 ++ .../store/sqlstore/remote_cluster_store.go | 33 ++ server/channels/store/store.go | 1 + .../storetest/mocks/RemoteClusterStore.go | 26 ++ .../store/storetest/remote_cluster_store.go | 103 ++++-- .../channels/store/timerlayer/timerlayer.go | 16 + server/i18n/en.json | 6 +- .../services/remotecluster/mocks_test.go | 57 +++ .../platform/services/remotecluster/ping.go | 51 ++- .../services/remotecluster/ping_test.go | 39 +- .../services/remotecluster/send_test.go | 15 +- .../remotecluster/sendprofileImage_test.go | 7 +- .../services/remotecluster/service.go | 25 +- .../services/remotecluster/service_test.go | 5 +- .../sharedchannel/mock_AppIface_test.go | 24 ++ .../services/sharedchannel/service.go | 5 +- .../sharedchannel/sync_send_remote.go | 17 +- server/public/model/shared_channel.go | 18 + server/public/plugin/api.go | 69 +++- .../plugin/api_timer_layer_generated.go | 63 ++++ server/public/plugin/client_rpc_generated.go | 341 ++++++++++++++++++ server/public/plugin/hooks.go | 24 ++ .../plugin/hooks_timer_layer_generated.go | 14 + server/public/plugin/plugintest/api.go | 170 +++++++++ server/public/plugin/plugintest/hooks.go | 38 ++ .../public/plugin/product_hooks_generated.go | 44 +++ 40 files changed, 1725 insertions(+), 168 deletions(-) diff --git a/server/channels/api4/shared_channel_test.go b/server/channels/api4/shared_channel_test.go index 3ad1886d5a..c955b29a79 100644 --- a/server/channels/api4/shared_channel_test.go +++ b/server/channels/api4/shared_channel_test.go @@ -45,7 +45,7 @@ func TestGetAllSharedChannels(t *testing.T) { CreatorId: th.BasicChannel.CreatorId, RemoteId: model.NewId(), } - _, err := th.App.SaveSharedChannel(th.Context, sc) + _, err := th.App.ShareChannel(th.Context, sc) require.NoError(t, err) savedIds = append(savedIds, channel.Id) } @@ -125,7 +125,7 @@ func TestGetRemoteClusterById(t *testing.T) { CreatorId: th.BasicChannel.CreatorId, RemoteId: rc.RemoteId, } - sc, err := th.App.SaveSharedChannel(th.Context, sc) + sc, err := th.App.ShareChannel(th.Context, sc) require.NoError(t, err) // create a shared channel remote to connect them diff --git a/server/channels/app/app_iface.go b/server/channels/app/app_iface.go index 83d4ffa30a..f6cf66c541 100644 --- a/server/channels/app/app_iface.go +++ b/server/channels/app/app_iface.go @@ -269,6 +269,12 @@ type AppIface interface { MoveChannel(c request.CTX, team *model.Team, channel *model.Channel, user *model.User) *model.AppError // NotifySessionsExpired is called periodically from the job server to notify any mobile sessions that have expired. NotifySessionsExpired() error + // OnSharedChannelsPing is called by the Shared Channels service for a registered plugin wto check that the plugin + // is still responding and has a connection to any upstream services it needs (e.g. MS Graph API). + OnSharedChannelsPing(rc *model.RemoteCluster) bool + // OnSharedChannelsSyncMsg is called by the Shared Channels service for a registered plugin when there is new content + // that needs to be synchronized. + OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) // OverrideIconURLIfEmoji changes the post icon override URL prop, if it has an emoji icon, // so that it points to the URL (relative) of the emoji - static if emoji is default, /api if custom. OverrideIconURLIfEmoji(c request.CTX, post *model.Post) @@ -339,6 +345,8 @@ type AppIface interface { // SyncRolesAndMembership updates the SchemeAdmin status and membership of all of the members of the given // syncable. SyncRolesAndMembership(c request.CTX, syncableID string, syncableType model.GroupSyncableType, includeRemovedMembers bool) + // SyncSharedChannel forces a shared channel to send any changed content to all remote clusters. + SyncSharedChannel(channelID string) error // SyncSyncableRoles updates the SchemeAdmin field value of the given syncable's members based on the configuration of // the member's group memberships and the configuration of those groups to the syncable. This method should only // be invoked on group-synced (aka group-constrained) syncables. @@ -376,6 +384,12 @@ type AppIface interface { UpdateDNDStatusOfUsers() // UpdateProductNotices is called periodically from a scheduled worker to fetch new notices and update the cache UpdateProductNotices() *model.AppError + // UpdateSharedChannelCursor updates the cursor for the specified channelID and remoteID. + // This can be used to manually set the point of last sync, either forward to skip older posts, + // or backward to re-sync history. + // This call by itself does not force a re-sync - a change to channel contents or a call to + // SyncSharedChannel are needed to force a sync. + UpdateSharedChannelCursor(channelID, remoteID string, cursor model.GetPostsSinceForSyncCursor) error // UpdateViewedProductNotices is called from the frontend to mark a set of notices as 'viewed' by user UpdateViewedProductNotices(userID string, noticeIds []string) *model.AppError // UpdateViewedProductNoticesForNewUser is called when new user is created to mark all current notices for this @@ -546,7 +560,6 @@ type AppIface interface { DeleteRemoteCluster(remoteClusterId string) (bool, *model.AppError) DeleteRetentionPolicy(policyID string) *model.AppError DeleteScheme(schemeId string) (*model.Scheme, *model.AppError) - DeleteSharedChannel(channelID string) (bool, error) DeleteSharedChannelRemote(id string) (bool, error) DeleteSidebarCategory(c request.CTX, userID, teamID, categoryId string) *model.AppError DeleteToken(token *model.Token) *model.AppError @@ -888,6 +901,7 @@ type AppIface interface { InviteGuestsToChannelsGracefully(teamID string, guestsInvite *model.GuestsInvite, senderId string) ([]*model.EmailInviteWithError, *model.AppError) InviteNewUsersToTeam(emailList []string, teamID, senderId string) *model.AppError InviteNewUsersToTeamGracefully(memberInvite *model.MemberInvite, teamID, senderId string, reminderInterval string) ([]*model.EmailInviteWithError, *model.AppError) + InviteRemoteToChannel(channelID, remoteID, userID string) error IsCRTEnabledForUser(c request.CTX, userID string) bool IsConfigReadOnly() bool IsFirstUserAccount() bool @@ -968,6 +982,7 @@ type AppIface interface { RegenerateOAuthAppSecret(app *model.OAuthApp) (*model.OAuthApp, *model.AppError) RegenerateTeamInviteId(teamID string) (*model.Team, *model.AppError) RegisterPluginCommand(pluginID string, command *model.Command) error + RegisterPluginForSharedChannels(opts model.RegisterPluginOpts) (remoteID string, err error) ReloadConfig() error RemoveAllDeactivatedMembersFromChannel(c request.CTX, channel *model.Channel) *model.AppError RemoveChannelsFromRetentionPolicy(policyID string, channelIDs []string) *model.AppError @@ -1016,7 +1031,6 @@ type AppIface interface { SaveBrandImage(rctx request.CTX, imageData *multipart.FileHeader) *model.AppError SaveComplianceReport(rctx request.CTX, job *model.Compliance) (*model.Compliance, *model.AppError) SaveReactionForPost(c request.CTX, reaction *model.Reaction) (*model.Reaction, *model.AppError) - SaveSharedChannel(c request.CTX, sc *model.SharedChannel) (*model.SharedChannel, error) SaveSharedChannelRemote(remote *model.SharedChannelRemote) (*model.SharedChannelRemote, error) SaveUserTermsOfService(userID, termsOfServiceId string, accepted bool) *model.AppError SchemesIterator(scope string, batchSize int) func() []*model.Scheme @@ -1092,6 +1106,7 @@ type AppIface interface { SetTeamIcon(teamID string, imageData *multipart.FileHeader) *model.AppError SetTeamIconFromFile(team *model.Team, file io.Reader) *model.AppError SetTeamIconFromMultiPartFile(teamID string, file multipart.File) *model.AppError + ShareChannel(c request.CTX, sc *model.SharedChannel) (*model.SharedChannel, error) SlackImport(c request.CTX, fileData multipart.File, fileSize int64, teamID string) (*model.AppError, *bytes.Buffer) SoftDeleteTeam(teamID string) *model.AppError Srv() *Server @@ -1112,7 +1127,10 @@ type AppIface interface { ToggleMuteChannel(c request.CTX, channelID, userID string) (*model.ChannelMember, *model.AppError) TotalWebsocketConnections() int TriggerWebhook(c request.CTX, payload *model.OutgoingWebhookPayload, hook *model.OutgoingWebhook, post *model.Post, channel *model.Channel) + UninviteRemoteFromChannel(channelID, remoteID string) error UnregisterPluginCommand(pluginID, teamID, trigger string) + UnregisterPluginForSharedChannels(pluginID string) error + UnshareChannel(channelID string) (bool, error) UpdateActive(c request.CTX, user *model.User, active bool) (*model.User, *model.AppError) UpdateChannelMemberNotifyProps(c request.CTX, data map[string]string, channelID string, userID string) (*model.ChannelMember, *model.AppError) UpdateChannelMemberRoles(c request.CTX, channelID string, userID string, newRoles string) (*model.ChannelMember, *model.AppError) diff --git a/server/channels/app/channel.go b/server/channels/app/channel.go index 8d2856197e..930a26489b 100644 --- a/server/channels/app/channel.go +++ b/server/channels/app/channel.go @@ -560,7 +560,7 @@ func (a *App) createDirectChannelWithUser(c request.CTX, user, otherUser *model. Type: channel.Type, } - if _, err := a.SaveSharedChannel(c, sc); err != nil { + if _, err := a.ShareChannel(c, sc); err != nil { return nil, model.NewAppError("CreateDirectChannel", "app.sharedchannel.dm_channel_creation.internal_error", nil, "", http.StatusInternalServerError).Wrap(err) } } diff --git a/server/channels/app/helper_test.go b/server/channels/app/helper_test.go index a4904ea86a..69cc543679 100644 --- a/server/channels/app/helper_test.go +++ b/server/channels/app/helper_test.go @@ -374,7 +374,7 @@ func (th *TestHelper) createChannel(c request.CTX, team *model.Team, channelType if channel.IsShared() { id := model.NewId() - _, err := th.App.SaveSharedChannel(c, &model.SharedChannel{ + _, err := th.App.ShareChannel(c, &model.SharedChannel{ ChannelId: channel.Id, TeamId: channel.TeamId, Home: false, diff --git a/server/channels/app/opentracing/opentracing_layer.go b/server/channels/app/opentracing/opentracing_layer.go index e5800cf40a..6980d0310a 100644 --- a/server/channels/app/opentracing/opentracing_layer.go +++ b/server/channels/app/opentracing/opentracing_layer.go @@ -3526,28 +3526,6 @@ func (a *OpenTracingAppLayer) DeleteScheme(schemeId string) (*model.Scheme, *mod return resultVar0, resultVar1 } -func (a *OpenTracingAppLayer) DeleteSharedChannel(channelID string) (bool, error) { - origCtx := a.ctx - span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.DeleteSharedChannel") - - a.ctx = newCtx - a.app.Srv().Store().SetContext(newCtx) - defer func() { - a.app.Srv().Store().SetContext(origCtx) - a.ctx = origCtx - }() - - defer span.Finish() - resultVar0, resultVar1 := a.app.DeleteSharedChannel(channelID) - - if resultVar1 != nil { - span.LogFields(spanlog.Error(resultVar1)) - ext.Error.Set(span, true) - } - - return resultVar0, resultVar1 -} - func (a *OpenTracingAppLayer) DeleteSharedChannelRemote(id string) (bool, error) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.DeleteSharedChannelRemote") @@ -11883,6 +11861,28 @@ func (a *OpenTracingAppLayer) InviteNewUsersToTeamGracefully(memberInvite *model return resultVar0, resultVar1 } +func (a *OpenTracingAppLayer) InviteRemoteToChannel(channelID string, remoteID string, userID string) error { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.InviteRemoteToChannel") + + a.ctx = newCtx + a.app.Srv().Store().SetContext(newCtx) + defer func() { + a.app.Srv().Store().SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + resultVar0 := a.app.InviteRemoteToChannel(channelID, remoteID, userID) + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + func (a *OpenTracingAppLayer) IsCRTEnabledForUser(c request.CTX, userID string) bool { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.IsCRTEnabledForUser") @@ -12799,6 +12799,45 @@ func (a *OpenTracingAppLayer) NotifySharedChannelUserUpdate(user *model.User) { a.app.NotifySharedChannelUserUpdate(user) } +func (a *OpenTracingAppLayer) OnSharedChannelsPing(rc *model.RemoteCluster) bool { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.OnSharedChannelsPing") + + a.ctx = newCtx + a.app.Srv().Store().SetContext(newCtx) + defer func() { + a.app.Srv().Store().SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + resultVar0 := a.app.OnSharedChannelsPing(rc) + + return resultVar0 +} + +func (a *OpenTracingAppLayer) OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.OnSharedChannelsSyncMsg") + + a.ctx = newCtx + a.app.Srv().Store().SetContext(newCtx) + defer func() { + a.app.Srv().Store().SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + resultVar0, resultVar1 := a.app.OnSharedChannelsSyncMsg(msg, rc) + + if resultVar1 != nil { + span.LogFields(spanlog.Error(resultVar1)) + ext.Error.Set(span, true) + } + + return resultVar0, resultVar1 +} + func (a *OpenTracingAppLayer) OpenInteractiveDialog(request model.OpenDialogRequest) *model.AppError { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.OpenInteractiveDialog") @@ -13718,6 +13757,28 @@ func (a *OpenTracingAppLayer) RegisterPluginCommand(pluginID string, command *mo return resultVar0 } +func (a *OpenTracingAppLayer) RegisterPluginForSharedChannels(opts model.RegisterPluginOpts) (remoteID string, err error) { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.RegisterPluginForSharedChannels") + + a.ctx = newCtx + a.app.Srv().Store().SetContext(newCtx) + defer func() { + a.app.Srv().Store().SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + resultVar0, resultVar1 := a.app.RegisterPluginForSharedChannels(opts) + + if resultVar1 != nil { + span.LogFields(spanlog.Error(resultVar1)) + ext.Error.Set(span, true) + } + + return resultVar0, resultVar1 +} + func (a *OpenTracingAppLayer) ReloadConfig() error { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.ReloadConfig") @@ -14848,28 +14909,6 @@ func (a *OpenTracingAppLayer) SaveReactionForPost(c request.CTX, reaction *model return resultVar0, resultVar1 } -func (a *OpenTracingAppLayer) SaveSharedChannel(c request.CTX, sc *model.SharedChannel) (*model.SharedChannel, error) { - origCtx := a.ctx - span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SaveSharedChannel") - - a.ctx = newCtx - a.app.Srv().Store().SetContext(newCtx) - defer func() { - a.app.Srv().Store().SetContext(origCtx) - a.ctx = origCtx - }() - - defer span.Finish() - resultVar0, resultVar1 := a.app.SaveSharedChannel(c, sc) - - if resultVar1 != nil { - span.LogFields(spanlog.Error(resultVar1)) - ext.Error.Set(span, true) - } - - return resultVar0, resultVar1 -} - func (a *OpenTracingAppLayer) SaveSharedChannelRemote(remote *model.SharedChannelRemote) (*model.SharedChannelRemote, error) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SaveSharedChannelRemote") @@ -16566,6 +16605,28 @@ func (a *OpenTracingAppLayer) SetTeamIconFromMultiPartFile(teamID string, file m return resultVar0 } +func (a *OpenTracingAppLayer) ShareChannel(c request.CTX, sc *model.SharedChannel) (*model.SharedChannel, error) { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.ShareChannel") + + a.ctx = newCtx + a.app.Srv().Store().SetContext(newCtx) + defer func() { + a.app.Srv().Store().SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + resultVar0, resultVar1 := a.app.ShareChannel(c, sc) + + if resultVar1 != nil { + span.LogFields(spanlog.Error(resultVar1)) + ext.Error.Set(span, true) + } + + return resultVar0, resultVar1 +} + func (a *OpenTracingAppLayer) SlackImport(c request.CTX, fileData multipart.File, fileSize int64, teamID string) (*model.AppError, *bytes.Buffer) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SlackImport") @@ -16772,6 +16833,28 @@ func (a *OpenTracingAppLayer) SyncRolesAndMembership(c request.CTX, syncableID s a.app.SyncRolesAndMembership(c, syncableID, syncableType, includeRemovedMembers) } +func (a *OpenTracingAppLayer) SyncSharedChannel(channelID string) error { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SyncSharedChannel") + + a.ctx = newCtx + a.app.Srv().Store().SetContext(newCtx) + defer func() { + a.app.Srv().Store().SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + resultVar0 := a.app.SyncSharedChannel(channelID) + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + func (a *OpenTracingAppLayer) SyncSyncableRoles(syncableID string, syncableType model.GroupSyncableType) *model.AppError { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SyncSyncableRoles") @@ -17063,6 +17146,28 @@ func (a *OpenTracingAppLayer) TriggerWebhook(c request.CTX, payload *model.Outgo a.app.TriggerWebhook(c, payload, hook, post, channel) } +func (a *OpenTracingAppLayer) UninviteRemoteFromChannel(channelID string, remoteID string) error { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UninviteRemoteFromChannel") + + a.ctx = newCtx + a.app.Srv().Store().SetContext(newCtx) + defer func() { + a.app.Srv().Store().SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + resultVar0 := a.app.UninviteRemoteFromChannel(channelID, remoteID) + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + func (a *OpenTracingAppLayer) UnregisterPluginCommand(pluginID string, teamID string, trigger string) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UnregisterPluginCommand") @@ -17078,6 +17183,50 @@ func (a *OpenTracingAppLayer) UnregisterPluginCommand(pluginID string, teamID st a.app.UnregisterPluginCommand(pluginID, teamID, trigger) } +func (a *OpenTracingAppLayer) UnregisterPluginForSharedChannels(pluginID string) error { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UnregisterPluginForSharedChannels") + + a.ctx = newCtx + a.app.Srv().Store().SetContext(newCtx) + defer func() { + a.app.Srv().Store().SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + resultVar0 := a.app.UnregisterPluginForSharedChannels(pluginID) + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + +func (a *OpenTracingAppLayer) UnshareChannel(channelID string) (bool, error) { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UnshareChannel") + + a.ctx = newCtx + a.app.Srv().Store().SetContext(newCtx) + defer func() { + a.app.Srv().Store().SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + resultVar0, resultVar1 := a.app.UnshareChannel(channelID) + + if resultVar1 != nil { + span.LogFields(spanlog.Error(resultVar1)) + ext.Error.Set(span, true) + } + + return resultVar0, resultVar1 +} + func (a *OpenTracingAppLayer) UpdateActive(c request.CTX, user *model.User, active bool) (*model.User, *model.AppError) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UpdateActive") @@ -17866,6 +18015,28 @@ func (a *OpenTracingAppLayer) UpdateSharedChannel(sc *model.SharedChannel) (*mod return resultVar0, resultVar1 } +func (a *OpenTracingAppLayer) UpdateSharedChannelCursor(channelID string, remoteID string, cursor model.GetPostsSinceForSyncCursor) error { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UpdateSharedChannelCursor") + + a.ctx = newCtx + a.app.Srv().Store().SetContext(newCtx) + defer func() { + a.app.Srv().Store().SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + resultVar0 := a.app.UpdateSharedChannelCursor(channelID, remoteID, cursor) + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + func (a *OpenTracingAppLayer) UpdateSharedChannelRemoteCursor(id string, cursor model.GetPostsSinceForSyncCursor) error { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UpdateSharedChannelRemoteCursor") diff --git a/server/channels/app/plugin_api.go b/server/channels/app/plugin_api.go index 24f2bb8d8d..dcae0dccb3 100644 --- a/server/channels/app/plugin_api.go +++ b/server/channels/app/plugin_api.go @@ -1279,3 +1279,39 @@ func (api *PluginAPI) SendPushNotification(notification *model.PushNotification, // Ignoring skipSessionId because it's only used internally to clear push notifications return api.app.sendPushNotificationToAllSessions(notification, userID, "") } + +func (api *PluginAPI) RegisterPluginForSharedChannels(opts model.RegisterPluginOpts) (remoteID string, err error) { + return api.app.RegisterPluginForSharedChannels(opts) +} + +func (api *PluginAPI) UnregisterPluginForSharedChannels(pluginID string) error { + return api.app.UnregisterPluginForSharedChannels(pluginID) +} + +func (api *PluginAPI) ShareChannel(sc *model.SharedChannel) (*model.SharedChannel, error) { + return api.app.ShareChannel(api.ctx, sc) +} + +func (api *PluginAPI) UpdateSharedChannel(sc *model.SharedChannel) (*model.SharedChannel, error) { + return api.app.UpdateSharedChannel(sc) +} + +func (api *PluginAPI) UnshareChannel(channelID string) (unshared bool, err error) { + return api.app.UnshareChannel(channelID) +} + +func (api *PluginAPI) UpdateSharedChannelCursor(channelID, remoteID string, cusror model.GetPostsSinceForSyncCursor) error { + return api.app.UpdateSharedChannelCursor(channelID, remoteID, cusror) +} + +func (api *PluginAPI) SyncSharedChannel(channelID string) error { + return api.app.SyncSharedChannel(channelID) +} + +func (api *PluginAPI) InviteRemoteToChannel(channelID string, remoteID, userID string) error { + return api.app.InviteRemoteToChannel(channelID, remoteID, userID) +} + +func (api *PluginAPI) UninviteRemoteFromChannel(channelID string, remoteID string) error { + return api.app.UninviteRemoteFromChannel(channelID, remoteID) +} diff --git a/server/channels/app/remote_cluster.go b/server/channels/app/remote_cluster.go index 8114dbd9e0..ec1fc75956 100644 --- a/server/channels/app/remote_cluster.go +++ b/server/channels/app/remote_cluster.go @@ -4,6 +4,7 @@ package app import ( + "database/sql" "net/http" "github.com/pkg/errors" @@ -14,6 +15,57 @@ import ( "github.com/mattermost/mattermost/server/public/model" ) +func (a *App) RegisterPluginForSharedChannels(opts model.RegisterPluginOpts) (remoteID string, err error) { + // check for pluginID already registered + rc, err := a.Srv().Store().RemoteCluster().GetByPluginID(opts.PluginID) + if err != nil { + if !errors.Is(err, sql.ErrNoRows) { + // anything other than not_found is unrecoverable + return "", err + } + } + + // if plugin is already registered then treat this as an update. + if rc != nil { + rc.DisplayName = opts.Displayname + rc.Options = opts.GetOptionFlags() + + if _, err = a.Srv().Store().RemoteCluster().Update(rc); err != nil { + return "", err + } + return rc.RemoteId, nil + } + + rc = &model.RemoteCluster{ + Name: opts.Displayname, + DisplayName: opts.Displayname, + Token: model.NewId(), + CreatorId: opts.CreatorID, + PluginID: opts.PluginID, + Options: opts.GetOptionFlags(), + } + + rcSaved, err := a.Srv().Store().RemoteCluster().Save(rc) + if err != nil { + return "", err + } + + return rcSaved.RemoteId, nil +} + +func (a *App) UnregisterPluginForSharedChannels(pluginID string) error { + rc, err := a.Srv().Store().RemoteCluster().GetByPluginID(pluginID) + if err != nil { + return err + } + + _, appErr := a.DeleteRemoteCluster(rc.RemoteId) + if appErr != nil { + return appErr + } + return nil +} + func (a *App) AddRemoteCluster(rc *model.RemoteCluster) (*model.RemoteCluster, *model.AppError) { rc, err := a.Srv().Store().RemoteCluster().Save(rc) if err != nil { diff --git a/server/channels/app/server.go b/server/channels/app/server.go index 1f9d4169c8..07e279ec6e 100644 --- a/server/channels/app/server.go +++ b/server/channels/app/server.go @@ -625,8 +625,8 @@ func (s *Server) startInterClusterServices(license *model.License) error { } var err error - - rcs, err := remotecluster.NewRemoteClusterService(s) + appInstance := New(ServerConnector(s.Channels())) + rcs, err := remotecluster.NewRemoteClusterService(s, appInstance) if err != nil { return err } @@ -653,7 +653,6 @@ func (s *Server) startInterClusterServices(license *model.License) error { return nil } - appInstance := New(ServerConnector(s.Channels())) scs, err := sharedchannel.NewSharedChannelService(s, appInstance) if err != nil { return err diff --git a/server/channels/app/shared_channel.go b/server/channels/app/shared_channel.go index 47d0158140..26626604a8 100644 --- a/server/channels/app/shared_channel.go +++ b/server/channels/app/shared_channel.go @@ -13,6 +13,10 @@ import ( "github.com/mattermost/mattermost/server/v8/channels/store" ) +var ( + errNotFound = errors.New("not found") +) + func (a *App) checkChannelNotShared(c request.CTX, channelId string) error { // check that channel exists. if _, err := a.GetChannel(c, channelId); err != nil { @@ -57,13 +61,27 @@ func (a *App) CheckCanInviteToSharedChannel(channelId string) error { return nil } +func (a *App) notifyClientsForSharedChannelUpdate(teamID, channelID string) { + messageWs := model.NewWebSocketEvent(model.WebsocketEventChannelConverted, teamID, "", "", nil, "") + messageWs.Add("channel_id", channelID) + a.Publish(messageWs) +} + // SharedChannels -func (a *App) SaveSharedChannel(c request.CTX, sc *model.SharedChannel) (*model.SharedChannel, error) { +func (a *App) ShareChannel(c request.CTX, sc *model.SharedChannel) (*model.SharedChannel, error) { if err := a.checkChannelNotShared(c, sc.ChannelId); err != nil { return nil, err } - return a.Srv().Store().SharedChannel().Save(sc) + + // stores a SharedChannel and set the share flag on the channel. + scNew, err := a.Srv().Store().SharedChannel().Save(sc) + if err != nil { + return nil, err + } + + a.notifyClientsForSharedChannelUpdate(scNew.TeamId, scNew.ChannelId) + return scNew, nil } func (a *App) GetSharedChannel(channelID string) (*model.SharedChannel, error) { @@ -87,15 +105,105 @@ func (a *App) GetSharedChannelsCount(opts model.SharedChannelFilterOpts) (int64, } func (a *App) UpdateSharedChannel(sc *model.SharedChannel) (*model.SharedChannel, error) { - return a.Srv().Store().SharedChannel().Update(sc) + scUpdated, err := a.Srv().Store().SharedChannel().Update(sc) + if err != nil { + return nil, err + } + a.notifyClientsForSharedChannelUpdate(scUpdated.TeamId, scUpdated.ChannelId) + return scUpdated, nil } -func (a *App) DeleteSharedChannel(channelID string) (bool, error) { - return a.Srv().Store().SharedChannel().Delete(channelID) +func (a *App) UnshareChannel(channelID string) (bool, error) { + // fetch the SharedChannel first + sc, err := a.GetSharedChannel(channelID) + if err != nil { + return false, err + } + + // deletes the ShareChannel, unsets the share flag on the channel, deletes all remotes for the channel + deleted, err := a.Srv().Store().SharedChannel().Delete(channelID) + if err != nil { + return false, err + } + a.notifyClientsForSharedChannelUpdate(sc.TeamId, sc.ChannelId) + return deleted, nil } // SharedChannelRemotes +func (a *App) InviteRemoteToChannel(channelID, remoteID, userID string) error { + syncService := a.Srv().GetSharedChannelSyncService() + if syncService == nil || !syncService.Active() { + return model.NewAppError("InviteRemoteToChannel", "api.command_share.service_disabled", + nil, "", http.StatusBadRequest) + } + + hasRemote, err := a.HasRemote(channelID, remoteID) + if err != nil { + return model.NewAppError("InviteRemoteToChannel", "api.command_share.fetch_remote.error", + map[string]any{"Error": err.Error()}, "", http.StatusInternalServerError) + } + if hasRemote { + // already invited + return nil + } + + // Check if channel is shared or not. + hasChan, err := a.HasSharedChannel(channelID) + if err != nil { + return model.NewAppError("InviteRemoteToChannel", "api.command_share.check_channel_exist.error", + map[string]any{"ChannelID": channelID, "Error": err.Error()}, "", http.StatusInternalServerError) + } + if !hasChan { + return model.NewAppError("InviteRemoteToChannel", "api.command_share.channel_not_shared.error", + map[string]any{"ChannelID": channelID}, "", http.StatusBadRequest) + } + + // don't allow invitation to shared channel originating from remote. + // (also blocks cyclic invitations) + if err := a.CheckCanInviteToSharedChannel(channelID); err != nil { + return model.NewAppError("InviteRemoteToChannel", "api.command_share.channel_invite_not_home.error", nil, "", http.StatusInternalServerError) + } + + rc, appErr := a.GetRemoteCluster(remoteID) + if appErr != nil { + return model.NewAppError("InviteRemoteToChannel", "api.command_share.remote_id_invalid.error", + map[string]any{"Error": appErr.Error()}, "", http.StatusInternalServerError).Wrap(appErr) + } + + channel, errApp := a.GetChannel(request.EmptyContext(a.Log()), channelID) + if errApp != nil { + return model.NewAppError("InviteRemoteToChannel", "api.command_share.channel_invite.error", + map[string]any{"Name": rc.DisplayName, "Error": errApp.Error()}, "", http.StatusInternalServerError).Wrap(appErr) + } + // send channel invite to remote cluster. Will notify clients of channel change. + if err := syncService.SendChannelInvite(channel, userID, rc); err != nil { + return model.NewAppError("InviteRemoteToChannel", "api.command_share.channel_invite.error", + map[string]any{"Name": rc.DisplayName, "Error": err.Error()}, "", http.StatusInternalServerError).Wrap(err) + } + return nil +} + +func (a *App) UninviteRemoteFromChannel(channelID, remoteID string) error { + scr, err := a.GetSharedChannelRemoteByIds(channelID, remoteID) + if err != nil || scr.ChannelId != channelID { + return model.NewAppError("UninviteRemoteFromChannel", "api.command_share.channel_remote_id_not_exists", + map[string]any{"RemoteId": remoteID}, "", http.StatusInternalServerError) + } + + deleted, err := a.Srv().Store().SharedChannel().DeleteRemote(scr.Id) + if err != nil || !deleted { + code := http.StatusInternalServerError + if err == nil { + err = errNotFound + code = http.StatusBadRequest + } + return model.NewAppError("UninviteRemoteFromChannel", "api.command_share.could_not_uninvite.error", + map[string]any{"RemoteId": remoteID, "Error": err.Error()}, "", code) + } + return nil +} + func (a *App) SaveSharedChannelRemote(remote *model.SharedChannelRemote) (*model.SharedChannelRemote, error) { if err := a.checkChannelIsShared(remote.ChannelId); err != nil { return nil, err @@ -164,3 +272,66 @@ func (a *App) onUserProfileChange(userID string) { } syncService.NotifyUserProfileChanged(userID) } + +// Sync + +// UpdateSharedChannelCursor updates the cursor for the specified channelID and remoteID. +// This can be used to manually set the point of last sync, either forward to skip older posts, +// or backward to re-sync history. +// This call by itself does not force a re-sync - a change to channel contents or a call to +// SyncSharedChannel are needed to force a sync. +func (a *App) UpdateSharedChannelCursor(channelID, remoteID string, cursor model.GetPostsSinceForSyncCursor) error { + src, err := a.Srv().Store().SharedChannel().GetRemoteByIds(channelID, remoteID) + if err != nil { + return fmt.Errorf("cursor update failed - cannot fetch shared channel remote: %w", err) + } + return a.Srv().Store().SharedChannel().UpdateRemoteCursor(src.Id, cursor) +} + +// SyncSharedChannel forces a shared channel to send any changed content to all remote clusters. +func (a *App) SyncSharedChannel(channelID string) error { + syncService := a.Srv().GetSharedChannelSyncService() + if syncService == nil || !syncService.Active() { + return model.NewAppError("InviteRemoteToChannel", "api.command_share.service_disabled", + nil, "", http.StatusBadRequest) + } + + syncService.NotifyChannelChanged(channelID) + return nil +} + +// Hooks + +var ErrPluginUnavailable = errors.New("plugin unavialable") + +// OnSharedChannelsSyncMsg is called by the Shared Channels service for a registered plugin when there is new content +// that needs to be synchronized. +func (a *App) OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) { + pluginsEnvironment := a.GetPluginsEnvironment() + if pluginsEnvironment == nil { + return model.SyncResponse{}, fmt.Errorf("cannot deliver sync msg to plugin %s: %w", rc.PluginID, ErrPluginUnavailable) + } + + pluginHooks, err := pluginsEnvironment.HooksForPlugin(rc.PluginID) + if err != nil { + return model.SyncResponse{}, fmt.Errorf("cannot deliver sync msg to plugin %s: %w", rc.PluginID, err) + } + + return pluginHooks.OnSharedChannelsSyncMsg(msg, rc) +} + +// OnSharedChannelsPing is called by the Shared Channels service for a registered plugin wto check that the plugin +// is still responding and has a connection to any upstream services it needs (e.g. MS Graph API). +func (a *App) OnSharedChannelsPing(rc *model.RemoteCluster) bool { + pluginsEnvironment := a.GetPluginsEnvironment() + if pluginsEnvironment == nil { + return false + } + + pluginHooks, err := pluginsEnvironment.HooksForPlugin(rc.PluginID) + if err != nil { + return false + } + + return pluginHooks.OnSharedChannelsPing(rc) +} diff --git a/server/channels/app/shared_channel_test.go b/server/channels/app/shared_channel_test.go index 6d03d6d3e0..c565bf1988 100644 --- a/server/channels/app/shared_channel_test.go +++ b/server/channels/app/shared_channel_test.go @@ -38,7 +38,7 @@ func TestApp_CheckCanInviteToSharedChannel(t *testing.T) { CreatorId: th.BasicUser.Id, RemoteId: d.remoteId, } - _, err := th.App.SaveSharedChannel(th.Context, sc) + _, err := th.App.ShareChannel(th.Context, sc) require.NoError(t, err) } diff --git a/server/channels/app/slashcommands/command_share.go b/server/channels/app/slashcommands/command_share.go index b4ca40c056..e1b3cc182b 100644 --- a/server/channels/app/slashcommands/command_share.go +++ b/server/channels/app/slashcommands/command_share.go @@ -127,11 +127,13 @@ func (sp *ShareProvider) DoCommand(a *app.App, c request.CTX, args *model.Comman return responsef(args.T("api.command_share.permission_required", map[string]any{"Permission": "manage_shared_channels"})) } - if a.Srv().GetSharedChannelSyncService() == nil { + syncService := a.Srv().GetSharedChannelSyncService() + if syncService == nil || !syncService.Active() { return responsef(args.T("api.command_share.service_disabled")) } - if a.Srv().GetRemoteClusterService() == nil { + rcService := a.Srv().GetRemoteClusterService() + if rcService == nil || !rcService.Active() { return responsef(args.T("api.command_remote.service_disabled")) } @@ -157,7 +159,7 @@ func (sp *ShareProvider) DoCommand(a *app.App, c request.CTX, args *model.Comman } func (sp *ShareProvider) doShareChannel(a *app.App, c request.CTX, args *model.CommandArgs, margs map[string]string) *model.CommandResponse { - // check that channel exists. + // fetch defaults for missing channel props channel, errApp := a.GetChannel(c, args.ChannelId) if errApp != nil { return responsef(args.T("api.command_share.share_channel.error", map[string]any{"Error": errApp.Error()})) @@ -196,22 +198,15 @@ func (sp *ShareProvider) doShareChannel(a *app.App, c request.CTX, args *model.C CreatorId: args.UserId, } - if _, err := a.SaveSharedChannel(c, sc); err != nil { + if _, err := a.ShareChannel(c, sc); err != nil { return responsef(args.T("api.command_share.share_channel.error", map[string]any{"Error": err.Error()})) } - notifyClientsForChannelUpdate(a, sc) - return responsef("##### " + args.T("api.command_share.channel_shared")) } func (sp *ShareProvider) doUnshareChannel(a *app.App, args *model.CommandArgs, margs map[string]string) *model.CommandResponse { - sc, appErr := a.GetSharedChannel(args.ChannelId) - if appErr != nil { - return responsef(args.T("api.command_share.shared_channel_unshare.error", map[string]any{"Error": appErr.Error()})) - } - - deleted, err := a.DeleteSharedChannel(args.ChannelId) + deleted, err := a.UnshareChannel(args.ChannelId) if err != nil { return responsef(args.T("api.command_share.shared_channel_unshare.error", map[string]any{"Error": err.Error()})) } @@ -219,8 +214,6 @@ func (sp *ShareProvider) doUnshareChannel(a *app.App, args *model.CommandArgs, m return responsef(args.T("api.command_share.not_shared_channel_unshare")) } - notifyClientsForChannelUpdate(a, sc) - return responsef("##### " + args.T("api.command_share.shared_channel_unavailable")) } @@ -241,7 +234,7 @@ func (sp *ShareProvider) doInviteRemote(a *app.App, c request.CTX, args *model.C // Check if channel is shared or not. hasChan, err := a.HasSharedChannel(args.ChannelId) if err != nil { - return responsef(args.T("api.command_share.check_channel_exist.error", map[string]any{"Error": err.Error()})) + return responsef(args.T("api.command_share.check_channel_exist.error", map[string]any{"ChannelID": args.ChannelId, "Error": err.Error()})) } if !hasChan { // If it doesn't exist, then create it. @@ -253,24 +246,13 @@ func (sp *ShareProvider) doInviteRemote(a *app.App, c request.CTX, args *model.C }() } - // don't allow invitation to shared channel originating from remote. - // (also blocks cyclic invitations) - if err := a.CheckCanInviteToSharedChannel(args.ChannelId); err != nil { - return responsef(args.T("api.command_share.channel_invite_not_home.error")) - } - rc, appErr := a.GetRemoteCluster(remoteID) if appErr != nil { return responsef(args.T("api.command_share.remote_id_invalid.error", map[string]any{"Error": appErr.Error()})) } - channel, errApp := a.GetChannel(c, args.ChannelId) - if errApp != nil { - return responsef(args.T("api.command_share.channel_invite.error", map[string]any{"Name": rc.DisplayName, "Error": errApp.Error()})) - } - // send channel invite to remote cluster - if err := a.Srv().GetSharedChannelSyncService().SendChannelInvite(channel, args.UserId, rc); err != nil { - return responsef(args.T("api.command_share.channel_invite.error", map[string]any{"Name": rc.DisplayName, "Error": err.Error()})) + if err = a.InviteRemoteToChannel(args.ChannelId, remoteID, args.UserId); err != nil { + return responsef(appErr.Error()) } return responsef("##### " + args.T("api.command_share.invitation_sent", map[string]any{"Name": rc.DisplayName, "SiteURL": rc.SiteURL})) @@ -282,15 +264,11 @@ func (sp *ShareProvider) doUninviteRemote(a *app.App, args *model.CommandArgs, m return responsef(args.T("api.command_share.remote_not_valid")) } - scr, err := a.GetSharedChannelRemoteByIds(args.ChannelId, remoteID) - if err != nil || scr.ChannelId != args.ChannelId { - return responsef(args.T("api.command_share.channel_remote_id_not_exists", map[string]any{"RemoteId": remoteID})) + err := a.UninviteRemoteFromChannel(args.ChannelId, remoteID) + if err != nil { + return responsef(err.Error()) } - deleted, err := a.DeleteSharedChannelRemote(scr.Id) - if err != nil || !deleted { - return responsef(args.T("api.command_share.could_not_uninvite.error", map[string]any{"RemoteId": remoteID, "Error": err.Error()})) - } return responsef("##### " + args.T("api.command_share.remote_uninvited", map[string]any{"RemoteId": remoteID})) } @@ -323,9 +301,3 @@ func (sp *ShareProvider) doStatus(a *app.App, args *model.CommandArgs, _ map[str } return responsef(sb.String()) } - -func notifyClientsForChannelUpdate(a *app.App, sharedChannel *model.SharedChannel) { - messageWs := model.NewWebSocketEvent(model.WebsocketEventChannelConverted, sharedChannel.TeamId, "", "", nil, "") - messageWs.Add("channel_id", sharedChannel.ChannelId) - a.Publish(messageWs) -} diff --git a/server/channels/app/slashcommands/command_share_test.go b/server/channels/app/slashcommands/command_share_test.go index 8274c58684..59bb153713 100644 --- a/server/channels/app/slashcommands/command_share_test.go +++ b/server/channels/app/slashcommands/command_share_test.go @@ -26,15 +26,19 @@ func TestShareProviderDoCommand(t *testing.T) { th.addPermissionToRole(model.PermissionManageSharedChannels.Id, th.BasicUser.Roles) - mockSyncService := app.NewMockSharedChannelService(nil) + mockSyncService := app.NewMockSharedChannelService(nil, app.MockOptionSharedChannelServiceWithActive(true)) th.Server.SetSharedChannelSyncService(mockSyncService) - mockRemoteCluster, err := remotecluster.NewRemoteClusterService(th.Server) + remoteClusterService, err := remotecluster.NewRemoteClusterService(th.Server, th.App) require.NoError(t, err) - th.Server.SetRemoteClusterService(mockRemoteCluster) + th.Server.SetRemoteClusterService(remoteClusterService) testCluster := &testlib.FakeClusterInterface{} th.Server.Platform().SetCluster(testCluster) + err = remoteClusterService.Start() + require.NoError(t, err) + defer remoteClusterService.Shutdown() + commandProvider := ShareProvider{} channel := th.CreateChannel(th.BasicTeam, WithShared(false)) @@ -53,7 +57,7 @@ func TestShareProviderDoCommand(t *testing.T) { event, err := model.WebSocketEventFromJSON(bytes.NewReader(msg.Data)) return err == nil && event.EventType() == model.WebsocketEventChannelConverted }) - assert.Len(t, channelConvertedMessages, 1) + assert.Len(t, channelConvertedMessages, 1) // one msg for share creation }) t.Run("unshare command sends a websocket channel converted event", func(t *testing.T) { @@ -64,13 +68,17 @@ func TestShareProviderDoCommand(t *testing.T) { mockSyncService := app.NewMockSharedChannelService(nil) th.Server.SetSharedChannelSyncService(mockSyncService) - mockRemoteCluster, err := remotecluster.NewRemoteClusterService(th.Server) + remoteClusterService, err := remotecluster.NewRemoteClusterService(th.Server, th.App) require.NoError(t, err) - th.Server.SetRemoteClusterService(mockRemoteCluster) + th.Server.SetRemoteClusterService(remoteClusterService) testCluster := &testlib.FakeClusterInterface{} th.Server.Platform().SetCluster(testCluster) + err = remoteClusterService.Start() + require.NoError(t, err) + defer remoteClusterService.Shutdown() + commandProvider := ShareProvider{} channel := th.CreateChannel(th.BasicTeam, WithShared(true)) args := &model.CommandArgs{ @@ -88,6 +96,6 @@ func TestShareProviderDoCommand(t *testing.T) { event, err := model.WebSocketEventFromJSON(bytes.NewReader(msg.Data)) return err == nil && event.EventType() == model.WebsocketEventChannelConverted }) - require.Len(t, channelConvertedMessages, 1) + require.Len(t, channelConvertedMessages, 2) // one msg for share creation, one for unshare. }) } diff --git a/server/channels/app/slashcommands/helper_test.go b/server/channels/app/slashcommands/helper_test.go index e91e66e642..ea0c4ec0f3 100644 --- a/server/channels/app/slashcommands/helper_test.go +++ b/server/channels/app/slashcommands/helper_test.go @@ -274,7 +274,7 @@ func (th *TestHelper) createChannel(team *model.Team, channelType model.ChannelT if channel.IsShared() { id := model.NewId() - _, err := th.App.SaveSharedChannel(th.Context, &model.SharedChannel{ + _, err := th.App.ShareChannel(th.Context, &model.SharedChannel{ ChannelId: channel.Id, TeamId: channel.TeamId, Home: false, diff --git a/server/channels/store/opentracinglayer/opentracinglayer.go b/server/channels/store/opentracinglayer/opentracinglayer.go index 4d7f01142c..e6c51d9609 100644 --- a/server/channels/store/opentracinglayer/opentracinglayer.go +++ b/server/channels/store/opentracinglayer/opentracinglayer.go @@ -7643,6 +7643,24 @@ func (s *OpenTracingLayerRemoteClusterStore) GetAll(filter model.RemoteClusterQu return result, err } +func (s *OpenTracingLayerRemoteClusterStore) GetByPluginID(pluginID string) (*model.RemoteCluster, error) { + origCtx := s.Root.Store.Context() + span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "RemoteClusterStore.GetByPluginID") + s.Root.Store.SetContext(newCtx) + defer func() { + s.Root.Store.SetContext(origCtx) + }() + + defer span.Finish() + result, err := s.RemoteClusterStore.GetByPluginID(pluginID) + if err != nil { + span.LogFields(spanlog.Error(err)) + ext.Error.Set(span, true) + } + + return result, err +} + func (s *OpenTracingLayerRemoteClusterStore) Save(rc *model.RemoteCluster) (*model.RemoteCluster, error) { origCtx := s.Root.Store.Context() span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "RemoteClusterStore.Save") diff --git a/server/channels/store/retrylayer/retrylayer.go b/server/channels/store/retrylayer/retrylayer.go index e3cbf142d6..c35d44cfc7 100644 --- a/server/channels/store/retrylayer/retrylayer.go +++ b/server/channels/store/retrylayer/retrylayer.go @@ -8684,6 +8684,27 @@ func (s *RetryLayerRemoteClusterStore) GetAll(filter model.RemoteClusterQueryFil } +func (s *RetryLayerRemoteClusterStore) GetByPluginID(pluginID string) (*model.RemoteCluster, error) { + + tries := 0 + for { + result, err := s.RemoteClusterStore.GetByPluginID(pluginID) + if err == nil { + return result, nil + } + if !isRepeatableError(err) { + return result, err + } + tries++ + if tries >= 3 { + err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures") + return result, err + } + timepkg.Sleep(100 * timepkg.Millisecond) + } + +} + func (s *RetryLayerRemoteClusterStore) Save(rc *model.RemoteCluster) (*model.RemoteCluster, error) { tries := 0 diff --git a/server/channels/store/sqlstore/remote_cluster_store.go b/server/channels/store/sqlstore/remote_cluster_store.go index 7e577d8e1c..3d2f45a4e4 100644 --- a/server/channels/store/sqlstore/remote_cluster_store.go +++ b/server/channels/store/sqlstore/remote_cluster_store.go @@ -4,6 +4,7 @@ package sqlstore import ( + "database/sql" "fmt" "strings" @@ -50,6 +51,19 @@ func (s sqlRemoteClusterStore) Save(remoteCluster *model.RemoteCluster) (*model. return nil, err } + // check for pluginID collisions - on collision treat as idempotent + if remoteCluster.PluginID != "" { + rc, err := s.GetByPluginID(remoteCluster.PluginID) + if err == nil { + // if this plugin id already exists, just return it + return rc, nil + } + if !errors.Is(err, sql.ErrNoRows) { + // anything other than NotFound is unexpected + return nil, errors.Wrapf(err, "failed to lookup RemoteCluster by pluginID %s", remoteCluster.PluginID) + } + } + query := `INSERT INTO RemoteClusters (RemoteId, RemoteTeamId, Name, DisplayName, SiteURL, CreateAt, LastPingAt, Token, RemoteToken, Topics, CreatorId, PluginID, Options) @@ -69,6 +83,7 @@ func (s sqlRemoteClusterStore) Update(remoteCluster *model.RemoteCluster) (*mode return nil, err } + // not all fields can be updated. query := `UPDATE RemoteClusters SET Token = :Token, RemoteTeamId = :RemoteTeamId, @@ -129,6 +144,24 @@ func (s sqlRemoteClusterStore) Get(remoteId string) (*model.RemoteCluster, error return &rc, nil } +func (s sqlRemoteClusterStore) GetByPluginID(pluginID string) (*model.RemoteCluster, error) { + query := s.getQueryBuilder(). + Select(remoteClusterFields("")...). + From("RemoteClusters"). + Where(sq.Eq{"PluginID": pluginID}) + + queryString, args, err := query.ToSql() + if err != nil { + return nil, errors.Wrap(err, "remote_cluster_get_by_pluginid_tosql") + } + + var rc model.RemoteCluster + if err := s.GetReplicaX().Get(&rc, queryString, args...); err != nil { + return nil, errors.Wrapf(err, "failed to find RemoteCluster by plugin_id") + } + return &rc, nil +} + func (s sqlRemoteClusterStore) GetAll(filter model.RemoteClusterQueryFilter) ([]*model.RemoteCluster, error) { query := s.getQueryBuilder(). Select(remoteClusterFields("rc")...). diff --git a/server/channels/store/store.go b/server/channels/store/store.go index 0a6c6fe427..0cb531cb4d 100644 --- a/server/channels/store/store.go +++ b/server/channels/store/store.go @@ -543,6 +543,7 @@ type RemoteClusterStore interface { Update(rc *model.RemoteCluster) (*model.RemoteCluster, error) Delete(remoteClusterId string) (bool, error) Get(remoteClusterId string) (*model.RemoteCluster, error) + GetByPluginID(pluginID string) (*model.RemoteCluster, error) GetAll(filter model.RemoteClusterQueryFilter) ([]*model.RemoteCluster, error) UpdateTopics(remoteClusterId string, topics string) (*model.RemoteCluster, error) SetLastPingAt(remoteClusterId string) error diff --git a/server/channels/store/storetest/mocks/RemoteClusterStore.go b/server/channels/store/storetest/mocks/RemoteClusterStore.go index 0cb4737cbd..f9c29193d4 100644 --- a/server/channels/store/storetest/mocks/RemoteClusterStore.go +++ b/server/channels/store/storetest/mocks/RemoteClusterStore.go @@ -90,6 +90,32 @@ func (_m *RemoteClusterStore) GetAll(filter model.RemoteClusterQueryFilter) ([]* return r0, r1 } +// GetByPluginID provides a mock function with given fields: pluginID +func (_m *RemoteClusterStore) GetByPluginID(pluginID string) (*model.RemoteCluster, error) { + ret := _m.Called(pluginID) + + var r0 *model.RemoteCluster + var r1 error + if rf, ok := ret.Get(0).(func(string) (*model.RemoteCluster, error)); ok { + return rf(pluginID) + } + if rf, ok := ret.Get(0).(func(string) *model.RemoteCluster); ok { + r0 = rf(pluginID) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*model.RemoteCluster) + } + } + + if rf, ok := ret.Get(1).(func(string) error); ok { + r1 = rf(pluginID) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // Save provides a mock function with given fields: rc func (_m *RemoteClusterStore) Save(rc *model.RemoteCluster) (*model.RemoteCluster, error) { ret := _m.Called(rc) diff --git a/server/channels/store/storetest/remote_cluster_store.go b/server/channels/store/storetest/remote_cluster_store.go index 9b40cbbb86..d35004aa59 100644 --- a/server/channels/store/storetest/remote_cluster_store.go +++ b/server/channels/store/storetest/remote_cluster_store.go @@ -15,10 +15,6 @@ import ( "github.com/stretchr/testify/require" ) -const ( - testPluginID = "com.sample.blap" -) - func TestRemoteClusterStore(t *testing.T, rctx request.CTX, ss store.Store) { t.Run("RemoteClusterGetAllInChannel", func(t *testing.T) { testRemoteClusterGetAllInChannel(t, rctx, ss) }) t.Run("RemoteClusterGetAllNotInChannel", func(t *testing.T) { testRemoteClusterGetAllNotInChannel(t, rctx, ss) }) @@ -30,12 +26,17 @@ func TestRemoteClusterStore(t *testing.T, rctx request.CTX, ss store.Store) { t.Run("RemoteClusterUpdateTopics", func(t *testing.T) { testRemoteClusterUpdateTopics(t, rctx, ss) }) } +func makeSiteURL() string { + return "www.example.com/" + model.NewId() +} + func testRemoteClusterSave(t *testing.T, rctx request.CTX, ss store.Store) { t.Run("Save", func(t *testing.T) { rc := &model.RemoteCluster{ Name: "some_remote", - SiteURL: "somewhere.com", + SiteURL: makeSiteURL(), CreatorId: model.NewId(), + PluginID: model.NewId(), } rcSaved, err := ss.RemoteCluster().Save(rc) @@ -44,12 +45,13 @@ func testRemoteClusterSave(t *testing.T, rctx request.CTX, ss store.Store) { require.Equal(t, rc.SiteURL, rcSaved.SiteURL) require.Greater(t, rc.CreateAt, int64(0)) require.Equal(t, rc.LastPingAt, int64(0)) + require.Equal(t, rc.PluginID, rcSaved.PluginID) require.Equal(t, rc.Options, model.Bitmask(0)) }) t.Run("Save missing display name", func(t *testing.T) { rc := &model.RemoteCluster{ - SiteURL: "somewhere.com", + SiteURL: makeSiteURL(), CreatorId: model.NewId(), } _, err := ss.RemoteCluster().Save(rc) @@ -59,34 +61,84 @@ func testRemoteClusterSave(t *testing.T, rctx request.CTX, ss store.Store) { t.Run("Save missing creator id", func(t *testing.T) { rc := &model.RemoteCluster{ Name: "some_remote_2", - SiteURL: "somewhere.com", + SiteURL: makeSiteURL(), } _, err := ss.RemoteCluster().Save(rc) require.Error(t, err) }) + t.Run("Save pluginID collision", func(t *testing.T) { + const testPluginID = "com.example.collision" + + rc := &model.RemoteCluster{ + Name: "some_remote", + SiteURL: makeSiteURL(), + CreatorId: model.NewId(), + PluginID: testPluginID, + } + _, err := ss.RemoteCluster().Save(rc) + require.NoError(t, err) + + rc2 := &model.RemoteCluster{ + Name: "another_remote", + SiteURL: makeSiteURL(), + CreatorId: model.NewId(), + PluginID: testPluginID, + } + + rcSaved, err := ss.RemoteCluster().Save(rc2) + require.NoError(t, err) + require.NotNil(t, rcSaved) + + // original remotecluster should be returned + require.Equal(t, rc.Name, rcSaved.Name) + require.Equal(t, rc.SiteURL, rcSaved.SiteURL) + require.Greater(t, rc.CreateAt, int64(0)) + require.Equal(t, rc.PluginID, rcSaved.PluginID) + }) + + t.Run("Save multiple with blank pluginID", func(t *testing.T) { + rc := &model.RemoteCluster{ + Name: model.NewId(), + SiteURL: makeSiteURL(), + CreatorId: model.NewId(), + } + _, err := ss.RemoteCluster().Save(rc) + require.NoError(t, err) + + rc2 := &model.RemoteCluster{ + Name: model.NewId(), + SiteURL: makeSiteURL(), + CreatorId: model.NewId(), + } + _, err = ss.RemoteCluster().Save(rc2) + require.NoError(t, err) + }) + t.Run("Save for plugin with options", func(t *testing.T) { rc := &model.RemoteCluster{ Name: "plugin_remote", - SiteURL: "plugin.example.com", + SiteURL: makeSiteURL(), CreatorId: model.NewId(), - PluginID: testPluginID, + PluginID: model.NewId(), Options: model.BitflagOptionAutoShareDMs, } rcSaved, err := ss.RemoteCluster().Save(rc) require.NoError(t, err) - require.Equal(t, testPluginID, rcSaved.PluginID) + require.Equal(t, rc.PluginID, rcSaved.PluginID) require.Equal(t, model.BitflagOptionAutoShareDMs, rcSaved.Options) require.True(t, rcSaved.IsOptionFlagSet(model.BitflagOptionAutoShareDMs)) rc.Name = "plugin_remote_2" + rc.SiteURL = makeSiteURL() + rc.PluginID = model.NewId() rc.SiteURL = "plugin2.example.com" rc.UnsetOptionFlag(model.BitflagOptionAutoShareDMs) rcSaved, err = ss.RemoteCluster().Save(rc) require.NoError(t, err) - require.Equal(t, testPluginID, rcSaved.PluginID) + require.Equal(t, rc.PluginID, rcSaved.PluginID) require.Equal(t, model.Bitmask(0), rcSaved.Options) require.False(t, rcSaved.IsOptionFlagSet(model.BitflagOptionAutoShareDMs)) }) @@ -96,7 +148,7 @@ func testRemoteClusterDelete(t *testing.T, rctx request.CTX, ss store.Store) { t.Run("Delete", func(t *testing.T) { rc := &model.RemoteCluster{ Name: "shortlived_remote", - SiteURL: "nowhere.com", + SiteURL: makeSiteURL(), CreatorId: model.NewId(), } rcSaved, err := ss.RemoteCluster().Save(rc) @@ -118,9 +170,9 @@ func testRemoteClusterGet(t *testing.T, rctx request.CTX, ss store.Store) { t.Run("Get", func(t *testing.T) { rc := &model.RemoteCluster{ Name: "shortlived_remote_2", - SiteURL: "nowhere.com", + SiteURL: makeSiteURL(), CreatorId: model.NewId(), - PluginID: testPluginID, + PluginID: model.NewId(), } rc.SetOptionFlag(model.BitflagOptionAutoShareDMs) rcSaved, err := ss.RemoteCluster().Save(rc) @@ -129,7 +181,7 @@ func testRemoteClusterGet(t *testing.T, rctx request.CTX, ss store.Store) { rcGet, err := ss.RemoteCluster().Get(rcSaved.RemoteId) require.NoError(t, err) require.Equal(t, rcSaved.RemoteId, rcGet.RemoteId) - require.Equal(t, testPluginID, rcGet.PluginID) + require.Equal(t, rcSaved.PluginID, rcGet.PluginID) require.True(t, rcGet.IsOptionFlagSet(model.BitflagOptionAutoShareDMs)) }) @@ -147,10 +199,10 @@ func testRemoteClusterGetAll(t *testing.T, rctx request.CTX, ss store.Store) { pingLongAgo := model.GetMillis() - (model.RemoteOfflineAfterMillis * 3) data := []*model.RemoteCluster{ - {Name: "offline_remote", CreatorId: userId, SiteURL: "somewhere.com", LastPingAt: pingLongAgo, Topics: " shared incident "}, - {Name: "some_online_remote", CreatorId: userId, SiteURL: "nowhere.com", LastPingAt: now, Topics: " shared incident "}, - {Name: "another_online_remote", CreatorId: model.NewId(), SiteURL: "underwhere.com", LastPingAt: now, Topics: ""}, - {Name: "another_offline_remote", CreatorId: model.NewId(), SiteURL: "knowhere.com", LastPingAt: pingLongAgo, Topics: " shared "}, + {Name: "offline_remote", CreatorId: userId, SiteURL: makeSiteURL(), LastPingAt: pingLongAgo, Topics: " shared incident "}, + {Name: "some_online_remote", CreatorId: userId, SiteURL: makeSiteURL(), LastPingAt: now, Topics: " shared incident "}, + {Name: "another_online_remote", CreatorId: model.NewId(), SiteURL: makeSiteURL(), LastPingAt: now, Topics: ""}, + {Name: "another_offline_remote", CreatorId: model.NewId(), SiteURL: makeSiteURL(), LastPingAt: pingLongAgo, Topics: " shared "}, {Name: "brand_new_offline_remote", CreatorId: userId, SiteURL: "", LastPingAt: 0, Topics: " bogus shared stuff "}, } @@ -245,6 +297,11 @@ func testRemoteClusterGetAll(t *testing.T, rctx request.CTX, ss store.Store) { } func testRemoteClusterGetAllInChannel(t *testing.T, rctx request.CTX, ss store.Store) { + const ( + testPluginID_1 = "com.sample.blap" + testPluginID_2 = "com.sample.bloop" + ) + require.NoError(t, clearRemoteClusters(ss)) now := model.GetMillis() @@ -272,8 +329,8 @@ func testRemoteClusterGetAllInChannel(t *testing.T, rctx request.CTX, ss store.S // Create some remote clusters rcData := []*model.RemoteCluster{ - {Name: "AAAA_Inc", CreatorId: userId, SiteURL: "aaaa.com", RemoteId: model.NewId(), LastPingAt: now, PluginID: testPluginID}, - {Name: "BBBB_Inc", CreatorId: userId, SiteURL: "bbbb.com", RemoteId: model.NewId(), LastPingAt: 0, PluginID: testPluginID}, + {Name: "AAAA_Inc", CreatorId: userId, SiteURL: "aaaa.com", RemoteId: model.NewId(), LastPingAt: now, PluginID: testPluginID_1}, + {Name: "BBBB_Inc", CreatorId: userId, SiteURL: "bbbb.com", RemoteId: model.NewId(), LastPingAt: 0, PluginID: testPluginID_2}, {Name: "CCCC_Inc", CreatorId: userId, SiteURL: "cccc.com", RemoteId: model.NewId(), LastPingAt: now}, {Name: "DDDD_Inc", CreatorId: userId, SiteURL: "dddd.com", RemoteId: model.NewId(), LastPingAt: now}, {Name: "EEEE_Inc", CreatorId: userId, SiteURL: "eeee.com", RemoteId: model.NewId(), LastPingAt: 0}, @@ -305,8 +362,8 @@ func testRemoteClusterGetAllInChannel(t *testing.T, rctx request.CTX, ss store.S require.Len(t, list, 2, "channel 1 should have 2 remote clusters") ids := getIds(list) require.ElementsMatch(t, []string{rcData[0].RemoteId, rcData[1].RemoteId}, ids) - require.Equal(t, testPluginID, rcData[0].PluginID) - require.Equal(t, testPluginID, rcData[1].PluginID) + require.Equal(t, testPluginID_1, rcData[0].PluginID) + require.Equal(t, testPluginID_2, rcData[1].PluginID) }) t.Run("Channel 1 online only", func(t *testing.T) { diff --git a/server/channels/store/timerlayer/timerlayer.go b/server/channels/store/timerlayer/timerlayer.go index fcab4b4d32..c5f3b4bab5 100644 --- a/server/channels/store/timerlayer/timerlayer.go +++ b/server/channels/store/timerlayer/timerlayer.go @@ -6905,6 +6905,22 @@ func (s *TimerLayerRemoteClusterStore) GetAll(filter model.RemoteClusterQueryFil return result, err } +func (s *TimerLayerRemoteClusterStore) GetByPluginID(pluginID string) (*model.RemoteCluster, error) { + start := time.Now() + + result, err := s.RemoteClusterStore.GetByPluginID(pluginID) + + elapsed := float64(time.Since(start)) / float64(time.Second) + if s.Root.Metrics != nil { + success := "false" + if err == nil { + success = "true" + } + s.Root.Metrics.ObserveStoreMethodDuration("RemoteClusterStore.GetByPluginID", success, elapsed) + } + return result, err +} + func (s *TimerLayerRemoteClusterStore) Save(rc *model.RemoteCluster) (*model.RemoteCluster, error) { start := time.Now() diff --git a/server/i18n/en.json b/server/i18n/en.json index 27441ecf20..e29285a99b 100644 --- a/server/i18n/en.json +++ b/server/i18n/en.json @@ -1389,6 +1389,10 @@ "id": "api.command_share.channel_invite_not_home.error", "translation": "Cannot invite secure connection to a shared channel originating somewhere else." }, + { + "id": "api.command_share.channel_not_shared.error", + "translation": "Cannot invite remote because channel {{.ChannelID}} is not shared." + }, { "id": "api.command_share.channel_remote_id_not_exists", "translation": "Shared channel secure connection `{{.RemoteId}}` does not exist for this channel." @@ -1407,7 +1411,7 @@ }, { "id": "api.command_share.check_channel_exist.error", - "translation": "Error while checking if shared channel exists: {{.Error}}" + "translation": "Error while checking if shared channel {{.ChannelID}} exists: {{.Error}}" }, { "id": "api.command_share.could_not_uninvite.error", diff --git a/server/platform/services/remotecluster/mocks_test.go b/server/platform/services/remotecluster/mocks_test.go index e45f70a53b..0176631d2c 100644 --- a/server/platform/services/remotecluster/mocks_test.go +++ b/server/platform/services/remotecluster/mocks_test.go @@ -5,6 +5,7 @@ package remotecluster import ( "context" + "sync" "testing" "github.com/mattermost/mattermost/server/public/model" @@ -47,10 +48,12 @@ func (ms *mockServer) GetStore() store.Store { return true }) anyUserId := mock.AnythingOfType("string") + anyId := mock.AnythingOfType("string") remoteClusterStoreMock := &mocks.RemoteClusterStore{} remoteClusterStoreMock.On("GetByTopic", "share").Return(ms.remotes, nil) remoteClusterStoreMock.On("GetAll", anyQueryFilter).Return(ms.remotes, nil) + remoteClusterStoreMock.On("SetLastPingAt", anyId).Return(nil) userStoreMock := &mocks.UserStore{} userStoreMock.On("Get", context.Background(), anyUserId).Return(ms.user, nil) @@ -60,3 +63,57 @@ func (ms *mockServer) GetStore() store.Store { storeMock.On("User").Return(userStoreMock) return storeMock } + +type mockApp struct { + offlinePluginIDs []string + + mux sync.Mutex + totalPingCount int + totalPingErrors int + pingCounts map[string]int +} + +func newMockApp(t *testing.T, offlinePluginIDs []string) *mockApp { + return &mockApp{ + offlinePluginIDs: offlinePluginIDs, + pingCounts: make(map[string]int), + } +} + +func (ma *mockApp) OnSharedChannelsPing(rc *model.RemoteCluster) bool { + ma.mux.Lock() + defer ma.mux.Unlock() + + for _, id := range ma.offlinePluginIDs { + if rc.PluginID == id { + ma.totalPingErrors++ + return false + } + } + + ma.totalPingCount++ + + count := ma.pingCounts[rc.PluginID] + ma.pingCounts[rc.PluginID] = count + 1 + + return true +} + +func (ma *mockApp) GetTotalPingCount() int { + ma.mux.Lock() + defer ma.mux.Unlock() + return ma.totalPingCount +} + +func (ma *mockApp) GetTotalPingErrorCount() int { + ma.mux.Lock() + defer ma.mux.Unlock() + return ma.totalPingErrors +} + +func (ma *mockApp) GetPingCount(pluginID string) int { + ma.mux.Lock() + defer ma.mux.Unlock() + + return ma.pingCounts[pluginID] +} diff --git a/server/platform/services/remotecluster/ping.go b/server/platform/services/remotecluster/ping.go index 29e5d7fcc8..c90dd4838d 100644 --- a/server/platform/services/remotecluster/ping.go +++ b/server/platform/services/remotecluster/ping.go @@ -5,6 +5,7 @@ package remotecluster import ( "encoding/json" + "errors" "fmt" "time" @@ -28,6 +29,7 @@ func (rcs *Service) pingGenerator(pingChan chan *model.RemoteCluster, done <-cha defer close(pingChan) for { + pingFreq := rcs.GetPingFreq() start := time.Now() // get all remotes, including any previously offline. @@ -35,7 +37,7 @@ func (rcs *Service) pingGenerator(pingChan chan *model.RemoteCluster, done <-cha if err != nil { rcs.server.Log().Log(mlog.LvlRemoteClusterServiceError, "Ping remote cluster failed (could not get list of remotes)", mlog.Err(err)) select { - case <-time.After(PingFreq): + case <-time.After(pingFreq): continue case <-done: return @@ -50,8 +52,8 @@ func (rcs *Service) pingGenerator(pingChan chan *model.RemoteCluster, done <-cha // try to maintain frequency elapsed := time.Since(start) - if elapsed < PingFreq { - sleep := time.Until(start.Add(PingFreq)) + if elapsed < pingFreq { + sleep := time.Until(start.Add(pingFreq)) select { case <-time.After(sleep): case <-done: @@ -77,6 +79,7 @@ func (rcs *Service) pingEmitter(pingChan <-chan *model.RemoteCluster, done <-cha rcs.server.Log().Log(mlog.LvlRemoteClusterServiceWarn, "Remote cluster ping failed", mlog.String("remote", rc.DisplayName), mlog.String("remoteId", rc.RemoteId), + mlog.String("pluginId", rc.PluginID), mlog.Err(err), ) } @@ -93,25 +96,36 @@ func (rcs *Service) pingEmitter(pingChan <-chan *model.RemoteCluster, done <-cha } } +var ErrPluginPingFail = errors.New("plugin ping failed") + // pingRemote make a synchronous ping to a remote cluster. Return is error if ping is // unsuccessful and nil on success. func (rcs *Service) pingRemote(rc *model.RemoteCluster) error { - frame, err := makePingFrame(rc) - if err != nil { - return err - } - url := fmt.Sprintf("%s/%s", rc.SiteURL, PingURL) - - resp, err := rcs.sendFrameToRemote(PingTimeout, rc, frame, url) - if err != nil { - return err - } - rc.LastPingAt = model.GetMillis() - ping := model.RemoteClusterPing{} - err = json.Unmarshal(resp, &ping) - if err != nil { - return err + + if rc.PluginID != "" { + ping.SentAt = model.GetMillis() + if ok := rcs.app.OnSharedChannelsPing(rc); !ok { + return ErrPluginPingFail + } + ping.RecvAt = model.GetMillis() + } else { + frame, err := makePingFrame(rc) + if err != nil { + return err + } + url := fmt.Sprintf("%s/%s", rc.SiteURL, PingURL) + + resp, err := rcs.sendFrameToRemote(PingTimeout, rc, frame, url) + if err != nil { + return err + } + rc.LastPingAt = model.GetMillis() + + err = json.Unmarshal(resp, &ping) + if err != nil { + return err + } } if err := rcs.server.GetStore().RemoteCluster().SetLastPingAt(rc.RemoteId); err != nil { @@ -135,6 +149,7 @@ func (rcs *Service) pingRemote(rc *model.RemoteCluster) error { rcs.server.Log().Log(mlog.LvlRemoteClusterServiceDebug, "Remote cluster ping", mlog.String("remote", rc.DisplayName), mlog.String("remoteId", rc.RemoteId), + mlog.String("pluginId", rc.PluginID), mlog.Int("SentAt", ping.SentAt), mlog.Int("RecvAt", ping.RecvAt), mlog.Int("Diff", ping.RecvAt-ping.SentAt), diff --git a/server/platform/services/remotecluster/ping_test.go b/server/platform/services/remotecluster/ping_test.go index df729683bb..0714abe5e3 100644 --- a/server/platform/services/remotecluster/ping_test.go +++ b/server/platform/services/remotecluster/ping_test.go @@ -11,6 +11,7 @@ import ( "sync" "sync/atomic" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -66,9 +67,10 @@ func TestPing(t *testing.T) { })) defer ts.Close() - mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL)) + mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL, false)) + mockApp := newMockApp(t, nil) - service, err := NewRemoteClusterService(mockServer) + service, err := NewRemoteClusterService(mockServer, mockApp) require.NoError(t, err) err = service.Start() @@ -115,9 +117,10 @@ func TestPing(t *testing.T) { })) defer ts.Close() - mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL)) + mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL, false)) + mockApp := newMockApp(t, nil) - service, err := NewRemoteClusterService(mockServer) + service, err := NewRemoteClusterService(mockServer, mockApp) require.NoError(t, err) err = service.Start() @@ -132,6 +135,34 @@ func TestPing(t *testing.T) { t.Logf("%d web requests counted; %d expected", atomic.LoadInt32(&countWebReq), NumRemotes) }) + + t.Run("Plugin ping", func(t *testing.T) { + mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, model.NewId(), true)) + offline := []string{mockServer.remotes[0].PluginID, mockServer.remotes[1].PluginID} + + mockApp := newMockApp(t, offline) + + service, err := NewRemoteClusterService(mockServer, mockApp) + require.NoError(t, err) + + // high ping frequency so we don't delay unit tests. + service.SetPingFreq(time.Millisecond * 50) + + err = service.Start() + require.NoError(t, err) + defer service.Shutdown() + + checkPingCount := func() bool { + return mockApp.GetTotalPingCount() >= NumRemotes + } + + checkErrorCount := func() bool { + return mockApp.GetTotalPingErrorCount() >= 2 + } + + assert.Eventually(t, checkPingCount, time.Second*5, 10*time.Millisecond) + assert.Eventually(t, checkErrorCount, time.Second*5, 10*time.Millisecond) + }) } func checkRecent(millis int64, within int64) bool { diff --git a/server/platform/services/remotecluster/send_test.go b/server/platform/services/remotecluster/send_test.go index 360d557c62..f4407a0da8 100644 --- a/server/platform/services/remotecluster/send_test.go +++ b/server/platform/services/remotecluster/send_test.go @@ -82,9 +82,10 @@ func TestBroadcastMsg(t *testing.T) { })) defer ts.Close() - mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL)) + mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL, false)) + mockApp := newMockApp(t, nil) - service, err := NewRemoteClusterService(mockServer) + service, err := NewRemoteClusterService(mockServer, mockApp) require.NoError(t, err) err = service.Start() @@ -138,9 +139,10 @@ func TestBroadcastMsg(t *testing.T) { })) defer ts.Close() - mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL)) + mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL, false)) + mockApp := newMockApp(t, nil) - service, err := NewRemoteClusterService(mockServer) + service, err := NewRemoteClusterService(mockServer, mockApp) require.NoError(t, err) err = service.Start() @@ -169,10 +171,13 @@ func TestBroadcastMsg(t *testing.T) { }) } -func makeRemoteClusters(num int, siteURL string) []*model.RemoteCluster { +func makeRemoteClusters(num int, siteURL string, isPlugin bool) []*model.RemoteCluster { var remotes []*model.RemoteCluster for i := 0; i < num; i++ { rc := makeRemoteCluster(fmt.Sprintf("test cluster %d", i+1), siteURL, TestTopics) + if isPlugin { + rc.PluginID = model.NewId() + } remotes = append(remotes, rc) } return remotes diff --git a/server/platform/services/remotecluster/sendprofileImage_test.go b/server/platform/services/remotecluster/sendprofileImage_test.go index ab47234796..c99d453a2e 100644 --- a/server/platform/services/remotecluster/sendprofileImage_test.go +++ b/server/platform/services/remotecluster/sendprofileImage_test.go @@ -100,9 +100,12 @@ func TestService_sendProfileImageToRemote(t *testing.T) { provider := testImageProvider{} - mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL)) + mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL, false)) mockServer.SetUser(user) - service, err := NewRemoteClusterService(mockServer) + + mockApp := newMockApp(t, nil) + + service, err := NewRemoteClusterService(mockServer, mockApp) require.NoError(t, err) err = service.Start() diff --git a/server/platform/services/remotecluster/service.go b/server/platform/services/remotecluster/service.go index e9b1b07f78..88773b3179 100644 --- a/server/platform/services/remotecluster/service.go +++ b/server/platform/services/remotecluster/service.go @@ -50,6 +50,10 @@ type ServerIface interface { GetMetrics() einterfaces.MetricsInterface } +type AppIface interface { + OnSharedChannelsPing(rc *model.RemoteCluster) bool +} + // RemoteClusterServiceIFace is used to allow mocking where a remote cluster service is used (for testing). // Unfortunately it lives here because the shared channel service, app layer, and server interface all need it. // Putting it in app layer means shared channel service must import app package. @@ -78,6 +82,7 @@ type ConnectionStateListener func(rc *model.RemoteCluster, online bool) // Service provides inter-cluster communication via topic based messages. In product these are called "Secured Connections". type Service struct { server ServerIface + app AppIface httpClient *http.Client send []chan any @@ -88,10 +93,11 @@ type Service struct { topicListeners map[string]map[string]TopicListener // maps topic id to a map of listenerid->listener connectionStateListeners map[string]ConnectionStateListener // maps listener id to listener done chan struct{} + pingFreq time.Duration } // NewRemoteClusterService creates a RemoteClusterService instance. In product this is called a "Secured Connection". -func NewRemoteClusterService(server ServerIface) (*Service, error) { +func NewRemoteClusterService(server ServerIface, app AppIface) (*Service, error) { transport := &http.Transport{ Proxy: http.ProxyFromEnvironment, DialContext: (&net.Dialer{ @@ -115,6 +121,7 @@ func NewRemoteClusterService(server ServerIface) (*Service, error) { service := &Service{ server: server, + app: app, httpClient: client, topicListeners: make(map[string]map[string]TopicListener), connectionStateListeners: make(map[string]ConnectionStateListener), @@ -124,6 +131,7 @@ func NewRemoteClusterService(server ServerIface) (*Service, error) { for i := range service.send { service.send[i] = make(chan any, SendChanBuffer) } + service.pingFreq = PingFreq return service, nil } @@ -154,6 +162,21 @@ func (rcs *Service) Active() bool { return rcs.active } +// GetPingFreq gets the frequency of pings to each remote. +func (rcs *Service) GetPingFreq() time.Duration { + rcs.mux.Lock() + defer rcs.mux.Unlock() + return rcs.pingFreq +} + +// SetPingFreq sets the frequency of pings to each remote. Defaults to `PingFreq`. +// This is typically used to set a higher frequency for testing. +func (rcs *Service) SetPingFreq(freq time.Duration) { + rcs.mux.Lock() + defer rcs.mux.Unlock() + rcs.pingFreq = freq +} + // AddTopicListener registers a callback func (rcs *Service) AddTopicListener(topic string, listener TopicListener) string { rcs.mux.Lock() diff --git a/server/platform/services/remotecluster/service_test.go b/server/platform/services/remotecluster/service_test.go index 9773a24444..1f0bff10c2 100644 --- a/server/platform/services/remotecluster/service_test.go +++ b/server/platform/services/remotecluster/service_test.go @@ -29,9 +29,10 @@ func TestService_AddTopicListener(t *testing.T) { return nil } - mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, "")) + mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, "", false)) + mockApp := newMockApp(t, nil) - service, err := NewRemoteClusterService(mockServer) + service, err := NewRemoteClusterService(mockServer, mockApp) require.NoError(t, err) l1id := service.AddTopicListener("test", l1) diff --git a/server/platform/services/sharedchannel/mock_AppIface_test.go b/server/platform/services/sharedchannel/mock_AppIface_test.go index f69ee15d6b..81dbd2c9aa 100644 --- a/server/platform/services/sharedchannel/mock_AppIface_test.go +++ b/server/platform/services/sharedchannel/mock_AppIface_test.go @@ -314,6 +314,30 @@ func (_m *MockAppIface) NotifySharedChannelUserUpdate(user *model.User) { _m.Called(user) } +// OnSharedChannelsSyncMsg provides a mock function with given fields: msg, rc +func (_m *MockAppIface) OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) { + ret := _m.Called(msg, rc) + + var r0 model.SyncResponse + var r1 error + if rf, ok := ret.Get(0).(func(*model.SyncMsg, *model.RemoteCluster) (model.SyncResponse, error)); ok { + return rf(msg, rc) + } + if rf, ok := ret.Get(0).(func(*model.SyncMsg, *model.RemoteCluster) model.SyncResponse); ok { + r0 = rf(msg, rc) + } else { + r0 = ret.Get(0).(model.SyncResponse) + } + + if rf, ok := ret.Get(1).(func(*model.SyncMsg, *model.RemoteCluster) error); ok { + r1 = rf(msg, rc) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // PatchChannelModerationsForChannel provides a mock function with given fields: c, channel, channelModerationsPatch func (_m *MockAppIface) PatchChannelModerationsForChannel(c request.CTX, channel *model.Channel, channelModerationsPatch []*model.ChannelModerationPatch) ([]*model.ChannelModeration, *model.AppError) { ret := _m.Called(c, channel, channelModerationsPatch) diff --git a/server/platform/services/sharedchannel/service.go b/server/platform/services/sharedchannel/service.go index bb99cf8e18..c11fa4d972 100644 --- a/server/platform/services/sharedchannel/service.go +++ b/server/platform/services/sharedchannel/service.go @@ -63,6 +63,7 @@ type AppIface interface { GetProfileImage(user *model.User) ([]byte, bool, *model.AppError) InvalidateCacheForUser(userID string) NotifySharedChannelUserUpdate(user *model.User) + OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) } // errNotFound allows checking against Store.ErrNotFound errors without making Store a dependency. @@ -108,7 +109,7 @@ func NewSharedChannelService(server ServerIface, app AppIface) (*Service, error) // Start is called by the server on server start-up. func (scs *Service) Start() error { rcs := scs.server.GetRemoteClusterService() - if rcs == nil { + if rcs == nil || !rcs.Active() { return errors.New("Shared Channel Service cannot activate: requires Remote Cluster Service") } @@ -128,7 +129,7 @@ func (scs *Service) Start() error { // Shutdown is called by the server on server shutdown. func (scs *Service) Shutdown() error { rcs := scs.server.GetRemoteClusterService() - if rcs == nil { + if rcs == nil || !rcs.Active() { return errors.New("Shared Channel Service cannot shutdown: requires Remote Cluster Service") } diff --git a/server/platform/services/sharedchannel/sync_send_remote.go b/server/platform/services/sharedchannel/sync_send_remote.go index 9124087cf4..a0d5ea3c4e 100644 --- a/server/platform/services/sharedchannel/sync_send_remote.go +++ b/server/platform/services/sharedchannel/sync_send_remote.go @@ -527,13 +527,17 @@ func (scs *Service) sendProfileImageSyncData(sd *syncData) { } } -// sendSyncMsgToRemote synchronously sends the sync message to the remote cluster. +// sendSyncMsgToRemote synchronously sends the sync message to the remote cluster (or plugin). func (scs *Service) sendSyncMsgToRemote(msg *model.SyncMsg, rc *model.RemoteCluster, f sendSyncMsgResultFunc) error { rcs := scs.server.GetRemoteClusterService() if rcs == nil { return fmt.Errorf("cannot update remote cluster %s for channel id %s; Remote Cluster Service not enabled", rc.Name, msg.ChannelId) } + if rc.PluginID != "" { + return scs.sendSyncMsgToPlugin(msg, rc, f) + } + b, err := json.Marshal(msg) if err != nil { return err @@ -568,6 +572,17 @@ func (scs *Service) sendSyncMsgToRemote(msg *model.SyncMsg, rc *model.RemoteClus return err } +// sendSyncMsgToRemote synchronously sends the sync message to a plugin. +func (scs *Service) sendSyncMsgToPlugin(msg *model.SyncMsg, rc *model.RemoteCluster, f sendSyncMsgResultFunc) error { + syncResp, errResp := scs.app.OnSharedChannelsSyncMsg(msg, rc) + + if f != nil { + f(syncResp, errResp) + } + + return errResp +} + func sanitizeSyncData(sd *syncData) { for id, user := range sd.users { sd.users[id] = sanitizeUserForSync(user) diff --git a/server/public/model/shared_channel.go b/server/public/model/shared_channel.go index 98f49b1e76..c0bdeea823 100644 --- a/server/public/model/shared_channel.go +++ b/server/public/model/shared_channel.go @@ -297,3 +297,21 @@ type SyncResponse struct { ReactionsLastUpdateAt int64 `json:"reactions_last_update_at"` ReactionErrors []string `json:"reaction_errors"` } + +// RegisterPluginOpts is passed by plugins to the `RegisterPluginForSharedChannels` plugin API +// to provide options for registering as a shared channels remote. +type RegisterPluginOpts struct { + Displayname string // a displayname used in status reports + PluginID string // id of this plugin registering + CreatorID string // id of the user/bot registering + AutoShareDMs bool // when true, all DMs are automatically shared to this remote +} + +// GetOptionFlags returns a Bitmask of option flags as specified by the boolean options. +func (po RegisterPluginOpts) GetOptionFlags() Bitmask { + var flags Bitmask + if po.AutoShareDMs { + flags |= BitflagOptionAutoShareDMs + } + return flags +} diff --git a/server/public/plugin/api.go b/server/public/plugin/api.go index a5e3533f07..ff4e2ca57f 100644 --- a/server/public/plugin/api.go +++ b/server/public/plugin/api.go @@ -32,7 +32,7 @@ type API interface { // Minimum server version: 5.2 RegisterCommand(command *model.Command) error - // UnregisterCommand unregisters a command previously registered via RegisterCommand. + // UnregisterCommand unregisters a command previously register via RegisterCommand. // // @tag Command // Minimum server version: 5.2 @@ -1207,6 +1207,73 @@ type API interface { // @tag User // Minimum server version: 9.3 UpdateUserAuth(userID string, userAuth *model.UserAuth) (*model.UserAuth, *model.AppError) + + // RegisterPluginForSharedChannels registers the plugin as a `Remote` for SharedChannels. + // The plugin will receive synchronization messages via the `OnSharedChannelsSyncMsg` hook. + // This API is idempotent - when called repeatedly with the same `RegisterPluginOpts.PluginID` + // it will return the same remoteID. + // + // @tag SharedChannels + // Minimum server version: 9.5 + RegisterPluginForSharedChannels(opts model.RegisterPluginOpts) (remoteID string, err error) + + // UnregisterPluginForSharedChannels unregisters the plugin as a `Remote` for SharedChannels. + // The plugin will no longer receive synchronization messages via the `OnSharedChannelsSyncMsg` hook. + // + // @tag SharedChannels + // Minimum server version: 9.5 + UnregisterPluginForSharedChannels(pluginID string) error + + // ShareChannel marks a channel for sharing via shared channels. Note, this does not automatically + // invite any remote clusters to the channel - use `InviteRemote` to invite a remote , or this plugin, + // to the shared channel and start synchronization. + // + // @tag SharedChannels + // Minimum server version: 9.5 + ShareChannel(sc *model.SharedChannel) (*model.SharedChannel, error) + + // UpdateSharedChannel updates a shared channel. This can be used to change the share name, + // display name, purpose, header, etc. + // + // @tag SharedChannels + // Minimum server version: 9.5 + UpdateSharedChannel(sc *model.SharedChannel) (*model.SharedChannel, error) + + // UnshareChannel unmarks a channel for sharing. The channel will no longer be shared and + // all remotes will be uninvited to the channel. + // + // @tag SharedChannels + // Minimum server version: 9.5 + UnshareChannel(channelID string) (unshared bool, err error) + + // UpdateSharedChannelCursor updates the cursor for the specified channel and RemoteID (passed by + // the plugin when registering). This can be used to manually set the point of last sync, either + // forward to skip older posts, or backward to re-sync history. This call by itself does not force + // a re-sync - a change to channel contents or a call to SyncSharedChannel are needed to force a sync. + // + // @tag SharedChannels + // Minimum server version: 9.5 + UpdateSharedChannelCursor(channelID, remoteID string, cusror model.GetPostsSinceForSyncCursor) error + + // SyncSharedChannel forces a shared channel to send any changed content to all remotes. + // + // @tag SharedChannels + // Minimum server version: 9.5 + SyncSharedChannel(channelID string) error + + // InviteRemoteToChannel invites a remote, or this plugin, as a target for synchronizing. Once invited, the + // remote will start to receive synchronization messages for any changed content in the specified channel. + // + // @tag SharedChannels + // Minimum server version: 9.5 + InviteRemoteToChannel(channelID string, remoteID string, userID string) error + + // UninviteRemoteFromChannel uninvites a remote, or this plugin, such that it will stop receiving sychronization + // messages for the channel. + // + // @tag SharedChannels + // Minimum server version: 9.5 + UninviteRemoteFromChannel(channelID string, remoteID string) error } var handshake = plugin.HandshakeConfig{ diff --git a/server/public/plugin/api_timer_layer_generated.go b/server/public/plugin/api_timer_layer_generated.go index 266ea5d20a..f39a411760 100644 --- a/server/public/plugin/api_timer_layer_generated.go +++ b/server/public/plugin/api_timer_layer_generated.go @@ -1287,3 +1287,66 @@ func (api *apiTimerLayer) UpdateUserAuth(userID string, userAuth *model.UserAuth api.recordTime(startTime, "UpdateUserAuth", _returnsB == nil) return _returnsA, _returnsB } + +func (api *apiTimerLayer) RegisterPluginForSharedChannels(opts model.RegisterPluginOpts) (remoteID string, err error) { + startTime := timePkg.Now() + _returnsA, _returnsB := api.apiImpl.RegisterPluginForSharedChannels(opts) + api.recordTime(startTime, "RegisterPluginForSharedChannels", _returnsB == nil) + return _returnsA, _returnsB +} + +func (api *apiTimerLayer) UnregisterPluginForSharedChannels(pluginID string) error { + startTime := timePkg.Now() + _returnsA := api.apiImpl.UnregisterPluginForSharedChannels(pluginID) + api.recordTime(startTime, "UnregisterPluginForSharedChannels", _returnsA == nil) + return _returnsA +} + +func (api *apiTimerLayer) ShareChannel(sc *model.SharedChannel) (*model.SharedChannel, error) { + startTime := timePkg.Now() + _returnsA, _returnsB := api.apiImpl.ShareChannel(sc) + api.recordTime(startTime, "ShareChannel", _returnsB == nil) + return _returnsA, _returnsB +} + +func (api *apiTimerLayer) UpdateSharedChannel(sc *model.SharedChannel) (*model.SharedChannel, error) { + startTime := timePkg.Now() + _returnsA, _returnsB := api.apiImpl.UpdateSharedChannel(sc) + api.recordTime(startTime, "UpdateSharedChannel", _returnsB == nil) + return _returnsA, _returnsB +} + +func (api *apiTimerLayer) UnshareChannel(channelID string) (unshared bool, err error) { + startTime := timePkg.Now() + _returnsA, _returnsB := api.apiImpl.UnshareChannel(channelID) + api.recordTime(startTime, "UnshareChannel", _returnsB == nil) + return _returnsA, _returnsB +} + +func (api *apiTimerLayer) UpdateSharedChannelCursor(channelID, remoteID string, cusror model.GetPostsSinceForSyncCursor) error { + startTime := timePkg.Now() + _returnsA := api.apiImpl.UpdateSharedChannelCursor(channelID, remoteID, cusror) + api.recordTime(startTime, "UpdateSharedChannelCursor", _returnsA == nil) + return _returnsA +} + +func (api *apiTimerLayer) SyncSharedChannel(channelID string) error { + startTime := timePkg.Now() + _returnsA := api.apiImpl.SyncSharedChannel(channelID) + api.recordTime(startTime, "SyncSharedChannel", _returnsA == nil) + return _returnsA +} + +func (api *apiTimerLayer) InviteRemoteToChannel(channelID string, remoteID string, userID string) error { + startTime := timePkg.Now() + _returnsA := api.apiImpl.InviteRemoteToChannel(channelID, remoteID, userID) + api.recordTime(startTime, "InviteRemoteToChannel", _returnsA == nil) + return _returnsA +} + +func (api *apiTimerLayer) UninviteRemoteFromChannel(channelID string, remoteID string) error { + startTime := timePkg.Now() + _returnsA := api.apiImpl.UninviteRemoteFromChannel(channelID, remoteID) + api.recordTime(startTime, "UninviteRemoteFromChannel", _returnsA == nil) + return _returnsA +} diff --git a/server/public/plugin/client_rpc_generated.go b/server/public/plugin/client_rpc_generated.go index 8f300fa74a..6acb35eb49 100644 --- a/server/public/plugin/client_rpc_generated.go +++ b/server/public/plugin/client_rpc_generated.go @@ -947,6 +947,77 @@ func (s *hooksRPCServer) UserHasBeenDeactivated(args *Z_UserHasBeenDeactivatedAr return nil } +func init() { + hookNameToId["OnSharedChannelsSyncMsg"] = OnSharedChannelsSyncMsgID +} + +type Z_OnSharedChannelsSyncMsgArgs struct { + A *model.SyncMsg + B *model.RemoteCluster +} + +type Z_OnSharedChannelsSyncMsgReturns struct { + A model.SyncResponse + B error +} + +func (g *hooksRPCClient) OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) { + _args := &Z_OnSharedChannelsSyncMsgArgs{msg, rc} + _returns := &Z_OnSharedChannelsSyncMsgReturns{} + if g.implemented[OnSharedChannelsSyncMsgID] { + if err := g.client.Call("Plugin.OnSharedChannelsSyncMsg", _args, _returns); err != nil { + g.log.Error("RPC call OnSharedChannelsSyncMsg to plugin failed.", mlog.Err(err)) + } + } + return _returns.A, _returns.B +} + +func (s *hooksRPCServer) OnSharedChannelsSyncMsg(args *Z_OnSharedChannelsSyncMsgArgs, returns *Z_OnSharedChannelsSyncMsgReturns) error { + if hook, ok := s.impl.(interface { + OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) + }); ok { + returns.A, returns.B = hook.OnSharedChannelsSyncMsg(args.A, args.B) + returns.B = encodableError(returns.B) + } else { + return encodableError(fmt.Errorf("Hook OnSharedChannelsSyncMsg called but not implemented.")) + } + return nil +} + +func init() { + hookNameToId["OnSharedChannelsPing"] = OnSharedChannelsPingID +} + +type Z_OnSharedChannelsPingArgs struct { + A *model.RemoteCluster +} + +type Z_OnSharedChannelsPingReturns struct { + A bool +} + +func (g *hooksRPCClient) OnSharedChannelsPing(rc *model.RemoteCluster) bool { + _args := &Z_OnSharedChannelsPingArgs{rc} + _returns := &Z_OnSharedChannelsPingReturns{} + if g.implemented[OnSharedChannelsPingID] { + if err := g.client.Call("Plugin.OnSharedChannelsPing", _args, _returns); err != nil { + g.log.Error("RPC call OnSharedChannelsPing to plugin failed.", mlog.Err(err)) + } + } + return _returns.A +} + +func (s *hooksRPCServer) OnSharedChannelsPing(args *Z_OnSharedChannelsPingArgs, returns *Z_OnSharedChannelsPingReturns) error { + if hook, ok := s.impl.(interface { + OnSharedChannelsPing(rc *model.RemoteCluster) bool + }); ok { + returns.A = hook.OnSharedChannelsPing(args.A) + } else { + return encodableError(fmt.Errorf("Hook OnSharedChannelsPing called but not implemented.")) + } + return nil +} + type Z_RegisterCommandArgs struct { A *model.Command } @@ -6026,3 +6097,273 @@ func (s *apiRPCServer) UpdateUserAuth(args *Z_UpdateUserAuthArgs, returns *Z_Upd } return nil } + +type Z_RegisterPluginForSharedChannelsArgs struct { + A model.RegisterPluginOpts +} + +type Z_RegisterPluginForSharedChannelsReturns struct { + A string + B error +} + +func (g *apiRPCClient) RegisterPluginForSharedChannels(opts model.RegisterPluginOpts) (remoteID string, err error) { + _args := &Z_RegisterPluginForSharedChannelsArgs{opts} + _returns := &Z_RegisterPluginForSharedChannelsReturns{} + if err := g.client.Call("Plugin.RegisterPluginForSharedChannels", _args, _returns); err != nil { + log.Printf("RPC call to RegisterPluginForSharedChannels API failed: %s", err.Error()) + } + return _returns.A, _returns.B +} + +func (s *apiRPCServer) RegisterPluginForSharedChannels(args *Z_RegisterPluginForSharedChannelsArgs, returns *Z_RegisterPluginForSharedChannelsReturns) error { + if hook, ok := s.impl.(interface { + RegisterPluginForSharedChannels(opts model.RegisterPluginOpts) (remoteID string, err error) + }); ok { + returns.A, returns.B = hook.RegisterPluginForSharedChannels(args.A) + returns.B = encodableError(returns.B) + } else { + return encodableError(fmt.Errorf("API RegisterPluginForSharedChannels called but not implemented.")) + } + return nil +} + +type Z_UnregisterPluginForSharedChannelsArgs struct { + A string +} + +type Z_UnregisterPluginForSharedChannelsReturns struct { + A error +} + +func (g *apiRPCClient) UnregisterPluginForSharedChannels(pluginID string) error { + _args := &Z_UnregisterPluginForSharedChannelsArgs{pluginID} + _returns := &Z_UnregisterPluginForSharedChannelsReturns{} + if err := g.client.Call("Plugin.UnregisterPluginForSharedChannels", _args, _returns); err != nil { + log.Printf("RPC call to UnregisterPluginForSharedChannels API failed: %s", err.Error()) + } + return _returns.A +} + +func (s *apiRPCServer) UnregisterPluginForSharedChannels(args *Z_UnregisterPluginForSharedChannelsArgs, returns *Z_UnregisterPluginForSharedChannelsReturns) error { + if hook, ok := s.impl.(interface { + UnregisterPluginForSharedChannels(pluginID string) error + }); ok { + returns.A = hook.UnregisterPluginForSharedChannels(args.A) + returns.A = encodableError(returns.A) + } else { + return encodableError(fmt.Errorf("API UnregisterPluginForSharedChannels called but not implemented.")) + } + return nil +} + +type Z_ShareChannelArgs struct { + A *model.SharedChannel +} + +type Z_ShareChannelReturns struct { + A *model.SharedChannel + B error +} + +func (g *apiRPCClient) ShareChannel(sc *model.SharedChannel) (*model.SharedChannel, error) { + _args := &Z_ShareChannelArgs{sc} + _returns := &Z_ShareChannelReturns{} + if err := g.client.Call("Plugin.ShareChannel", _args, _returns); err != nil { + log.Printf("RPC call to ShareChannel API failed: %s", err.Error()) + } + return _returns.A, _returns.B +} + +func (s *apiRPCServer) ShareChannel(args *Z_ShareChannelArgs, returns *Z_ShareChannelReturns) error { + if hook, ok := s.impl.(interface { + ShareChannel(sc *model.SharedChannel) (*model.SharedChannel, error) + }); ok { + returns.A, returns.B = hook.ShareChannel(args.A) + returns.B = encodableError(returns.B) + } else { + return encodableError(fmt.Errorf("API ShareChannel called but not implemented.")) + } + return nil +} + +type Z_UpdateSharedChannelArgs struct { + A *model.SharedChannel +} + +type Z_UpdateSharedChannelReturns struct { + A *model.SharedChannel + B error +} + +func (g *apiRPCClient) UpdateSharedChannel(sc *model.SharedChannel) (*model.SharedChannel, error) { + _args := &Z_UpdateSharedChannelArgs{sc} + _returns := &Z_UpdateSharedChannelReturns{} + if err := g.client.Call("Plugin.UpdateSharedChannel", _args, _returns); err != nil { + log.Printf("RPC call to UpdateSharedChannel API failed: %s", err.Error()) + } + return _returns.A, _returns.B +} + +func (s *apiRPCServer) UpdateSharedChannel(args *Z_UpdateSharedChannelArgs, returns *Z_UpdateSharedChannelReturns) error { + if hook, ok := s.impl.(interface { + UpdateSharedChannel(sc *model.SharedChannel) (*model.SharedChannel, error) + }); ok { + returns.A, returns.B = hook.UpdateSharedChannel(args.A) + returns.B = encodableError(returns.B) + } else { + return encodableError(fmt.Errorf("API UpdateSharedChannel called but not implemented.")) + } + return nil +} + +type Z_UnshareChannelArgs struct { + A string +} + +type Z_UnshareChannelReturns struct { + A bool + B error +} + +func (g *apiRPCClient) UnshareChannel(channelID string) (unshared bool, err error) { + _args := &Z_UnshareChannelArgs{channelID} + _returns := &Z_UnshareChannelReturns{} + if err := g.client.Call("Plugin.UnshareChannel", _args, _returns); err != nil { + log.Printf("RPC call to UnshareChannel API failed: %s", err.Error()) + } + return _returns.A, _returns.B +} + +func (s *apiRPCServer) UnshareChannel(args *Z_UnshareChannelArgs, returns *Z_UnshareChannelReturns) error { + if hook, ok := s.impl.(interface { + UnshareChannel(channelID string) (unshared bool, err error) + }); ok { + returns.A, returns.B = hook.UnshareChannel(args.A) + returns.B = encodableError(returns.B) + } else { + return encodableError(fmt.Errorf("API UnshareChannel called but not implemented.")) + } + return nil +} + +type Z_UpdateSharedChannelCursorArgs struct { + A string + B string + C model.GetPostsSinceForSyncCursor +} + +type Z_UpdateSharedChannelCursorReturns struct { + A error +} + +func (g *apiRPCClient) UpdateSharedChannelCursor(channelID, remoteID string, cusror model.GetPostsSinceForSyncCursor) error { + _args := &Z_UpdateSharedChannelCursorArgs{channelID, remoteID, cusror} + _returns := &Z_UpdateSharedChannelCursorReturns{} + if err := g.client.Call("Plugin.UpdateSharedChannelCursor", _args, _returns); err != nil { + log.Printf("RPC call to UpdateSharedChannelCursor API failed: %s", err.Error()) + } + return _returns.A +} + +func (s *apiRPCServer) UpdateSharedChannelCursor(args *Z_UpdateSharedChannelCursorArgs, returns *Z_UpdateSharedChannelCursorReturns) error { + if hook, ok := s.impl.(interface { + UpdateSharedChannelCursor(channelID, remoteID string, cusror model.GetPostsSinceForSyncCursor) error + }); ok { + returns.A = hook.UpdateSharedChannelCursor(args.A, args.B, args.C) + returns.A = encodableError(returns.A) + } else { + return encodableError(fmt.Errorf("API UpdateSharedChannelCursor called but not implemented.")) + } + return nil +} + +type Z_SyncSharedChannelArgs struct { + A string +} + +type Z_SyncSharedChannelReturns struct { + A error +} + +func (g *apiRPCClient) SyncSharedChannel(channelID string) error { + _args := &Z_SyncSharedChannelArgs{channelID} + _returns := &Z_SyncSharedChannelReturns{} + if err := g.client.Call("Plugin.SyncSharedChannel", _args, _returns); err != nil { + log.Printf("RPC call to SyncSharedChannel API failed: %s", err.Error()) + } + return _returns.A +} + +func (s *apiRPCServer) SyncSharedChannel(args *Z_SyncSharedChannelArgs, returns *Z_SyncSharedChannelReturns) error { + if hook, ok := s.impl.(interface { + SyncSharedChannel(channelID string) error + }); ok { + returns.A = hook.SyncSharedChannel(args.A) + returns.A = encodableError(returns.A) + } else { + return encodableError(fmt.Errorf("API SyncSharedChannel called but not implemented.")) + } + return nil +} + +type Z_InviteRemoteToChannelArgs struct { + A string + B string + C string +} + +type Z_InviteRemoteToChannelReturns struct { + A error +} + +func (g *apiRPCClient) InviteRemoteToChannel(channelID string, remoteID string, userID string) error { + _args := &Z_InviteRemoteToChannelArgs{channelID, remoteID, userID} + _returns := &Z_InviteRemoteToChannelReturns{} + if err := g.client.Call("Plugin.InviteRemoteToChannel", _args, _returns); err != nil { + log.Printf("RPC call to InviteRemoteToChannel API failed: %s", err.Error()) + } + return _returns.A +} + +func (s *apiRPCServer) InviteRemoteToChannel(args *Z_InviteRemoteToChannelArgs, returns *Z_InviteRemoteToChannelReturns) error { + if hook, ok := s.impl.(interface { + InviteRemoteToChannel(channelID string, remoteID string, userID string) error + }); ok { + returns.A = hook.InviteRemoteToChannel(args.A, args.B, args.C) + returns.A = encodableError(returns.A) + } else { + return encodableError(fmt.Errorf("API InviteRemoteToChannel called but not implemented.")) + } + return nil +} + +type Z_UninviteRemoteFromChannelArgs struct { + A string + B string +} + +type Z_UninviteRemoteFromChannelReturns struct { + A error +} + +func (g *apiRPCClient) UninviteRemoteFromChannel(channelID string, remoteID string) error { + _args := &Z_UninviteRemoteFromChannelArgs{channelID, remoteID} + _returns := &Z_UninviteRemoteFromChannelReturns{} + if err := g.client.Call("Plugin.UninviteRemoteFromChannel", _args, _returns); err != nil { + log.Printf("RPC call to UninviteRemoteFromChannel API failed: %s", err.Error()) + } + return _returns.A +} + +func (s *apiRPCServer) UninviteRemoteFromChannel(args *Z_UninviteRemoteFromChannelArgs, returns *Z_UninviteRemoteFromChannelReturns) error { + if hook, ok := s.impl.(interface { + UninviteRemoteFromChannel(channelID string, remoteID string) error + }); ok { + returns.A = hook.UninviteRemoteFromChannel(args.A, args.B) + returns.A = encodableError(returns.A) + } else { + return encodableError(fmt.Errorf("API UninviteRemoteFromChannel called but not implemented.")) + } + return nil +} diff --git a/server/public/plugin/hooks.go b/server/public/plugin/hooks.go index fd93e90e7f..dce8d40b1e 100644 --- a/server/public/plugin/hooks.go +++ b/server/public/plugin/hooks.go @@ -55,6 +55,8 @@ const ( MessageHasBeenDeletedID = 37 MessagesWillBeConsumedID = 38 ServeMetricsID = 39 + OnSharedChannelsSyncMsgID = 40 + OnSharedChannelsPingID = 41 TotalHooksID = iota ) @@ -330,4 +332,26 @@ type Hooks interface { // // Minimum server version: 9.2 ServeMetrics(c *Context, w http.ResponseWriter, r *http.Request) + + // OnSharedChannelsSyncMsg is invoked for plugins that wish to receive synchronization messages from the + // Shared Channels service for which they have been invited via InviteRemote. Each SyncMsg may contain + // multiple updates (posts, reactions, attachments, users) for a single channel. + // + // The cursor will be advanced based on the SyncResponse returned. + // + // Minimum server version: 9.5 + OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) + + // OnSharedChannelsPing is invoked for plugins to indicate the health of the plugin and the connection + // to the upstream service (e.g. MS Graph APIs). + // + // Return true to indicate all is well. + // + // Return false to indicate there is a problem with the plugin or connection to upstream service. + // Some number of failed pings will result in the plugin being marked offline and it will stop receiving + // OnSharedChannelsSyncMsg calls until it comes back online. The plugin will also appear offline in the status + // report via the `secure-connection status` slash command. + // + // Minimum server version: 9.5 + OnSharedChannelsPing(rc *model.RemoteCluster) bool } diff --git a/server/public/plugin/hooks_timer_layer_generated.go b/server/public/plugin/hooks_timer_layer_generated.go index d059b1ac11..f0a4bdc194 100644 --- a/server/public/plugin/hooks_timer_layer_generated.go +++ b/server/public/plugin/hooks_timer_layer_generated.go @@ -250,3 +250,17 @@ func (hooks *hooksTimerLayer) ServeMetrics(c *Context, w http.ResponseWriter, r hooks.hooksImpl.ServeMetrics(c, w, r) hooks.recordTime(startTime, "ServeMetrics", true) } + +func (hooks *hooksTimerLayer) OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) { + startTime := timePkg.Now() + _returnsA, _returnsB := hooks.hooksImpl.OnSharedChannelsSyncMsg(msg, rc) + hooks.recordTime(startTime, "OnSharedChannelsSyncMsg", _returnsB == nil) + return _returnsA, _returnsB +} + +func (hooks *hooksTimerLayer) OnSharedChannelsPing(rc *model.RemoteCluster) bool { + startTime := timePkg.Now() + _returnsA := hooks.hooksImpl.OnSharedChannelsPing(rc) + hooks.recordTime(startTime, "OnSharedChannelsPing", true) + return _returnsA +} diff --git a/server/public/plugin/plugintest/api.go b/server/public/plugin/plugintest/api.go index c09622a85c..370dd52aaf 100644 --- a/server/public/plugin/plugintest/api.go +++ b/server/public/plugin/plugintest/api.go @@ -2780,6 +2780,20 @@ func (_m *API) InstallPlugin(file io.Reader, replace bool) (*model.Manifest, *mo return r0, r1 } +// InviteRemoteToChannel provides a mock function with given fields: channelID, remoteID, userID +func (_m *API) InviteRemoteToChannel(channelID string, remoteID string, userID string) error { + ret := _m.Called(channelID, remoteID, userID) + + var r0 error + if rf, ok := ret.Get(0).(func(string, string, string) error); ok { + r0 = rf(channelID, remoteID, userID) + } else { + r0 = ret.Error(0) + } + + return r0 +} + // IsEnterpriseReady provides a mock function with given fields: func (_m *API) IsEnterpriseReady() bool { ret := _m.Called() @@ -3309,6 +3323,30 @@ func (_m *API) RegisterCommand(command *model.Command) error { return r0 } +// RegisterPluginForSharedChannels provides a mock function with given fields: opts +func (_m *API) RegisterPluginForSharedChannels(opts model.RegisterPluginOpts) (string, error) { + ret := _m.Called(opts) + + var r0 string + var r1 error + if rf, ok := ret.Get(0).(func(model.RegisterPluginOpts) (string, error)); ok { + return rf(opts) + } + if rf, ok := ret.Get(0).(func(model.RegisterPluginOpts) string); ok { + r0 = rf(opts) + } else { + r0 = ret.Get(0).(string) + } + + if rf, ok := ret.Get(1).(func(model.RegisterPluginOpts) error); ok { + r1 = rf(opts) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // RemovePlugin provides a mock function with given fields: id func (_m *API) RemovePlugin(id string) *model.AppError { ret := _m.Called(id) @@ -3731,6 +3769,60 @@ func (_m *API) SetUserStatusTimedDND(userId string, endtime int64) (*model.Statu return r0, r1 } +// ShareChannel provides a mock function with given fields: sc +func (_m *API) ShareChannel(sc *model.SharedChannel) (*model.SharedChannel, error) { + ret := _m.Called(sc) + + var r0 *model.SharedChannel + var r1 error + if rf, ok := ret.Get(0).(func(*model.SharedChannel) (*model.SharedChannel, error)); ok { + return rf(sc) + } + if rf, ok := ret.Get(0).(func(*model.SharedChannel) *model.SharedChannel); ok { + r0 = rf(sc) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*model.SharedChannel) + } + } + + if rf, ok := ret.Get(1).(func(*model.SharedChannel) error); ok { + r1 = rf(sc) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// SyncSharedChannel provides a mock function with given fields: channelID +func (_m *API) SyncSharedChannel(channelID string) error { + ret := _m.Called(channelID) + + var r0 error + if rf, ok := ret.Get(0).(func(string) error); ok { + r0 = rf(channelID) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// UninviteRemoteFromChannel provides a mock function with given fields: channelID, remoteID +func (_m *API) UninviteRemoteFromChannel(channelID string, remoteID string) error { + ret := _m.Called(channelID, remoteID) + + var r0 error + if rf, ok := ret.Get(0).(func(string, string) error); ok { + r0 = rf(channelID, remoteID) + } else { + r0 = ret.Error(0) + } + + return r0 +} + // UnregisterCommand provides a mock function with given fields: teamID, trigger func (_m *API) UnregisterCommand(teamID string, trigger string) error { ret := _m.Called(teamID, trigger) @@ -3745,6 +3837,44 @@ func (_m *API) UnregisterCommand(teamID string, trigger string) error { return r0 } +// UnregisterPluginForSharedChannels provides a mock function with given fields: pluginID +func (_m *API) UnregisterPluginForSharedChannels(pluginID string) error { + ret := _m.Called(pluginID) + + var r0 error + if rf, ok := ret.Get(0).(func(string) error); ok { + r0 = rf(pluginID) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// UnshareChannel provides a mock function with given fields: channelID +func (_m *API) UnshareChannel(channelID string) (bool, error) { + ret := _m.Called(channelID) + + var r0 bool + var r1 error + if rf, ok := ret.Get(0).(func(string) (bool, error)); ok { + return rf(channelID) + } + if rf, ok := ret.Get(0).(func(string) bool); ok { + r0 = rf(channelID) + } else { + r0 = ret.Get(0).(bool) + } + + if rf, ok := ret.Get(1).(func(string) error); ok { + r1 = rf(channelID) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // UpdateBotActive provides a mock function with given fields: botUserId, active func (_m *API) UpdateBotActive(botUserId string, active bool) (*model.Bot, *model.AppError) { ret := _m.Called(botUserId, active) @@ -3999,6 +4129,46 @@ func (_m *API) UpdatePreferencesForUser(userID string, preferences []model.Prefe return r0 } +// UpdateSharedChannel provides a mock function with given fields: sc +func (_m *API) UpdateSharedChannel(sc *model.SharedChannel) (*model.SharedChannel, error) { + ret := _m.Called(sc) + + var r0 *model.SharedChannel + var r1 error + if rf, ok := ret.Get(0).(func(*model.SharedChannel) (*model.SharedChannel, error)); ok { + return rf(sc) + } + if rf, ok := ret.Get(0).(func(*model.SharedChannel) *model.SharedChannel); ok { + r0 = rf(sc) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*model.SharedChannel) + } + } + + if rf, ok := ret.Get(1).(func(*model.SharedChannel) error); ok { + r1 = rf(sc) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// UpdateSharedChannelCursor provides a mock function with given fields: channelID, remoteID, cusror +func (_m *API) UpdateSharedChannelCursor(channelID string, remoteID string, cusror model.GetPostsSinceForSyncCursor) error { + ret := _m.Called(channelID, remoteID, cusror) + + var r0 error + if rf, ok := ret.Get(0).(func(string, string, model.GetPostsSinceForSyncCursor) error); ok { + r0 = rf(channelID, remoteID, cusror) + } else { + r0 = ret.Error(0) + } + + return r0 +} + // UpdateTeam provides a mock function with given fields: team func (_m *API) UpdateTeam(team *model.Team) (*model.Team, *model.AppError) { ret := _m.Called(team) diff --git a/server/public/plugin/plugintest/hooks.go b/server/public/plugin/plugintest/hooks.go index b9b84d22b7..9e8adc2322 100644 --- a/server/public/plugin/plugintest/hooks.go +++ b/server/public/plugin/plugintest/hooks.go @@ -311,6 +311,44 @@ func (_m *Hooks) OnSendDailyTelemetry() { _m.Called() } +// OnSharedChannelsPing provides a mock function with given fields: rc +func (_m *Hooks) OnSharedChannelsPing(rc *model.RemoteCluster) bool { + ret := _m.Called(rc) + + var r0 bool + if rf, ok := ret.Get(0).(func(*model.RemoteCluster) bool); ok { + r0 = rf(rc) + } else { + r0 = ret.Get(0).(bool) + } + + return r0 +} + +// OnSharedChannelsSyncMsg provides a mock function with given fields: msg, rc +func (_m *Hooks) OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) { + ret := _m.Called(msg, rc) + + var r0 model.SyncResponse + var r1 error + if rf, ok := ret.Get(0).(func(*model.SyncMsg, *model.RemoteCluster) (model.SyncResponse, error)); ok { + return rf(msg, rc) + } + if rf, ok := ret.Get(0).(func(*model.SyncMsg, *model.RemoteCluster) model.SyncResponse); ok { + r0 = rf(msg, rc) + } else { + r0 = ret.Get(0).(model.SyncResponse) + } + + if rf, ok := ret.Get(1).(func(*model.SyncMsg, *model.RemoteCluster) error); ok { + r1 = rf(msg, rc) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // OnWebSocketConnect provides a mock function with given fields: webConnID, userID func (_m *Hooks) OnWebSocketConnect(webConnID string, userID string) { _m.Called(webConnID, userID) diff --git a/server/public/plugin/product_hooks_generated.go b/server/public/plugin/product_hooks_generated.go index 6d7dc11a3a..7452b3f6e6 100644 --- a/server/public/plugin/product_hooks_generated.go +++ b/server/public/plugin/product_hooks_generated.go @@ -139,6 +139,14 @@ type ServeMetricsIFace interface { ServeMetrics(c *Context, w http.ResponseWriter, r *http.Request) } +type OnSharedChannelsSyncMsgIFace interface { + OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) +} + +type OnSharedChannelsPingIFace interface { + OnSharedChannelsPing(rc *model.RemoteCluster) bool +} + type HooksAdapter struct { implemented map[int]struct{} productHooks any @@ -431,6 +439,24 @@ func NewAdapter(productHooks any) (*HooksAdapter, error) { return nil, errors.New("hook has ServeMetrics method but does not implement plugin.ServeMetrics interface") } + // Assessing the type of the productHooks if it individually implements OnSharedChannelsSyncMsg interface. + tt = reflect.TypeOf((*OnSharedChannelsSyncMsgIFace)(nil)).Elem() + + if ft.Implements(tt) { + a.implemented[OnSharedChannelsSyncMsgID] = struct{}{} + } else if _, ok := ft.MethodByName("OnSharedChannelsSyncMsg"); ok { + return nil, errors.New("hook has OnSharedChannelsSyncMsg method but does not implement plugin.OnSharedChannelsSyncMsg interface") + } + + // Assessing the type of the productHooks if it individually implements OnSharedChannelsPing interface. + tt = reflect.TypeOf((*OnSharedChannelsPingIFace)(nil)).Elem() + + if ft.Implements(tt) { + a.implemented[OnSharedChannelsPingID] = struct{}{} + } else if _, ok := ft.MethodByName("OnSharedChannelsPing"); ok { + return nil, errors.New("hook has OnSharedChannelsPing method but does not implement plugin.OnSharedChannelsPing interface") + } + return a, nil } @@ -712,3 +738,21 @@ func (a *HooksAdapter) ServeMetrics(c *Context, w http.ResponseWriter, r *http.R a.productHooks.(ServeMetricsIFace).ServeMetrics(c, w, r) } + +func (a *HooksAdapter) OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) { + if _, ok := a.implemented[OnSharedChannelsSyncMsgID]; !ok { + panic("product hooks must implement OnSharedChannelsSyncMsg") + } + + return a.productHooks.(OnSharedChannelsSyncMsgIFace).OnSharedChannelsSyncMsg(msg, rc) + +} + +func (a *HooksAdapter) OnSharedChannelsPing(rc *model.RemoteCluster) bool { + if _, ok := a.implemented[OnSharedChannelsPingID]; !ok { + panic("product hooks must implement OnSharedChannelsPing") + } + + return a.productHooks.(OnSharedChannelsPingIFace).OnSharedChannelsPing(rc) + +}