feat: Add OnBeforeSend hook (#1325)

This adds an optional `OnBeforeSend` hook to plugin clients that choose to implement the interface. When implemented, the plugin client will receive a call before every message is sent, allowing it to modify the message, perform a side-effect, or return an error.

This keeps the potentially general useful for all plugins, with the first use case being to check whether a row should be counted as premium or not in upcoming paid plugins.

Author: hermanschaaf

internal/servers/plugin/v3/plugin.go

@@ -130,8 +130,13 @@ func (s *Server) Sync(req *pb.Sync_Request, stream pb.Plugin_SyncServer) error {
 			syncErr = fmt.Errorf("failed to sync records: %w", err)
 		}
 	}()
-
+	var err error
 	for msg := range msgs {
+		msg, err = s.Plugin.OnBeforeSend(ctx, msg)
+		if err != nil {
+			syncErr = fmt.Errorf("failed before sending message: %w", err)
+			return syncErr
+		}
 		pbMsg := &pb.Sync_Response{}
 		switch m := msg.(type) {
 		case *message.SyncMigrateTable:

plugin/plugin.go

@@ -124,6 +124,23 @@ func (p *Plugin) Version() string {
 	return p.version
 }
 
+type OnBeforeSender interface {
+	OnBeforeSend(context.Context, message.SyncMessage) (message.SyncMessage, error)
+}
+
+// OnBeforeSend gets called before every message is sent to the destination. A plugin client
+// that implements the OnBeforeSender interface will have this method called.
+func (p *Plugin) OnBeforeSend(ctx context.Context, msg message.SyncMessage) (message.SyncMessage, error) {
+	// This method is called once for every message, so it is on the hot path, and we should be careful about its performance.
+	// However, most recent versions of Go have optimized type assertions and type switches to be very fast, so
+	// we use them here without expecting a significant impact on performance.
+	// See: https://stackoverflow.com/questions/28024884/does-a-type-assertion-type-switch-have-bad-performance-is-slow-in-go
+	if v, ok := p.client.(OnBeforeSender); ok {
+		return v.OnBeforeSend(ctx, msg)
+	}
+	return msg, nil
+}
+
 // IsStaticLinkingEnabled whether static linking is to be enabled
 func (p *Plugin) IsStaticLinkingEnabled() bool {
 	return p.staticLinking