DataService

概要

DataServiceはCQRSパターンのクエリ側であり、DynamoDBに保存されたデータに対する効率的な読み取り操作を提供します。クエリに最適化されたデータテーブル（読み取りモデル）からのすべての読み取り操作を処理します。

Read-Your-Writes（自己書き込み読み取り）整合性

DataServiceはDynamoDB Streamsを介して非同期に更新されるDynamoDBデータテーブルから読み取ります。ユーザーが自分の書き込みをすぐに確認できる必要がある場合（Streamが反映される前）は、代わりにRepositoryを使用してください。RepositoryはRYWセッションキャッシュを追加するDataServiceのドロップイン代替品です。設定と使用方法についてはCommandServiceのRead-Your-Writesを参照してください。

DataServiceを使用する前に、CommandServiceセクションで説明されているようにCommandModuleをセットアップする必要があります。

メソッド

async getItem(key: DetailKey): Promise<DataModel | undefined>

getItemメソッドは、指定された詳細キー/主キーを持つアイテムの属性セットを返します。一致するアイテムがない場合、getItemはundefinedを返します。

例：

import { DataService, DataModel } from '@mbc-cqrs-serverless/core';
import { Injectable, NotFoundException } from '@nestjs/common';

@Injectable()
export class CatService {
  constructor(private readonly dataService: DataService) {}

  async getCat(pk: string, sk: string): Promise<CatDataEntity> {
    const item = await this.dataService.getItem({ pk, sk });

    if (!item) {
      throw new NotFoundException('Cat not found');
    }
    return new CatDataEntity(item as CatDataEntity);
  }
}

async listItemsByPk(pk: string, opts?: ListItemsOptions): Promise<DataListEntity>

listItemsByPkメソッドは、パーティションキーに一致する1つ以上のアイテムを返します。フィルタリング、ページネーション、ソートをサポートしています。

基本的な使い方

主キー（pk）でアイテムを一覧取得：

const res = await this.dataService.listItemsByPk(pk);
return new CatListEntity(res);

ソートキーフィルター付き

主キー（pk）でアイテムを一覧取得し、ソートキー（sk）にフィルター式を使用します。例えば、ソートキーがCAT#で始まるアイテムを取得し、100件に制限：

import { KEY_SEPARATOR } from '@mbc-cqrs-serverless/core';

const query = {
  sk: {
    skExpression: 'begins_with(sk, :typeCode)',
    skAttributeValues: {
      ':typeCode': `CAT${KEY_SEPARATOR}`,
    },
  },
  limit: 100,
};
const res = await this.dataService.listItemsByPk(pk, query);
return new CatDataListEntity(res);

ページネーション

startFromSkとlimitを使用してページネーションを実装：

async listCatsWithPagination(
  tenantCode: string,
  pageSize: number,
  lastSk?: string
): Promise<{ items: CatDataEntity[]; lastSk?: string }> {
  const pk = `CAT#${tenantCode}`;

  const result = await this.dataService.listItemsByPk(pk, {
    limit: pageSize,
    startFromSk: lastSk,
  });

  return {
    items: result.items.map(item => new CatDataEntity(item)),
    lastSk: result.lastSk,
  };
}

ソートキー演算子

以下のソートキー式がサポートされています：

演算子	式	説明
等しい	sk = :value	完全一致
前方一致	begins_with(sk, :prefix)	プレフィックス一致
範囲	sk BETWEEN :start AND :end	範囲クエリ
より小さい	sk < :value	より小さい比較
より大きい	sk > :value	より大きい比較

範囲クエリの例：

const query = {
  sk: {
    skExpression: 'sk BETWEEN :start AND :end',
    skAttributeValues: {
      ':start': 'ORDER#2024-01-01',
      ':end': 'ORDER#2024-12-31',
    },
  },
};
const res = await this.dataService.listItemsByPk(pk, query);

async publish(cmd: CommandModel): Promise<DataModel>

publishメソッドはコマンドデータをデータテーブルに発行します。これは通常、フレームワークのデータ同期ハンドラーによって内部的に呼び出されますが、カスタム同期ロジックを実装する際に直接使用することもできます。

注記

このメソッドは主にフレームワーク内部で使用されます。ほとんどの場合、データ同期を自動的に処理するCommandService.publishSync()またはCommandService.publishAsync()を使用してください。

import {
  CommandModel,
  DataModel,
  DataService,
  DataSyncHandler,
  IDataSyncHandler,
} from "@mbc-cqrs-serverless/core";
import { Injectable } from "@nestjs/common";

// Custom data sync handler example (カスタムデータ同期ハンドラーの例)
@DataSyncHandler('your-command-table-name')
@Injectable()
export class CustomDataSyncHandler implements IDataSyncHandler {
  constructor(private readonly dataService: DataService) {}

  async up(cmd: CommandModel): Promise<DataModel> {
    // Publish command to data table (コマンドをデータテーブルに発行)
    const dataModel = await this.dataService.publish(cmd);

    // Additional synchronization to external systems (外部システムへの追加同期)
    await this.syncToExternalSystem(dataModel);

    return dataModel;
  }

  async down(cmd: CommandModel): Promise<void> {
    // Handle rollback if needed (必要に応じてロールバックを処理)
  }

  private async syncToExternalSystem(data: DataModel): Promise<void> {
    // Custom sync logic (カスタム同期ロジック)
  }
}

このメソッドは：

• CommandModelをDataModel形式に変換
• ソートキーからバージョンサフィックスを削除
• 元の作成メタデータを保持（createdAt、createdBy、createdIp）
• 更新メタデータを更新（updatedAt、updatedBy、updatedIp）
• データをデータテーブルに保存

共通パターン

コードで検索

テナント内でユニークなコードでアイテムを検索：

async findByCode(tenantCode: string, code: string): Promise<CatDataEntity | undefined> {
  const pk = `CAT#${tenantCode}`;
  const sk = `CAT#${code}`;

  const item = await this.dataService.getItem({ pk, sk });
  return item ? new CatDataEntity(item) : undefined;
}

タイプフィルター付き一覧

タイプでフィルターされたアイテムを一覧取得：

async listByType(tenantCode: string, type: string): Promise<CatDataEntity[]> {
  const pk = `CAT#${tenantCode}`;

  const result = await this.dataService.listItemsByPk(pk, {
    sk: {
      skExpression: 'begins_with(sk, :type)',
      skAttributeValues: {
        ':type': `${type}#`,
      },
    },
  });

  return result.items.map(item => new CatDataEntity(item));
}

エラーハンドリング

一般的なクエリエラーを適切に処理：

import {
  Injectable,
  InternalServerErrorException,
  Logger,
  NotFoundException,
} from '@nestjs/common';
import { DataService } from '@mbc-cqrs-serverless/core';

@Injectable()
export class CatService {
  private readonly logger = new Logger(CatService.name);

  constructor(private readonly dataService: DataService) {}

  async getItemSafely(pk: string, sk: string): Promise<CatDataEntity> {
    try {
      const item = await this.dataService.getItem({ pk, sk });

      if (!item) {
        throw new NotFoundException(`Item not found: ${pk}/${sk}`);
      }

      return new CatDataEntity(item);
    } catch (error) {
      if (error instanceof NotFoundException) {
        throw error;
      }
      // Log and rethrow unexpected errors (予期しないエラーをログに記録して再スロー)
      this.logger.error('Unexpected error querying item:', error);
      throw new InternalServerErrorException('Failed to retrieve item');
    }
  }
}

型定義

DetailKey

interface DetailKey {
  pk: string;  // パーティションキー
  sk: string;  // ソートキー
}

ListItemsOptions

ListItemsOptionsはメソッドシグネチャ内のインライン型定義であり、個別にエクスポートされるインターフェースではありません。型構造は以下の通りです：

{
  sk?: {
    skExpression: string;
    skAttributeValues: Record<string, string>;
    skAttributeNames?: Record<string, string>;
  };
  startFromSk?: string;
  limit?: number;       // Default: 10 (デフォルト: 10)
  order?: 'asc' | 'desc';
}

DataListEntity

DataListEntityはリストクエリ結果をラップするクラス（インターフェースではありません）です。簡単にインスタンス化するためのコンストラクタを提供します：

class DataListEntity {
  items: DataEntity[];   // Array of data entities (データエンティティの配列)
  lastSk?: string;       // Sort key for pagination cursor (ページネーションカーソル用のソートキー)
  total?: number;        // Total count (if available) (利用可能な場合の合計数)

  constructor(data: Partial<DataListEntity>);
}

コンストラクタは部分オブジェクトを受け取り、クエリ結果からインスタンスを作成できます：

const result = await this.dataService.listItemsByPk(pk);
const listEntity = new DataListEntity(result);

HistoryService

HistoryService は履歴テーブルへのアクセスを提供し、コマンドのすべての過去バージョンを保存します。監査証跡やロールバックのシナリオで特定の履歴バージョンを取得する必要がある場合に使用します。

import {
  HistoryService,
  addSortKeyVersion,
} from '@mbc-cqrs-serverless/core';
import { Injectable, NotFoundException } from '@nestjs/common';

@Injectable()
export class ProductService {
  constructor(private readonly historyService: HistoryService) {}

  async findVersion(pk: string, sk: string, version: number) {
    const skWithVersion = addSortKeyVersion(sk, version);
    const item = await this.historyService.getItem({ pk, sk: skWithVersion });

    if (!item) {
      throw new NotFoundException(`Version ${version} not found`);
    }
    return item;
  }
}

async getItem(key: DetailKey): Promise<DataModel | undefined>

履歴テーブルから特定バージョンのアイテムを取得します。ソートキーにはバージョンサフィックスが必要です（addSortKeyVersion を使用して構築してください）。バージョンが存在しない場合は undefined を返します。

注記

履歴テーブルはデータテーブルと同じキー構造を使用しますが、ソートキーにはバージョンサフィックスが含まれます: sk@version。キーの構築には常に addSortKeyVersion(sk, version) を使用してください。

ベストプラクティス

1. プロジェクション式を使用: データ転送を削減するために必要な属性のみを取得
2. ページネーションを実装: メモリ問題を避けるため、大きな結果セットは常にページネーション
3. 頻繁にアクセスするデータをキャッシュ: 静的または変更頻度の低いデータのキャッシュを検討
4. 適切なキー設計を使用: クエリパターンを効率的にサポートするようにキーを設計
5. 見つからない場合の処理: 使用する前に常にアイテムが存在するか確認

関連ドキュメント

• コマンドサービス - CommandServiceを使ったコマンド操作
• ユニットテスト - モックを使用したDataServiceのユニットテスト
• サービスパターン - 完全なサービス層パターン
• キーパターン - 効率的なクエリのためのDynamoDBキー設計
• エンティティパターン - DataEntityを使ったエンティティ定義
• シリアライズヘルパー - DynamoDB構造をAPIレスポンス形式に変換
• データ同期ハンドラー例 - 同期ハンドラーでのDataService.publish()の使用