← Use cases

Real-time per-entity aggregations, queryable as plain SQL.

Maintain rolling per-customer, per-tenant, or per-device counters (count, sum, max, avg) continuously, with no Materialize, ksqlDB, or ClickHouse cluster to operate. The aggregate is a row in queen_streams.state — your dashboard, BI tool, billing service, or alerting rule queries it as plain SQL on the PostgreSQL you already run.

Numbers are from a verified run with 30 customers × ~100 events each over 12 seconds, tumbling windows of 5 seconds. Reproduce with examples/use-case-aggregations.js.

1. The problem

You need "per-customer activity in the last N seconds / minutes / hours" continuously, accessible from:

• A real-time dashboard (Grafana panel, custom React UI).
• A billing service that meters usage per window.
• An alerting rule that fires if any customer crosses a threshold.
• An ML feature pipeline that needs rolling stats as inputs.

The standard answers in this space all add their own infrastructure:

2. The pattern

A producer pushes events partitioned by entity (customerId, tenantId, deviceId). A streaming query keys by partition (default), tumbles into windows of N seconds, and runs aggregate() over each window. The per-window accumulator is persisted as a row in queen_streams.state at the end of every cycle:

Each window's accumulator is a JSONB document like { count: 47, sum: 312.5, max: 99.0, windowStart, windowEnd }. Workers holding the partition lease are the only writers, so there's no cross-worker contention on a customer's row. State is durable, replicated by your PG backup, and queryable from anywhere that speaks SQL.

3. Quickstart

// npm install queen-mq        (Node 22+)
import { Queen, Stream } from 'queen-mq'

const url = 'http://localhost:6632'
const q   = new Queen({ url, handleSignals: false })

const Q_EVENTS = 'events'

await q.queue(Q_EVENTS).config({
  leaseTime:                 30,
  retryLimit:                3,
  retentionEnabled:          true,
  retentionSeconds:          3600
}).create()

// Producer: 3000 events spread across 30 customers.
for (let burst = 0; burst < 30; burst++) {
  const batch = []
  for (let i = 0; i < 25; i++) {
    const customerId = `cust-${Math.floor(Math.random() * 30)}`
    batch.push({
      partition: customerId,
      item:      { data: { customerId, amount: Math.random() * 100, t: Date.now() } }
    })
  }
  // Group by partition so each push hits one partition cleanly.
  const byPart = new Map()
  for (const { partition, item } of batch) {
    const arr = byPart.get(partition) || []
    arr.push(item)
    byPart.set(partition, arr)
  }
  await Promise.all([...byPart].map(([p, items]) =>
    q.queue(Q_EVENTS).partition(p).push(items)
  ))
  await new Promise(r => setTimeout(r, 50))
}

// 5-second tumbling windows, count + sum + max per customer.
const stream = await Stream
  .from(q.queue(Q_EVENTS))
  // Default key = partition_id = customerId. No explicit keyBy needed.
  .windowTumbling({ seconds: 5, idleFlushMs: 1500 })
  .aggregate({
    count: () => 1,
    sum:   m => m.amount,
    max:   m => m.amount
  })
  .to(q.queue('events.totals_per_customer_per_5s'))   // optional sink queue
  .run({
    queryId:       'customer-rolling-stats',
    url,
    batchSize:     200,
    maxPartitions: 8,
    reset:         true
  })

// Verified run, 3000 events across 30 customers, 12s producer:
//   Σ window.count across all emits  = 3000   (matches input exactly)
//   Σ window.sum   across all emits  = 150803.70  (matches input to the penny)
//   query registered in queen_streams.queries with config_hash 376ce9fa...

4. Querying the live aggregate via SQL

While a window is still open for a customer, the partial accumulator is sitting in queen_streams.state. You can query it directly:

SELECT
    s.key                                AS customer_or_partition,
    (s.value->'acc'->>'count')::int      AS events_in_window,
    (s.value->'acc'->>'sum')::numeric    AS sum_in_window,
    (s.value->'acc'->>'max')::numeric    AS max_in_window,
    to_timestamp(((s.value->>'windowStart')::bigint / 1000.0)) AS window_start,
    to_timestamp(((s.value->>'windowEnd')::bigint / 1000.0))   AS window_end
FROM queen_streams.state s
JOIN queen_streams.queries q ON q.id = s.query_id
WHERE q.name = 'customer-rolling-stats'
ORDER BY events_in_window DESC
LIMIT 20;

Hook this query into Grafana with the standard PostgreSQL datasource and you have a live per-customer dashboard. No exporter, no second database, no schema migration in a different system.

5. Emitting closed windows to a queue

Pair this with the .to(sinkQueue) on the stream and every closed window becomes a regular Queen message on a sink queue. That queue can fan out to multiple downstream consumers (billing, alerting, anomaly detection) using consumer groups, with full replay-from-timestamp support.

await q.queue('events.totals_per_customer_per_5s')
  .group('billing-meter')
  .batch(50)
  .each()
  .consume(async (msg) => {
    // msg.data = { count, sum, max }   one record per closed window
    await billingSystem.recordUsage({
      customer:  msg.partition,        // partitions carry forward the customerId
      windowEnd: msg.data.windowEnd,
      count:     msg.data.count,
      revenue:   msg.data.sum
    })
  })

Other consumer groups (analytics-dashboard, anomaly-detector) read the same sink queue independently. Adding a new downstream is one q.queue(...).group('new-name').consume(...) — no schema change to the streaming query.

6. vs. Materialize / ksqlDB / Tinybird

When you pick this approach over a dedicated streaming SQL engine, you're making two trade-offs:

Pick Materialize / ksqlDB when you genuinely need streaming SQL with joins, CDC, or advanced temporal operators. Pick Queen streams when your aggregations are per-entity windowed counters / sums / max — the vast majority of "real-time per-customer X" requirements — and you don't want to operate a second database for them.

7. Common pitfalls

8. References

← Back to use cases