Implementando uma notificação via PUBSUB para após a excecução do worker notificar o usuário que o worker foi finalizado

Neste post, vamos explorar como implementar um sistema de notificações utilizando o Pub/Sub para informar os usuários sobre a finalização de um worker. Essa abordagem é essencial para garantir que os usuários estejam sempre atualizados sobre o status de suas tarefas, melhorando a experiência geral do sistema.

🚀 O que é Pub/Sub?

Pub/Sub, ou Publish/Subscribe, é um padrão de comunicação assíncrono que permite que diferentes partes de um sistema se comuniquem de forma eficiente. Nesse modelo, os publicadores enviam mensagens sem saber quem são os assinantes, enquanto os assinantes recebem mensagens sem saber quem as enviou. Isso promove um desacoplamento entre os componentes do sistema, facilitando a escalabilidade e a manutenção.

💡 Dica Importante: O uso de Pub/Sub é ideal para sistemas que requerem alta disponibilidade e escalabilidade, como aplicações em nuvem.

⚡ Características Principais

• Escalabilidade: O modelo Pub/Sub permite que você adicione novos publicadores e assinantes sem impactar o sistema existente.
• Performance: Mensagens são processadas de forma assíncrona, melhorando a performance geral do sistema.
• Simplicidade: A implementação do padrão é relativamente simples e pode ser adaptada a diferentes linguagens e frameworks.

💻 Implementação Prática

Vamos ver um exemplo prático de como implementar uma notificação via Pub/Sub após a execução de um worker:

defmodule MeuWorker do
  use GenServer

  def start_link(_) do
    GenServer.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  def init(:ok) do
    {:ok, %{}}
  end

  def perform_task(task) do
    # Simulação de execução de tarefa
    Process.sleep(2000)
    notify_user(task)
  end

  defp notify_user(task) do
    PubSub.publish(MyApp.PubSub, "notificacoes", %{task: task, status: "finalizado"})
  end
end

1. O Worker Oban Estruturado
Nesta implementação, o perform/1 foca na lógica de negócio, enquanto capturamos qualquer exceção para notificar o usuário antes de permitir que o Oban siga sua estratégia de retry.

defmodule MyApp.Workers.ProcessarDocumento do
  use Oban.Worker, queue: :default, max_attempts: 3
  alias Phoenix.PubSub

  @topic "notificacoes"

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"user_id" => user_id, "doc_id" => doc_id}}) do
    # 1. Notificar início (Opcional, para feedback visual imediato)
    broadcast(user_id, %{status: "processando", doc_id: doc_id})

    try do
      # Simulação de lógica complexa
      resultado = executar_processamento_pesado(doc_id)

      # 2. Notificar Sucesso
      broadcast(user_id, %{
        status: "sucesso",
        doc_id: doc_id,
        message: "Documento processado com êxito!",
        data: resultado
      })

      :ok
    rescue
      e ->
        # 3. Notificar Erro ao Usuário (mantendo-o informado)
        broadcast(user_id, %{
          status: "erro",
          doc_id: doc_id,
          message: "Falha temporária. Tentaremos novamente em breve.",
          error: Exception.message(e)
        })

        # Re-raise para o Oban saber que falhou e agendar o retry
        reraise e, __STACKTRACE__
    end
  end

  defp executar_processamento_pesado(id) do
    Process.sleep(2000)
    if :rand.uniform() > 0.8, do: raise "Erro de conexão com o Storage!"
    %{url: "https://cdn.meuapp.com/docs/#{id}.pdf"}
  end

  defp broadcast(user_id, payload) do
    PubSub.broadcast(MyApp.PubSub, "#{@topic}:#{user_id}", payload)
  end
end

Integrando com LiveView (Exemplo de Consumo)
O seu LiveView (ou canal) se inscreveria no tópico específico do usuário para reagir a essas mensagens:

defmodule MyAppWeb.NotificacaoLive do
  use MyAppWeb, :live_view

  def mount(_params, _session, socket) do
    if connected?(socket) do
      # Subscreve apenas nas notificações DESTE usuário
      Phoenix.PubSub.subscribe(MyApp.PubSub, "notificacoes:#{socket.assigns.current_user.id}")
    end
    {:ok, assign(socket, :notificacoes, [])}
  end

  # Trata as mensagens vindas do Worker
  def handle_info(%{status: status, message: msg} = info, socket) do
    new_notif = %{status: status, text: msg, id: info[:doc_id]}
    {:noreply, assign(socket, :notificacoes, [new_notif | socket.assigns.notificacoes])}
  end
end

✅ Boa Prática: Sempre documente seu código adequadamente para facilitar a manutenção e a colaboração.

🎯 Casos de Uso

Esta solução é ideal para:

1. Notificações de finalização de tarefas em sistemas de processamento em lote.
2. Atualizações em tempo real para usuários em aplicações web interativas.
3. Integração com sistemas de monitoramento e alertas.

💬 "A comunicação assíncrona é a chave para a escalabilidade em sistemas modernos."

⚠️ Atenção: Certifique-se de que o sistema de Pub/Sub esteja configurado corretamente para evitar perda de mensagens.

Dica Pro: Usando Oban Web ou Telemetry

Se você precisar tratar erros de forma ainda mais global (ex: logar no Sentry ou atualizar o banco de dados toda vez que um job morre definitivamente), em vez de try/catch dentro do worker, você pode usar os Oban Plugins ou escutar os eventos de [:oban, :job, :exception].

📝 Conclusão

Neste artigo, abordamos a implementação de um sistema de notificações utilizando o padrão Pub/Sub. Vimos como essa abordagem pode melhorar a comunicação entre diferentes componentes de um sistema, garantindo que os usuários sejam informados sobre a finalização de tarefas de forma eficiente.

Agora é sua vez! Experimente implementar essa solução em seu próximo projeto e veja como ela pode melhorar a experiência do usuário.

📚 Fontes de Referência

• Documentação oficial: Phoenix PubSub
• Artigos técnicos: Pub/Sub Pattern in Elixir
• Livros recomendados: Programming Elixir 1.6
• Recursos adicionais: Vídeo sobre Pub/Sub

✨Conteúdo gerado por uma IA da RCO TechLab Blog - Este artigo foi criado com assistência de inteligência artificial para fornecer informações técnicas precisas e atualizadas.