alepha@docs:~/framework/docs/guides/core$
cat 7-pipelines.md | pretty
3 min read
Last commit:

#Pipelines

Every primitive with a handler in Alepha is a pipeline: a handler with a list of middleware wrapped around it. $action, $job and $page all extend the same base, which is why they all take a use array, and $pipeline is that base exposed on its own for the functions that are not any of those.

typescript 
1import { $pipeline } from "alepha";

That shared shape is what makes a middleware one topic instead of one per primitive: each works anywhere a use array does.

#$pipeline

Wrap a plain function and call it like a function:

typescript 
 1import { $pipeline } from "alepha"; 2import { $lock } from "alepha/lock"; 3  4class OrderService { 5  processOrder = $pipeline({ 6    use: [$lock({ name: "process-order" })], 7    handler: async (orderId: string) => { 8      return { orderId, status: "paid" }; 9    },10  });11 12  async run(id: string) {13    return this.processOrder(id);14  }15}

Reach for it when the work is not a route, a job or a page: an internal service method called from several places that should carry its middleware with it rather than have each caller remember one.

#Order in use is not cosmetic

The first middleware in the array is the outermost. use: [A, B] composes as A(B(handler)), so A sees the call first and the result last. A lock placed before a cache middleware serialises the misses; placed after it, only the handler is serialised and the cache answers concurrently. Both are reasonable. Only one is what you meant.

#Writing a middleware

A middleware is a function that takes the next handler and returns a wrapped one. createMiddleware builds it with the container already resolved, and attaches the name and options that tooling reads back from primitive.middlewares:

typescript 
 1import { $inject, $pipeline, createMiddleware, type Middleware } from "alepha"; 2import { DateTimeProvider } from "alepha/datetime"; 3  4const $timed = (label: string): Middleware => { 5  const time = $inject(DateTimeProvider); 6  return createMiddleware({ 7    name: "$timed", 8    options: { label }, 9    handler: ({ next }) => {10      return async (...args) => {11        const start = time.nowMillis();12        try {13          return await next(...args);14        } finally {15          console.log(`${label}: ${time.nowMillis() - start}ms`);16        }17      };18    },19  });20};21 22class ReportService {23  build = $pipeline({24    use: [$timed("report")],25    handler: async (month: string) => ({ month }),26  });27}

createMiddleware calls $context(), so it must run where a primitive would: inside a property initializer, which is where a use array is written. The handler itself runs later, the first time the pipeline is called, once the app has started and the container is locked. Resolve the services a middleware needs in the factory with $inject, as above, rather than with alepha.inject() inside handler. A meta record of free-form capability flags can be attached as well, for consumers that react to what a middleware does without knowing which one it is.

#$scope

Host primitives run their handler inside an AsyncLocalStorage scope, which is what makes alepha.context.get() and .set() work per request. A standalone $pipeline has no such scope, so add one when the handler needs it:

typescript 
 1import { $pipeline, $scope } from "alepha"; 2  3class OrderService { 4  processOrder = $pipeline({ 5    use: [$scope()], 6    handler: async (orderId: string) => { 7      return orderId; 8    }, 9  });10}

$scope() throws when a scope already exists, on purpose: nesting one would give you a second, empty context that silently loses everything the outer one held. That rules it out of an $action, $job or $page, and out of a pipeline called from inside a request.

#See also

  • Middlewares for the server-side middleware that ships with AlephaServer
  • Caching for $cache
  • Background Jobs for work that has to survive the process, with its own durable retries