1. About the first point _handle_pubsub is synchronous because Pulsar client supports only sync listeners, therefore I need to use sync _handle_pubsub method, but because I want to utilize asynchronous tasks and my whole app is async I manage those tasks there with getting the running loop which is the main loop for the app or creating new one if there is no loop running. Ofc while I've added this loop creation, potentially there will be no such case due to the fact that it is FastAPI async app so it will always have a running loop which is created and managed by FastAPI itself.
2. Regarding event producers, I thought about some sort of scheduler that will handle sending of all the Pulsar events in the background, therefore I could make responses faster, without actually waiting the send it self. This way it could improve the efficiency of the app without any tradeoff on resource usage. And due to faster response FE could make an optimistic update while waiting updates from Worker. Maybe you could help me with that? Here is my Pulsar manager so you would have the full context (ProfileService and PulsarManager):
import enum
import uuid
from datetime import datetime
from decimal import Decimal
from typing import Dict, List, Literal, Optional, Union

import pulsar
from diamonds_domain.config import settings
from diamonds_domain.constants import ActionType
from pydantic import BaseModel, Field

__all__ = [
    "PulsarManager",
    "pulsar_manager",
    "PulsarUpdateEvent",
    "PulsarClaimEvent",
    "PulsarCreateProfileEvent",
    "PulsarCommandEvent",
    "PulsarCommandType",
    "PulsarTopicType",
    "get_pulsar_manager",
]


@enum.unique
class PulsarTopicType(str, enum.Enum):
    COMMAND = "command"
    UPDATE = "update"


@enum.unique
class PulsarCommandType(str, enum.Enum):
    CREATE_PROFILE = "create_profile"
    CLAIM = "claim"


class PulsarUpdateEvent(BaseModel):
    profile_uuid: uuid.UUID
    auth_user_uuid: uuid.UUID
    ref_code: str
    ref_quantity: int
    name: str
    direct_points: Decimal
    ref_points: Decimal
    crystal_balls: int
    streak: int
    actions: Dict[ActionType, datetime]


class PulsarClaimEvent(BaseModel):
    command: Literal[PulsarCommandType.CLAIM] = PulsarCommandType.CLAIM
    action: ActionType
    value: Decimal
    auth_user_uuid: uuid.UUID
    idempotency_key: uuid.UUID


class PulsarCreateProfileEvent(BaseModel):
    command: Literal[PulsarCommandType.CREATE_PROFILE] = (
        PulsarCommandType.CREATE_PROFILE
    )
    auth_user_uuid: uuid.UUID
    ref_code: str
    name: str
    direct_referrer_uuid: Optional[uuid.UUID] = None
    indirect_referrer_uuid: Optional[uuid.UUID] = None


class PulsarCommandEvent(BaseModel):
    payload: Union[
        PulsarClaimEvent,
        PulsarCreateProfileEvent,
    ] = Field(discriminator="command")


class PulsarManager:
    def __init__(self) -> None:
        self._commands_topic = f"persistent://{settings.PULSAR_TENANT_NAME}/{settings.PULSAR_NAMESPACE}/{settings.PULSAR_COMMANDS_TOPIC_NAME}"
        self._updates_topic = f"persistent://{settings.PULSAR_TENANT_NAME}/{settings.PULSAR_NAMESPACE}/{settings.PULSAR_UPDATES_TOPIC_NAME}"
        self._client = pulsar.Client(
            settings.PULSAR_SERVICE_URL,
            authentication=pulsar.AuthenticationToken(settings.PULSAR_TOKEN),
        )
        commands_producer = self._client.create_producer(self._commands_topic)
        updates_producer = self._client.create_producer(self._updates_topic)
        self._type_to_producer_map = {
            PulsarTopicType.COMMAND: commands_producer,
            PulsarTopicType.UPDATE: updates_producer,
        }
        self._updates_subscription_name = f"updates-subscription-{uuid.uuid4()}"
        self._commands_subscription_name = "commands-subscription"
        self._commands_consumer = None
        self._updates_consumer = None

    def _get_producer(self, topic_type: PulsarTopicType):
        return self._type_to_producer_map[topic_type]

    def set_commands_consumer_with_callback(self, listener):
        if self._commands_consumer is None:
            self._commands_consumer = self._client.subscribe(
                self._commands_topic,
                self._commands_subscription_name,
                pulsar.ConsumerType.Shared,
                message_listener=listener,
            )

    def set_updates_consumer_with_callback(self, listener):
        if self._updates_consumer is None:
            self._updates_consumer = self._client.subscribe(
                self._updates_topic,
                self._updates_subscription_name,
                pulsar.ConsumerType.Exclusive,
                message_listener=listener,
            )

    def unsubscribe(self):
        if self._updates_consumer:
            self._updates_consumer.unsubscribe()
        if self._commands_consumer:
            self._commands_consumer.unsubscribe()

    def produce_update(
        self,
        event: PulsarUpdateEvent,
    ) -> None:
        data = event.model_dump_json()
        producer = self._get_producer(PulsarTopicType.UPDATE)
        producer.send(data.encode("utf-8"))

    def produce_command(
        self,
        event: PulsarCommandEvent,
    ) -> None:
        data = event.model_dump_json()
        producer = self._get_producer(PulsarTopicType.COMMAND)
        producer.send(
            data.encode("utf-8"), partition_key=str(event.payload.auth_user_uuid)
        )

    def produce_commands(
        self,
        events: List[PulsarCommandEvent],
    ) -> None:
        for event in events:
            self.produce_command(event)

    def consume_update(self, msg) -> PulsarUpdateEvent:
        try:
            event = PulsarUpdateEvent.model_validate_json(msg.data().decode("utf-8"))
            self._updates_consumer.acknowledge(msg)
            return event
        except Exception:
            self._updates_consumer.negative_acknowledge(msg)

    def consume_command(self, msg) -> PulsarCommandEvent:
        try:
            event = PulsarCommandEvent.model_validate_json(msg.data().decode("utf-8"))
            self._commands_consumer.acknowledge(msg)
            return event
        except Exception:
            self._commands_consumer.negative_acknowledge(msg)


pulsar_manager = None


def get_pulsar_manager():
    global pulsar_manager
    if pulsar_manager is None:
        pulsar_manager = PulsarManager()
    return pulsar_manager