Database

Stream

The Stream<T> class represents a sequence of documents that can be transformed, filtered, joined, grouped, and aggregated. It extends Query<T[]> and serves as the foundation for collection-based operations in AQL.

Class Definition

class Stream<T> extends Query<T[]> {
  // Type casting
  cast<U>(): Stream<U>;

  // Field access
  key<K extends keyof T>(key: K, def?: unknown): Stream<T[K]>;
  default<U>(val: Value<U>): Stream<Exclude<T, undefined | null> | U>;

  // Transformation
  map<U>(mapper: (val: ValueProxy<T>) => U): Stream<ExtractType<U>>;
  filter(predicate: (val: ValueProxy<T>) => ValueProxyOrValue<boolean>): this;
  pluck(...fields: string[]): Stream<Partial<T>>;
  without(...fields: string[]): Stream<Partial<T>>;

  // Combining
  union<U>(other: Stream<U>): Stream<T | U>;

  // Joining
  join<U, V>(right: Stream<U>, predicate, mapper): Stream<ExtractType<V>>;
  joinInner<U, V>(right: Stream<U>, predicate, mapper): Stream<ExtractType<V>>;
  lookup<U, TK extends keyof T>(right: Selection<U>, localKey: TK, otherKey: keyof U): Stream<...>;

  // Grouping
  group<U, K>(index: K, mapper: (stream: Stream<T>, group: ValueProxy) => U): Stream<ExtractType<U>>;

  // Ordering and pagination
  orderBy(index: string, direction?: "asc" | "desc"): this;
  slice(offset: Value<number>, count?: Value<number>): this;
  nth(n: Value<number>): Datum<T | null>;

  // Aggregation
  count(field?: keyof T): Datum<number>;
  sum(field?: keyof T): Datum<number>;
  avg(field?: keyof T): Datum<number>;
  min(field?: keyof T): Datum<number>;
  max(field?: keyof T): Datum<number>;
  distinct(): Datum<T[]>;
  distinct<TK extends keyof T>(field: TK): Stream<T[TK]>;

  // Change feeds
  changes(): Query<Changes<T>[]>;
}

Field Access

Access a specific field from all documents in the stream with key(), or provide a default value for null fields with default().

const users = schema.instance().table("users");

// Extract a single field
const emails = await users.key("email");

// Provide a default value
const phones = await users.key("phone").default("N/A");

Transform Data

`map()`

Transform each document using a mapping function. The callback receives a ValueProxy<T> for constructing expressions.

const profiles = await users.map((user) => ({
  id: user.key("_id"),
  displayName: user.key("name"),
  contact: user.key("email"),
}));

`filter()`

Select documents matching a predicate. The callback returns a boolean expression.

const active = await users.filter((user) =>
  user.key("status").eq("active")
    .and(user.key("age").ge(18)),
);

`pluck()` and `without()`

Select or exclude specific fields from all documents.

// Keep only specific fields
const slim = await users.pluck("_id", "name", "email");

// Remove sensitive fields
const safe = await users.without("password", "ssn");

Combine Streams

`union()`

Concatenate two streams without deduplication.

const allPremium = await users
  .filter((u) => u.key("plan").eq("premium"))
  .union(
    users.filter((u) => u.key("plan").eq("enterprise")),
  );

Join Data

`join()` - Left Join

Perform a left join between two streams. Every document from the left stream appears in the result, with null for the right side when no match exists.

const usersWithOrders = await users.join(
  schema.instance().table("orders"),
  (user, order) => order.key("userId").eq(user.key("_id")),
  (user, order) => ({
    userName: user.key("name"),
    orderTotal: order.key("amount").default(0),
  }),
);

`joinInner()` - Inner Join

Perform an inner join. Only documents with matches on both sides appear in the result.

const matched = await users.joinInner(
  schema.instance().table("orders"),
  (user, order) => order.key("userId").eq(user.key("_id")),
  (user, order) => ({
    userName: user.key("name"),
    amount: order.key("amount"),
  }),
);

`lookup()`

Replace a foreign key field with the referenced document from another table. See Lookup for detailed usage.

const postsWithAuthors = await schema
  .instance()
  .table("posts")
  .lookup(schema.instance().table("users"), "authorId", "_id");

Group Data

The group method partitions the stream by an index value, then applies a mapping function to each group. The mapper receives the group's sub-stream and the group key.

const statsByRole = await users.group("role", (stream, role) => ({
  role: role,
  total: stream.count(),
  avgAge: stream.avg("age"),
}));

Order and Paginate

`orderBy()`

Sort the stream by an index field.

const sorted = await users.orderBy("name", "asc");
const newest = await users.orderBy("createdAt", "desc");

`slice()`

Extract a subsection of the stream for pagination.

const page1 = await users.orderBy("createdAt").slice(0, 10);
const page2 = await users.orderBy("createdAt").slice(10, 10);

`nth()`

Retrieve a single document by its position in the stream. Returns a Datum<T | null>.

const first = await users.orderBy("createdAt").nth(0);

Aggregate Data

Stream provides several aggregation methods that return a Datum<number>:

const totalUsers = await users.count();
const distinctEmails = await users.count("email");
const totalSales = await schema.instance().table("orders").sum("amount");
const avgOrder = await schema.instance().table("orders").avg("amount");
const maxOrder = await schema.instance().table("orders").max("amount");
const minOrder = await schema.instance().table("orders").min("amount");

`distinct()`

Retrieve deduplicated documents or distinct values of a specific field.

// All distinct documents
const unique = await users.distinct();

// Distinct values of a field
const roles = await users.distinct("role");

Change Feeds

The changes method subscribes to real-time change notifications on the stream.

const changes = await users.changes();

for (const change of changes) {
  console.log(change.changeType, change.oldValue, change.newValue);
}

See Change Feeds for more details.

SingleSelection

The SingleSelection<T> class represents a reference to a single document in a table. It extends Datum<T> and provides methods for updating, replacing, deleting, and monitoring changes to an individual document.

Change Feeds

Change feeds provide real-time notifications when documents are inserted, updated, or deleted. Both Stream<T> and SingleSelection<T> expose a changes() method that returns change events.