Files
team/server/public/plugin/client_rpc.go
Doug LauderandGitHub d78f7d7048 Cherry-pick Shared Channel plugin API fixes to release-11.7 (#36597)
* MM-68339: Add XML struct tags and multi-remote registration for shared channels plugin API  (#36126)

* Add XML struct tags and multi-remote registration for shared channels plugin API

  Phase 1: Add xml struct tags to model types used in SyncMsg (Post, User,
  Reaction, Status, PostAcknowledgement, FileInfo, SyncResponse,
  MembershipChangeMsg). Add custom MarshalXML/UnmarshalXML for SyncMsg
  (Users map, MentionTransforms map), StringMap, and StringInterface.
  Exclude Post.Metadata, PrevStatus, and server-internal FileInfo fields
  from XML. JSON serialization is unaffected.

  Phase 2: Lift the one-remote-per-plugin constraint so plugins can
  register multiple remotes with different SiteURLs. Add SiteURL field to
  RegisterPluginOpts (defaults to "plugin_<PluginID>" for backward
  compatibility). Add GetAllByPluginID and GetBySiteURL store methods.
  Rewrite registration to dedup by SiteURL instead of PluginID. Add
  UnregisterPluginRemoteForSharedChannels for single-remote removal with
  plugin ownership validation. Validate SiteURL is non-empty in
  RemoteCluster.IsValid. Simplify IsPlugin() to check PluginID only.

(cherry picked from commit a760aefcd0)

* MM-68339: slugify RemoteCluster.Name in plugin registration (#36309)

The new TestRegisterPluginForSharedChannels tests added in #36126 broke
  master CI because RegisterPluginForSharedChannels assigned opts.Displayname
  directly to RemoteCluster.Name, which IsValid validates against the slug
  regex ^[a-zA-Z0-9.\-_]+$. Display names with spaces (e.g. "legacy plugin")
  fail validation. The tests didn't run in the PR's final CI shard and the
  issue surfaced post-merge.

  Add CleanRemoteName to the public model, mirroring CleanTeamName and
  CleanUsername: lowercase, replace spaces and other disallowed characters
  with hyphens, trim, truncate to RemoteNameMaxLength, fall back to NewId
  when the result is empty. Use it in RegisterPluginForSharedChannels so
  Name is always slug-valid while DisplayName keeps the human-readable label.
  This also lets real plugins register with display names containing spaces.

(cherry picked from commit 078f6b31b5)

* MM-68622: start inter-cluster services before plugin activation (#36366)

* MM-68622: start inter-cluster services before plugin activation

  Move startInterClusterServices from the end of Server.Start() to the
  beginning, before Channels().Start() initializes plugins. This lets
  plugins call shared channels APIs (ShareChannel, InviteRemoteToChannel,
  UninviteRemoteFromChannel, UnshareChannel, UpdateSharedChannel,
  CheckCanInviteToSharedChannel) during OnActivate instead of failing with
  "Shared Channels Service is disabled".

  Side-effect analysis:

  * Plugin API gating: getSharedChannelsService in
    channels/app/shared_channel.go:32 only requires the service to be
    non-nil. The plugin-facing wrappers all pass ensureIsActive=false, so
    Active() is bypassed. Once SetSharedChannelService runs, calls succeed
    on both leader and follower nodes. This is the fix path.

  * Multi-node leader timing: the enterprise cluster impl in
    enterprise/cluster/cluster.go:70 initializes currentLeader="", so
    IsLeader() returns false before StartInterNodeCommunication runs. The
    immediate onClusterLeaderChange in scs.Start at
    platform/services/sharedchannel/service.go:151 therefore takes the
    pause path, which is a no-op since the service was never active. When
    memberlist.Create fires NotifyJoin for the local node,
    addPotentialLeader runs and InvokeClusterLeaderChangedListeners drives
    the registered listener to resume() the sync loop on the elected
    leader. End state matches the prior ordering.

  * Single-node: IsLeader() returns true unconditionally per
    channels/app/platform/cluster.go:33, so SharedChannelSyncHandler is
    active during plugin OnActivate. Events emitted by plugins during
    activation (posts to shared channels, DM creation) now flow through
    sync where they were previously dropped. This is intended correctness,
    not a regression.

  * Transport and handlers: api4 remote-cluster routes are registered
    before Server.Start, so HTTP handlers exist when rcs.Start runs early.
    rcs and scs do not send cluster-broadcast messages during Start; they
    only register topic listeners on the rcs transport, which is
    independent of cluster gossip. registerClusterHandlers ordering is
    unaffected.

  * Config: scs reads only ConnectedWorkspacesSettings and the License at
    construction, both stable from the initial config load. ReloadConfig
    at server.go:912 has no bearing on inter-cluster service init.

  Errors from startInterClusterServices remain logged and non-fatal,
  matching prior behavior.

(cherry picked from commit 4e967cb7ad)

* MM-68536: Show actual remote names in system console channel list (#36298)

* MM-68536: Show actual remote names in system console channel list

  The system console "Channels" list rendered SharedChannelIndicator
  without a remoteNames prop, so every shared channel showed the generic
  "Shared with trusted organizations" fallback even after PR 35908.

  Add a small connected wrapper that selects remote names via
  getRemoteNamesForChannel and dispatches fetchChannelRemotes on mount,
  mirroring the pattern used by the LHS sidebar. The shared_channel_remote_updated
  websocket event added in PR 35908 already refreshes the same Redux slice,
  so the system console list now stays in sync automatically.

(cherry picked from commit 15d7fbd8cb)

* [MM-68697] Preserve sender file ID in plugin-relayed shared channel attachments (#36468)

Set ReqFileId on the UploadSession created in
  ReceiveSharedChannelAttachmentSyncMsg so the attachment is persisted
  under the sender's file ID. Without this, the receiving server stored
  the bytes under a freshly generated ID while the synced post's FileIds
  still referenced the sender's ID, leaving the attachment invisible in
  the UI even though the file and FileInfo row existed on disk.

  Mirrors the existing cluster-to-cluster path in
  platform/services/sharedchannel/attachment.go.

  Also tightens TestPluginAPIReceiveSharedChannelAttachmentSyncMsg to
  pass a sender-side fi.Id and assert the saved FileInfo keeps it. The
  prior assertion only checked that some ID was assigned, which the
  buggy code also satisfied.

(cherry picked from commit a72b1db4c5)

* MM-68705 - Order-tolerant Shared Channel plugin API's for receiving attachments (#36486)

* MM-68705 - Make ReceiveSharedChannelAttachmentSyncMsg order tolerant

  Allow plugin remotes to invoke ReceiveSharedChannelAttachmentSyncMsg
  and ReceiveSharedChannelSyncMsg in either order without losing the
  post/file binding, and make repeat deliveries of the same file id
  idempotent.

  Two localised changes inside ReceiveSharedChannelAttachmentSyncMsg:

  - Idempotency check before CreateUploadSession. If a FileInfo with
    the sender's id already exists for the same channel and creator,
    return it instead of inserting a duplicate (which would violate
    the FileInfo PK and force the caller to retry indefinitely after
    any transient ack failure). A mismatched channel or creator on the
    same id is rejected.

  - Lazy bind to post after UploadData. When the matching post is
    already present (post-then-file ordering), CreatePost will have
    stripped the unmatched file id from Post.FileIds; AttachToPost
    plus Post.Overwrite restore the binding. When the post is not
    yet present (file-then-post ordering), the FileInfo is left
    unbound so the eventual post arrival's CreatePost path binds it.
    The post is fetched first because AttachToPost is a blind UPDATE
    that does not validate post existence, so calling it before the
    post exists would orphan the FileInfo.

  No other paths change. Cluster (non-plugin) shared-channel attachments,
  ReceiveSharedChannelProfileImageSyncMsg, and UI uploads are unaffected.

  New tests in shared_channel_test.go cover both orderings, repeated
  receive success and rejection on channel/creator mismatch, and the
  empty-PostId no-op.

(cherry picked from commit 5c58953e8e)

* MM-68708 - Fix TestCreatePost shared DM/GM subtests when env pins feature flag (#36488)

The two shared DM/GM subtests in TestCreatePost rely on
  FeatureFlags.EnableSharedChannelsDMs being false, but the config store
  reapplies MM_FEATUREFLAGS_* env overrides on every Set, so an
  UpdateConfig pin gets clobbered. Use t.Setenv to force the flag false
  for the duration of each subtest, and drop the parent's
  mainHelper.Parallel(t) so Go's "no Setenv under a parallel ancestor"
  rule is satisfied. Sibling subtests still parallelize via their own
  mainHelper.Parallel(t) calls.

(cherry picked from commit a98e01ebac)

* MM-68838: Ping a restored plugin remote immediately on re-register (#36592)

* MM-68838: ping restored plugin remote immediately on re-register

  RegisterPluginForSharedChannels' restore branch updated the row but did
  not call PingNow, leaving the restored remote offline until the next
  pingLoop tick (up to PingFreq, default 1 minute). The new-connection
  branch already calls PingNow; the restore branch now mirrors it so
  sync attempts immediately after a plugin restart no longer fail with
  "offline remote cluster".

* MM-68838: gob-encode error returns in apiRPCServer.ReceiveSharedChannelAttachmentSyncMsg

  The apiRPCServer wrapper for ReceiveSharedChannelAttachmentSyncMsg
  assigned the hook's error return directly to the gob-encoded response
  struct. When the framework's App.ReceiveSharedChannelAttachmentSyncMsg
  returned an error wrapped with %w (*fmt.wrapError, an unexported type),
  gob refused to encode it and the RPC server broke the connection with
  "type not registered for interface: fmt.wrapError".

  Every subsequent plugin/server RPC call then returned the zero-value
  response struct, causing plugins that dereferenced the nil returns to
  crash.

  Apply the existing encodableError() helper so the returned error
  becomes a gob-safe ErrorString, matching every other apiRPCServer
  method in this file.

(cherry picked from commit ea7f488c64)
2026-05-19 13:55:15 -04:00

1501 lines
46 KiB
Go

// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.txt for license information.
//go:generate go run interface_generator/main.go
package plugin
import (
"bytes"
"database/sql"
"database/sql/driver"
"encoding/gob"
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"net/rpc"
"net/url"
"os"
"reflect"
"sync"
"github.com/hashicorp/go-plugin"
"github.com/lib/pq"
"github.com/mattermost/mattermost/server/public/model"
"github.com/mattermost/mattermost/server/public/shared/mlog"
)
// Plugin RPC Architecture
//
// Mattermost plugins run as separate OS processes for isolation and safety, using
// HashiCorp's go-plugin library. Communication between the server and plugins is
// bidirectional via RPC:
//
// ┌─────────────────────────┐ ┌─────────────────────────┐
// │ Mattermost Server │ │ Plugin Process │
// │ │ │ │
// │ ┌───────────────────┐ │ hooks (calls) │ ┌───────────────────┐ │
// │ │ hooksRPCClient │──┼───────────────────►│ │ hooksRPCServer │ │
// │ └───────────────────┘ │ │ └───────────────────┘ │
// │ │ │ │
// │ ┌───────────────────┐ │ API (callbacks) │ ┌───────────────────┐ │
// │ │ apiRPCServer │◄─┼────────────────────┼──│ apiRPCClient │ │
// │ └───────────────────┘ │ │ └───────────────────┘ │
// └─────────────────────────┘ └─────────────────────────┘
//
// - Server → Plugin (Hooks): hooksRPCClient serializes hook calls and sends them
// to hooksRPCServer in the plugin process, which delegates to the plugin implementation.
//
// - Plugin → Server (API): apiRPCClient in the plugin serializes API calls and sends
// them to apiRPCServer in the server, which delegates to the Mattermost API.
//
// The MuxBroker enables multiplexed streaming connections over a single RPC connection,
// which is essential for efficiently streaming HTTP bodies, file uploads, and other
// large data transfers without buffering everything in memory.
var hookNameToId = make(map[string]int)
// hooksRPCClient is the client-side RPC proxy that runs in the Mattermost server process and connects to the [hooksRPCServer] on the plugin side.
// It implements the Hooks interface and forwards hook invocations to plugins running in
// separate processes via RPC.
//
// When Mattermost needs to call a plugin hook (e.g., MessageWillBePosted), it calls the
// corresponding method on hooksRPCClient, which serializes the arguments and makes an
// RPC call to the plugin process where hooksRPCServer receives and handles it.
//
// The struct also holds references to the API and Driver implementations that will be
// exposed to the plugin via apiRPCServer when the plugin is activated.
type hooksRPCClient struct {
client *rpc.Client
log *mlog.Logger
muxBroker *plugin.MuxBroker
apiImpl API
driver Driver
implemented [TotalHooksID]bool
doneWg sync.WaitGroup
}
// hooksRPCServer is the server-side RPC handler that runs in the plugin process and receives requests from [hooksRPCClient].
// It receives hook invocations from hooksRPCClient (in the Mattermost server) and
// delegates them to the actual plugin implementation.
//
// During plugin activation (OnActivate), it establishes a reverse RPC connection
// back to the server, creating an apiRPCClient that the plugin uses to call
// Mattermost APIs.
type hooksRPCServer struct {
impl any
muxBroker *plugin.MuxBroker
apiRPCClient *apiRPCClient
}
// Implements hashicorp/go-plugin/plugin.Plugin interface to connect the hooks of a plugin
type hooksPlugin struct {
hooks any
apiImpl API
driverImpl Driver
log *mlog.Logger
}
func (p *hooksPlugin) Server(b *plugin.MuxBroker) (any, error) {
return &hooksRPCServer{impl: p.hooks, muxBroker: b}, nil
}
func (p *hooksPlugin) Client(b *plugin.MuxBroker, client *rpc.Client) (any, error) {
return &hooksRPCClient{
client: client,
log: p.log,
muxBroker: b,
apiImpl: p.apiImpl,
driver: p.driverImpl,
}, nil
}
// apiRPCClient is the client-side RPC proxy that runs in the plugin process and connects to the [apiRPCServer] on the Mattermost server side.
// It implements the API interface and allows plugins to call Mattermost server
// APIs (e.g., GetUser, CreatePost) by forwarding requests via RPC to apiRPCServer.
//
// This is created during plugin activation and injected into the plugin via SetAPI().
type apiRPCClient struct {
client *rpc.Client
muxBroker *plugin.MuxBroker
}
// apiRPCServer is the server-side RPC handler that runs in the Mattermost server process and receives requests from [apiRPCClient].
// It receives API calls from plugins (via apiRPCClient) and delegates them to the actual
// Mattermost API implementation.
//
// This enables plugins to interact with Mattermost functionality like users, posts,
// channels, and configuration through a well-defined API boundary.
type apiRPCServer struct {
impl API
muxBroker *plugin.MuxBroker
}
// ErrorString is a fallback for sending unregistered implementations of the error interface across
// rpc. For example, the errorString type from the github.com/pkg/errors package cannot be
// registered since it is not exported, but this precludes common error handling paradigms.
// ErrorString merely preserves the string description of the error, while satisfying the error
// interface itself to allow other registered types (such as model.AppError) to be sent unmodified.
type ErrorString struct {
Code int // Code to map to various error variables
Err string
}
func (e ErrorString) Error() string {
return e.Err
}
func encodableError(err error) error {
if err == nil {
return nil
}
if _, ok := err.(*model.AppError); ok {
return err
}
if _, ok := err.(*pq.Error); ok {
return err
}
ret := &ErrorString{
Err: err.Error(),
}
switch err {
case io.EOF:
ret.Code = 1
case sql.ErrNoRows:
ret.Code = 2
case sql.ErrConnDone:
ret.Code = 3
case sql.ErrTxDone:
ret.Code = 4
case driver.ErrSkip:
ret.Code = 5
case driver.ErrBadConn:
ret.Code = 6
case driver.ErrRemoveArgument:
ret.Code = 7
}
return ret
}
func decodableError(err error) error {
if encErr, ok := err.(*ErrorString); ok {
switch encErr.Code {
case 1:
return io.EOF
case 2:
return sql.ErrNoRows
case 3:
return sql.ErrConnDone
case 4:
return sql.ErrTxDone
case 5:
return driver.ErrSkip
case 6:
return driver.ErrBadConn
case 7:
return driver.ErrRemoveArgument
}
}
return err
}
// Registering some types used by MM for encoding/gob used by rpc
func init() {
// Use RegisterName with the old type name to maintain backward compatibility
// with plugins compiled against older server versions where SlackAttachment
// was the concrete struct type (now renamed to MessageAttachment with
// SlackAttachment as a type alias). Without this, gob decoding fails because
// old plugins encode the type as "[]*model.SlackAttachment" but the server
// would only recognize "[]*model.MessageAttachment". The full type string
// including "[]*" prefix is required because gob uses reflect.Type.String()
// as the wire format type name.
gob.RegisterName("[]*model.SlackAttachment", []*model.MessageAttachment{})
gob.Register([]any{})
gob.Register(map[string]any{})
gob.Register(&model.AppError{})
gob.Register(&pq.Error{})
gob.Register(&ErrorString{})
gob.Register(&model.AutocompleteDynamicListArg{})
gob.Register(&model.AutocompleteStaticListArg{})
gob.Register(&model.AutocompleteTextArg{})
gob.Register(&model.PreviewPost{})
gob.Register(model.PropertyOptions[*model.PluginPropertyOption]{})
}
// These enforce compile time checks to make sure types implement the interface
// If you are getting an error here, you probably need to run `make pluginapi` to
// autogenerate RPC glue code
var (
_ plugin.Plugin = &hooksPlugin{}
_ Hooks = &hooksRPCClient{}
)
//
// Below are special cases for hooks or APIs that can not be auto generated
//
func (g *hooksRPCClient) Implemented() (impl []string, err error) {
err = g.client.Call("Plugin.Implemented", struct{}{}, &impl)
for _, hookName := range impl {
if hookId, ok := hookNameToId[hookName]; ok {
g.implemented[hookId] = true
}
}
return
}
// Implemented replies with the names of the hooks that are implemented.
func (s *hooksRPCServer) Implemented(args struct{}, reply *[]string) error {
ifaceType := reflect.TypeFor[Hooks]()
implType := reflect.TypeOf(s.impl)
selfType := reflect.TypeFor[*hooksRPCServer]()
var methods []string
for i := 0; i < ifaceType.NumMethod(); i++ {
method := ifaceType.Method(i)
m, ok := implType.MethodByName(method.Name)
if !ok {
continue
} else if m.Type.NumIn() != method.Type.NumIn()+1 {
continue
} else if m.Type.NumOut() != method.Type.NumOut() {
continue
}
match := true
for j := 0; j < method.Type.NumIn(); j++ {
if m.Type.In(j+1) != method.Type.In(j) {
match = false
break
}
}
for j := 0; j < method.Type.NumOut(); j++ {
if m.Type.Out(j) != method.Type.Out(j) {
match = false
break
}
}
if !match {
continue
}
if _, ok := selfType.MethodByName(method.Name); !ok {
continue
}
methods = append(methods, method.Name)
}
*reply = methods
return encodableError(nil)
}
type Z_OnActivateArgs struct {
APIMuxId uint32
DriverMuxId uint32
}
type Z_OnActivateReturns struct {
A error
}
func (g *hooksRPCClient) OnActivate() error {
muxId := g.muxBroker.NextId()
g.doneWg.Go(func() {
g.muxBroker.AcceptAndServe(muxId, &apiRPCServer{
impl: g.apiImpl,
muxBroker: g.muxBroker,
})
})
nextID := g.muxBroker.NextId()
g.doneWg.Go(func() {
g.muxBroker.AcceptAndServe(nextID, &dbRPCServer{
dbImpl: g.driver,
})
})
_args := &Z_OnActivateArgs{
APIMuxId: muxId,
DriverMuxId: nextID,
}
_returns := &Z_OnActivateReturns{}
if err := g.client.Call("Plugin.OnActivate", _args, _returns); err != nil {
g.log.Error("RPC call to OnActivate plugin failed.", mlog.Err(err))
}
return _returns.A
}
func (s *hooksRPCServer) OnActivate(args *Z_OnActivateArgs, returns *Z_OnActivateReturns) error {
connection, err := s.muxBroker.Dial(args.APIMuxId)
if err != nil {
return err
}
conn2, err := s.muxBroker.Dial(args.DriverMuxId)
if err != nil {
return err
}
s.apiRPCClient = &apiRPCClient{
client: rpc.NewClient(connection),
muxBroker: s.muxBroker,
}
dbClient := &dbRPCClient{
client: rpc.NewClient(conn2),
}
if mmplugin, ok := s.impl.(interface {
SetAPI(api API)
SetDriver(driver Driver)
}); ok {
mmplugin.SetAPI(s.apiRPCClient)
mmplugin.SetDriver(dbClient)
}
if mmplugin, ok := s.impl.(interface {
OnConfigurationChange() error
}); ok {
if err := mmplugin.OnConfigurationChange(); err != nil {
fmt.Fprintf(os.Stderr, "[ERROR] call to OnConfigurationChange failed, error: %v", err.Error())
}
}
// Capture output of standard logger because go-plugin
// redirects it.
log.SetOutput(os.Stderr)
if hook, ok := s.impl.(interface {
OnActivate() error
}); ok {
returns.A = encodableError(hook.OnActivate())
}
return nil
}
type Z_LoadPluginConfigurationArgsArgs struct{}
type Z_LoadPluginConfigurationArgsReturns struct {
A []byte
}
func (g *apiRPCClient) LoadPluginConfiguration(dest any) error {
_args := &Z_LoadPluginConfigurationArgsArgs{}
_returns := &Z_LoadPluginConfigurationArgsReturns{}
if err := g.client.Call("Plugin.LoadPluginConfiguration", _args, _returns); err != nil {
log.Printf("RPC call to LoadPluginConfiguration API failed: %s", err.Error())
}
if err := json.Unmarshal(_returns.A, dest); err != nil {
log.Printf("LoadPluginConfiguration API failed to unmarshal: %s", err.Error())
}
return nil
}
func (s *apiRPCServer) LoadPluginConfiguration(args *Z_LoadPluginConfigurationArgsArgs, returns *Z_LoadPluginConfigurationArgsReturns) error {
var config any
if hook, ok := s.impl.(interface {
LoadPluginConfiguration(dest any) error
}); ok {
if err := hook.LoadPluginConfiguration(&config); err != nil {
return err
}
}
b, err := json.Marshal(config)
if err != nil {
return err
}
returns.A = b
return nil
}
func init() {
hookNameToId["ServeHTTP"] = ServeHTTPID
}
// Using a subset of http.Request prevents a known incompatibility when decoding Go v1.23+ gob-encoded x509.Certificate
// structs from Go v1.22 compiled plugins. These come from http.Request.TLS field (*tls.ConnectionState).
type HTTPRequestSubset struct {
Method string
URL *url.URL
Proto string
ProtoMajor int
ProtoMinor int
Header http.Header
Host string
RemoteAddr string
RequestURI string
Body io.ReadCloser
}
func (r *HTTPRequestSubset) GetHTTPRequest() *http.Request {
return &http.Request{
Method: r.Method,
URL: r.URL,
Proto: r.Proto,
ProtoMajor: r.ProtoMajor,
ProtoMinor: r.ProtoMinor,
Header: r.Header,
Host: r.Host,
RemoteAddr: r.RemoteAddr,
RequestURI: r.RequestURI,
Body: r.Body,
}
}
type Z_ServeHTTPArgs struct {
ResponseWriterStream uint32
Request *HTTPRequestSubset
Context *Context
RequestBodyStream uint32
}
func (g *hooksRPCClient) ServeHTTP(c *Context, w http.ResponseWriter, r *http.Request) {
if !g.implemented[ServeHTTPID] {
http.NotFound(w, r)
return
}
serveHTTPStreamId := g.muxBroker.NextId()
go func() {
connection, err := g.muxBroker.Accept(serveHTTPStreamId)
if err != nil {
g.log.Error("Plugin failed to ServeHTTP, muxBroker couldn't accept connection", mlog.Uint("serve_http_stream_id", serveHTTPStreamId), mlog.Err(err))
return
}
defer connection.Close()
rpcServer := rpc.NewServer()
if err := rpcServer.RegisterName("Plugin", &httpResponseWriterRPCServer{w: w, log: g.log}); err != nil {
g.log.Error("Plugin failed to ServeHTTP, couldn't register RPC name", mlog.Err(err))
return
}
rpcServer.ServeConn(connection)
}()
requestBodyStreamId := uint32(0)
if r.Body != nil {
requestBodyStreamId = g.muxBroker.NextId()
go func() {
bodyConnection, err := g.muxBroker.Accept(requestBodyStreamId)
if err != nil {
g.log.Error("Plugin failed to ServeHTTP, muxBroker couldn't Accept request body connection", mlog.Err(err))
return
}
defer bodyConnection.Close()
serveIOReader(r.Body, bodyConnection)
}()
}
forwardedRequest := &HTTPRequestSubset{
Method: r.Method,
URL: r.URL,
Proto: r.Proto,
ProtoMajor: r.ProtoMajor,
ProtoMinor: r.ProtoMinor,
Header: r.Header,
Host: r.Host,
RemoteAddr: r.RemoteAddr,
RequestURI: r.RequestURI,
}
if err := g.client.Call("Plugin.ServeHTTP", Z_ServeHTTPArgs{
Context: c,
ResponseWriterStream: serveHTTPStreamId,
Request: forwardedRequest,
RequestBodyStream: requestBodyStreamId,
}, nil); err != nil {
g.log.Error("Plugin failed to ServeHTTP, RPC call failed", mlog.Err(err))
http.Error(w, "500 internal server error", http.StatusInternalServerError)
}
}
func (s *hooksRPCServer) ServeHTTP(args *Z_ServeHTTPArgs, returns *struct{}) error {
connection, err := s.muxBroker.Dial(args.ResponseWriterStream)
if err != nil {
fmt.Fprintf(os.Stderr, "[ERROR] Can't connect to remote response writer stream, error: %v", err.Error())
return err
}
w := connectHTTPResponseWriter(connection)
defer w.Close()
r := args.Request
if args.RequestBodyStream != 0 {
connection, err := s.muxBroker.Dial(args.RequestBodyStream)
if err != nil {
fmt.Fprintf(os.Stderr, "[ERROR] Can't connect to remote request body stream, error: %v", err.Error())
return err
}
r.Body = connectIOReader(connection)
} else {
r.Body = io.NopCloser(&bytes.Buffer{})
}
defer r.Body.Close()
httpReq := r.GetHTTPRequest()
if hook, ok := s.impl.(interface {
ServeHTTP(c *Context, w http.ResponseWriter, r *http.Request)
}); ok {
hook.ServeHTTP(args.Context, w, httpReq)
} else {
http.NotFound(w, httpReq)
}
return nil
}
// PluginHTTPStream - Streaming version of PluginHTTP that uses MuxBroker for streaming request/response bodies.
// This avoids buffering large payloads in memory.
// Legacy buffered structs (kept for backward compatibility with old servers)
type Z_PluginHTTPArgs struct {
Request *HTTPRequestSubset
RequestBody []byte
}
type Z_PluginHTTPReturns struct {
Response *http.Response
ResponseBody []byte
}
// New streaming structs
type Z_PluginHTTPStreamArgs struct {
ResponseBodyStream uint32
Request *HTTPRequestSubset
RequestBodyStream uint32
}
type Z_PluginHTTPStreamReturns struct {
StatusCode int
Header http.Header
}
func (g *apiRPCClient) PluginHTTP(request *http.Request) *http.Response {
// Try to use the streaming version first (if server supports it)
// Fall back to buffered version if not available (signaled by nil)
response, err := g.pluginHTTPStream(request)
if err != nil {
// If we error for some other reason other than stream not being
// implemented just report and fail
log.Print(err.Error())
return nil
}
if response != nil {
return response
}
// Fallback to buffered version
return g.pluginHTTPBuffered(request)
}
// pluginHTTPStream attempts to use the new streaming endpoint
func (g *apiRPCClient) pluginHTTPStream(request *http.Request) (*http.Response, error) {
// Set up request body stream
requestBodyStreamId := uint32(0)
if request.Body != nil {
requestBodyStreamId = g.muxBroker.NextId()
go func() {
bodyConnection, err := g.muxBroker.Accept(requestBodyStreamId)
if err != nil {
log.Printf("Plugin failed to accept request body connection for PluginHTTPStream: %s", err.Error())
return
}
defer bodyConnection.Close()
serveIOReader(request.Body, bodyConnection)
}()
}
// Set up response body stream
responseBodyStreamId := g.muxBroker.NextId()
responsePipe := make(chan io.ReadCloser, 1)
go func() {
connection, err := g.muxBroker.Accept(responseBodyStreamId)
if err != nil {
log.Printf("Plugin failed to accept response body connection for PluginHTTPStream: %s", err.Error())
responsePipe <- nil
return
}
// Don't close connection here - it will be closed when response body is read
responsePipe <- connectIOReader(connection)
}()
forwardedRequest := &HTTPRequestSubset{
Method: request.Method,
URL: request.URL,
Proto: request.Proto,
ProtoMajor: request.ProtoMajor,
ProtoMinor: request.ProtoMinor,
Header: request.Header,
Host: request.Host,
RemoteAddr: request.RemoteAddr,
RequestURI: request.RequestURI,
}
_args := &Z_PluginHTTPStreamArgs{
ResponseBodyStream: responseBodyStreamId,
Request: forwardedRequest,
RequestBodyStream: requestBodyStreamId,
}
_returns := &Z_PluginHTTPStreamReturns{}
if err := g.client.Call("Plugin.PluginHTTPStream", _args, _returns); err != nil {
// If the method doesn't exist, return nil to trigger fallback
if err.Error() == "rpc: can't find method Plugin.PluginHTTPStream" {
return nil, nil
}
return nil, fmt.Errorf("RPC call to PluginHTTPStream API failed: %w", err)
}
// Wait for response body reader
responseBody := <-responsePipe
if responseBody == nil {
return nil, fmt.Errorf("Failed to get response body stream for PluginHTTPStream")
}
// Create response with streamed body
response := &http.Response{
StatusCode: _returns.StatusCode,
Header: _returns.Header,
Body: responseBody,
Proto: request.Proto,
ProtoMajor: request.ProtoMajor,
ProtoMinor: request.ProtoMinor,
}
return response, nil
}
// pluginHTTPBuffered is the original buffered implementation
func (g *apiRPCClient) pluginHTTPBuffered(request *http.Request) *http.Response {
forwardedRequest := &HTTPRequestSubset{
Method: request.Method,
URL: request.URL,
Proto: request.Proto,
ProtoMajor: request.ProtoMajor,
ProtoMinor: request.ProtoMinor,
Header: request.Header,
Host: request.Host,
RemoteAddr: request.RemoteAddr,
RequestURI: request.RequestURI,
}
_args := &Z_PluginHTTPArgs{
Request: forwardedRequest,
}
if request.Body != nil {
requestBody, err := io.ReadAll(request.Body)
if err != nil {
log.Printf("RPC call to PluginHTTP API failed: %s", err.Error())
return nil
}
request.Body.Close()
request.Body = nil
_args.RequestBody = requestBody
}
_returns := &Z_PluginHTTPReturns{}
if err := g.client.Call("Plugin.PluginHTTP", _args, _returns); err != nil {
log.Printf("RPC call to PluginHTTP API failed: %s", err.Error())
return nil
}
_returns.Response.Body = io.NopCloser(bytes.NewBuffer(_returns.ResponseBody))
return _returns.Response
}
func (s *apiRPCServer) PluginHTTPStream(args *Z_PluginHTTPStreamArgs, returns *Z_PluginHTTPStreamReturns) error {
responseConnection, err := s.muxBroker.Dial(args.ResponseBodyStream)
if err != nil {
return encodableError(fmt.Errorf("can't connect to remote response body stream: %w", err))
}
// Connect to request body stream
r := args.Request
if args.RequestBodyStream != 0 {
requestConnection, err := s.muxBroker.Dial(args.RequestBodyStream)
if err != nil {
return encodableError(fmt.Errorf("can't connect to remote request body stream: %w", err))
}
r.Body = connectIOReader(requestConnection)
} else {
r.Body = io.NopCloser(&bytes.Buffer{})
}
httpReq := r.GetHTTPRequest()
// Call the PluginHTTP implementation
if hook, ok := s.impl.(interface {
PluginHTTP(request *http.Request) *http.Response
}); ok {
response := hook.PluginHTTP(httpReq)
if response != nil {
returns.StatusCode = response.StatusCode
returns.Header = response.Header
// Connect to response body stream and stream the response body
go func() {
defer r.Body.Close()
if response.Body != nil {
// Stream the response body through the connection
if _, err := io.Copy(responseConnection, response.Body); err != nil {
log.Printf("error streaming response body: %s", err.Error())
}
response.Body.Close()
}
responseConnection.Close()
}()
} else {
r.Body.Close()
}
} else {
r.Body.Close()
return encodableError(fmt.Errorf("API PluginHTTP called but not implemented"))
}
return nil
}
// Server-side handler for old buffered PluginHTTP (for backward compatibility)
func (s *apiRPCServer) PluginHTTP(args *Z_PluginHTTPArgs, returns *Z_PluginHTTPReturns) error {
args.Request.Body = io.NopCloser(bytes.NewBuffer(args.RequestBody))
if hook, ok := s.impl.(interface {
PluginHTTP(request *http.Request) *http.Response
}); ok {
response := hook.PluginHTTP(args.Request.GetHTTPRequest())
responseBody, err := io.ReadAll(response.Body)
if err != nil {
return encodableError(fmt.Errorf("RPC call to PluginHTTP API failed: %s", err.Error()))
}
response.Body.Close()
response.Body = nil
returns.Response = response
returns.ResponseBody = responseBody
} else {
return encodableError(fmt.Errorf("API PluginHTTP called but not implemented"))
}
return nil
}
func init() {
hookNameToId["FileWillBeUploaded"] = FileWillBeUploadedID
}
type Z_FileWillBeUploadedArgs struct {
A *Context
B *model.FileInfo
UploadedFileStream uint32
ReplacementFileStream uint32
}
type Z_FileWillBeUploadedReturns struct {
A *model.FileInfo
B string
}
func (g *hooksRPCClient) FileWillBeUploaded(c *Context, info *model.FileInfo, file io.Reader, output io.Writer) (*model.FileInfo, string) {
if !g.implemented[FileWillBeUploadedID] {
return info, ""
}
uploadedFileStreamId := g.muxBroker.NextId()
go func() {
uploadedFileConnection, err := g.muxBroker.Accept(uploadedFileStreamId)
if err != nil {
g.log.Error("Plugin failed to serve upload file stream. MuxBroker could not Accept connection", mlog.Err(err))
return
}
defer uploadedFileConnection.Close()
serveIOReader(file, uploadedFileConnection)
}()
replacementDone := make(chan bool)
replacementFileStreamId := g.muxBroker.NextId()
go func() {
defer close(replacementDone)
replacementFileConnection, err := g.muxBroker.Accept(replacementFileStreamId)
if err != nil {
g.log.Error("Plugin failed to serve replacement file stream. MuxBroker could not Accept connection", mlog.Err(err))
return
}
defer replacementFileConnection.Close()
if _, err := io.Copy(output, replacementFileConnection); err != nil {
g.log.Error("Error reading replacement file.", mlog.Err(err))
}
}()
_args := &Z_FileWillBeUploadedArgs{c, info, uploadedFileStreamId, replacementFileStreamId}
_returns := &Z_FileWillBeUploadedReturns{A: _args.B}
if err := g.client.Call("Plugin.FileWillBeUploaded", _args, _returns); err != nil {
g.log.Error("RPC call FileWillBeUploaded to plugin failed.", mlog.Err(err))
}
// Ensure the io.Copy from the replacementFileConnection above completes.
<-replacementDone
return _returns.A, _returns.B
}
func (s *hooksRPCServer) FileWillBeUploaded(args *Z_FileWillBeUploadedArgs, returns *Z_FileWillBeUploadedReturns) error {
uploadFileConnection, err := s.muxBroker.Dial(args.UploadedFileStream)
if err != nil {
fmt.Fprintf(os.Stderr, "[ERROR] Can't connect to remote upload file stream, error: %v", err.Error())
return err
}
defer uploadFileConnection.Close()
fileReader := connectIOReader(uploadFileConnection)
defer fileReader.Close()
replacementFileConnection, err := s.muxBroker.Dial(args.ReplacementFileStream)
if err != nil {
fmt.Fprintf(os.Stderr, "[ERROR] Can't connect to remote replacement file stream, error: %v", err.Error())
return err
}
defer replacementFileConnection.Close()
returnFileWriter := replacementFileConnection
if hook, ok := s.impl.(interface {
FileWillBeUploaded(c *Context, info *model.FileInfo, file io.Reader, output io.Writer) (*model.FileInfo, string)
}); ok {
returns.A, returns.B = hook.FileWillBeUploaded(args.A, args.B, fileReader, returnFileWriter)
} else {
return fmt.Errorf("hook FileWillBeUploaded called but not implemented")
}
return nil
}
// MessageWillBePosted is in this file because of the difficulty of identifying which fields need special behaviour.
// The special behaviour needed is decoding the returned post into the original one to avoid the unintentional removal
// of fields by older plugins.
func init() {
hookNameToId["MessageWillBePosted"] = MessageWillBePostedID
}
type Z_MessageWillBePostedArgs struct {
A *Context
B *model.Post
}
type Z_MessageWillBePostedReturns struct {
A *model.Post
B string
}
func (g *hooksRPCClient) MessageWillBePosted(c *Context, post *model.Post) (*model.Post, string) {
_args := &Z_MessageWillBePostedArgs{c, post}
_returns := &Z_MessageWillBePostedReturns{A: _args.B}
if g.implemented[MessageWillBePostedID] {
if err := g.client.Call("Plugin.MessageWillBePosted", _args, _returns); err != nil {
g.log.Error("RPC call MessageWillBePosted to plugin failed.", mlog.Err(err))
}
}
return _returns.A, _returns.B
}
func (s *hooksRPCServer) MessageWillBePosted(args *Z_MessageWillBePostedArgs, returns *Z_MessageWillBePostedReturns) error {
if hook, ok := s.impl.(interface {
MessageWillBePosted(c *Context, post *model.Post) (*model.Post, string)
}); ok {
returns.A, returns.B = hook.MessageWillBePosted(args.A, args.B)
} else {
return encodableError(fmt.Errorf("hook MessageWillBePosted called but not implemented"))
}
return nil
}
// MessageWillBeUpdated is in this file because of the difficulty of identifying which fields need special behaviour.
// The special behaviour needed is decoding the returned post into the original one to avoid the unintentional removal
// of fields by older plugins.
func init() {
hookNameToId["MessageWillBeUpdated"] = MessageWillBeUpdatedID
}
type Z_MessageWillBeUpdatedArgs struct {
A *Context
B *model.Post
C *model.Post
}
type Z_MessageWillBeUpdatedReturns struct {
A *model.Post
B string
}
func (g *hooksRPCClient) MessageWillBeUpdated(c *Context, newPost, oldPost *model.Post) (*model.Post, string) {
_args := &Z_MessageWillBeUpdatedArgs{c, newPost, oldPost}
_default_returns := &Z_MessageWillBeUpdatedReturns{A: _args.B}
if g.implemented[MessageWillBeUpdatedID] {
_returns := &Z_MessageWillBeUpdatedReturns{}
if err := g.client.Call("Plugin.MessageWillBeUpdated", _args, _returns); err != nil {
g.log.Error("RPC call MessageWillBeUpdated to plugin failed.", mlog.Err(err))
return _default_returns.A, _default_returns.B
}
return _returns.A, _returns.B
}
return _default_returns.A, _default_returns.B
}
func (s *hooksRPCServer) MessageWillBeUpdated(args *Z_MessageWillBeUpdatedArgs, returns *Z_MessageWillBeUpdatedReturns) error {
if hook, ok := s.impl.(interface {
MessageWillBeUpdated(c *Context, newPost, oldPost *model.Post) (*model.Post, string)
}); ok {
returns.A, returns.B = hook.MessageWillBeUpdated(args.A, args.B, args.C)
} else {
return encodableError(fmt.Errorf("hook MessageWillBeUpdated called but not implemented"))
}
return nil
}
// MessagesWillBeConsumed is in this file because of the difficulty of identifying which fields need special behaviour.
// The special behaviour needed is decoding the returned post into the original one to avoid the unintentional removal
// of fields by older plugins.
func init() {
hookNameToId["MessagesWillBeConsumed"] = MessagesWillBeConsumedID
}
type Z_MessagesWillBeConsumedArgs struct {
A []*model.Post
}
type Z_MessagesWillBeConsumedReturns struct {
A []*model.Post
}
func (g *hooksRPCClient) MessagesWillBeConsumed(posts []*model.Post) []*model.Post {
_args := &Z_MessagesWillBeConsumedArgs{posts}
_returns := &Z_MessagesWillBeConsumedReturns{}
if g.implemented[MessagesWillBeConsumedID] {
if err := g.client.Call("Plugin.MessagesWillBeConsumed", _args, _returns); err != nil {
g.log.Error("RPC call MessagesWillBeConsumed to plugin failed.", mlog.Err(err))
}
}
return _returns.A
}
func (s *hooksRPCServer) MessagesWillBeConsumed(args *Z_MessagesWillBeConsumedArgs, returns *Z_MessagesWillBeConsumedReturns) error {
if hook, ok := s.impl.(interface {
MessagesWillBeConsumed(posts []*model.Post) []*model.Post
}); ok {
returns.A = hook.MessagesWillBeConsumed(args.A)
} else {
return encodableError(fmt.Errorf("hook MessagesWillBeConsumed called but not implemented"))
}
return nil
}
type Z_LogDebugArgs struct {
A string
B []any
}
type Z_LogDebugReturns struct{}
func (g *apiRPCClient) LogDebug(msg string, keyValuePairs ...any) {
stringifiedPairs := stringifyToObjects(keyValuePairs)
_args := &Z_LogDebugArgs{msg, stringifiedPairs}
_returns := &Z_LogDebugReturns{}
if err := g.client.Call("Plugin.LogDebug", _args, _returns); err != nil {
log.Printf("RPC call to LogDebug API failed: %s", err.Error())
}
}
func (s *apiRPCServer) LogDebug(args *Z_LogDebugArgs, returns *Z_LogDebugReturns) error {
if hook, ok := s.impl.(interface {
LogDebug(msg string, keyValuePairs ...any)
}); ok {
hook.LogDebug(args.A, args.B...)
} else {
return encodableError(fmt.Errorf("API LogDebug called but not implemented"))
}
return nil
}
type Z_LogInfoArgs struct {
A string
B []any
}
type Z_LogInfoReturns struct{}
func (g *apiRPCClient) LogInfo(msg string, keyValuePairs ...any) {
stringifiedPairs := stringifyToObjects(keyValuePairs)
_args := &Z_LogInfoArgs{msg, stringifiedPairs}
_returns := &Z_LogInfoReturns{}
if err := g.client.Call("Plugin.LogInfo", _args, _returns); err != nil {
log.Printf("RPC call to LogInfo API failed: %s", err.Error())
}
}
func (s *apiRPCServer) LogInfo(args *Z_LogInfoArgs, returns *Z_LogInfoReturns) error {
if hook, ok := s.impl.(interface {
LogInfo(msg string, keyValuePairs ...any)
}); ok {
hook.LogInfo(args.A, args.B...)
} else {
return encodableError(fmt.Errorf("API LogInfo called but not implemented"))
}
return nil
}
type Z_LogWarnArgs struct {
A string
B []any
}
type Z_LogWarnReturns struct{}
func (g *apiRPCClient) LogWarn(msg string, keyValuePairs ...any) {
stringifiedPairs := stringifyToObjects(keyValuePairs)
_args := &Z_LogWarnArgs{msg, stringifiedPairs}
_returns := &Z_LogWarnReturns{}
if err := g.client.Call("Plugin.LogWarn", _args, _returns); err != nil {
log.Printf("RPC call to LogWarn API failed: %s", err.Error())
}
}
func (s *apiRPCServer) LogWarn(args *Z_LogWarnArgs, returns *Z_LogWarnReturns) error {
if hook, ok := s.impl.(interface {
LogWarn(msg string, keyValuePairs ...any)
}); ok {
hook.LogWarn(args.A, args.B...)
} else {
return encodableError(fmt.Errorf("API LogWarn called but not implemented"))
}
return nil
}
type Z_LogErrorArgs struct {
A string
B []any
}
type Z_LogErrorReturns struct{}
func (g *apiRPCClient) LogError(msg string, keyValuePairs ...any) {
stringifiedPairs := stringifyToObjects(keyValuePairs)
_args := &Z_LogErrorArgs{msg, stringifiedPairs}
_returns := &Z_LogErrorReturns{}
if err := g.client.Call("Plugin.LogError", _args, _returns); err != nil {
log.Printf("RPC call to LogError API failed: %s", err.Error())
}
}
func (s *apiRPCServer) LogError(args *Z_LogErrorArgs, returns *Z_LogErrorReturns) error {
if hook, ok := s.impl.(interface {
LogError(msg string, keyValuePairs ...any)
}); ok {
hook.LogError(args.A, args.B...)
} else {
return encodableError(fmt.Errorf("API LogError called but not implemented"))
}
return nil
}
type Z_LogAuditRecArgs struct {
A *model.AuditRecord
}
type Z_LogAuditRecReturns struct {
}
// Custom audit logging methods with gob safety checks
func (g *apiRPCClient) LogAuditRec(rec *model.AuditRecord) {
gobSafeRec := makeAuditRecordGobSafe(*rec)
_args := &Z_LogAuditRecArgs{&gobSafeRec}
_returns := &Z_LogAuditRecReturns{}
if err := g.client.Call("Plugin.LogAuditRec", _args, _returns); err != nil {
log.Printf("RPC call to LogAuditRec API failed: %s", err.Error())
}
}
func (s *apiRPCServer) LogAuditRec(args *Z_LogAuditRecArgs, returns *Z_LogAuditRecReturns) error {
if hook, ok := s.impl.(interface {
LogAuditRec(rec *model.AuditRecord)
}); ok {
hook.LogAuditRec(args.A)
} else {
return encodableError(fmt.Errorf("API LogAuditRec called but not implemented"))
}
return nil
}
type Z_LogAuditRecWithLevelArgs struct {
A *model.AuditRecord
B mlog.Level
}
type Z_LogAuditRecWithLevelReturns struct {
}
func (g *apiRPCClient) LogAuditRecWithLevel(rec *model.AuditRecord, level mlog.Level) {
gobSafeRec := makeAuditRecordGobSafe(*rec)
_args := &Z_LogAuditRecWithLevelArgs{&gobSafeRec, level}
_returns := &Z_LogAuditRecWithLevelReturns{}
if err := g.client.Call("Plugin.LogAuditRecWithLevel", _args, _returns); err != nil {
log.Printf("RPC call to LogAuditRecWithLevel API failed: %s", err.Error())
}
}
func (s *apiRPCServer) LogAuditRecWithLevel(args *Z_LogAuditRecWithLevelArgs, returns *Z_LogAuditRecWithLevelReturns) error {
if hook, ok := s.impl.(interface {
LogAuditRecWithLevel(rec *model.AuditRecord, level mlog.Level)
}); ok {
hook.LogAuditRecWithLevel(args.A, args.B)
} else {
return encodableError(fmt.Errorf("API LogAuditRecWithLevel called but not implemented"))
}
return nil
}
type Z_InstallPluginArgs struct {
PluginStreamID uint32
B bool
}
type Z_InstallPluginReturns struct {
A *model.Manifest
B *model.AppError
}
func (g *apiRPCClient) InstallPlugin(file io.Reader, replace bool) (*model.Manifest, *model.AppError) {
pluginStreamID := g.muxBroker.NextId()
go func() {
uploadPluginConnection, err := g.muxBroker.Accept(pluginStreamID)
if err != nil {
log.Print("Plugin failed to upload plugin. MuxBroker could not Accept connection", mlog.Err(err))
return
}
defer uploadPluginConnection.Close()
serveIOReader(file, uploadPluginConnection)
}()
_args := &Z_InstallPluginArgs{pluginStreamID, replace}
_returns := &Z_InstallPluginReturns{}
if err := g.client.Call("Plugin.InstallPlugin", _args, _returns); err != nil {
log.Print("RPC call InstallPlugin to plugin failed.", mlog.Err(err))
}
return _returns.A, _returns.B
}
func (s *apiRPCServer) InstallPlugin(args *Z_InstallPluginArgs, returns *Z_InstallPluginReturns) error {
hook, ok := s.impl.(interface {
InstallPlugin(file io.Reader, replace bool) (*model.Manifest, *model.AppError)
})
if !ok {
return encodableError(fmt.Errorf("API InstallPlugin called but not implemented"))
}
receivePluginConnection, err := s.muxBroker.Dial(args.PluginStreamID)
if err != nil {
fmt.Fprintf(os.Stderr, "[ERROR] Can't connect to remote plugin stream, error: %v", err.Error())
return err
}
pluginReader := connectIOReader(receivePluginConnection)
defer pluginReader.Close()
returns.A, returns.B = hook.InstallPlugin(pluginReader, args.B)
return nil
}
type Z_ReceiveSharedChannelAttachmentSyncMsgArgs struct {
A string // remoteID
B string // channelID
C *model.FileInfo // fi
DataStreamID uint32
}
type Z_ReceiveSharedChannelAttachmentSyncMsgReturns struct {
A *model.FileInfo
B error
}
func (g *apiRPCClient) ReceiveSharedChannelAttachmentSyncMsg(remoteID, channelID string, fi *model.FileInfo, data io.Reader) (*model.FileInfo, error) {
dataStreamID := g.muxBroker.NextId()
go func() {
dataConnection, err := g.muxBroker.Accept(dataStreamID)
if err != nil {
log.Print("Failed to stream attachment data. MuxBroker could not Accept connection", mlog.Err(err))
return
}
defer dataConnection.Close()
serveIOReader(data, dataConnection)
}()
_args := &Z_ReceiveSharedChannelAttachmentSyncMsgArgs{remoteID, channelID, fi, dataStreamID}
_returns := &Z_ReceiveSharedChannelAttachmentSyncMsgReturns{}
if err := g.client.Call("Plugin.ReceiveSharedChannelAttachmentSyncMsg", _args, _returns); err != nil {
log.Print("RPC call ReceiveSharedChannelAttachmentSyncMsg to plugin failed.", mlog.Err(err))
}
return _returns.A, _returns.B
}
func (s *apiRPCServer) ReceiveSharedChannelAttachmentSyncMsg(args *Z_ReceiveSharedChannelAttachmentSyncMsgArgs, returns *Z_ReceiveSharedChannelAttachmentSyncMsgReturns) error {
hook, ok := s.impl.(interface {
ReceiveSharedChannelAttachmentSyncMsg(remoteID, channelID string, fi *model.FileInfo, data io.Reader) (*model.FileInfo, error)
})
if !ok {
return encodableError(fmt.Errorf("API ReceiveSharedChannelAttachmentSyncMsg called but not implemented"))
}
receiveDataConnection, err := s.muxBroker.Dial(args.DataStreamID)
if err != nil {
fmt.Fprintf(os.Stderr, "[ERROR] Can't connect to remote data stream, error: %v", err.Error())
return err
}
dataReader := connectIOReader(receiveDataConnection)
defer dataReader.Close()
returns.A, returns.B = hook.ReceiveSharedChannelAttachmentSyncMsg(args.A, args.B, args.C, dataReader)
returns.B = encodableError(returns.B)
return nil
}
type Z_UploadDataArgs struct {
A *model.UploadSession
PluginStreamID uint32
}
type Z_UploadDataReturns struct {
A *model.FileInfo
B error
}
func (g *apiRPCClient) UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo, error) {
pluginStreamID := g.muxBroker.NextId()
go func() {
pluginConnection, err := g.muxBroker.Accept(pluginStreamID)
if err != nil {
log.Print("Failed to upload data. MuxBroker could not Accept connection", mlog.Err(err))
return
}
defer pluginConnection.Close()
serveIOReader(rd, pluginConnection)
}()
_args := &Z_UploadDataArgs{us, pluginStreamID}
_returns := &Z_UploadDataReturns{}
if err := g.client.Call("Plugin.UploadData", _args, _returns); err != nil {
log.Print("RPC call UploadData to plugin failed.", mlog.Err(err))
}
return _returns.A, _returns.B
}
func (s *apiRPCServer) UploadData(args *Z_UploadDataArgs, returns *Z_UploadDataReturns) error {
hook, ok := s.impl.(interface {
UploadData(us *model.UploadSession, rd io.Reader) (*model.FileInfo, error)
})
if !ok {
return encodableError(fmt.Errorf("API UploadData called but not implemented"))
}
receivePluginConnection, err := s.muxBroker.Dial(args.PluginStreamID)
if err != nil {
fmt.Fprintf(os.Stderr, "[ERROR] Can't connect to remote plugin stream, error: %v", err.Error())
return err
}
pluginReader := connectIOReader(receivePluginConnection)
defer pluginReader.Close()
returns.A, returns.B = hook.UploadData(args.A, pluginReader)
return nil
}
func init() {
hookNameToId["ServeMetrics"] = ServeMetricsID
}
type Z_ServeMetricsArgs struct {
ResponseWriterStream uint32
Request *HTTPRequestSubset
Context *Context
RequestBodyStream uint32
}
func (g *hooksRPCClient) ServeMetrics(c *Context, w http.ResponseWriter, r *http.Request) {
if !g.implemented[ServeMetricsID] {
http.NotFound(w, r)
return
}
serveMetricsStreamId := g.muxBroker.NextId()
go func() {
connection, err := g.muxBroker.Accept(serveMetricsStreamId)
if err != nil {
g.log.Error("Plugin failed to ServeMetrics, muxBroker couldn't accept connection", mlog.Uint("serve_http_stream_id", serveMetricsStreamId), mlog.Err(err))
return
}
defer connection.Close()
rpcServer := rpc.NewServer()
if err := rpcServer.RegisterName("Plugin", &httpResponseWriterRPCServer{w: w, log: g.log}); err != nil {
g.log.Error("Plugin failed to ServeMetrics, couldn't register RPC name", mlog.Err(err))
return
}
rpcServer.ServeConn(connection)
}()
requestBodyStreamId := uint32(0)
if r.Body != nil {
requestBodyStreamId = g.muxBroker.NextId()
go func() {
bodyConnection, err := g.muxBroker.Accept(requestBodyStreamId)
if err != nil {
g.log.Error("Plugin failed to ServeMetrics, muxBroker couldn't Accept request body connection", mlog.Err(err))
return
}
defer bodyConnection.Close()
serveIOReader(r.Body, bodyConnection)
}()
}
forwardedRequest := &HTTPRequestSubset{
Method: r.Method,
URL: r.URL,
Proto: r.Proto,
ProtoMajor: r.ProtoMajor,
ProtoMinor: r.ProtoMinor,
Header: r.Header,
Host: r.Host,
RemoteAddr: r.RemoteAddr,
RequestURI: r.RequestURI,
}
if err := g.client.Call("Plugin.ServeMetrics", Z_ServeMetricsArgs{
Context: c,
ResponseWriterStream: serveMetricsStreamId,
Request: forwardedRequest,
RequestBodyStream: requestBodyStreamId,
}, nil); err != nil {
g.log.Error("Plugin failed to ServeMetrics, RPC call failed", mlog.Err(err))
http.Error(w, "500 internal server error", http.StatusInternalServerError)
}
}
func (s *hooksRPCServer) ServeMetrics(args *Z_ServeMetricsArgs, returns *struct{}) error {
connection, err := s.muxBroker.Dial(args.ResponseWriterStream)
if err != nil {
fmt.Fprintf(os.Stderr, "[ERROR] Can't connect to remote response writer stream, error: %v", err.Error())
return err
}
w := connectHTTPResponseWriter(connection)
defer w.Close()
r := args.Request
if args.RequestBodyStream != 0 {
connection, err := s.muxBroker.Dial(args.RequestBodyStream)
if err != nil {
fmt.Fprintf(os.Stderr, "[ERROR] Can't connect to remote request body stream, error: %v", err.Error())
return err
}
r.Body = connectIOReader(connection)
} else {
r.Body = io.NopCloser(&bytes.Buffer{})
}
defer r.Body.Close()
httpReq := r.GetHTTPRequest()
if hook, ok := s.impl.(interface {
ServeMetrics(c *Context, w http.ResponseWriter, r *http.Request)
}); ok {
hook.ServeMetrics(args.Context, w, httpReq)
} else {
http.NotFound(w, httpReq)
}
return nil
}
// ChannelMemberWillBeAdded is hand-written to preserve the original ChannelMember as the default
// return value, avoiding unintentional field removal by older plugins.
func init() {
hookNameToId["ChannelMemberWillBeAdded"] = ChannelMemberWillBeAddedID
}
type Z_ChannelMemberWillBeAddedArgs struct {
A *Context
B *model.ChannelMember
}
type Z_ChannelMemberWillBeAddedReturns struct {
A *model.ChannelMember
B string
}
func (g *hooksRPCClient) ChannelMemberWillBeAdded(c *Context, channelMember *model.ChannelMember) (*model.ChannelMember, string) {
_args := &Z_ChannelMemberWillBeAddedArgs{c, channelMember}
_returns := &Z_ChannelMemberWillBeAddedReturns{A: _args.B}
if g.implemented[ChannelMemberWillBeAddedID] {
if err := g.client.Call("Plugin.ChannelMemberWillBeAdded", _args, _returns); err != nil {
g.log.Error("RPC call ChannelMemberWillBeAdded to plugin failed.", mlog.Err(err))
}
}
return _returns.A, _returns.B
}
func (s *hooksRPCServer) ChannelMemberWillBeAdded(args *Z_ChannelMemberWillBeAddedArgs, returns *Z_ChannelMemberWillBeAddedReturns) error {
if hook, ok := s.impl.(interface {
ChannelMemberWillBeAdded(c *Context, channelMember *model.ChannelMember) (*model.ChannelMember, string)
}); ok {
returns.A, returns.B = hook.ChannelMemberWillBeAdded(args.A, args.B)
} else {
return encodableError(fmt.Errorf("hook ChannelMemberWillBeAdded called but not implemented"))
}
return nil
}
// TeamMemberWillBeAdded is hand-written to preserve the original TeamMember as the default
// return value, avoiding unintentional field removal by older plugins.
func init() {
hookNameToId["TeamMemberWillBeAdded"] = TeamMemberWillBeAddedID
}
type Z_TeamMemberWillBeAddedArgs struct {
A *Context
B *model.TeamMember
}
type Z_TeamMemberWillBeAddedReturns struct {
A *model.TeamMember
B string
}
func (g *hooksRPCClient) TeamMemberWillBeAdded(c *Context, teamMember *model.TeamMember) (*model.TeamMember, string) {
_args := &Z_TeamMemberWillBeAddedArgs{c, teamMember}
_returns := &Z_TeamMemberWillBeAddedReturns{A: _args.B}
if g.implemented[TeamMemberWillBeAddedID] {
if err := g.client.Call("Plugin.TeamMemberWillBeAdded", _args, _returns); err != nil {
g.log.Error("RPC call TeamMemberWillBeAdded to plugin failed.", mlog.Err(err))
}
}
return _returns.A, _returns.B
}
func (s *hooksRPCServer) TeamMemberWillBeAdded(args *Z_TeamMemberWillBeAddedArgs, returns *Z_TeamMemberWillBeAddedReturns) error {
if hook, ok := s.impl.(interface {
TeamMemberWillBeAdded(c *Context, teamMember *model.TeamMember) (*model.TeamMember, string)
}); ok {
returns.A, returns.B = hook.TeamMemberWillBeAdded(args.A, args.B)
} else {
return encodableError(fmt.Errorf("hook TeamMemberWillBeAdded called but not implemented"))
}
return nil
}