@travetto/model

Data Modeling Support

Datastore abstraction for core operations.

Install: @travetto/model

npm install @travetto/model

# or

yarn add @travetto/model

This module provides a set of contracts/interfaces to data model persistence, modification and retrieval. This module builds heavily upon the Schema, which is used for data model validation.

Contracts

The module is mainly composed of contracts. The contracts define the expected interface for various model patterns. The primary contracts are Basic, CRUD, Indexed, Expiry, Streaming and Bulk.

Basic

All Data Modeling Support implementations, must honor the Basic contract to be able to participate in the model ecosystem. This contract represents the bare minimum for a model service.

Code: Basic Contract

export interface ModelBasicSupport<C = unknown> {
  /**
   * Get underlying client
   */
  get client(): C;

  /**
   * Get by Id
   * @param id The identifier of the document to retrieve
   * @throws {NotFoundError} When an item is not found
   */
  get<T extends ModelType>(cls: Class<T>, id: string): Promise<T>;

  /**
   * Create new item
   * @param item The document to create
   * @throws {ExistsError} When an item with the provided id already exists
   */
  create<T extends ModelType>(cls: Class<T>, item: OptionalId<T>): Promise<T>;

  /**
   * Delete an item
   * @param id The id of the document to delete
   * @throws {NotFoundError} When an item is not found
   */
  delete<T extends ModelType>(cls: Class<T>, id: string): Promise<void>;
}

CRUD

The CRUD contract, builds upon the basic contract, and is built around the idea of simple data retrieval and storage, to create a foundation for other services that need only basic support. The model extension in Authentication, is an example of a module that only needs create, read and delete, and so any implementation of Data Modeling Support that honors this contract, can be used with the Authentication model extension.

Code: Crud Contract

export interface ModelCrudSupport extends ModelBasicSupport {

  /**
   * Id Source
   */
  idSource: ModelIdSource;

  /**
   * Update an item
   * @param item The document to update.
   * @throws {NotFoundError} When an item is not found
   */
  update<T extends ModelType>(cls: Class<T>, item: T): Promise<T>;

  /**
   * Create or update an item
   * @param item The document to upsert
   * @param view The schema view to validate against
   */
  upsert<T extends ModelType>(cls: Class<T>, item: OptionalId<T>): Promise<T>;

  /**
   * Update partial, respecting only top level keys.
   *
   * When invoking this method, any top level keys that are null/undefined are treated as removals/deletes.  Any properties
   * that point to sub objects/arrays are treated as wholesale replacements.
   *
   * @param id The document identifier to update
   * @param item The document to partially update.
   * @param view The schema view to validate against
   * @throws {NotFoundError} When an item is not found
   */
  updatePartial<T extends ModelType>(cls: Class<T>, item: Partial<T> & { id: string }, view?: string): Promise<T>;

  /**
   * List all items
   */
  list<T extends ModelType>(cls: Class<T>): AsyncIterable<T>;
}

Indexed

Additionally, an implementation may support the ability for basic Indexed queries. This is not the full featured query support of Data Model Querying, but allowing for indexed lookups. This does not support listing by index, but may be added at a later date.

Code: Indexed Contract

export interface ModelIndexedSupport extends ModelBasicSupport {
  /**
   * Get entity by index as defined by fields of idx and the body fields
   * @param cls The type to search by
   * @param idx The index name to search against
   * @param body The payload of fields needed to search
   */
  getByIndex<T extends ModelType>(cls: Class<T>, idx: string, body: DeepPartial<T>): Promise<T>;

  /**
   * Delete entity by index as defined by fields of idx and the body fields
   * @param cls The type to search by
   * @param idx The index name to search against
   * @param body The payload of fields needed to search
   */
  deleteByIndex<T extends ModelType>(cls: Class<T>, idx: string, body: DeepPartial<T>): Promise<void>;

  /**
   * List entity by ranged index as defined by fields of idx and the body fields
   * @param cls The type to search by
   * @param idx The index name to search against
   * @param body The payload of fields needed to search
   */
  listByIndex<T extends ModelType>(cls: Class<T>, idx: string, body?: DeepPartial<T>): AsyncIterable<T>;

  /**
   * Upsert by index, allowing the index to act as a primary key
   * @param cls The type to create for
   * @param idx The index name to use
   * @param body The document to potentially store
   */
  upsertByIndex<T extends ModelType>(cls: Class<T>, idx: string, body: OptionalId<T>): Promise<T>;
}

Expiry

Certain implementations will also provide support for automatic Expiry of data at runtime. This is extremely useful for temporary data as, and is used in the Caching module for expiring data accordingly.

Code: Expiry Contract

export interface ModelExpirySupport extends ModelCrudSupport {
  /**
   * Delete all expired by class
   *
   * @returns Returns the number of documents expired
   */
  deleteExpired<T extends ModelType>(cls: Class<T>): Promise<number>;
}

Stream

Some implementations also allow for the ability to read/write binary data as a Streaming. Given that all implementations can store Base64 encoded data, the key differentiator here, is native support for streaming data, as well as being able to store binary data of significant sizes. This pattern is currently used by Asset for reading and writing asset data.

Code: Stream Contract

export interface ModelStreamSupport {

  /**
   * Upsert stream to storage
   * @param location The location of the stream
   * @param input The actual stream to write
   * @param meta The stream metadata
   */
  upsertStream(location: string, input: Readable, meta: StreamMeta): Promise<void>;

  /**
   * Get stream from asset store
   * @param location The location of the stream
   */
  getStream(location: string): Promise<Readable>;

  /**
   * Get partial stream from asset store given a starting byte and an optional ending byte
   * @param location The location of the stream
   */
  getStreamPartial(location: string, start: number, end?: number): Promise<PartialStream>;

  /**
   * Get metadata for stream
   * @param location The location of the stream
   */
  describeStream(location: string): Promise<StreamMeta>;

  /**
   * Delete stream by location
   * @param location The location of the stream
   */
  deleteStream(location: string): Promise<void>;
}

Bulk

Finally, there is support for Bulk operations. This is not to simply imply issuing many commands at in parallel, but implementation support for an atomic/bulk operation. This should allow for higher throughput on data ingest, and potentially for atomic support on transactions.

Code: Bulk Contract

export interface ModelBulkSupport extends ModelCrudSupport {
  processBulk<T extends ModelType>(cls: Class<T>, operations: BulkOp<T>[]): Promise<BulkResponse>;
}

Declaration

Models are declared via the @Model decorator, which allows the system to know that this is a class that is compatible with the module. The only requirement for a model is the ModelType

Code: ModelType

export interface ModelType {
  /**
   * Unique identifier.
   *
   * If not provided, will be computed on create
   */
  id: string;
}

The id is the only required field for a model, as this is a hard requirement on naming and type. This may make using existing data models impossible if types other than strings are required. Additionally, the type field, is intended to record the base model type, but can be remapped. This is important to support polymorphism, not only in Data Modeling Support, but also in Schema.

Implementations

Service	Basic	CRUD	Indexed	Expiry	Stream	Bulk
DynamoDB Model Support	X	X	X	X
Elasticsearch Model Source	X	X	X	X		X
Firestore Model Support	X	X	X
MongoDB Model Support	X	X	X	X	X	X
Redis Model Support	X	X	X	X
S3 Model Support	X	X		X	X
SQL Model Service	X	X	X	X		X
MemoryModelService	X	X	X	X	X	X
FileModelService	X	X		X	X	X

Custom Model Service

In addition to the provided contracts, the module also provides common utilities and shared test suites. The common utilities are useful for repetitive functionality, that is unable to be shared due to not relying upon inheritance (this was an intentional design decision). This allows for all the Data Modeling Support implementations to completely own the functionality and also to be able to provide additional/unique functionality that goes beyond the interface.

Code: Memory Service

import { Readable } from 'node:stream';
import { StreamUtil, Class, TimeSpan } from '@travetto/base';
import { DeepPartial } from '@travetto/schema';
import { Injectable } from '@travetto/di';
import { Config } from '@travetto/config';
import { ModelCrudSupport } from '../service/crud';
import { ModelStreamSupport, PartialStream, StreamMeta } from '../service/stream';
import { ModelType, OptionalId } from '../types/model';
import { ModelExpirySupport } from '../service/expiry';
import { ModelRegistry } from '../registry/model';
import { ModelStorageSupport } from '../service/storage';
import { ModelCrudUtil } from '../internal/service/crud';
import { ModelExpiryUtil } from '../internal/service/expiry';
import { NotFoundError } from '../error/not-found';
import { ExistsError } from '../error/exists';
import { ModelIndexedSupport } from '../service/indexed';
import { ModelIndexedUtil } from '../internal/service/indexed';
import { ModelStorageUtil } from '../internal/service/storage';
import { ModelStreamUtil, StreamModel, STREAMS } from '../internal/service/stream';
import { IndexConfig } from '../registry/types';
const STREAM_META = `${STREAMS}_meta`;
type StoreType = Map<string, Buffer>;
@Config('model.memory')
export class MemoryModelConfig {
  autoCreate?: boolean;
  namespace?: string;
  cullRate?: number | TimeSpan;
}
function indexName<T extends ModelType>(cls: Class<T>, idx: IndexConfig<T> | string, suffix?: string): string {
  return [cls.Ⲑid, typeof idx === 'string' ? idx : idx.name, suffix].filter(x => !!x).join(':');
}
function getFirstId(data: Map<string, unknown> | Set<string>, value?: string | number): string | undefined {
  let id: string | undefined;
  if (data instanceof Set) {
    id = data.values().next().value;
  } else {
    id = [...data.entries()].find(([k, v]) => value === undefined || v === value)?.[0];
  }
  return id;
}
/**
 * Standard in-memory support
 */
@Injectable()
export class MemoryModelService implements ModelCrudSupport, ModelStreamSupport, ModelExpirySupport, ModelStorageSupport, ModelIndexedSupport {
    sorted: new Map<string, Map<string, Map<string, number>>>(),
    unsorted: new Map<string, Map<string, Set<string>>>()
  };
  idSource = ModelCrudUtil.uuidSource();
  get client(): Map<string, StoreType>;
  constructor(public readonly config: MemoryModelConfig) { }
  async postConstruct(): Promise<void>;
  // CRUD Support
  async get<T extends ModelType>(cls: Class<T>, id: string): Promise<T>;
  async create<T extends ModelType>(cls: Class<T>, item: OptionalId<T>): Promise<T>;
  async update<T extends ModelType>(cls: Class<T>, item: T): Promise<T>;
  async upsert<T extends ModelType>(cls: Class<T>, item: OptionalId<T>): Promise<T>;
  async updatePartial<T extends ModelType>(cls: Class<T>, item: Partial<T> & { id: string }, view?: string): Promise<T>;
  async delete<T extends ModelType>(cls: Class<T>, id: string): Promise<void>;
  async * list<T extends ModelType>(cls: Class<T>): AsyncIterable<T>;
  // Stream Support
  async upsertStream(location: string, input: Readable, meta: StreamMeta): Promise<void>;
  async getStream(location: string): Promise<Readable>;
  async getStreamPartial(location: string, start: number, end?: number): Promise<PartialStream>;
  async describeStream(location: string): Promise<StreamMeta>;
  async deleteStream(location: string): Promise<void>;
  // Expiry
  async deleteExpired<T extends ModelType>(cls: Class<T>): Promise<number>;
  // Storage Support
  async createStorage(): Promise<void>;
  async deleteStorage(): Promise<void>;
  async createModel<T extends ModelType>(cls: Class<T>): Promise<void>;
  async truncateModel<T extends ModelType>(cls: Class<T>): Promise<void>;
  // Indexed
  async getByIndex<T extends ModelType>(cls: Class<T>, idx: string, body: DeepPartial<T>): Promise<T>;
  async deleteByIndex<T extends ModelType>(cls: Class<T>, idx: string, body: DeepPartial<T>): Promise<void>;
  upsertByIndex<T extends ModelType>(cls: Class<T>, idx: string, body: OptionalId<T>): Promise<T>;
  async * listByIndex<T extends ModelType>(cls: Class<T>, idx: string, body?: DeepPartial<T>): AsyncIterable<T>;
}

To enforce that these contracts are honored, the module provides shared test suites to allow for custom implementations to ensure they are adhering to the contract's expected behavior.

Code: Memory Service Test Configuration

import { Suite } from '@travetto/test';

import { MemoryModelConfig, MemoryModelService } from '../src/provider/memory';
import { ModelCrudSuite } from '../support/test/crud';
import { ModelExpirySuite } from '../support/test/expiry';
import { ModelStreamSuite } from '../support/test/stream';
import { ModelIndexedSuite } from '../support/test/indexed';
import { ModelBasicSuite } from '../support/test/basic';
import { ModelPolymorphismSuite } from '../support/test/polymorphism';

@Suite()
export class MemoryBasicSuite extends ModelBasicSuite {
  serviceClass = MemoryModelService;
  configClass = MemoryModelConfig;
}

@Suite()
export class MemoryCrudSuite extends ModelCrudSuite {
  serviceClass = MemoryModelService;
  configClass = MemoryModelConfig;
}

@Suite()
export class MemoryStreamSuite extends ModelStreamSuite {
  serviceClass = MemoryModelService;
  configClass = MemoryModelConfig;
}

@Suite()
export class MemoryExpirySuite extends ModelExpirySuite {
  serviceClass = MemoryModelService;
  configClass = MemoryModelConfig;
}

@Suite()
export class MemoryIndexedSuite extends ModelIndexedSuite {
  serviceClass = MemoryModelService;
  configClass = MemoryModelConfig;
}

@Suite()
export class MemoryPolymorphicSuite extends ModelPolymorphismSuite {
  serviceClass = MemoryModelService;
  configClass = MemoryModelConfig;
}

CLI - model:export

The module provides the ability to generate an export of the model structure from all the various @Models within the application. This is useful for being able to generate the appropriate files to manually create the data schemas in production.

Terminal: Running model export

$ trv model:export --help

Usage: model:export [options] <provider:string> <models...:string>

Options:
  -e, --env <string>     Application environment
  -m, --module <module>  Module to run for
  -h, --help             display help for command

Providers
--------------------
  * SQL

Models
--------------------
  * samplemodel

CLI - model:install

The module provides the ability to install all the various @Models within the application given the current configuration being targeted. This is useful for being able to prepare the datastore manually.

Terminal: Running model install

$ trv model:install --help

Usage: model:install [options] <provider:string> <models...:string>

Options:
  -e, --env <string>     Application environment
  -m, --module <module>  Module to run for
  -h, --help             display help for command

Providers
--------------------
  * Memory
  * SQL

Models
--------------------
  * samplemodel