mirror of
https://github.com/hanzoai/team.git
synced 2026-09-30 01:57:35 +00:00
* 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 commita760aefcd0) * 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 commit078f6b31b5) * 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 commit4e967cb7ad) * 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 commit15d7fbd8cb) * [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 commita72b1db4c5) * 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 commit5c58953e8e) * 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 commita98e01ebac) * 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 commitea7f488c64)
1501 lines
46 KiB
Go
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
|
|
}
|