PipeSafe

Type-safe MongoDB
aggregations.

Catch errors at compile time, not runtime.
Tim Vyas · Lead Engineer, GetGround
Ryan Peggs · Technical Director, Suvera
MongoDB.Local London
7 May 2026
A few questions
> Have you written a MongoDB aggregation pipeline?
> Have you written it in TypeScript?
> Have you ever shipped one that broke in production?
> Have you ever said "TypeScript should have caught that"?
A pipeline you've probably written

How many bugs do you see?

schema.ts
interface Attendee { _id: ObjectId; name: string; email: string; conferenceId: ObjectId; talks: ObjectId[]; ticketTier: "standard" | "vip"; } interface Talk { _id: ObjectId; title: string; speakerId: ObjectId; durationMin: number; schedule: { track: "main" | "workshop" | "lightning"; room: string; startsAt: Date; }; }
vip_report.ts
const report = await attendees.aggregate([ { $match: { tier: "vip" } }, { $lookup: { from: "talk", localField: "talks", foreignField: "_id", as: "talkDocs" } }, { $project: { name: 1, talkCount: { $size: "$talkDocs" }, mainStageTalks: { $size: { $filter: { input: "$talkDocs", cond: { $eq: ["$$this.schedule.tracks", "main"] } } } } } }, { $group: { _id: "$conferenceId", total: { $sum: "$talkCount" } } }, { $sort: { total: -1 } } ]).toArray();
●  What actually breaks in production

Four bugs. None caught.

1
$match: { tier: "vip" }  — field is ticketTier.
Returns zero results. Quietly.
2
$group _id: "$conferenceId"  — but $project dropped it.
Groups by null. Single bucket. Looks plausible.
3
"$this.schedule.tracks"  — path is $this.schedule.track.
Filter never matches. mainStageTalks is 0 across the entire report.
4
$lookup from: "talk"  — collection is "talks".
Magic string. Empty join, every count is 0. No error.
What TypeScript catches today

TS sees Document[]. // That's it.

vipReport.ts — IDE
const report = await attendees .aggregate([ /* ... */ ]) .toArray(); // ^? Document[] report[0].total // any report[0].whatever // any report[0].nope // any
Your $match, your $project, your $group — all opaque to the type system.
The driver hands back Document[]. You hand-cast. You hope.
v1.0.0 — released today

PipeSafe

Type-safe MongoDB aggregations.
Inferred end-to-end.
Compile-time errors
Field-level autocomplete
Driver-agnostic
Open source · Apache 2.0 · on npm today
Live

Same pipeline.
With PipeSafe.

Switching to the editor →
How it works

Every stage is a typed transformation.

Input
Attendee
→
$match
Attendee
→
$lookup
Attendee & { talkDocs }
→
$project
{ name, talkCount }
→
Output
{ _id, total }[]
The pipeline's output type is just the composition of every stage. No casting. No any. No surprises.
The hard part we solved

If you've tried this in your own codebase
and given up — yes, that's why.

Variadic generics
Pipelines have arbitrary length. Each stage's output is the next stage's input.
$project reshape inference
Computed fields, dotted paths, exclusions, $$ROOT. All typed.
$lookup joins
Cross-collection types resolved at compile time, including pipeline-form lookups.
$group accumulators
$sum, $avg, $push, $addToSet — each accumulator's return type, inferred.
Bring your own driver / ORM

Same pipeline. Three call sites.

mongodb driver
import { pipeline, InferOutputType } from "@pipesafe/core"; const p = pipeline<Attendee>() .match({ ticketTier: "vip" }) .count("vips"); type R = InferPipelineOutput< typeOf P >; const r: R = await attendees .aggregate(p.getPipeline()) .toArray();
mongoose
import { pipeline, InferOutputType } from "@pipesafe/core"; const p = pipeline<Attendee>() .match({ ticketTier: "vip" }) .count("vips"); type R = InferOutputType< typeOf P >; const r: R = await Attendee .aggregate(p.getPipeline());
prisma
import { pipeline, InferOutputType } from "@pipesafe/core"; const p = pipeline<Attendee>() .match({ ticketTier: "vip" }) .count("vips"); type R = InferOutputType< typeOf P >; const r: R = await prisma.attendee .aggregateRaw({ pipeline: p.getPipeline() });
PipeSafe builds the pipeline. You run it how ever you already do.
Adoption surface

Drop-in. One pipeline at a time.

Before
existing.ts
await db.collection("attendees") .aggregate([ /* ... */ ]) .toArray();
After
migrated.ts
await db.collection("attendees") .aggregate( pipeline<Attendee>() .match({ ticketTier: "vip" }) .getPipeline() ) .toArray();
→
Not an ORM. Use the driver you already have.
→
No runtime cost. Erased after compile.
→
No external dependencies.
→
1.6 kB gzipped.
AI-generated pipelines

LLMs are great at proposing aggregations.
Terrible for hallucinating fields and syntax.

llm-generated.ts
// "Top 5 conferences by VIP attendee count" const p = pipeline<Attendee>() .match({ tier: "vip" }) // ^ Property 'tier' does not exist // on type Attendee. // Did you mean 'ticketTier'? .group({ _id: "$conferenceId", vips: { $sum: 1 } }) .sort({ vips: -1 }) .limit(5);
Red squiggles in the editor for every LLM suggestion.
Built into the LSP — the agent sees the diagnostic the moment it writes the line.
Iterate to a correct query without ever touching production data.
What's next

@pipesafe/manifold

DAG orchestration for typed pipelines.
Models. Projects. Dependency graphs.
dbt for MongoDB.
Coming June 2026 · ELv2
SOURCES STAGING MARTS tickets attendees talks stg_revenue stg_talk_audience revenue vip_recommendations talk_audience
Try it now
$npm install @pipesafe/core
github.com/pipesafe/pipesafe
pipesafe.co.uk
Apache 2.0 · v1.0.0 · Issues and PRs welcome

Questions?

Tim Vyas
@timvyas
Ryan Peggs
@ryanpeggs
$ npm install @pipesafe/core