Código explicado: preparando a mensagem

Objetivo desta página: ler uma função real sem precisar entender todo o projeto de uma vez. O foco é prepareTelemetry() em apps/mqtt-to-db/src/message-processor.ts. Ela é o passo preparar do caminho da medição. Se os nomes dos arquivos ainda parecem desconexos, consulte antes o mapa do código.

O que entra e o que sai

Antes desta função, packages/mqtt/src/subscriber.ts já classificou a mensagem como válida (ok: true) ou inválida (ok: false). prepareTelemetry() recebe essa classificação e devolve um objeto pronto para a camada de persistência. Ela não abre conexão nem grava no banco.

Mensagem recebida + opções → prepareTelemetry() → dados preparados → persistTelemetry()

Em TypeScript, o campo ok funciona como uma etiqueta: ao testar if (!message.ok), o código sabe qual variante da mensagem está usando. Isso se chama união discriminada; aqui basta entender que existem dois caminhos explícitos.

Primeiro caminho: mensagem inválida

Trecho reduzido da função real (a tipagem e alguns campos foram omitidos para facilitar a leitura):

const receivedAt = options.receivedAt ?? new Date();

if (!message.ok) {
  return {
    ok: false,
    topic: message.topic,
    payloadRaw: message.raw,
    payloadJson: parsePayloadJson(message.raw),
    receivedAt,
    error: message.error,
  };
}

?? quer dizer: “use o valor da esquerda; se ele não foi fornecido, use o da direita”. Assim, receivedAt pode vir do runtime ou ser calculado agora. O texto original fica em payloadRaw para auditoria. parsePayloadJson() tenta guardar também uma representação JSON; se o texto nem for JSON válido, retorna Prisma.DbNull. A etapa seguinte grava o registro inválido em raw_messages, sem criar um measurement.

Segundo caminho: mensagem válida

No caminho ok: true, a validação do contrato já aconteceu. O trecho decisivo é:

measuredAt: message.payload.ts === undefined
  ? receivedAt
  : new Date(message.payload.ts),
device: {
  type: message.deviceType,
  externalId: message.externalId,
  activeOnCreate: options.autoActivate ?? false,
},
metrics: Object.entries(message.payload.metrics).map(([key, valueNum]) => ({
  key,
  valueNum,
  unit: inferMetricUnit(key),
})),

• measuredAt é a hora atribuída à medição; receivedAt é a hora de chegada ao persistidor. Quando o dispositivo não envia ts, as duas usam a mesma hora de chegada.
• device identifica a fonte por deviceType e externalId. Um dispositivo novo não é ativado automaticamente por padrão; a opção autoActivate pode mudar essa regra.
• Object.entries(...).map(...) transforma cada par “nome da métrica → valor” em um pequeno objeto. Para power_w: 1234.5, o resultado inclui key: 'power_w', valueNum: 1234.5 e unit: 'W'. A unidade vem de uma convenção do sufixo _w, não de uma leitura separada.

Além desses campos, a função preserva o tópico e o corpo original. A função seguinte, persistTelemetry() em apps/mqtt-to-db/src/persistence.ts, decide onde gravar conforme ok.

Continue na persistência explicada para acompanhar essa gravação. No coletor físico atual, o ts é produzido pela aplicação após a leitura; não pressupomos sincronização com um relógio interno do inversor.

Experimente prever o resultado

Se chegar uma mensagem válida, sem ts, com metrics: { power_w: 500 }:

1. Qual instante será usado como measuredAt?
2. Qual unidade será inferida para power_w?
3. prepareTelemetry() já terá escrito algo no banco?

Resposta: o instante de chegada (receivedAt), W, e não. Ela só prepara dados; a escrita acontece em persistence.ts.

Limite deste recorte

O trecho acima é uma explicação selecionada, não uma cópia integral nem uma API para executar. Para entender tratamento de fila, erros e encerramento, leia também runtime.ts e queue.ts. Não execute o serviço para estudar esta página: ele pode conectar ao broker e ao banco reais.

Base revista em 2026-10-06: packages/mqtt/src/subscriber.ts, contract.ts; apps/mqtt-to-db/src/message-processor.ts, index.ts, persistence.ts e runtime.ts.