Sample: Inbound-Mail Triage

Attach a business process to inbound e-mail: inbound-mail-foundation owns transport, dedup and PII — this sample’s mail-triage feature turns every ingested mail into a triage item via a worker-lane job.

What you learn here

• Consumer pattern: hook your business process onto the foundation’s ingest write-handler with r.job({ trigger: { on: InboundMailFoundationHandlers.ingestMessage } }) — the job receives the plaintext normalized mail (PII encryption happens inside the handler, after your trigger payload was captured).
• Idempotency discipline: the trigger also fires on idempotent replays (duplicate ingest returns success) — key your side-effects by providerMessageId so replays overwrite instead of duplicating.
• Provider swap: the test drives inbound-provider-inmemory; production mounts inbound-provider-imap (password/app-password, IMAP IDLE) against the same contract — the consumer code does not change.

Feature composition

config + tenant + compliance-profiles + tenant-lifecycle   (foundation requirements)
inbound-mail-foundation                                     (streams, projections, ingest)
inbound-provider-inmemory                                   (scriptable provider)
mail-triage  ← this sample                                  (the app-side consumer)

Flow

1. connect-account creates a mail account (shared mailbox, ownerUserId = null).
2. A mail is ingested — in production the watch-supervisor (IMAP IDLE push) or the reconciliation poll dispatches ingest-message; the inline projection materializes read_inbound_messages in the same transaction.
3. The handler-trigger fans the write out to the worker lane; the triage job records { from, subject, scope, threadHint }.
4. A replay of the same providerMessageId reports duplicate: true, creates no second row — and the keyed triage store stays at one item.

Tests

• E2E through createAllInOneEntrypoint (HTTP → dispatcher → BullMQ worker lane, no framework-internal shortcuts): connect → ingest → triage item appears.
• Replay scenario: duplicate ingest stays idempotent on both the projection and the consumer side.

---

Source code

The feature entry point — embedded straight from the source file, so the code here is exactly what runs. Multi-file samples keep their remaining files next to it on GitHub (link below):

// Inbound-Mail Triage Sample
//
// Shows the consumer pattern for `inbound-mail-foundation`: the
// foundation owns transport, dedup and PII handling — the app attaches
// its business process to the ingest write-handler via a worker-lane
// job. Every inbound mail becomes a triage item; a real app would
// create a ticket, run an AI classification, extract an invoice or
// draft a reply here.
//
// Feature composition (see the test):
//   inbound-mail-foundation   — streams, projections, ingest handler
//   inbound-provider-inmemory — scriptable provider (tests/demos)
//   mail-triage (this file)   — the app-side consumer
//
// The trigger fires AFTER every successful ingest-message write — that
// includes idempotent replays (duplicate ingest returns success). The
// triage store is therefore keyed by providerMessageId: replays
// overwrite instead of duplicating. Key your side-effects the same way.

import { InboundMailFoundationHandlers } from "@cosmicdrift/kumiko-bundled-features/inbound-mail-foundation";
import { defineFeature, type FeatureDefinition } from "@cosmicdrift/kumiko-framework/engine";

export type TriageItem = {
  readonly from: string;
  readonly subject: string;
  readonly scope: string;
  readonly threadHint: string | null;
};

// In-memory store so the integration-test can assert the job ran with
// the right payload. A real app would do domain writes / AI calls /
// notifications — the side-effect is opaque to the framework.
export const triageInbox = new Map<string, TriageItem>();

export function createMailTriageFeature(): FeatureDefinition {
  return defineFeature("mail-triage", (r) => {
    r.describe(
      "Sample consumer for inbound-mail-foundation: a worker-lane job triggered by the ingest-message write-handler turns every inbound mail into a triage item.",
    );
    r.requires("inbound-mail-foundation");

    // The job receives the dispatcher-validated ingest payload — i.e.
    // the PLAINTEXT normalized mail (PII encryption happens inside the
    // handler right before the event append, so consumers on the
    // handler-trigger path never deal with ciphertext).
    r.job(
      "triage-inbound",
      {
        trigger: { on: InboundMailFoundationHandlers.ingestMessage },
        runIn: "worker",
      },
      async (payload) => {
        const providerMessageId = payload["providerMessageId"] as string;
        triageInbox.set(providerMessageId, {
          from: payload["from"] as string,
          subject: payload["subject"] as string,
          scope: payload["scope"] as string,
          threadHint: (payload["messageIdHeader"] as string | null) ?? null,
        });
      },
    );
  });
}

📄 On GitHub: samples/recipes/inbound-mail-triage/src/feature.ts