Middleware Pipelines

Middleware pipelines let you compose named, ordered stages that run on state changes. Unlike flat middleware, pipelines support short-circuiting, async stages, and before/after phasing.

Define a Pipeline

mesh.pipeline("auth", [
  {
    name: "validate",
    async handler(ctx, next) {
      if (!ctx.state.auth.token) throw new Error("No auth token");
      await next();
    }
  },
  {
    name: "log",
    handler(ctx, next) {
      console.log(`[auth] ${ctx.event.type} - ${ctx.event.name}`);
      return next();
    }
  }
], {
  filter: { type: "resource.*" },  // Only run for resource events
  phase: "before"                   // Run before existing flat middleware
});

How It Works

Each stage receives:

• ctx — a PipelineContext with event, mesh, state, stageIndex, and stageName
• next — a function to continue to the next stage

Call next() to continue. Return without calling next() to short-circuit the pipeline.

mesh.pipeline("rate-limit", [
  {
    name: "check-limit",
    handler(ctx, next) {
      if (requestCount > 100) {
        console.warn("Rate limit exceeded — short-circuiting");
        return; // Don't call next() — pipeline stops here
      }
      return next();
    }
  },
  {
    name: "proceed",
    handler(ctx, next) {
      console.log("Request allowed");
      return next();
    }
  }
]);

Async Stages

Each stage can be async. The pipeline awaits each stage before continuing:

mesh.pipeline("audit", [
  {
    name: "log-to-server",
    async handler(ctx, next) {
      await fetch("/api/audit", {
        method: "POST",
        body: JSON.stringify({ event: ctx.event, timestamp: Date.now() })
      });
      return next();
    }
  }
]);

Filters

Use filter to run the pipeline only for specific events:

// Filter by event type
mesh.pipeline("resource-logger", stages, {
  filter: { type: "resource.*" }
});

// Filter by event name
mesh.pipeline("cart-handler", stages, {
  filter: { name: "cart.*" }
});

// Filter with RegExp
mesh.pipeline("api-logger", stages, {
  filter: { type: /resource\.|mutation\./ }
});

// Combine type and name
mesh.pipeline("specific", stages, {
  filter: { type: "action.completed", name: "cart.*" }
});

Phases

Control when the pipeline runs relative to existing flat middleware:

// Run before flat middleware (default)
mesh.pipeline("auth", stages, { phase: "before" });

// Run after flat middleware
mesh.pipeline("logger", stages, { phase: "after" });

Remove a Pipeline

mesh.removePipeline("auth");

Duplicate Guard

Registering a pipeline with the same name throws DuplicateRegistrationError:

mesh.pipeline("auth", stages1);
mesh.pipeline("auth", stages2); // Throws DuplicateRegistrationError

Error Isolation

Pipeline errors are caught and logged. They never break state mutations:

mesh.pipeline("fragile", [
  {
    name: "might-fail",
    handler(ctx, next) {
      throw new Error("Something went wrong");
      // This error is caught and logged — state mutation continues normally
    }
  }
]);

Example: Auth + Logging Pipeline

mesh.pipeline("auth-guard", [
  {
    name: "check-token",
    handler(ctx, next) {
      const token = ctx.state.auth?.token;
      if (!token) {
        console.warn("No auth token — blocking resource fetch");
        return; // Short-circuit
      }
      return next();
    }
  },
  {
    name: "add-auth-header",
    handler(ctx, next) {
      // Modify the event or mesh state as needed
      console.log("Auth token present — proceeding");
      return next();
    }
  }
], {
  filter: { type: "resource.fetch" },
  phase: "before"
});

Important Notes

TIP

Pipelines are ideal for cross-cutting concerns: authentication, rate limiting, audit logging, and request validation.

WARNING

Pipeline stages run in registration order. The first stage runs first. If a stage doesn't call next(), subsequent stages are skipped.

TIP

Pipelines and flat middleware coexist. Use phase to control ordering.

Next Steps

• Middleware — Flat middleware and event listeners
• Guards — Block operations by name pattern
• Plugins — Extend the mesh with custom functionality