feat(leadmail): use skip_queue=true and adopt Zeptomail Message-ID for threading

LeadMail server now supports skip_queue=true to send synchronously and
return the actual outbound Message-ID (which Zeptomail rewrites for VERP
bounce tracking). This is the critical piece for Chatwoot's reply
threading: Email::SendOnEmailService reads Mail::Message#message_id after
deliver and stores it as the message's source_id. When the recipient
replies with In-Reply-To: <that-id>, Chatwoot's threading lookup finds the
matching source_id and threads into the same conversation.

Changes:

- Send skip_queue: true on every request. Verification still runs (now
  inline) but is cache-hit fast for any recipient we've previously sent to
  — Chatwoot replies are nearly always to such addresses (came in via
  IMAP, contact created on first inbound). User UX unaffected because all
  send paths run via Sidekiq.

- After a successful response, override message.message_id with
  data.message_id from the response. This is what Email::SendOnEmailService
  then writes to source_id.

- Parse LeadMail's structured error response (422/502 with error.code +
  error.message) so failures surface readable diagnostics in Sentry rather
  than raw HTML/JSON bodies.

app/mailers/leadmail_delivery.rb

@@ -9,14 +9,23 @@ class LeadmailDelivery
     payload = build_payload(message)
     response = http_client.post('emails/send', payload.to_json)
 
-    unless response.success?
-      raise LeadmailDeliveryError, "LeadMail API error: #{response.status} - #{response.body}"
-    end
+    raise LeadmailDeliveryError, build_error_message(response) unless response.success?
 
     response_data = JSON.parse(response.body)
-    message.header['X-LeadMail-Log-ID'] = response_data.dig('data', 'log_id')
+    log_id = response_data.dig('data', 'log_id')
+    actual_message_id = response_data.dig('data', 'message_id')
+
+    # Override the Mail::Message's message_id with whatever LeadMail/Zeptomail
+    # actually used as the outbound Message-ID. Email::SendOnEmailService reads
+    # message.message_id after deliver and stores it as source_id; the recipient's
+    # eventual reply has In-Reply-To matching this exact ID, so source_id must
+    # match for threading. Synchronous send (skip_queue=true) is what makes the
+    # actual Message-ID available here at deliver time.
+    message.message_id = actual_message_id if actual_message_id.present?
+
+    message.header['X-LeadMail-Log-ID'] = log_id
     Rails.logger.info(
-      "Email sent via LeadMail: log_id=#{response_data.dig('data', 'log_id')} " \
+      "Email sent via LeadMail: log_id=#{log_id} message_id=#{actual_message_id.inspect} " \
       "from=#{payload[:from].inspect} to=#{message.to.inspect}"
     )
     response
@@ -27,6 +36,22 @@ class LeadmailDelivery
 
   private
 
+  # Parse LeadMail's structured error response. Format per the API spec:
+  #   422 -> {success: false, error: {code: 'VERIFICATION_FAILED'|'NO_VALID_RECIPIENTS', message: '…'}, data: {log_id, status, message_id: nil}}
+  #   502 -> {success: false, error: {code: 'TRANSPORT_ERROR', message: '…'}, data: {…}}
+  # Older or unexpected error shapes fall back to the raw body.
+  def build_error_message(response)
+    body = JSON.parse(response.body) rescue nil
+    if body.is_a?(Hash) && body['error'].is_a?(Hash)
+      code = body['error']['code'] || 'UNKNOWN'
+      msg = body['error']['message'] || ''
+      log_id = body.dig('data', 'log_id')
+      "LeadMail API error: HTTP #{response.status} code=#{code} log_id=#{log_id.inspect} - #{msg}"
+    else
+      "LeadMail API error: HTTP #{response.status} - #{response.body.to_s[0, 300]}"
+    end
+  end
+
   def build_payload(message)
     {
       from: extract_from(message),
@@ -40,11 +65,20 @@ class LeadmailDelivery
       attachments: extract_attachments(message),
       # === LeadChat: microsoft_shared (start) ===
       # Forward threading headers so conversation replies sent via LeadMail thread
-      # correctly in the recipient's mail client. The LeadMail API should preserve
-      # these as RFC822 headers on the outbound MIME.
+      # correctly in the recipient's mail client. The LeadMail API preserves
+      # in_reply_to/references on the outbound MIME and normalises angle-bracket
+      # wrapping. Message-ID is best-effort: ZeptoMail rewrites it for VERP, but
+      # LeadMail returns the actual one used in data.message_id (see deliver!).
       in_reply_to: message.in_reply_to.presence,
       references: message.references.presence,
       message_id: message.message_id.presence,
+      # skip_queue=true makes LeadMail send synchronously and return the actual
+      # Zeptomail-assigned Message-ID in the response. Verification still runs
+      # but is cache-hit fast for any address we've sent to before. Required for
+      # Chatwoot threading: source_id must equal the outbound Message-ID so the
+      # recipient's reply (with In-Reply-To: <that-id>) threads back into the
+      # same conversation.
+      skip_queue: true,
       # === LeadChat: microsoft_shared (end) ===
       options: {
         on_verification_failure: 'strip',