feat(plugins): Implement event system w/ POC events

Implements the partially created event system with 3 events implemented as proof of concepts.

The 3 events are:
- `blacklistedIpBlocked`: emitted when a request from a blacklisted IP
- `accessRuleCreated`: emitted when a new access rule is created
- `blacklistToggled`: emitted when the blacklist is toggled for a given access rule

Why these events? Because these are the ones I forsee myself needing in the next version of the zoraxy_crowdsec_bouncer

Events are dispatched via a global event manager `plugins.EventSystem.Emit`

src/mod/plugins/event_system.go

@@ -0,0 +1,181 @@
+package plugins
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"imuslab.com/zoraxy/mod/info/logger"
+	zoraxyPlugin "imuslab.com/zoraxy/mod/plugins/zoraxy_plugin"
+)
+
+// eventManager manages event subscriptions and dispatching
+type eventManager struct {
+	subscriptions map[zoraxyPlugin.EventName][]string    // EventType -> []PluginID
+	pluginLookup  func(pluginID string) (*Plugin, error) // Function to get plugin by ID
+	logger        *logger.Logger                         // Logger for the event manager
+	mutex         sync.RWMutex                           // Mutex for concurrent access
+}
+
+var (
+	// EventSystem is the singleton instance of the event manager
+	EventSystem *eventManager
+	once        sync.Once
+)
+
+// InitEventManager initializes the event manager with the plugin manager
+func InitEventManager(pluginLookup func(pluginID string) (*Plugin, error), logger *logger.Logger) {
+	once.Do(func() {
+		EventSystem = &eventManager{
+			subscriptions: make(map[zoraxyPlugin.EventName][]string),
+			pluginLookup:  pluginLookup,
+			logger:        logger,
+		}
+	})
+}
+
+// Subscribe adds a plugin to the subscription list for an event type
+func (em *eventManager) Subscribe(pluginID string, eventType zoraxyPlugin.EventName) error {
+	em.mutex.Lock()
+	defer em.mutex.Unlock()
+
+	if _, exists := em.subscriptions[eventType]; !exists {
+		em.subscriptions[eventType] = []string{}
+	}
+
+	// Check if already subscribed
+	for _, id := range em.subscriptions[eventType] {
+		if id == pluginID {
+			return nil // Already subscribed
+		}
+	}
+
+	em.subscriptions[eventType] = append(em.subscriptions[eventType], pluginID)
+	return nil
+}
+
+// Unsubscribe removes a plugin from the subscription list for an event type
+
+func (em *eventManager) Unsubscribe(pluginID string, eventType zoraxyPlugin.EventName) error {
+	em.mutex.Lock()
+	defer em.mutex.Unlock()
+
+	if subscribers, exists := em.subscriptions[eventType]; exists {
+		for i, id := range subscribers {
+			if id == pluginID {
+				// Remove from slice
+				em.subscriptions[eventType] = append(subscribers[:i], subscribers[i+1:]...)
+				break
+			}
+		}
+	}
+
+	return nil
+}
+
+// UnsubscribeAll removes a plugin from all event subscriptions
+func (em *eventManager) UnsubscribeAll(pluginID string) error {
+	em.mutex.Lock()
+	defer em.mutex.Unlock()
+
+	for eventType, subscribers := range em.subscriptions {
+		for i, id := range subscribers {
+			if id == pluginID {
+				em.subscriptions[eventType] = append(subscribers[:i], subscribers[i+1:]...)
+				break
+			}
+		}
+	}
+
+	return nil
+}
+
+// Emit dispatches an event to all subscribed plugins
+func (em *eventManager) Emit(payload zoraxyPlugin.EventPayload) error {
+	eventName := payload.GetName()
+
+	em.mutex.RLock()
+	subscribers, exists := em.subscriptions[eventName]
+	em.mutex.RUnlock()
+
+	if !exists || len(subscribers) == 0 {
+		return nil // No subscribers
+	}
+
+	// Create the event
+	event := zoraxyPlugin.Event{
+		Name:      eventName,
+		Timestamp: time.Now(),
+		Data:      payload,
+	}
+
+	// Dispatch to all subscribers asynchronously
+	for _, pluginID := range subscribers {
+		go em.dispatchToPlugin(pluginID, event)
+	}
+
+	return nil
+}
+
+// dispatchToPlugin sends an event to a specific plugin
+func (em *eventManager) dispatchToPlugin(pluginID string, event zoraxyPlugin.Event) {
+	plugin, err := em.pluginLookup(pluginID)
+	if err != nil {
+		em.logger.PrintAndLog("event-system", "Failed to get plugin for event dispatch: "+pluginID, err)
+		return
+	}
+
+	if !plugin.Enabled || plugin.AssignedPort == 0 {
+		// Plugin is not running, skip
+		return
+	}
+
+	subscriptionPath := plugin.Spec.SubscriptionPath
+	if subscriptionPath == "" {
+		// No subscription path configured, skip
+		return
+	}
+	if !strings.HasPrefix(subscriptionPath, "/") {
+		subscriptionPath = "/" + subscriptionPath
+	}
+
+	// Prepare the URL
+	url := fmt.Sprintf("http://127.0.0.1:%d%s/%s", plugin.AssignedPort, subscriptionPath, event.Name)
+
+	// Marshal the event to JSON
+	eventData, err := json.Marshal(event)
+	if err != nil {
+		em.logger.PrintAndLog("event-system", "Failed to marshal event for plugin "+pluginID, err)
+		return
+	}
+
+	// Create HTTP request
+	req, err := http.NewRequest("POST", url, bytes.NewBuffer(eventData))
+	if err != nil {
+		em.logger.PrintAndLog("event-system", "Failed to create HTTP request for plugin "+pluginID, err)
+		return
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Zoraxy-Event-Type", string(event.Name))
+
+	// Send the request with timeout
+	client := &http.Client{
+		Timeout: 10 * time.Second,
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		em.logger.PrintAndLog("event-system", "Failed to send event to plugin "+pluginID, err)
+		return
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		em.logger.PrintAndLog("event-system", "Plugin "+pluginID+" returned non-200 status for event: "+resp.Status, nil)
+	}
+}

src/mod/plugins/lifecycle.go

@@ -151,6 +151,19 @@ func (m *Manager) StartPlugin(pluginID string) error {
 	thisPlugin.process = cmd
 	thisPlugin.Enabled = true
 
+	// Register event subscriptions
+	if thisPlugin.Spec.SubscriptionsEvents != nil {
+		for eventName := range thisPlugin.Spec.SubscriptionsEvents {
+			eventType := zoraxyPlugin.EventName(eventName)
+			err := EventSystem.Subscribe(thisPlugin.Spec.ID, eventType)
+			if err != nil {
+				m.Log("Failed to subscribe plugin "+thisPlugin.Spec.Name+" to event "+eventName, err)
+			} else {
+				m.Log("Subscribed plugin "+thisPlugin.Spec.Name+" to event "+eventName, nil)
+			}
+		}
+	}
+
 	//Create a new static forwarder router for each of the static capture paths
 	thisPlugin.StartAllStaticPathRouters()
 
@@ -288,6 +301,11 @@ func (m *Manager) StopPlugin(pluginID string) error {
 	if err != nil {
 		m.Log("Failed to revoke API keys for plugin "+thisPlugin.Spec.Name, err)
 	}
+	//Unsubscribe from all events
+	err = EventSystem.UnsubscribeAll(thisPlugin.Spec.ID)
+	if err != nil {
+		m.Log("Failed to unsubscribe plugin "+thisPlugin.Spec.Name+" from events", err)
+	}
 
 	return nil
 }

src/mod/plugins/typdef.go

@@ -60,7 +60,7 @@ type Manager struct {
 	LoadedPlugins      map[string]*Plugin   //Storing *Plugin
 	tagPluginMap       sync.Map             //Storing *radix.Tree for each plugin tag
 	tagPluginListMutex sync.RWMutex         //Mutex for the tagPluginList
-	tagPluginList      map[string][]*Plugin //Storing the plugin list for each tag, only concurrent READ is allowed
+	tagPluginList      map[string][]*Plugin //Storing the plugin list for each tag, only concurrent read is allowed
 	Options            *ManagerOptions
 
 	PluginHash map[string]string //The hash of the plugin file, used to check if the plugin file is changed

src/mod/plugins/zoraxy_plugin/event.go

@@ -0,0 +1,69 @@
+package zoraxy_plugin
+
+import (
+	"time"
+)
+
+// EventName represents the type of event
+type EventName string
+
+// EventPayload interface for all event payloads
+type EventPayload interface {
+	// GetName returns the event type
+	GetName() EventName
+}
+
+// Event represents a system event
+type Event struct {
+	Name      EventName    `json:"type"`
+	Timestamp time.Time    `json:"timestamp"`
+	Data      EventPayload `json:"data"`
+}
+
+const (
+	// EventBlacklistedIPBlocked is emitted when a blacklisted IP is blocked
+	EventBlacklistedIPBlocked EventName = "blacklistedIpBlocked"
+	// EventBlacklistToggled is emitted when the blacklist is toggled for an access rule
+	EventBlacklistToggled EventName = "blacklistToggled"
+	// EventAccessRuleCreated is emitted when a new access ruleset is created
+	EventAccessRuleCreated EventName = "accessRuleCreated"
+
+	// Add more event types as needed
+)
+
+// BlacklistedIPBlockedEvent represents an event when a blacklisted IP is blocked
+type BlacklistedIPBlockedEvent struct {
+	IP           string `json:"ip"`
+	Comment      string `json:"comment"`
+	RequestedURL string `json:"requested_url"`
+	Hostname     string `json:"hostname"`
+	UserAgent    string `json:"user_agent"`
+	Method       string `json:"method"`
+}
+
+func (e *BlacklistedIPBlockedEvent) GetName() EventName {
+	return EventBlacklistedIPBlocked
+}
+
+// BlacklistToggledEvent represents an event when the blacklist is disabled for an access rule
+type BlacklistToggledEvent struct {
+	RuleID  string `json:"rule_id"`
+	Enabled bool   `json:"enabled"` // Whether the blacklist is enabled or disabled
+}
+
+func (e *BlacklistToggledEvent) GetName() EventName {
+	return EventBlacklistToggled
+}
+
+// AccessRuleCreatedEvent represents an event when a new access ruleset is created
+type AccessRuleCreatedEvent struct {
+	ID               string `json:"id"`
+	Name             string `json:"name"`
+	Desc             string `json:"description"`
+	BlacklistEnabled bool   `json:"blacklist_enabled"`
+	WhitelistEnabled bool   `json:"whitelist_enabled"`
+}
+
+func (e *AccessRuleCreatedEvent) GetName() EventName {
+	return EventAccessRuleCreated
+}

src/mod/plugins/zoraxy_plugin/zoraxy_plugin.go

@@ -167,12 +167,12 @@ func RecvConfigureSpec() (*ConfigureSpec, error) {
 					return nil, err
 				}
 			} else {
-				return nil, fmt.Errorf("No port specified after -configure flag")
+				return nil, fmt.Errorf("no port specified after -configure flag")
 			}
 			return &configSpec, nil
 		}
 	}
-	return nil, fmt.Errorf("No -configure flag found")
+	return nil, fmt.Errorf("no -configure flag found")
 }
 
 /*