Fila no Postgres

Guardar o evento na mesma transação que grava a mudança e enviar em background, com novas tentativas. Feature webhook_postgres.

Cargo.toml
termo_core = { version = "=0.1.5", features = ["webhook_postgres"] }

O padrão outbox: o evento é gravado numa tabela do banco do próprio app dentro da mesma transação da mudança, e uma tarefa em background publica no hub.

iniciar_envio_automatico_dos_eventos_guardados

Chame uma vez ao subir o app. Cria a tabela se não existir e inicia a tarefa que envia os eventos guardados.

pub async fn iniciar_envio_automatico_dos_eventos_guardados(
    pool: &PgPool,
    hub: Option<ConfiguracaoDoHub>,
) -> Result<JoinHandle<()>, sqlx::Error>
main.rs
termo_core::definir_segredo_do_app(segredo_lido_do_banco);
termo_core::webhook::iniciar_envio_automatico_dos_eventos_guardados(
    &pool,
    termo_core::webhook::ConfiguracaoDoHub::carregar_do_ambiente_atual(),
)
.await?;
  • Aviso imediato por LISTEN/NOTIFY no canal termo_eventos_<schema>; se não der para ouvir, varre a tabela a cada 5 s.
  • Envia em lotes de 50, em ordem de criação, com FOR UPDATE SKIP LOCKED (várias instâncias do app não enviam o mesmo evento).
  • Falha que vale tentar de novo espera 5 s, 30 s, 2 min, 5 min e depois 10 min entre tentativas.
  • Recusa definitiva do hub (400, 413, 422) descarta o evento com log.
  • Evento sem entrega há mais de 1 h (RETENCAO_EM_SEGUNDOS) é descartado com log.
  • Sem segredo do app, os eventos esperam na tabela até ele ser cadastrado.
  • Com hub igual a None, os eventos só são impressos no terminal.

guardar_evento_para_publicar_na_mesma_transacao

Guarda o evento dentro da transação que já grava a mudança. Devolve o id do envelope.

pub async fn guardar_evento_para_publicar_na_mesma_transacao<'c, E, T>(
    executor: E,
    entidade_id: impl Into<String>,
    dados: T,
) -> Result<Uuid, sqlx::Error>
where
    E: PgExecutor<'c>,
    T: Evento,
Exemplo
use termo_core::webhook::eventos::termo_auth::ColaboradorCriadoV2;

let mut transacao = pool.begin().await?;
sqlx::query("INSERT INTO colaborador ...").execute(&mut *transacao).await?;
termo_core::webhook::guardar_evento_para_publicar_na_mesma_transacao(
    &mut *transacao,
    colaborador.id.to_string(),
    ColaboradorCriadoV2 { /* ... */ },
)
.await?;
transacao.commit().await?;

guardar_envelope_para_publicar_na_mesma_transacao

O mesmo, quando você já montou o Envelope (ex.: para escolher o id ou o ocorrido_em).

pub async fn guardar_envelope_para_publicar_na_mesma_transacao<'c, E, T>(
    executor: E,
    envelope: &Envelope<T>,
) -> Result<Uuid, sqlx::Error>
where
    E: PgExecutor<'c>,
    T: Evento,

criar_tabela_da_fila_de_saida_se_nao_existir

Já chamada por iniciar_envio_automatico_dos_eventos_guardados. Use só se quiser criar a tabela antes (ex.: em migração).

pub async fn criar_tabela_da_fila_de_saida_se_nao_existir(pool: &PgPool) -> Result<(), sqlx::Error>

pub const TABELA_DA_FILA: &str = "termo_eventos_pendente";
pub const RETENCAO_EM_SEGUNDOS: f64 = 3600.0;
termo_eventos_pendente
CREATE TABLE IF NOT EXISTS termo_eventos_pendente (
    id uuid PRIMARY KEY,
    tipo text NOT NULL,
    versao integer NOT NULL,
    entidade_id text NOT NULL,
    envelope jsonb NOT NULL,
    tentativas integer NOT NULL DEFAULT 0,
    proxima_tentativa_em timestamptz NOT NULL DEFAULT now(),
    reservado_ate timestamptz,
    ultimo_erro text,
    criado_em timestamptz NOT NULL DEFAULT now()
);

Nesta página