Notifications from one API: template, batch, webhook.

# One notification, one template, the variables the template names.
curl -X POST https://app.usenotix.dev/api/v1/emails \
  -H "Authorization: Bearer $NOTIX_API_KEY" \
  -H "Idempotency-Key: comment-cmt_8812-notify-usr_123" \
  -H "Content-Type: application/json" \
  -d '{
    "from": "Acme <notifications@acme.com>",
    "to": "user@example.com",
    "templateId": "tpl_new_comment",
    "variables": {
      "commenter": "Amara",
      "excerpt": "Can we move the deadline to Friday?",
      "threadUrl": "https://acme.com/t/8812"
    }
  }'

# 200 { "emailId": "eml_abc123" }

import { Notix } from "notix-js";

const notix = new Notix(process.env.NOTIX_API_KEY);

// One template per notification type; the key names the event and the
// recipient, so a retried worker never sends the same notice twice.
export async function notifyNewComment(comment: Comment, recipient: User) {
  const { data, error } = await notix.emails.send(
    {
      from: "Acme <notifications@acme.com>",
      to: recipient.email,
      templateId: "tpl_new_comment",
      variables: {
        commenter: comment.author.name,
        excerpt: comment.body.slice(0, 140),
        threadUrl: `https://acme.com/t/${comment.threadId}`,
      },
    },
    { idempotencyKey: `comment-${comment.id}-notify-${recipient.id}` },
  );
  if (error) throw new Error(error.message);
  return data.emailId;
}

import os
from notix import Notix

notix = Notix(os.environ["NOTIX_API_KEY"])


def notify_new_comment(comment, recipient):
    data, error = notix.emails.send(
        {
            "from": "Acme <notifications@acme.com>",
            "to": recipient.email,
            "templateId": "tpl_new_comment",
            "variables": {
                "commenter": comment.author.name,
                "excerpt": comment.body[:140],
                "threadUrl": f"https://acme.com/t/{comment.thread_id}",
            },
        },
        options={"idempotency_key": f"comment-{comment.id}-notify-{recipient.id}"},
    )
    if error:
        raise RuntimeError(error["message"])
    return data["emailId"]

# An event that fans out to many people: one call, up to 100 emails,
# one idempotency key for the whole batch.
curl -X POST https://app.usenotix.dev/api/v1/emails/batch \
  -H "Authorization: Bearer $NOTIX_API_KEY" \
  -H "Idempotency-Key: deploy-dpl_551-notify" \
  -H "Content-Type: application/json" \
  -d '[
    { "from": "Acme <notifications@acme.com>", "to": "ada@example.com",
      "templateId": "tpl_deploy_finished", "variables": { "app": "api", "version": "2.14.0" } },
    { "from": "Acme <notifications@acme.com>", "to": "tunde@example.com",
      "templateId": "tpl_deploy_finished", "variables": { "app": "api", "version": "2.14.0" } },
    { "from": "Acme <notifications@acme.com>", "to": "chidi@example.com",
      "templateId": "tpl_weekly_digest", "variables": { "count": "12" },
      "scheduledAt": "2026-09-15T08:00:00+01:00" }
  ]'

# 200 { "data": [ { "emailId": "eml_1" }, { "emailId": "eml_2" }, { "emailId": "eml_3" } ] }

// Fan out in pages of 100; each page gets its own key.
export async function notifyDeployFinished(deploy: Deploy, watchers: User[]) {
  for (let i = 0; i < watchers.length; i += 100) {
    const page = watchers.slice(i, i + 100);
    const { data, error } = await notix.emails.batch(
      page.map((user) => ({
        from: "Acme <notifications@acme.com>",
        to: user.email,
        templateId: "tpl_deploy_finished",
        variables: { app: deploy.app, version: deploy.version },
      })),
      { idempotencyKey: `deploy-${deploy.id}-notify-${i / 100}` },
    );
    if (error) throw new Error(error.message);
    await db.notification.createMany({
      data: data.data.map((row, n) => ({ userId: page[n].id, emailId: row.emailId, state: "queued" })),
    });
  }
}

// Low-priority notices: hold them for the morning digest instead.
export async function queueForDigest(user: User, items: string[]) {
  const nextMorning = nextLocalTime(user.timeZone, 8);
  return notix.emails.send(
    {
      from: "Acme <notifications@acme.com>",
      to: user.email,
      templateId: "tpl_weekly_digest",
      variables: { count: String(items.length) },
      scheduledAt: nextMorning.toISOString(),
    },
    { idempotencyKey: `digest-${user.id}-${nextMorning.toISOString().slice(0, 10)}` },
  );
}

def notify_deploy_finished(deploy, watchers):
    for i in range(0, len(watchers), 100):
        page = watchers[i : i + 100]
        data, error = notix.emails.batch(
            [
                {
                    "from": "Acme <notifications@acme.com>",
                    "to": user.email,
                    "templateId": "tpl_deploy_finished",
                    "variables": {"app": deploy.app, "version": deploy.version},
                }
                for user in page
            ],
            options={"idempotency_key": f"deploy-{deploy.id}-notify-{i // 100}"},
        )
        if error:
            raise RuntimeError(error["message"])
        for user, row in zip(page, data["data"]):
            Notification.objects.create(user=user, email_id=row["emailId"], state="queued")

// Next.js route handler. The event id is the email id you stored, so
// the notification row updates without a lookup by address.
import { notix } from "@/lib/notix";

export async function POST(request: Request) {
  const raw = await request.text();
  const event = notix.webhooks(process.env.NOTIX_WEBHOOK_SECRET!).constructEvent(raw, {
    headers: request.headers,
  });

  const state = {
    "email.delivered": "delivered",
    "email.bounced": "bounced",
    "email.complained": "complained",
    "email.suppressed": "suppressed",
    "email.opened": "opened",
  }[event.type];

  if (state) {
    await db.notification.updateMany({ where: { emailId: event.data.id }, data: { state } });
    if (event.type === "email.bounced" && event.data.bounce.type === "Permanent") {
      // The address is now suppressed; ask for a new one in the product.
      await db.user.update({ where: { email: event.data.to[0] }, data: { emailInvalid: true } });
    }
  }
  return new Response("ok");
}

import os
from notix import Notix

notix = Notix(os.environ["NOTIX_API_KEY"])
webhooks = notix.webhooks(os.environ["NOTIX_WEBHOOK_SECRET"])

STATES = {
    "email.delivered": "delivered",
    "email.bounced": "bounced",
    "email.complained": "complained",
    "email.suppressed": "suppressed",
    "email.opened": "opened",
}


@app.post("/hooks/notix")
def notix_hook():
    event = webhooks.construct_event(request.data, headers=request.headers)
    state = STATES.get(event["type"])
    if state:
        Notification.objects.filter(email_id=event["data"]["id"]).update(state=state)
        if event["type"] == "email.bounced" and event["data"]["bounce"]["type"] == "Permanent":
            User.objects.filter(email=event["data"]["to"][0]).update(email_invalid=True)
    return "", 200