Когда залогиненый пользователь коннектится, надо хранить где-нибудь идентификатор его соединения. Можно в базе или, например, в Redis. Надо ее забывать что один пользователь может иметь сколько угодно соединений и с любого количества устройств. Поэтому обычно используют ещё очереди и подписываются на сообщения из очередей с тегом по Id пользователя.