Events: Horizontal scaling support in EventStream.subscribe

[martastain]

This pull request introduces several key improvements to the handling of event subscribers and the organization of hooks in the EventStream class.

Event Behavior: Single Instance vs. All Instances

Single Instance Handling

When an event is created and a subscriber is attached to the topic, the function is triggered only on the current server instance.

Use Case: Situations where the event should be processed once, e.g., "Entity status change sends a notification."

All Instances Handling:

In scaled deployments of Ayon (multiple server instances), some events need to trigger functions on all instances.

Use Case: "Addon settings change requires all addon instances to reload their cache."

Implementation

Added a new parameter, all_nodes: bool, to the EventStream.subscribe method (Default: False)
When set to True, the subscribed function will be triggered on all server instances.

Hooks are now split into two groups:

• local_hooks: Implementation remains unchanged.
• global_hooks: Implementation added in ayon_server.api.messaging.

Testing Notes

Use multiple replicas in a Docker Compose environment.

server:
    image: ynput/ayon:dev
    restart: unless-stopped
    deploy:
      replicas: 3
    ports: 
      -"5000-5002:5000"

Subscribe to an event topic from an addon using:

EventStream.subscribe(topic_name, handler, all_nodes=True)

Verify correct behavior for both single-instance and all-instance event handling.

Keep in mind Ayon server is still not fully horizontally scalable, so other functionality may break when the server runs with multiple replicas.

[iLLiCiTiT]

With this you can get rid of multiple if topic not in *hooks* conditions and avoid possible missing keys issues.

Suggested change

	local_hooks: dict[str, dict[str, HandlerType]] = {}
	global_hooks: dict[str, dict[str, HandlerType]] = {}
	local_hooks: dict[str, dict[str, HandlerType]] = collections.defaultdict(dict)
	global_hooks: dict[str, dict[str, HandlerType]] = collections.defaultdict(dict)

Not requirement tho...

[iLLiCiTiT]

Suggested change

	if all_nodes:
	if topic not in cls.global_hooks:
	cls.global_hooks[topic] = {}
	cls.global_hooks[topic][token] = handler
	return token
	else:
	if topic not in cls.local_hooks:
	cls.local_hooks[topic] = {}
	cls.local_hooks[topic][token] = handler
	return token
	hooks = cls.global_hooks if all_nodes else cls.local_hooks
	topic_hooks = hooks.setdefault(topic, {})
	topic_hooks[token] = handler
	return token

[iLLiCiTiT]

Works, but can be improved (multiple different ways):

        for hooks in (cls.global_hooks, cls.local_hooks):
            for topic, mapping in tuple(hooks.items()):
                mapping.pop(token, None)
                if not mapping:
                    hooks.pop(topic)