Building Relay

Part 3 · Chapter 3.9

The email nobody was sending

You will produce: The outbox pattern a third time, over a column chapter 3.6 already wrote — and Mailpit, because only a received message can prove an email carries no secret · about 60 minutes including the exercise

Source: SRS — Software Requirements Specification · Journey map

Chương 3.6 dựng cơ chế tự động tắt webhook. Một endpoint hỏng suốt một tiếng qua hai mươi lượt thử sẽ bị ngắt, và một dòng được ghi lại nói rằng tổ chức ấy đang được nợ một lời giải thích:

services/api/src/db/repository.ts (excerpt)
await tx.insert(webhookDisableNotifications).values({
  id: randomUUID(),

  // NOT SET, and that is the point. FR-WHK-07 asks for the organisation to be
  // notified by email and this platform has no email; `delivered_at` exists in
  // order to be null until a transport does.
});

Hai chương sau, những dòng ấy vẫn nằm đó và delivered_at vẫn null trên từng dòng một. FR-WHK-07 mới giao được một nửa: nền tảng biết mình nợ một email và không có cách nào gửi đi.

Chương này gửi nó. Phần thú vị không nằm ở SMTP.

Cái bảng vốn đã có sẵn đúng hình dạng

flowchart TB
    subgraph c33["chương 3.3 · event"]
      o1["outbox<br/>published_at"] --> o1x["NATS"]
    end
    subgraph c35["chương 3.5 · delivery"]
      o2["webhook_deliveries<br/>state · next_attempt_at"] --> o2x["endpoint của khách hàng"]
    end
    subgraph c38["chương 3.8 · notification"]
      o3["webhook_disable_notifications<br/>delivered_at"] --> o3x["SMTP"]
    end
    note["cái thứ ba KHÔNG cần migration:<br/>chương 3.6 đã viết delivered_at<br/>và để nó null suốt"]
    o3 -.-> note
    style o3 fill:#064e3b,color:#fff,stroke:#059669
Ba outbox, ba chương — và cái thứ ba không tốn một thay đổi schema nào.

Chương 3.3 rút bảng outbox sang NATS, lấy khóa theo published_at IS NULL. Chương 3.5 rút webhook_deliveries sang endpoint của khách hàng, lấy khóa theo trạng thái và thời điểm đến hạn. Chương này rút webhook_disable_notifications sang SMTP, lấy khóa theo delivered_at IS NULL.

Cột đó tồn tại từ chương 3.6 và luôn null trên mọi dòng từng được ghi. Nghĩa là vị từ nhận việc mà phương tiện gửi cần đã được viết xuống từ hai chương trước khi nó ra đời — và đống tồn đọng mà hai chương ấy tích lại không phải một bài toán migration. Nó là công việc chưa gửi, theo đúng định nghĩa của chính vị từ đó:

services/api/src/db/repository.ts (excerpt)
WHERE n.delivered_at IS NULL
ORDER BY n.disabled_at, n.id
LIMIT $1
  FOR UPDATE OF n SKIP LOCKED

FOR UPDATE OF n SKIP LOCKED là cơ chế của chương 3.3, không đổi: hai instance api có thể cùng rút và bỏ qua những dòng bên kia đã nhận thay vì chặn nhau. OF n là phần thêm — truy vấn nhận việc join sang environments, applicationswebhook_endpoints để dựng nội dung thư, và khóa cả những dòng đó lại sẽ khiến việc gửi một email chặn mất một khách hàng đang sửa endpoint của họ.

Mailpit, và hiến pháp VII

Hiến pháp VII đòi container thứ năm phải tự biện minh. Đây là lời biện minh, và nó là FR-WHK-07:

Thông báo KHÔNG ĐƯỢC chứa signing secret của endpoint, một API key, hay bất kỳ credential nào khác.

flowchart LR
    facts["DisableFacts<br/>url · environment · attempts<br/>KHÔNG có trường nào cho bí mật"]
    mail["disableNotification()"]
    smtp["Mailpit · SMTP"]
    api["Mailpit HTTP API"]
    test["lời khẳng định"]
    facts --> mail --> smtp --> api --> test
    stub["một STUB sẽ để test đọc lại<br/>chính đối tượng bên gửi truyền vào —<br/>nên một bí mật nằm trong header mà<br/>stub không mô hình hóa sẽ lọt qua"]
    test -.-> stub
    style smtp fill:#064e3b,color:#fff,stroke:#059669
    style stub fill:#7f1d1d,color:#fff,stroke:#dc2626
Vì sao lời khẳng định này cần một server chứ không phải một stub.

Đó là một khẳng định về nội dung của một email. Một mailer stub ghi lại chính đối tượng mà bên gửi truyền vào, nên một test khẳng định trên stub là đang đọc lại đầu vào của chính mình — và một mailer nhét bí mật vào một header mà stub không mô hình hóa sẽ vượt qua. Thứ duy nhất phán quyết được là một thông điệp mà một server đã thực sự nhận, kèm header, kèm cách mã hóa, kèm tất cả.

Vậy nên Mailpit: một SMTP server nhận mọi thứ, không gửi đi đâu cả, và phơi ra một HTTP API để đọc thứ nó bắt được. Cổng 1102518025, lệch khỏi mặc định, khớp quy ước 15432/16379/14222 mà mọi kho lưu trữ khác trong file compose này đều theo — để lane không đụng phải container của chính lập trình viên. Một healthcheck, vì thiếu nó thì docker compose up -d --wait chỉ chờ đang chạy chứ không chờ sẵn sàng, và suite có thể đọc API trước khi nó kịp phục vụ. Không volume: một hộp thư test sống sót qua lần khởi động lại là một hộp thư test rò rỉ trạng thái giữa các lượt chạy.

Nội dung thư, và đường nối khiến FR-WHK-07 thành sự thật

Mailer gồm hai mảnh, và chính cách chia ấy là biện pháp bảo mật.

services/api/src/notifications/mailer.ts (excerpt)
export interface DisableFacts {
  endpointUrl: string;
  environmentName: string;
  disabledAt: Date;
  runStartedAt: Date;
  attempts: number;
  /** Null when the endpoint never answered — a refused connection has no status. */
  lastStatus: number | null;
  lastError: string | null;
}

disableNotification(facts) biến chừng đó thành một tiêu đề và một thân thư, không chạm tới SMTP, không chạm đồng hồ, không chạm database — nên nội dung email nói gì được quyết bởi một unit test. createMailer() là mảnh nói chuyện với server, và mỏng tới mức chẳng còn gì trong đó để làm sai.

DisableFacts không có trường nào cho một bí mật. Mailer không thể làm rò rỉ thứ nó chưa từng được trao, và điều đó biến FR-WHK-07 thành thuộc tính của kiểu dữ liệu thay vì một bộ lọc trên đầu ra — mà bộ lọc là thứ bạn viết khi hình dạng đã thua từ trước.

Phép quét vẫn chạy, cả trong unit test lẫn trên thứ Mailpit nhận được, bởi "không thể xảy ra" là một lời khẳng định còn một phép quét là bằng chứng.

Người nhận, phân giải tại thời điểm gửi

services/api/src/db/repository.ts (excerpt)
SELECT DISTINCT h.email
  FROM memberships m
  JOIN humans h ON h.id = m.human_id
 WHERE m.organisation_id = $1
   AND h.email IS NOT NULL

Tổ chức đến từ chính dòng dữ liệu, không phải từ chủ sở hữu hiện tại của endpoint. Chương 3.6 đã phi chuẩn hóa organisation_id lên bản ghi thông báo đúng vì việc này, với lý do viết thẳng vào schema ngay lúc đó:

dòng này ghi lại một nghĩa vụ ĐÚNG NHƯ NÓ ĐÃ TỒN TẠI khi endpoint bị tắt, và một application chuyển sang tổ chức khác về sau không được phép lặng lẽ chuyển hướng một thông báo vốn đã nợ người khác

Chương này là đoạn code đầu tiên phụ thuộc vào điều đó, hai chương sau khi cột ấy được thêm. Một phép phi chuẩn hóa có lý do viết sẵn mà chưa ai đọc là một canh bạc; đây là lúc canh bạc ấy được trả.

Mọi thành viên có địa chỉ, không riêng chủ sở hữu. memberships.role là một trong owner, admin hoặc member, và chọn ra một tập con ở đây sẽ là chương này tự bịa ra một mô hình tùy chọn thông báo — thứ thuộc về sản phẩm.

Một thư cho mỗi người nhận, không bao giờ một thư mang nhiều địa chỉ. Địa chỉ email của đồng nghiệp một khách hàng là dữ liệu của khách hàng đó, và một header To mà mọi người nhận đều đọc được là một sự tiết lộ không ai yêu cầu.

Không có ai để viết thư tới

humans.email là nullable. Một người đăng nhập qua nhà cung cấp không trả về địa chỉ thì không có địa chỉ nào, nên một tổ chức mà mọi thành viên đều không thể gửi thư tới là trạng thái schema cho phép — một nhánh thật sự, không phải một câu if phòng thủ.

services/api/src/notifications/notification-relay.ts (excerpt)
if (recipients.length === 0) {
  logger.log("error", "notifications.unaddressable", {
    organisation_id: row.organisationId,
    notification_id: row.id,
    detail:
      "webhook disablement could not be notified: no member has an email address",
  });
  return;
}

Trả về bình thường sẽ đánh dấu dòng đó là đã gửi, và điều đó là có chủ đích. Không có địa chỉ nào để thử lại, và để nó tiếp tục nhận được nghĩa là cứ năm giây lại nhận đúng cái dòng không gửi được ấy, mãi mãi. Thứ thay thế cho email là dòng log: nghĩa vụ được giải trừ tới mức có thể, và việc nó không thể được đáp ứng thì được ghi lại chứ không nuốt đi.

Đột biến không thể gãy

Bộ sabotage của chương có một đột biến nhắm vào file này: đánh dấu delivered_at trước khi lệnh gửi trả về. Nó phải làm gãy cái test nói rằng dấu ấy đến sau.

Nó pass.

Lần theo điều đó lại tìm ra một lỗi tệ hơn thứ đột biến đang nhắm tới.

flowchart TB
    subgraph before["một transaction cho cả lô"]
      b1["nhận việc cũ nhất trước<br/>dòng A · địa chỉ hỏng<br/>dòng B · dòng C"]
      b2["gửi A → NÉM LỖI"]
      b3["transaction cuộn ngược"]
      b4["A, B và C đều chưa được đánh dấu"]
      b5["lượt sau lại nhận A đầu tiên"]
      b1 --> b2 --> b3 --> b4 --> b5
      b5 -.->|"mãi mãi"| b1
    end
    subgraph after["cô lập theo từng dòng"]
      a1["nhận việc cũ nhất trước"]
      a2["A ném lỗi → onError, không đánh dấu"]
      a3["B và C gửi đi → đánh dấu"]
      a4["A thử lại lượt sau<br/>B và C đã đi rồi"]
      a1 --> a2 --> a3 --> a4
    end
    style b5 fill:#7f1d1d,color:#fff,stroke:#dc2626
    style a3 fill:#064e3b,color:#fff,stroke:#059669
Một địa chỉ mà mail server từ chối, và hàng đợi dừng lại mãi mãi.

Các dòng được nhận theo thứ tự cũ nhất trước. Một dòng luôn ném lỗi — một địa chỉ bị mail server từ chối thẳng, chuyện hết sức bình thường với một địa chỉ — vì vậy luôn được nhận đầu tiên, luôn làm hủy cả lô, và mọi thông báo đứng sau nó không bao giờ được gửi. Nghẽn đầu dòng vĩnh viễn cho cả hàng đợi, chỉ từ một người nhận hỏng.

Bản vá là cô lập theo từng dòng:

services/api/src/db/repository.ts (excerpt)
for (const raw of claimed.rows) {
  try {
    await deliver(row);
    delivered.push(row.id);
  } catch (error) {
    // Not marked. The row stays claimable and the next pass tries it again,
    // which is the whole reason this is a table rather than a call.
    onError(row, error);
  }
}

Giờ không có gì cuộn ngược nữa, nên thứ tự là có thật và đột biến gãy đúng như nó phải gãy. Và có một test mang tên đúng cái thất bại mà chẳng ai đi tìm: một địa chỉ Mailpit từ chối, một địa chỉ tốt đứng ngay sau, và cái tốt vẫn đi.

Mail server biến mất thì tốn những gì

Không gì cả, và có một test nói vậy.

Tắt một endpoint là ghi một dòng rồi trả về; việc có ai gửi được email hay không là chuyện của riêng vòng lặp này. Với relay trỏ vào một cổng chẳng trả lời gì: không ném lỗi, không gửi được, dòng dữ liệu vẫn nhận được, và một endpoint thứ hai vẫn bị tắt bình thường trong lúc đó. Việc chuyển tin nhắn, API và việc phát webhook không hề chạm tới SMTP.

Đó là lý do phương tiện gửi là một cái bảng cộng một vòng lặp chứ không phải một lời gọi nằm trong đường tắt endpoint. Một cú timeout SMTP bên trong sweepDisabledEndpoints sẽ biến sự cố mail thành sự cố webhook — đánh đổi phụ thuộc kém quan trọng nhất hệ thống lấy một trong những phụ thuộc quan trọng nhất.

Relay poll năm giây một lần thay vì 200ms như event relay. Chẳng có gì phải chờ nó: FR-WHK-07 yêu cầu tổ chức được thông báo, chứ không yêu cầu được thông báo trong vòng một giây, và một mail server bị một service chẳng có gì để gửi poll năm lần mỗi giây là một service đang bất lịch sự với các phụ thuộc của mình.

Trọn vẹn chương này

Tất cả những gì ở trên, đúng như repository đang giữ.

Phương tiện gửi

Ba file mới. mailer.ts tách phần nội dung khỏi phần gửi đi, vì nội dung email NÓI gì thì do một unit test quyết, còn SMTP làm gì thì không.

services/api/src/notifications/mailer.ts
import { createTransport, type Transporter } from "nodemailer";
 
// The disablement notification (chapter 3.8, FR-WHK-07, FR-WHK-07).
//
// TWO PIECES, SEPARATED ON PURPOSE. `disableNotification` turns facts into a
// message and touches nothing — no SMTP, no clock, no database — so what the
// email SAYS is decided by a unit test. `createMailer` is the part that talks to
// a server, and it is thin enough that there is nothing in it to get wrong.
//
// THE SEAM IS THE SECURITY CONTROL. `DisableFacts` has no field for a secret, so
// the mailer cannot leak one it was never given — FR-WHK-07 is enforced by the
// shape of the input rather than by a filter over the output, and a filter is
// what you write when the shape already lost. The test scans the message anyway,
// because "cannot happen" is a claim and a scan is evidence.
 
/** Everything the message is allowed to know. */
export interface DisableFacts {
  endpointUrl: string;
  environmentName: string;
  disabledAt: Date;
  runStartedAt: Date;
  attempts: number;
  /** Null when the endpoint never answered — a refused connection has no status. */
  lastStatus: number | null;
  lastError: string | null;
}
 
export interface Mail {
  subject: string;
  text: string;
}
 
/** How long the failing run went on, in whole minutes. Rounded up, because "0
 * minutes" reads as "no time passed" for a run that lasted forty seconds. */
function durationMinutes(from: Date, to: Date): number {
  return Math.max(1, Math.ceil((to.getTime() - from.getTime()) / 60_000));
}
 
export function disableNotification(facts: DisableFacts): Mail {
  const host = new URL(facts.endpointUrl).host;
  // What it was failing WITH. A status when there was one, the transport error
  // when the request never got far enough to have one. Printing `null` would be
  // accurate and unusable.
  const cause =
    facts.lastStatus !== null
      ? `HTTP ${facts.lastStatus}`
      : (facts.lastError ?? "no response");
  const minutes = durationMinutes(facts.runStartedAt, facts.disabledAt);
 
  return {
    subject: `Relay disabled your webhook endpoint at ${host}`,
    text: [
      `Relay has stopped delivering webhooks to:`,
      ``,
      `    ${facts.endpointUrl}`,
      ``,
      `Environment: ${facts.environmentName}`,
      `Failing for: ${minutes} minute${minutes === 1 ? "" : "s"}`,
      `Attempts:    ${facts.attempts}`,
      `Last result: ${cause}`,
      ``,
      // The instruction, not just the state. A notification that reports a
      // problem without naming the action is a notification that becomes a
      // support ticket.
      `Deliveries will not resume on their own. Fix the endpoint, then`,
      `re-enable it from the webhook settings for this environment.`,
      ``,
      // Said explicitly, because the absence is the thing a reader will wonder
      // about. Nothing here identifies the endpoint beyond its own URL.
      `This message contains no signing secret and no credential. If you`,
      `need the secret to verify deliveries, read it from the dashboard.`,
    ].join("\n"),
  };
}
 
export const DEFAULT_SMTP_URL = "smtp://localhost:1025";
 
export interface Mailer {
  send: (to: string, mail: Mail) => Promise<void>;
  close: () => void;
}
 
/** A full URL with a default here, never a host and a port the caller composes.
 * `harness.ts` records what the other shape costs: a caller that builds its own
 * URL is a second source of truth for an address, which is how the e2e suite
 * first failed. */
export function createMailer(
  url: string = process.env["RELAY_SMTP_URL"] ?? DEFAULT_SMTP_URL,
  from = "Relay <relay@relay.example>",
): Mailer {
  const transport: Transporter = createTransport(url);
  return {
    send: async (to, mail) => {
      await transport.sendMail({ from, to, ...mail });
    },
    close: () => {
      transport.close();
    },
  };
}
services/api/src/notifications/notification-relay.ts
import type { Logger } from "@relay/service-kit";
 
import type { Db } from "../db/client";
import {
  drainDisableNotifications,
  organisationRecipients,
  type DisableNotificationRow,
} from "../db/repository";
import { disableNotification, type Mailer } from "./mailer";
 
// The notification relay (chapter 3.8, FR-WHK-07 to FR-WHK-07).
//
// THE OUTBOX A THIRD TIME, and deliberately the same shape as chapter 3.3's:
// claim undelivered rows oldest-first with `FOR UPDATE SKIP LOCKED`, do the
// side effect, mark what succeeded, and put the mark in a `finally`. A reader
// who understood the event relay understands this one, which is the argument
// for reaching for a pattern the codebase already has rather than a queue
// library it does not (constitution VII).
//
// WHAT IS DIFFERENT is that the side effect is an email, which cannot be undone
// and cannot be deduplicated by the recipient. That pushes every ambiguous case
// the same way: send once too few rather than once too many is WRONG here —
// FR-WHK-07 exists because an endpoint went quiet and nobody was told — so a
// crash between the send and the mark resends, and the chapter says so.
//
// It also makes a failing row a ROW's problem rather than the batch's. The event
// relay lets one bad publish abort its batch, because a broker is up or down;
// one address a mail server refuses is one address, and aborting on it would
// abort every batch for ever — it is claimed first, being oldest. The repository
// catches per row and this callback is where the failure surfaces.
//
// NOT ON THE REQUEST PATH, for chapter 3.3's reason. Disabling an endpoint
// writes a row and returns; whether a mail server is reachable is this loop's
// problem. An SMTP timeout inside the dispatcher's disablement check would make
// a mail outage into a webhook outage.
 
/** Rows per pass. Smaller than the event relay's hundred because each row is a
 * network round trip to a mail server rather than a publish to a local broker,
 * and a batch is a transaction. */
const BATCH_SIZE = 20;
 
/** Slower than the event relay's 200ms, and it should be. Nothing is waiting on
 * this: FR-WHK-07 asks that the organisation be told, not that it be told within
 * the second, and a mail server polled five times a second by a service with
 * nothing to send is a service being rude to its dependencies. */
const IDLE_INTERVAL_MS = 5_000;
 
export interface NotificationRelay {
  start(): void;
  stop(): Promise<void>;
  /** One pass, for tests and for the walk script — the same code path `start`
   * runs, so nothing is proven about a loop only tests exercise. */
  drainOnce(): Promise<number>;
}
 
export function createNotificationRelay({
  db,
  mailer,
  logger,
  batchSize = BATCH_SIZE,
  intervalMs = IDLE_INTERVAL_MS,
}: {
  db: Db;
  mailer: Mailer;
  logger: Logger;
  batchSize?: number;
  intervalMs?: number;
}): NotificationRelay {
  let running = false;
  let loop: Promise<void> = Promise.resolve();
 
  async function deliver(row: DisableNotificationRow): Promise<void> {
    // Resolved from the ROW's organisation, not from the endpoint's current
    // owner. Chapter 3.6 denormalised that column so this lookup could not
    // follow an application that moved after the disablement (FR-WHK-07).
    const recipients = await organisationRecipients(db, row.organisationId);
 
    if (recipients.length === 0) {
      // A REAL BRANCH, not a defensive `if`. `humans.email` is nullable — a
      // human who signed in through a provider that returned no address has
      // none — so an organisation whose every member is unaddressable is a
      // state the schema permits and this code will meet (FR-WHK-07).
      //
      // The row is still marked delivered. There is no address to retry to, and
      // leaving it claimable would mean this relay reclaimed the same
      // undeliverable row every five seconds for ever. What replaces the email
      // is this log line: the obligation is discharged as far as it can be, and
      // the fact that it could not be met is recorded rather than swallowed.
      logger.log("error", "notifications.unaddressable", {
        organisation_id: row.organisationId,
        notification_id: row.id,
        detail:
          "webhook disablement could not be notified: no member has an email address",
      });
      return;
    }
 
    const mail = disableNotification({
      endpointUrl: row.endpointUrl,
      environmentName: row.environmentName,
      disabledAt: row.disabledAt,
      runStartedAt: row.runStartedAt,
      attempts: row.runAttempts,
      lastStatus: row.lastStatus,
      lastError: row.lastError,
    });
 
    // Sequential, and one message per recipient rather than one message with
    // several addresses on it: a customer's colleagues' email addresses are
    // that customer's data, and putting them in a header every recipient can
    // read is a disclosure nobody asked for.
    for (const to of recipients) {
      await mailer.send(to, mail);
    }
    logger.log("info", "notifications.sent", {
      notification_id: row.id,
      recipients: recipients.length,
    });
  }
 
  async function drainOnce(): Promise<number> {
    return drainDisableNotifications(db, batchSize, deliver, (row, error) => {
      // One row's failure, one line, and the batch keeps going. A mail server
      // that is down produces one of these per claimed row and then a drain of
      // zero, which the loop treats as idle — correct, because there is nothing
      // this process can do but wait.
      logger.log("error", "notifications.send_failed", {
        notification_id: row.id,
        error: String(error),
      });
    });
  }
 
  async function run(): Promise<void> {
    while (running) {
      try {
        const sent = await drainOnce();
        if (sent > 0) {
          // A count and an id. Never an address: a recipient list in a log line
          // is a customer's people in an operator's terminal (NFR-SEC-06).
          logger.log("info", "notifications.drained", { count: sent });
          continue;
        }
      } catch (error) {
        // A mail server that is down lands here. Rows stay claimable and the
        // next pass tries again, which is the whole reason this is a table
        // rather than a call.
        logger.log("error", "notifications.drain_failed", {
          error: String(error),
        });
      }
      await new Promise((resolve) => setTimeout(resolve, intervalMs));
    }
  }
 
  return {
    start() {
      if (running) return;
      running = true;
      loop = run();
    },
    async stop() {
      running = false;
      await loop;
      mailer.close();
    },
    drainOnce,
  };
}
services/api/src/notifications/notifications.module.ts
import { Inject, Injectable, Module, type OnModuleDestroy } from "@nestjs/common";
 
import { createLogger } from "@relay/service-kit";
 
import { createDb, createPool, type Db } from "../db/client";
import { createMailer } from "./mailer";
import {
  createNotificationRelay,
  type NotificationRelay,
} from "./notification-relay";
 
// The notification relay's home (chapter 3.8). Same shape as the outbox
// module's, deliberately: a loop that reads a table, does a side effect, and
// shares no state with the request path — so promoting it out of this service
// would mean moving this file and nothing else.
 
export const NOTIFICATION_RELAY = "NOTIFICATION_RELAY";
 
/** Off for the suites that want a quiet database, on everywhere else. Same
 * switch and same reasoning as `RELAY_OUTBOX_RELAY`: most integration tests
 * assert on rows, and a background loop marking them delivered mid-assertion is
 * a race between test files rather than a property of the system.
 *
 * FLAPPING IS NOT SOLVED HERE, and is worth naming rather than discovering. An
 * endpoint that is disabled, re-enabled and disabled again produces two rows and
 * two emails, and nothing collapses them — the spec's own edge case says neither
 * must suppress the other, because a second outage really is a second thing to
 * be told about. An endpoint flapping hourly therefore sends hourly. Solving it
 * means a notification-preferences model, which is product. */
export function notificationRelayEnabled(): boolean {
  return (
    (process.env.RELAY_NOTIFICATION_RELAY ?? "on").toLowerCase() !== "off"
  );
}
 
@Injectable()
export class NotificationRelayService implements OnModuleDestroy {
  constructor(
    @Inject(NOTIFICATION_RELAY) private readonly relay: NotificationRelay,
  ) {}
 
  start(): void {
    if (notificationRelayEnabled()) this.relay.start();
  }
 
  async onModuleDestroy(): Promise<void> {
    await this.relay.stop();
  }
}
 
@Module({
  providers: [
    {
      provide: NOTIFICATION_RELAY,
      useFactory: (): NotificationRelay =>
        createNotificationRelay({
          db: createDb(createPool()) as Db,
          mailer: createMailer(),
          logger: createLogger("notifications"),
        }),
    },
    NotificationRelayService,
  ],
  exports: [NOTIFICATION_RELAY, NotificationRelayService],
})
export class NotificationsModule {}

Truy vấn nhận việc, và nơi nó được khởi động

Repository có thêm phần rút hàng đợi và phần tra cứu người nhận; main.ts khởi động vòng lặp bên cạnh hai vòng lặp còn lại.

services/api/src/db/repository.ts
@@ -306,6 +349,169 @@ export async function drainOutbox(
   });
 }
 
+// ---------------------------------------------------------------------------
+// The disablement notifications (chapter 3.8, FR-WHK-07 to FR-WHK-07). THE OUTBOX A
+// THIRD TIME — after chapter 3.3's events and chapter 3.5's deliveries — and
+// this one needed no migration at all: chapter 3.6 gave the table a
+// `delivered_at` column and left it null throughout, which is a claim predicate
+// already written down.
+//
+// The backlog 3.6 accumulated therefore drains on the first run with NO SPECIAL
+// HANDLING. By the predicate's own definition those rows are undelivered work,
+// and code that treated them as a migration would be code asserting they are
+// different when they are not (FR-WHK-07).
+//
+// Admin surface, like `drainOutbox`: one relay serves every environment, because
+// a notification is an obligation the platform owes rather than tenant traffic.
+// ---------------------------------------------------------------------------
+
+export interface DisableNotificationRow {
+  id: string;
+  organisationId: string;
+  /** What to call the environment in an email. `environments` has a `kind` —
+   * development, staging, production — and no name of its own, so on its own it
+   * is ambiguous for an organisation with four applications. The application's
+   * name and the kind together are the shortest thing a reader can act on. */
+  environmentName: string;
+  endpointUrl: string;
+  disabledAt: Date;
+  runStartedAt: Date;
+  runAttempts: number;
+  lastStatus: number | null;
+  lastError: string | null;
+}
+
+/** Claim up to `limit` undelivered notifications, hand each to `deliver`, and
+ * mark the ones that went out — all inside ONE transaction.
+ *
+ * AN EXPLICIT LIMIT, no default. Chapter 3.7's baseline found four suites broken
+ * by tests that asserted local facts about a global, oldest-first operation, and
+ * this is another global operation: a caller that wants only its own rows drained
+ * has to say how many, and a test that forgets ends up asserting about somebody
+ * else's fixture.
+ *
+ * SEND THEN MARK: a crash between the two resends one email, which is the
+ * accepted cost, and marking first would lose it silently — a notification
+ * nobody received is the failure FR-WHK-07 exists to prevent.
+ *
+ * PER-ROW ISOLATION, and this is where it departs from `drainOutbox` one screen
+ * up. That one lets a failing publish abort the batch, on the reasoning that the
+ * broker is either up or down and a partial batch means down. An email is not
+ * like that: one address a server refuses is one address, and letting it abort
+ * the batch would make it abort EVERY batch — claimed first because it is
+ * oldest, throwing, rolling back, and no notification behind it ever going out.
+ * Head-of-line blocking, permanent, from one bad recipient.
+ *
+ * So a row that throws is reported through `onError` and simply not marked. It
+ * stays claimable and the rows behind it still go.
+ *
+ * A NOTE ON WHY THE ORDERING IS OBSERVABLE AT ALL. It was not, at first: with
+ * the mark in a `finally` and the throw escaping the transaction callback, the
+ * transaction rolled back and undid the mark, so marking BEFORE the send and
+ * marking after it produced identical behaviour. The chapter's own sabotage
+ * mutation could not fail (research R44). Catching per row is what makes the two
+ * different, because now nothing rolls back.
+ */
+export async function drainDisableNotifications(
+  db: Db,
+  limit: number,
+  deliver: (row: DisableNotificationRow) => Promise<void>,
+  onError: (row: DisableNotificationRow, error: unknown) => void = () => {},
+): Promise<number> {
+  return db.transaction(async (tx) => {
+    const claimed = (await tx.execute(
+      sql`SELECT n.id                AS "id",
+                 n.organisation_id   AS "organisationId",
+                 a.name || ' / ' || e.kind AS "environmentName",
+                 w.url               AS "endpointUrl",
+                 n.disabled_at       AS "disabledAt",
+                 n.run_started_at    AS "runStartedAt",
+                 n.run_attempts      AS "runAttempts",
+                 n.last_status       AS "lastStatus",
+                 n.last_error        AS "lastError"
+            FROM webhook_disable_notifications n
+            JOIN environments e ON e.id = n.environment_id
+            JOIN applications a ON a.id = e.application_id
+            JOIN webhook_endpoints w ON w.id = n.endpoint_id
+           WHERE n.delivered_at IS NULL
+           ORDER BY n.disabled_at, n.id
+           LIMIT ${limit}
+             FOR UPDATE OF n SKIP LOCKED`,
+    )) as unknown as {
+      rows: (Omit<DisableNotificationRow, "disabledAt" | "runStartedAt"> & {
+        disabledAt: string | Date;
+        runStartedAt: string | Date;
+      })[];
+    };
+
+    const delivered: string[] = [];
+    for (const raw of claimed.rows) {
+      // Timestamps back as `Date`, not as whatever the driver felt like. Raw
+      // SQL through `execute` skips drizzle's column mapping, and a timestamptz
+      // arrives as a string — which reaches the mailer as an object with no
+      // `getTime`, one call later and one file away. Coerced here, at the
+      // boundary that produced it, rather than defended against downstream.
+      const row: DisableNotificationRow = {
+        ...raw,
+        disabledAt: new Date(raw.disabledAt),
+        runStartedAt: new Date(raw.runStartedAt),
+      };
+      try {
+        await deliver(row);
+        delivered.push(row.id);
+      } catch (error) {
+        // Not marked. The row stays claimable and the next pass tries it again,
+        // which is the whole reason this is a table rather than a call.
+        onError(row, error);
+      }
+    }
+
+    if (delivered.length > 0) {
+      // The builder rather than raw SQL, unlike the outbox drain one screen up.
+      // That one interpolates `ARRAY[…]::bigint[]` through `sql.raw` because its
+      // ids are integers; these are uuids, and `sql` renders a JS array as a
+      // comma-separated parameter list — which Postgres reads as a row
+      // expression and rejects with "record type has too many columns".
+      await tx
+        .update(webhookDisableNotifications)
+        .set({ deliveredAt: new Date() })
+        .where(inArray(webhookDisableNotifications.id, delivered));
+    }
+    return delivered.length;
+  });
+}
+
+/** The addresses to notify for an organisation, at SEND TIME (FR-WHK-07).
+ *
+ * Resolved from the row's `organisation_id`, which chapter 3.6 denormalised onto
+ * the notification precisely so this lookup could not follow the endpoint's
+ * CURRENT owner. An application that moved between organisations after the
+ * disablement must not silently retarget an obligation already owed to somebody
+ * else — 3.6 wrote the reason down and this is the first code to depend on it.
+ *
+ * `humans.email` is nullable, so this can legitimately return nothing. That is a
+ * branch the caller has to handle, not a case that cannot arise.
+ *
+ * EVERY member, not only the owners. `memberships.role` is one of owner, admin
+ * or member, and picking a subset here would be this chapter inventing a
+ * notification-preferences model — which is product, and belongs to whichever
+ * chapter builds preferences. Everyone who can see the endpoint hears that it
+ * stopped. */
+export async function organisationRecipients(
+  db: Db,
+  organisationId: string,
+): Promise<string[]> {
+  const result = (await db.execute(
+    sql`SELECT DISTINCT h.email AS "email"
+          FROM memberships m
+          JOIN humans h ON h.id = m.human_id
+         WHERE m.organisation_id = ${organisationId}
+           AND h.email IS NOT NULL
+         ORDER BY h.email`,
+  )) as unknown as { rows: { email: string }[] };
+  return result.rows.map((row) => row.email);
+}
+
 /** How far behind the relay is. The single number worth alarming on later, and
  * the one the chapter shows going up while the broker is down. */
 /** The name this consumer claims events under. One name, because the ledger is
services/api/src/main.ts
@@ -5,6 +5,7 @@ import { createLogger } from "@relay/service-kit";
 
 import { AppModule } from "./app.module";
 import { EventConsumerService } from "./consumer/consumer.module";
+import { NotificationRelayService } from "./notifications/notifications.module";
 import { OutboxRelayService } from "./outbox/outbox.module";
 import { DeliveryRelayService } from "./webhooks/webhooks.module";
 
@@ -20,6 +21,11 @@ async function bootstrap(): Promise<void> {
   // accumulating in Postgres instead of preventing the api from serving writes
   // (chapter 3.3, research R9).
   app.get(OutboxRelayService).start();
+  // Chapter 3.8: the disablement notifications chapter 3.6 wrote and nothing
+  // delivered. Its backlog drains on this first start as ordinary undelivered
+  // work — no migration and no special case, because `delivered_at IS NULL` was
+  // already true of every one of those rows.
+  app.get(NotificationRelayService).start();
   // And the second relay (chapter 3.5): the same loop over a different table,
   // publishing deliveries that have become due. Started here for 3.3's reason —
   // a retry schedule that only runs when someone remembers is not a schedule.

Container thứ năm

Mailpit, healthcheck của nó, và bản đăng ký nay đã có một cái gate chạy theo cả hai chiều.

compose.yaml
@@ -73,6 +73,33 @@ services:
       retries: 5
       start_period: 15s
 
+  mailpit:
+    image: axllent/mailpit:v1.28
+    # Chapter 3.8. An SMTP server that accepts everything and delivers nothing,
+    # with an HTTP API for reading what it caught.
+    #
+    # WHY A CONTAINER RATHER THAN A FAKE. FR-WHK-07 says an email must not contain a
+    # signing secret, and the only artefact that can settle that is the message a
+    # server RECEIVED — a stub records what the sender passed, which is the same
+    # object the assertion would be reading, so a mailer that dropped the secret
+    # into a header the stub does not model would pass. Constitution VII asks a
+    # fifth container to justify itself; this is the justification (research R9).
+    #
+    # No volume. Mailpit holds messages in memory, and a test inbox that survived
+    # a restart would be a test inbox that leaks state between runs — the same
+    # reasoning as Redis's, which the entry above records.
+    ports:
+      - "${RELAY_MAILPIT_HTTP_PORT:-8025}:8025"
+      - "${RELAY_MAILPIT_SMTP_PORT:-1025}:1025"
+    healthcheck:
+      # Without one, `docker compose up -d --wait` waits for RUNNING rather than
+      # for READY, and V9 can read the API before it is serving. `infra.test.ts`
+      # says the same thing about the other four.
+      test: ["CMD", "/mailpit", "readyz"]
+      interval: 5s
+      timeout: 3s
+      retries: 5
+
 
   # --- the services (chapter 3.5) -----------------------------------------
   # Behind `--profile services`, for the reason above.
@@ -85,6 +112,16 @@ services:
     environment:
       DATABASE_URL: postgres://relay:relay@postgres:5432/relay
       RELAY_NATS_URL: nats://nats:4222
+      # Chapter 3.8. Container names, not localhost — the api's own default is
+      # `redis://localhost:6379`, which inside this container is not the Redis
+      # service. And the tenant limiter FAILS OPEN by design (SAD §6.3), so a
+      # missing address would not crash anything: the composed stack would serve
+      # every request unlimited while reporting a limit. The constitution
+      # requires the full stack to start with one command, and this is what makes
+      # that true rather than merely quiet (research R24).
+      #
+      # RELAY_SMTP_URL joins in the transport phase, with the container it names.
+      RELAY_REDIS_URL: redis://redis:6379
       # Development values. Both are secrets in anything that is not a laptop,
       # and the api refuses to start in production without the first.
       RELAY_WEBHOOK_SECRET_KEY: ${RELAY_WEBHOOK_SECRET_KEY:-}
@@ -95,6 +132,7 @@ services:
     depends_on:
       postgres: { condition: service_healthy }
       nats: { condition: service_healthy }
+      redis: { condition: service_healthy }
     healthcheck:
       test: ["CMD", "wget", "-q", "-O", "-", "http://localhost:4000/healthz"]
       interval: 5s
packages/config/src/infra.ts
@@ -11,6 +11,10 @@ export const INFRA_SERVICES = [
   "redis",
   "nats",
   "clickhouse",
+  // Chapter 3.8. The fifth, and the only one that is not a store: Mailpit
+  // catches the SMTP the notification relay sends so a test can read what was
+  // RECEIVED rather than what was passed (FR-WHK-07).
+  "mailpit",
 ] as const;
 
 export const DURABLE_VOLUMES = [
packages/config/src/infra.test.ts
@@ -32,6 +32,27 @@ describe("the compose declaration agrees with @relay/config", () => {
     expect(healthchecks.length).toBeGreaterThanOrEqual(INFRA_SERVICES.length);
   });
 
+  it("REGISTERS every compose service, which is the direction this file lacked", () => {
+    // The assertion above runs one way: every registered service must appear in
+    // compose. Nothing ran the other way, so a container added to compose and
+    // never registered here was invisible — `INFRA_SERVICES` would quietly stop
+    // naming the local infrastructure while every test still passed. Chapter
+    // 3.8 added a fifth container and the gap is how it nearly went unnoticed.
+    //
+    // The services behind `--profile services` are Relay's own and are not
+    // infrastructure, so they are excluded by name rather than by pattern: a
+    // list is auditable and a pattern would silently absorb the next container.
+    const ours = new Set(["api", "gateway", "dispatcher"]);
+    // Only the `services:` block. Volume names sit at the same indentation one
+    // block down, and a match that swept the whole file would report
+    // `postgres-data` as an unregistered service.
+    const services = compose.slice(0, compose.indexOf("\nvolumes:"));
+    const declared = [...services.matchAll(/^ {2}([a-z][a-z0-9-]*):$/gm)]
+      .map((match) => match[1] as string)
+      .filter((service) => !ours.has(service));
+    expect([...declared].sort()).toEqual([...INFRA_SERVICES].sort());
+  });
+
   it("persists exactly the durable stores — and never Redis", () => {
     for (const volume of DURABLE_VOLUMES) {
       expect(compose).toContain(`${volume}:`);

Chương này không giao những gì

Bất cứ điều gì về hiện tượng bật tắt liên tục. Một endpoint bị tắt, bật lại, rồi tắt lần nữa sẽ sinh ra hai dòng và hai email, và không gì gộp chúng lại. Điều đó là có chủ đích — một sự cố thứ hai thật sự là một chuyện thứ hai cần được báo, và trường hợp biên trong spec nói rõ không thông báo nào được che lấp thông báo kia — nhưng một endpoint bật tắt mỗi giờ sẽ gửi thư mỗi giờ. Giải quyết nó nghĩa là một mô hình tùy chọn thông báo, thứ thuộc về sản phẩm chứ không phải hạ tầng.

Giới hạn số lần thử lại. Một dòng hỏng vĩnh viễn sẽ được thử lại vĩnh viễn, năm giây một lần. Nó không còn chặn ai nữa, đó là nửa phần cấp bách; một trạng thái dead-letter cho thông báo là nửa còn lại và không có ở đây.

Khả năng phát thư thực tế. Không SPF, không DKIM, không xử lý bounce, không nhà cung cấp. Mailpit nhận mọi thứ, điều đúng đắn với một hộp thư test và chẳng nói gì về việc một mail server thật có nhận hay không.

Bất kỳ thông báo nào khác. Đây là một email cho một sự kiện. Các ngưỡng quota của FR-RTL cần thêm ba cái nữa, và chúng đến cùng với quota.