React to another feature's events

Feature A sometimes needs to react after feature B changes data. Examples include sending a notification after an order is placed or updating a read model after a payment succeeds.

The supported pattern is:

1. The writing feature declares a typed domain event.
2. Its handler appends that event in the same transaction as the business write.
3. A multi-stream projection consumes the event after commit.

The consumer can update a projection table or perform an external side effect. Delivery is asynchronous and at least once, so the consumer must tolerate a retry.

Before you start

• You have a feature that owns the entity being changed.
• You have a consumer feature or a projection that should react to the event.
• You have read Events and projections.
• You know whether the consumer writes a table or only calls an external service.

1. Declare the event in the writing feature

Declare the event inside the feature that owns the aggregate. The event name is short; Kumiko qualifies it with the feature name.

const orderPlaced = r.defineEvent(
  "order-placed",
  z.object({
    id: z.uuid(),
    customer: z.string(),
    product: z.string(),
  }),
  { piiFields: "none" },
);

piiFields is required. Use "none" only when the payload contains no personal data. If it does contain personal data, map each field to its owning subject according to the crypto-shredding model.

What should happen

The feature contains a typed event definition. Run the app’s code generation step after adding the event so the local event type map includes its qualified name.

2. Append the event in the same write transaction

The handler writes the aggregate first, then appends the domain event. Both use the same transaction through ctx.db.

const { executor: orderExecutor } = createEntityExecutor("order", orderEntity);

r.writeHandler(
  "order:place",
  z.object({ customer: z.string().min(1), product: z.string().min(1) }),
  async (event, ctx) => {
    const result = await orderExecutor.create(event.payload, event.user, ctx.db);
    if (!result.isSuccess) return result;

    await ctx.events.append({
      aggregateId: String(result.data.id),
      aggregateType: "order",
      type: orderPlaced.name,
      payload: {
        id: String(result.data.id),
        customer: event.payload.customer,
        product: event.payload.product,
      },
    });

    return result;
  },
  { access: { roles: ["Admin", "Customer"] } },
);

Use the event definition’s .name. Do not hand-type a qualified event name. The append includes the aggregate ID, aggregate type, event type, and payload.

What should happen

A failed entity write returns a failed WriteResult and no domain event is appended. If the append fails, the surrounding transaction rolls back the entity write as well.

3. Consume the event after commit

Register a multi-stream projection in the consumer feature. A table-less projection is suitable for a notification, webhook, or external-system sync.

r.multiStreamProjection({
  name: "record-order-placed",
  apply: {
    [orderPlaced.name]: async (event) => {
      const payload = event.payload as {
        id: string;
        customer: string;
        product: string;
      };

      await sendOrderPlacedNotification({
        orderId: payload.id,
        customer: payload.customer,
      });
    },
  },
});

If the consumer writes a read model, add its projection table and use the transaction passed to apply. Keep the update idempotent, for example with an upsert keyed by the event’s aggregate ID or a business deduplication key.

What should happen

The event dispatcher sees the event after the original transaction commits and advances the projection’s cursor after successful processing. If apply throws, the event is retried according to the consumer’s error policy.

Why not import the other feature’s table

A direct table import bypasses the feature boundary. It can skip handler access checks, field filtering, tenant scoping, event creation, and optimistic locking. Use ctx.dispatch.query or ctx.dispatch.write when you need another feature’s handler. Use a domain event when you need an asynchronous reaction.

A consumer that only needs the event does not need to know the writer’s table layout at all.

Common mistakes

• Appending on a synthetic stream. Use the aggregate’s real ID and type.
• Hand-typing the event name. Pass orderPlaced.name.
• Putting external I/O in the original transaction. Send mail, webhooks, and external API calls from the post-commit consumer.
• Assuming exactly-once delivery. The cursor can retry an event. Make apply idempotent.
• Leaving personal data unclassified. Set the event’s piiFields stance before appending.
• Forgetting the dependency. If the consumer also uses another feature’s handler, declare r.requires("feature-name") and use its qualified handler name.

Live example

The tested cross-feature recipe appends order-placed on the order stream and consumes it with a table-less multi-stream projection:

export const pubsubOrderFeature = defineFeature("pubsubOrders", (r) => {
  r.entity({ name: "pubsub-order", ...orderEntity });

  // Define the event shape once. The returned `.name` is the qualified
  // name ("pubsub-orders:event:order-placed") — pass it to ctx.events.append.
  const orderPlaced = r.defineEvent(
    "order-placed",
    z.object({ id: z.string(), customer: z.string(), product: z.string() }),
    { piiFields: "none" },
  );

  const { executor: orderExecutor } = createEntityExecutor("pubsub-order", orderEntity);

  r.writeHandler({
    name: "order:place",
    schema: z.object({ customer: z.string().min(1), product: z.string().min(1) }),
    handler: async (event, ctx) => {
      const result = await orderExecutor.create(event.payload, event.user, ctx.db);
      if (result.isSuccess) {
        // emitEvent runs inside the executor's TX and is typed against
        // the orderPlaced event-def — a wrong payload shape is a compile
        // error, not a runtime Zod reject. The event lands on the order's
        // OWN stream (aggregateType "pubsub-order"), with the correct
        // version bumped automatically. Schema validation still runs at
        // append-time; a mismatch rolls back the whole write.
        await emitEvent(ctx, orderPlaced, {
          aggregateId: String(result.data.id),
          aggregateType: "pubsub-order",
          payload: {
            id: String(result.data.id),
            customer: event.payload.customer,
            product: event.payload.product,
          },
        });
      }
      return result;
    },
    access: { roles: ["Admin", "Customer"] },
  });

  // A job can react to the same event. `catchUp: true` gives the job its own
  // consumer cursor that starts at the beginning of the event store, so a job
  // added to a running app also processes the orders placed before it existed.
  // Without catchUp the job only sees events appended after its deploy.
  r.job(
    "invoice-order",
    { trigger: { on: orderPlaced.name, catchUp: true }, runIn: "worker" },
    async (payload) => {
      const order = orderPlaced.schema.parse(payload);
      invoicedOrderIds.push(order.id);
    },
  );

  // Async consumer — Marten-style. Fires after commit, via the event-

Full source: samples/recipes/cross-feature-events.

See also

• Events and projections, transaction and retry semantics.
• Commands and queries, the dispatcher contract.
• Features and composition, feature boundaries and dependencies.