encore-pubsub

Encore Pub/Sub

手順

Pub/Sub はサービス間の非同期メッセージングに使用します。プロデューサーは Topic にイベントを発行し、コンシューマーは Subscription をアタッチして反応します。リソースはパッケージレベルで宣言する必要があり、関数内で宣言することはできません。

トピック

import { Topic } from "encore.dev/pubsub";

interface OrderCreatedEvent {
  orderId: string;
  userId: string;
  total: number;
}

// Package level declaration
export const orderCreated = new Topic<OrderCreatedEvent>("order-created", {
  deliveryGuarantee: "at-least-once",
});

発行

await orderCreated.publish({
  orderId: "123",
  userId: "user-456",
  total: 99.99,
});

サブスクリプション

import { Subscription } from "encore.dev/pubsub";

const _ = new Subscription(orderCreated, "send-confirmation-email", {
  handler: async (event) => {
    await sendEmail(event.userId, event.orderId);
  },
});

メッセージ属性

メッセージ属性として扱うべきフィールド（フィルタリング/順序付け用）には Attribute<T> を使用します:

import { Topic, Attribute } from "encore.dev/pubsub";

interface CartEvent {
  cartId: Attribute<string>;  // Used for ordering
  userId: string;
  action: "add" | "remove";
  productId: string;
}

// Ordered topic: events with same cartId delivered in order
export const cartEvents = new Topic<CartEvent>("cart-events", {
  deliveryGuarantee: "at-least-once",
  orderingAttribute: "cartId",
});

トピック参照

静的解析を維持したまま、トピックへのアクセスを他のコードに渡します:

import { Publisher } from "encore.dev/pubsub";

const publisherRef = orderCreated.ref<Publisher>();

async function notifyOrder(ref: typeof publisherRef, orderId: string) {
  await ref.publish({ orderId, userId: "123", total: 99.99 });
}

配信保証

• at-least-once（デフォルト）: 重複配信の可能性があります → ハンドラーは冪等でなければなりません。
• exactly-once: より厳密ですが、スループットに上限があります（AWS 300 msg/s/topic、GCP 3000+ msg/s/region）。発行側では重複排除されません。

ガイドライン

• トピックとサブスクリプションはパッケージレベルで宣言する必要があります。
• サブスクリプションハンドラーは冪等でなければなりません（デフォルトでは at-least-once 配信です）。
• フィルタリング/順序付けを目的としたフィールドには Attribute<T> を使用し、任意のメタデータには使用しないでください。
• publish の呼び出し元で重い同期処理を行わないでください — publish はメッセージがキューに入った時点で戻ります。