Publish after the verdict

Where the publish lives — and why it is host-side

# echo_mq — EchoMQ.Events.publish/5 (host-side, after the verdict)
# Announce a lifecycle event AFTER its transition has decided. The transition
# scripts are byte-frozen, so the publish lives here in the host, not in the
# script — the scripts stay byte-unchanged. One PUBLISH, issued DIRECT (no Lua).
def publish(conn, queue, event, job_id, extra \ []) do
  # gate the id at the key builder (INV5) — raises on an ill-formed id BEFORE the
  # wire, so a malformed JOB id can never reach a PUBLISH
  _ = Keyspace.job_key(queue, job_id)
  payload = encode_event(event, job_id, extra)

  case Connector.command(conn, ["PUBLISH", channel(queue), payload]) do
    {:ok, _n} -> :ok                  # best-effort — n is the receiver count (may be 0)
    other -> other
  end
end

# the channel — emq:{q}:events, the SAME suffix the @update_progress seam publishes on
def channel(queue), do: Keyspace.queue_key(queue, "events")

Split the work: the script decides, the host announces. That one line of separation is what lets the bus broadcast every outcome without ever editing a frozen transition.

The payload — flat cjson, built by hand

# echo_mq — EchoMQ.Events.encode_event/3 (the flat cjson payload, built by hand)
# {"event":"<name>","job":"<id>", …extra} — flat string fields only. No JSON
# library on the bus; the object is concatenated, with \ and " escaped.
defp encode_event(event, job_id, extra) do
  fields =
    [{"event", to_string(event)}, {"job", job_id}] ++
      Enum.map(extra, fn {k, v} -> {to_string(k), to_string(v)} end)

  body =
    fields
    |> Enum.map(fn {k, v} -> ~s("#{k}":"#{escape(v)}") end)
    |> Enum.join(",")

  "{" <> body <> "}"
end

defp escape(v) do
  v
  |> String.replace("\\", "\\\\")   # backslash first
  |> String.replace("\"", "\\\"")   # then the quote
end

No JSON library on the bus — only a concatenated object with the two significant characters escaped. Small payloads, zero dependency.

Reading the name back — a scan, not a parser

# echo_mq — EchoMQ.Events.event_name/1 (read the name by scan; never mint an atom)
# The bus has no JSON parser and cjson key order is not guaranteed — so read the
# "event" field by substring scan. Resolve with to_existing_atom: an UNKNOWN
# name answers :unknown, never minting an atom from the wire (atoms aren't GC'd).
def event_name(payload) when is_binary(payload) do
  case Regex.run(~r/"event"\s*:\s*"([^"]+)"/, payload) do
    [_, name] ->
      try do
        String.to_existing_atom(name)     # only an atom that already exists
      rescue
        ArgumentError -> :unknown          # unknown name → :unknown, not a new atom
      end

    _ ->
      :unknown
  end
end

Read the name by scan, resolve it to an existing atom or :unknown — never mint an atom from the wire. A hand-built payload deserves a hand-built, safe read.

The publish round-trip

The id is gated before the payload is built — an ill-formed job id raises at job_key/2 and never reaches the wire. The read is a scan, so it tolerates any key order and refuses any name it does not already know.

Pattern & implementation

The pattern says emit on the change. The implementation emits from the host after the change, so the change's script never has to move.

Recap — the host announces

Next in the module: Fire-and-forget — what at-most-once costs, the resubscribe and the stream as the two mitigations, and the cooperative cancel token.

References