Aggregation

Roll many rows into numbers — counts, sums, percentiles, breakdowns — in a single call. Everything is pushed down to Postgres; you get plain objects back, never SQL.

import { count, percentile, timeBucket } from "@mauroandre/weave-sdk";

const series = await weave.appRequest.aggregate({
  where: { host: "api", ts: { gte: since } },
  groupBy: { ts: timeBucket("ts", "5min") },       // one row per 5-minute bucket
  select: { requests: count(), p95: percentile("durationMs", 0.95) },
  orderBy: { ts: "asc" },
});

Accumulators

Import the helpers from the SDK and put them in select:

count()                       // number of rows
sum(field) · avg(field)       // over a numeric field
min(field) · max(field)
distinct(field)               // count of distinct values
first(field)                  // one representative per group (earliest by created_at)
percentile(field, p)          // exact percentile — p strictly between 0 and 1 (p95 → 0.95)
histogram(field, [bounds])    // bucket counts (see below)

Every accumulator takes an optional { where } — a filtered aggregate, computed in the same pass:

select: {
  total:  count(),
  errors: count({ where: { status: { gte: 500 } } }),
}

This inner where is a reduced grammar: scalar operators, and/or/not, and dot-paths — no some/every/none. It's also the one filter slot that isn't typed against the entity, so a typo'd field or an unknown operator surfaces at runtime (loudly) rather than at compile time.

histogram turns N boundaries into N+1 buckets — one per interval plus an overflow bucket above the last boundary — and returns the counts as one array value:

select: { bars: histogram("durationMs", [100, 200, 500]) }
// → [ <100, 100–200, 200–500, ≥500 ]

groupBy — fields or a time bucket

groupBy: ["host", "route"]                    // by raw fields (aliases match the names)
groupBy: { bucket: timeBucket("ts", "1h") }   // by a time bucket — "30s" · "5min" · "1h" · "1d"

timeBucket aligns on epoch/UTC, so "1d" buckets don't drift with the session timezone. Omit groupBy entirely to aggregate the whole set into a single row.

Group by a reference too — it buckets by the target's foreign key. groupBy: ["department", "company"] groups by department_id + company_id; you can name the reference (department) or its id (departmentId). latestPer takes the reference name.

having and the aggregate's orderBy are different: they address your select aliases, not fields. having: { revenue: { gte: 100 } } refers to the revenue you selected — naming a field that isn't an alias is an error.

Grouping through relationships — paths

Any field slot in an aggregate — groupBy, an accumulator's field, a FILTER — takes a dot-path through what you own and what you reference. Weave joins the tables for you; you never write the join.

await weave.order.aggregate({
  groupBy: ["customer.region"],          // reference → a scalar on the target
  select: { n: count(), revenue: sum("total") },
});

A path steps through a reference (N:1) or an owned object, to any depth — avg("fulfilment.cost") (an owned object), ["customer.company.tier"] (two hops). A reference at the leaf buckets by its foreign key. Paths that share a prefix share one join. In the array form the output alias is the path string itself, so read it back as row["customer.region"].

Unnesting an owned list — `unnest`

Everything above rolls up parent rows. To roll up the elements of an owned list instead — an average or a distribution per element, not per parent — name the list in unnest. The aggregate then runs one row per element (Postgres' answer to Mongo's $unwind), and groupBy / an accumulator's field / a FILTER address the element's fields:

const perSku = await weave.order.aggregate({
  where:   { status: "paid" },     // filters the PARENT orders
  unnest:  "items",                // one row per line item
  groupBy: ["items.sku"],
  select: {
    lines:      count(),
    qty:        sum("items.qty"),
    backorders: count({ where: { "items.status": { eq: "backordered" } } }), // a band
    label:      first("items.name"),
  },
  orderBy: { "items.sku": "asc" },
});

The two filters play different roles: where filters the parents (which rows' elements count at all), while an accumulator's { where } filters the elements (the band — count(… FILTER …)). first gives one representative per group — the earliest element by created_at — for metadata that's constant within the group (the label tied to a sku).

With unnest the unit of aggregation is the element, so a parent contributes as many rows as it has elements. Counting parents under unnest therefore needs distinct("id") (the parent id repeats per element). One unnest per call — for several lists, run one aggregate each.

having, orderBy & top-N

having filters groups by their aggregates; orderBy sorts by an output alias; perPage / page take the top-N:

{
  groupBy: ["route"],
  select: { n: count() },
  having: { n: { gte: 100 } },   // only busy routes
  orderBy: { n: "desc" },
  perPage: 10,                    // top 10
}

Expressions over aggregates

div · mul · add · sub combine aliases arithmetically, and the result is usable in orderBy / having — so you filter and sort by a derived rate server-side, before pagination:

import { count, div } from "@mauroandre/weave-sdk";

await weave.appRequest.aggregate({
  groupBy: ["route"],
  select: {
    errors:    count({ where: { status: { gte: 500 } } }),
    total:     count(),
    errorRate: div("errors", "total"),        // references the aliases by name
  },
  orderBy: { errorRate: "desc" },             // the routes that fail most, proportionally
});

div guards against divide-by-zero (0 denominators come back as null). An operand can be an alias, a number, or an inline accumulator (div(count(...), count())).

The dashboard case: headline numbers plus several independent breakdowns of the same filtered set, in one request.

const { rows, facets } = await weave.order.aggregate({
  where: { status: "paid" },
  select: { revenue: sum("total") },
  facets: {
    byCategory: { groupBy: ["category"],  select: { r: sum("total") }, orderBy: { r: "desc" }, limit: 10 },
    byState:    { groupBy: ["shipState"], select: { n: count() },      orderBy: { n: "desc" } },
  },
});

rows[0].revenue as number;    // the headline
facets.byCategory;            // [{ category, r }, ...]

The values are numbers, the type isn't yet. A numeric accumulator (count, sum, avg, percentile, an expression, min/max over a numeric column) comes back as a real number — same as a findMany on the same column, no Number(...) needed. But an aggregate row is still typed Record<string, unknown>, so annotate at the boundary (as number) until the select's types are inferred. Group keys are raw JSON — a timeBucket key is an ISO string, not a Date.

Each facet is its own aggregate under the parent where. The return type follows your input: with facets, you get { rows, facets }; without, a plain array.

Rolling up over time — `accumulate`

Everything above reads raw rows and aggregates on read — exact, but it keeps every row. When you can't keep every row forever (telemetry, metrics, counters), roll them up as they arrive with accumulate. One call folds a data point into a running rollup keyed by a composite unique — atomically, in Postgres:

import { inc, max, min, setOnInsert } from "@mauroandre/weave-sdk";

await weave.metricRollup.accumulate(
  { workerId, name, ts: bucket },                      // the key — a declared unique
  {
    sampleCount: inc(1),                               // running counter
    cpuSum:      inc(cpu),      cpuMax: max(cpu),       // sum + peak
    memSum:      inc(mem),      memMin: min(mem),       // sum + valley
    firstSeen:   setOnInsert(bucket),                  // written once, kept forever
  },
); // → the resulting row

The golden rule: store what merges, derive the rest on read. Keep sum and count; compute the average when you read (cpuSum / sampleCount). Never store an average — two averages can't be merged, a sum and a count always can.

The ops

inc(n)          // col = col + n          — counters, sums (monotonic)
max(v) · min(v) // col = greatest/least   — peaks and valleys
setOnInsert(v)  // written on insert only — preserved on every later merge

inc/max/min merge in the database (+, greatest, least); setOnInsert writes on the first insert and is left untouched afterwards. There is no read-modify-write and no race — the whole thing is a single upsert.

The key must be a declared unique

accumulate upserts on the key, so the key has to be a unique the entity declares — a composite group (the rollup key) or a single .unique() column:

export default defineEntity(
  "metricRollup",
  {
    workerId: text().notNull(),
    name:     text().notNull(),
    ts:       timestamptz().notNull(),
    sampleCount: int4().notNull().default(0),
    cpuSum:      float8().notNull().default(0),
    cpuMax:      float8().notNull().default(0),
  },
  { unique: [["workerId", "name", "ts"]] },   // ← the ON-CONFLICT key
);

A key that doesn't match a declared unique is a clear error — nothing is written. Read the rollups back with the ordinary aggregators (sum, avg over the buckets), or plain findMany / latestPer for the raw rollup rows.

Batch ingest — createMany

Insert many rows in one transaction — the shape a batched producer wants:

await weave.appRequest.createMany([
  { host: "api", route: "/x", durationMs: 12, status: 200 },
  { host: "api", route: "/y", durationMs: 40, status: 500 },
]); // → the created rows, in input order

Retention — partition by time

A high-volume event table (requests, logs, audit trails) can't grow forever. Declare partitionBy + retention and Weave keeps a rolling window for you — natively, with zero maintenance on your side:

export default defineEntity(
  "appRequest",
  { host: text().notNull(), route: text().notNull(), ts: timestamptz().notNull(), status: int4().notNull() },
  { partitionBy: timeBucket("ts", "1d"), retention: "30d" },   // daily partitions, keep 30 days
);

Under the hood the table is RANGE-partitioned by ts. On each write Weave lazily creates the partition the incoming row falls into (so a late/backfilled batch still lands correctly), and once a new day opens it drops whole partitions past the retention window — a DROP TABLE, not a row-by-row DELETE that would bloat under load. You never run a cron or a cleanup job; it's internal to Weave. Reads (findMany, aggregate) span all partitions transparently — Postgres prunes by the ts predicate.

Things to know:

Append-only. The partition key rides in the primary key ((id, ts)), so a partitioned entity is insert-only — ingest with createMany, never updateOne. That's exactly what a raw event tier wants.
Past the window is skipped. A row whose ts is already older than retention is dropped on ingest (its partition is gone) — createMany skips it and logs the count; a single create of such a row is a clear error.
Partitions materialize on first write. A brand-new partitioned table has no partitions until the first row lands — the day's partition is created then, lazily. An empty table showing zero partitions is expected, not a problem. Insert through Weave (not a raw SQL INSERT), so it can create the partition the row needs.
Buckets align to UTC. A "1d" bucket is a UTC day, not a local-timezone day — the same boundary timeBucket("ts", "1d") uses in aggregate, so the raw partitioned tier and your rollups agree on what "a day" is. In practice a late-local-evening event can land in the next UTC day's partition; that's intended.

The partition field must be a timestamptz().notNull(). This is a general time-series capability — logs, metrics, events — not tied to any one domain.

Latest per group — `latestPer`

Greatest-n-per-group: one row per key, the latest wins. It feeds live-metrics widgets (“the most recent doc per worker”). You declare the group; Weave arranges the ordering:

await weave.worker.findMany(
  {},
  { latestPer: ["workerId"], orderBy: { ts: "desc" } }, // newest row per worker
);

Aggregation respects scopes just like reads — the scope's row filter is AND-ed into the aggregate (and every facet) before anything is grouped.

← PreviousThe CLI

Next →Testing

Aggregation

Aggregation

Accumulators

groupBy — fields or a time bucket

Grouping through relationships — paths

Unnesting an owned list — `unnest`

having, orderBy & top-N

Expressions over aggregates

Facets — many breakdowns in one pass

Rolling up over time — `accumulate`

The ops

The key must be a declared unique

Batch ingest — createMany

Retention — partition by time

Latest per group — `latestPer`