Migrating from Postmark Inbound Streams to GhostParse

// GhostParse's email.received data in the shape of Postmark's inbound webhook JSON, so a handler written for
// Postmark keeps working while you migrate. Pass it `event.data` from a verified webhook.
const who = (a) => (a ? (a.name ? `"${a.name}" <${a.address}>` : a.address) : "");
const full = (list = []) => list.map((a) => ({ Email: a.address, Name: a.name ?? "", MailboxHash: "" }));

export function toPostmark(data) {
  const recipient = data.recipients?.[0];
  return {
    From: data.from?.address ?? "",
    FromName: data.from?.name ?? "",
    FromFull: { Email: data.from?.address ?? "", Name: data.from?.name ?? "", MailboxHash: "" },
    To: data.to.map(who).join(", "),
    ToFull: full(data.to),
    Cc: (data.cc ?? []).map(who).join(", "),
    CcFull: full(data.cc),
    Bcc: "",
    BccFull: [],
    OriginalRecipient: recipient?.address ?? data.envelope?.rcpt_to?.[0] ?? "",
    // Postmark's "+hash" part of the address: GhostParse calls it the tag.
    MailboxHash: recipient?.tag ?? "",
    ReplyTo: (data.reply_to ?? []).map((a) => a.address).join(", "),
    Subject: data.subject ?? "",
    // Postmark's own id for the message; the Message-ID header is in Headers, as in Postmark.
    MessageID: data.id,
    Date: new Date(data.date ?? data.received_at).toUTCString().replace("GMT", "+0000"),
    TextBody: data.text ?? "",
    HtmlBody: data.html ?? "",
    StrippedTextReply: data.reply_text ?? "",
    Tag: "",
    Headers: Object.entries(data.headers ?? {}).map(([Name, Value]) => ({ Name, Value: String(Value) })),
    Attachments: data.attachments.map((a) => ({
      Name: a.filename ?? "attachment",
      Content: a.content ?? "", // base64, as in Postmark (empty if attachments go to your bucket)
      ContentType: a.content_type,
      ContentLength: a.size,
      ContentID: a.content_id ?? "",
    })),
  };
}

"""GhostParse's email.received data in the shape of Postmark's inbound webhook JSON, so a handler written for
Postmark keeps working while you migrate. Pass it event["data"] from a verified webhook."""
from datetime import datetime
from email.utils import format_datetime


def _who(a):
    return (f'"{a["name"]}" <{a["address"]}>' if a.get("name") else a["address"]) if a else ""


def _full(addresses):
    return [{"Email": a["address"], "Name": a.get("name") or "", "MailboxHash": ""} for a in addresses or []]


def to_postmark(data: dict) -> dict:
    sender = data.get("from") or {}
    recipient = (data.get("recipients") or [None])[0]
    when = datetime.fromisoformat((data.get("date") or data["received_at"]).replace("Z", "+00:00"))
    return {
        "From": sender.get("address", ""),
        "FromName": sender.get("name") or "",
        "FromFull": {"Email": sender.get("address", ""), "Name": sender.get("name") or "", "MailboxHash": ""},
        "To": ", ".join(_who(a) for a in data["to"]),
        "ToFull": _full(data["to"]),
        "Cc": ", ".join(_who(a) for a in data.get("cc") or []),
        "CcFull": _full(data.get("cc")),
        "Bcc": "",
        "BccFull": [],
        "OriginalRecipient": recipient["address"] if recipient else ((data.get("envelope") or {}).get("rcpt_to") or [""])[0],
        # Postmark's "+hash" part of the address: GhostParse calls it the tag.
        "MailboxHash": (recipient or {}).get("tag") or "",
        "ReplyTo": ", ".join(a["address"] for a in data.get("reply_to") or []),
        "Subject": data.get("subject") or "",
        # Postmark's own id for the message; the Message-ID header is in Headers, as in Postmark.
        "MessageID": data["id"],
        "Date": format_datetime(when),
        "TextBody": data.get("text") or "",
        "HtmlBody": data.get("html") or "",
        "StrippedTextReply": data.get("reply_text") or "",
        "Tag": "",
        "Headers": [{"Name": k, "Value": str(v)} for k, v in (data.get("headers") or {}).items()],
        "Attachments": [
            {
                "Name": a.get("filename") or "attachment",
                "Content": a.get("content") or "",  # base64, as in Postmark (empty if attachments go to your bucket)
                "ContentType": a["content_type"],
                "ContentLength": a["size"],
                "ContentID": a.get("content_id") or "",
            }
            for a in data["attachments"]
        ],
    }

import express from "express";
import { verifyWebhook } from "@wildye/email-parser";
import { toPostmark } from "./postmark.mjs";
import { handleInbound } from "./your-postmark-handler.js"; // the code you already have

const app = express();
// GhostParse signs every webhook: verify it with the raw body, then hand your old handler the Postmark shape.
app.post("/webhooks/ghostparse", express.raw({ type: "application/json", limit: "40mb" }), async (req, res) => {
  let event;
  try {
    event = verifyWebhook(req.body, req.headers, process.env.GHOSTPARSE_WEBHOOK_SECRET);
  } catch {
    return res.sendStatus(400);
  }
  if (event.type === "email.received") await handleInbound(toPostmark(event.data));
  res.sendStatus(200);
});

import os
from flask import Flask, request
from email_parser import verify_webhook, WebhookVerificationError
from postmark import to_postmark
from your_postmark_handler import handle_inbound  # the code you already have

app = Flask(__name__)

@app.post("/webhooks/ghostparse")
def ghostparse():
    # GhostParse signs every webhook: verify it with the raw body, then hand your old handler the Postmark shape.
    try:
        event = verify_webhook(request.get_data(), request.headers, os.environ["GHOSTPARSE_WEBHOOK_SECRET"])
    except WebhookVerificationError:
        return "", 400
    if event.type == "email.received":
        handle_inbound(to_postmark(event.data))
    return "", 200