# Data Consistency Patterns - Reference

This reference contains detailed implementations and code examples for data consistency patterns in distributed systems.

## Table of Contents

- [Saga Orchestrator Implementation](#saga-orchestrator-implementation)
- [Saga Example: Order Processing](#saga-example-order-processing)
- [Saga Choreography Pattern](#saga-choreography-pattern)
- [Idempotency Handler](#idempotency-handler)
- [Idempotency Key Generation](#idempotency-key-generation)
- [Optimistic Locking](#optimistic-locking)
- [CQRS Pattern](#cqrs-pattern)
- [Conflict Resolution](#conflict-resolution)
- [Testing Patterns](#testing-patterns)

---

## Saga Orchestrator Implementation

```typescript
// src/core/saga/saga-orchestrator.ts
// EN: Saga orchestrator for distributed transactions
// VI: Saga orchestrator cho distributed transactions
import { logger } from '@goodgo/logger';
import { eventPublisher } from '../events/event-publisher';

export interface SagaStep {
  name: string;
  execute: () => Promise<any>;
  compensate: (context: any) => Promise<void>;
  retry?: number;
}

export interface SagaContext {
  sagaId: string;
  steps: SagaStep[];
  currentStep: number;
  data: Record<string, any>;
  status: 'pending' | 'running' | 'completed' | 'compensating' | 'failed';
}

export class SagaOrchestrator {
  /**
   * EN: Execute saga with all steps
   * VI: Thuc thi saga voi tat ca cac buoc
   */
  async execute(context: SagaContext): Promise<void> {
    context.status = 'running';

    try {
      for (let i = 0; i < context.steps.length; i++) {
        context.currentStep = i;
        const step = context.steps[i];

        logger.info('Executing saga step', {
          sagaId: context.sagaId,
          step: step.name,
          stepIndex: i,
        });

        try {
          const result = await this.executeWithRetry(step, step.retry || 3);
          context.data[step.name] = result;

          // EN: Publish step completed event
          // VI: Publish event step da hoan thanh
          await eventPublisher.publish('saga.step.completed', {
            eventType: 'saga.step.completed',
            eventVersion: '1.0.0',
            data: {
              sagaId: context.sagaId,
              step: step.name,
              stepIndex: i,
            },
          });
        } catch (error) {
          logger.error('Saga step failed', {
            sagaId: context.sagaId,
            step: step.name,
            error: error.message,
          });

          // EN: Compensate all completed steps
          // VI: Compensate tat ca cac buoc da hoan thanh
          await this.compensate(context, i - 1);
          throw error;
        }
      }

      context.status = 'completed';
      logger.info('Saga completed successfully', { sagaId: context.sagaId });
    } catch (error) {
      context.status = 'failed';
      logger.error('Saga failed', {
        sagaId: context.sagaId,
        error: error.message,
      });
      throw error;
    }
  }

  private async executeWithRetry(step: SagaStep, maxRetries: number): Promise<any> {
    let lastError: Error;

    for (let attempt = 0; attempt <= maxRetries; attempt++) {
      try {
        return await step.execute();
      } catch (error) {
        lastError = error;
        if (attempt < maxRetries) {
          const delay = Math.pow(2, attempt) * 1000; // Exponential backoff
          await new Promise((resolve) => setTimeout(resolve, delay));
        }
      }
    }

    throw lastError!;
  }

  /**
   * EN: Compensate completed steps in reverse order
   * VI: Compensate cac buoc da hoan thanh theo thu tu nguoc
   */
  private async compensate(context: SagaContext, lastCompletedStep: number): Promise<void> {
    context.status = 'compensating';

    for (let i = lastCompletedStep; i >= 0; i--) {
      const step = context.steps[i];

      try {
        logger.info('Compensating saga step', {
          sagaId: context.sagaId,
          step: step.name,
          stepIndex: i,
        });

        await step.compensate(context.data);

        await eventPublisher.publish('saga.step.compensated', {
          eventType: 'saga.step.compensated',
          eventVersion: '1.0.0',
          data: {
            sagaId: context.sagaId,
            step: step.name,
            stepIndex: i,
          },
        });
      } catch (error) {
        logger.error('Compensation failed', {
          sagaId: context.sagaId,
          step: step.name,
          error: error.message,
        });
        // EN: Log but continue compensating other steps
        // VI: Log nhung tiep tuc compensate cac buoc khac
      }
    }
  }
}
```

---

## Saga Example: Order Processing

```typescript
// src/modules/order/order.saga.ts
// EN: Order processing saga
// VI: Saga xu ly order
import { SagaOrchestrator, SagaContext, SagaStep } from '../../core/saga/saga-orchestrator';
import { orderService } from './order.service';
import { paymentService } from '../payment/payment.service';
import { inventoryService } from '../inventory/inventory.service';
import { notificationService } from '../notification/notification.service';

export class OrderProcessingSaga {
  private orchestrator: SagaOrchestrator;

  constructor() {
    this.orchestrator = new SagaOrchestrator();
  }

  async processOrder(orderData: any): Promise<void> {
    const sagaId = `saga_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;

    const steps: SagaStep[] = [
      {
        name: 'create-order',
        execute: async () => {
          return await orderService.create(orderData);
        },
        compensate: async (context) => {
          await orderService.cancel(context.data['create-order'].id);
        },
      },
      {
        name: 'reserve-inventory',
        execute: async () => {
          const order = context.data['create-order'];
          return await inventoryService.reserve(order.items);
        },
        compensate: async (context) => {
          const reservation = context.data['reserve-inventory'];
          await inventoryService.release(reservation.id);
        },
      },
      {
        name: 'process-payment',
        execute: async () => {
          const order = context.data['create-order'];
          return await paymentService.charge(order.total, order.paymentMethod);
        },
        compensate: async (context) => {
          const payment = context.data['process-payment'];
          await paymentService.refund(payment.id);
        },
      },
      {
        name: 'send-confirmation',
        execute: async () => {
          const order = context.data['create-order'];
          await notificationService.sendOrderConfirmation(order.userId, order.id);
          // EN: Non-critical step, no compensation needed
          // VI: Buoc khong quan trong, khong can compensation
        },
        compensate: async () => {
          // EN: No compensation needed for notification
          // VI: Khong can compensation cho notification
        },
      },
    ];

    const context: SagaContext = {
      sagaId,
      steps,
      currentStep: 0,
      data: {},
      status: 'pending',
    };

    await this.orchestrator.execute(context);
  }
}
```

---

## Saga Choreography Pattern

```typescript
// EN: Choreography pattern - services react to events
// VI: Pattern choreography - services phan ung voi events
// src/modules/order/order.service.ts
export class OrderService {
  async createOrder(data: CreateOrderDto): Promise<Order> {
    const order = await this.orderRepository.create(data);

    // EN: Publish event for next step
    // VI: Publish event cho buoc tiep theo
    await eventPublisher.publish('order.created', {
      eventType: 'order.created',
      eventVersion: '1.0.0',
      data: {
        orderId: order.id,
        items: order.items,
        total: order.total,
      },
    });

    return order;
  }

  async cancelOrder(orderId: string): Promise<void> {
    await this.orderRepository.update(orderId, { status: 'cancelled' });
  }
}

// src/modules/inventory/inventory.consumer.ts
// EN: Inventory service reacts to order.created event
// VI: Inventory service phan ung voi order.created event
eventConsumer.on('order.created', {
  handle: async (event) => {
    try {
      const reservation = await inventoryService.reserve(event.data.items);

      // EN: Publish next step event
      // VI: Publish event buoc tiep theo
      await eventPublisher.publish('inventory.reserved', {
        eventType: 'inventory.reserved',
        eventVersion: '1.0.0',
        data: {
          orderId: event.data.orderId,
          reservationId: reservation.id,
        },
      });
    } catch (error) {
      // EN: Publish compensation event
      // VI: Publish event compensation
      await eventPublisher.publish('order.cancelled', {
        eventType: 'order.cancelled',
        eventVersion: '1.0.0',
        data: {
          orderId: event.data.orderId,
          reason: 'inventory_reservation_failed',
        },
      });
    }
  },
});

// src/modules/payment/payment.consumer.ts
eventConsumer.on('inventory.reserved', {
  handle: async (event) => {
    try {
      const order = await orderService.findById(event.data.orderId);
      const payment = await paymentService.charge(order.total, order.paymentMethod);

      await eventPublisher.publish('payment.processed', {
        eventType: 'payment.processed',
        eventVersion: '1.0.0',
        data: {
          orderId: event.data.orderId,
          paymentId: payment.id,
        },
      });
    } catch (error) {
      // EN: Trigger compensation
      // VI: Kich hoat compensation
      await eventPublisher.publish('order.cancelled', {
        eventType: 'order.cancelled',
        eventVersion: '1.0.0',
        data: {
          orderId: event.data.orderId,
          reason: 'payment_failed',
        },
      });
    }
  },
});
```

---

## Idempotency Handler

```typescript
// src/core/idempotency/idempotency.ts
// EN: Idempotency handler
// VI: Handler idempotency
import { PrismaClient } from '@prisma/client';
import { logger } from '@goodgo/logger';
import { v4 as uuidv4 } from 'uuid';

export class IdempotencyHandler {
  constructor(private prisma: PrismaClient) {}

  /**
   * EN: Execute operation with idempotency check
   * VI: Thuc thi operation voi idempotency check
   */
  async execute<T>(
    idempotencyKey: string,
    operation: () => Promise<T>,
    ttl: number = 3600 // EN: 1 hour default / VI: Mac dinh 1 gio
  ): Promise<T> {
    // EN: Check if already processed
    // VI: Kiem tra xem da duoc xu ly chua
    const existing = await this.prisma.idempotencyRecord.findUnique({
      where: { idempotencyKey },
    });

    if (existing) {
      logger.info('Idempotent operation skipped', { idempotencyKey });
      return existing.result as T;
    }

    // EN: Execute operation
    // VI: Thuc thi operation
    try {
      const result = await operation();

      // EN: Store result
      // VI: Luu ket qua
      await this.prisma.idempotencyRecord.create({
        data: {
          idempotencyKey,
          result: result as any,
          expiresAt: new Date(Date.now() + ttl * 1000),
        },
      });

      return result;
    } catch (error) {
      // EN: Store error for idempotency (optional)
      // VI: Luu loi cho idempotency (tuy chon)
      logger.error('Idempotent operation failed', { idempotencyKey, error });
      throw error;
    }
  }
}

// Usage
const idempotencyHandler = new IdempotencyHandler(prisma);

await idempotencyHandler.execute(
  `create-user-${userId}`,
  async () => {
    return await userService.create(userData);
  }
);
```

---

## Idempotency Key Generation

```typescript
// EN: Generate idempotency key from request
// VI: Tao idempotency key tu request
export function generateIdempotencyKey(
  userId: string,
  operation: string,
  requestData: any
): string {
  const hash = createHash('sha256')
    .update(JSON.stringify({ userId, operation, requestData }))
    .digest('hex');

  return `${operation}:${userId}:${hash.substr(0, 16)}`;
}
```

---

## Optimistic Locking

### Version-Based Locking

```typescript
// src/core/locking/optimistic-lock.ts
// EN: Optimistic locking with version field
// VI: Optimistic locking voi truong version
export class OptimisticLockService {
  /**
   * EN: Update with optimistic lock
   * VI: Update voi optimistic lock
   */
  async updateWithLock<T extends { version: number }>(
    repository: any,
    id: string,
    updateFn: (current: T) => Partial<T>,
    maxRetries: number = 3
  ): Promise<T> {
    for (let attempt = 0; attempt < maxRetries; attempt++) {
      // EN: Read current version
      // VI: Doc version hien tai
      const current = await repository.findById(id);

      if (!current) {
        throw new Error('Resource not found');
      }

      // EN: Apply update
      // VI: Ap dung update
      const updates = updateFn(current);

      try {
        // EN: Update with version check
        // VI: Update voi kiem tra version
        const updated = await repository.update(id, {
          ...updates,
          version: current.version + 1,
        }, {
          where: {
            id,
            version: current.version, // EN: Only update if version matches / VI: Chi update neu version khop
          },
        });

        return updated;
      } catch (error) {
        if (attempt === maxRetries - 1) {
          throw new Error('Optimistic lock conflict - max retries exceeded');
        }

        // EN: Retry with exponential backoff
        // VI: Retry voi exponential backoff
        await new Promise((resolve) =>
          setTimeout(resolve, Math.pow(2, attempt) * 100)
        );
      }
    }

    throw new Error('Optimistic lock failed');
  }
}
```

### Prisma Schema for Optimistic Locking

```prisma
// Add version field to entities requiring optimistic locking
model Entity {
  id        String   @id @default(cuid())
  version   Int      @default(1)
  data      String
  updatedAt DateTime @updatedAt
}
```

---

## CQRS Pattern

### Write Service

```typescript
// EN: Separate read and write models
// VI: Tach biet read va write models
// src/modules/user/user.write.service.ts
export class UserWriteService {
  async createUser(data: CreateUserDto): Promise<User> {
    const user = await this.userRepository.create(data);

    // EN: Publish event for read model update
    // VI: Publish event de cap nhat read model
    await eventPublisher.publish('user.created', {
      eventType: 'user.created',
      eventVersion: '1.0.0',
      data: {
        userId: user.id,
        email: user.email,
        name: user.name,
      },
    });

    return user;
  }
}
```

### Read Service

```typescript
// src/modules/user/user.read.service.ts
// EN: Read model updated via events
// VI: Read model duoc cap nhat qua events
export class UserReadService {
  async getUser(userId: string): Promise<UserReadModel> {
    return await this.userReadRepository.findById(userId);
  }
}
```

### Read Model Consumer

```typescript
// src/modules/user/user.read.consumer.ts
eventConsumer.on('user.created', {
  handle: async (event) => {
    // EN: Update read model
    // VI: Cap nhat read model
    await userReadRepository.create({
      userId: event.data.userId,
      email: event.data.email,
      name: event.data.name,
      // EN: Additional denormalized data for queries
      // VI: Du lieu denormalized them cho queries
    });
  },
});
```

---

## Conflict Resolution

```typescript
// src/core/consistency/conflict-resolver.ts
// EN: Conflict resolution strategies
// VI: Cac chien luoc giai quyet conflict
export enum ConflictResolutionStrategy {
  LAST_WRITE_WINS = 'last_write_wins',
  FIRST_WRITE_WINS = 'first_write_wins',
  MERGE = 'merge',
  MANUAL = 'manual',
}

export class ConflictResolver {
  resolve(
    strategy: ConflictResolutionStrategy,
    current: any,
    incoming: any
  ): any {
    switch (strategy) {
      case ConflictResolutionStrategy.LAST_WRITE_WINS:
        return incoming.timestamp > current.timestamp ? incoming : current;

      case ConflictResolutionStrategy.FIRST_WRITE_WINS:
        return incoming.timestamp < current.timestamp ? incoming : current;

      case ConflictResolutionStrategy.MERGE:
        return { ...current, ...incoming, timestamp: Date.now() };

      case ConflictResolutionStrategy.MANUAL:
        // EN: Store conflict for manual resolution
        // VI: Luu conflict de giai quyet thu cong
        this.storeConflict(current, incoming);
        return current;

      default:
        return current;
    }
  }

  private storeConflict(current: any, incoming: any): void {
    // Store conflict in database for manual resolution
    logger.warn('Conflict stored for manual resolution', { current, incoming });
  }
}
```

### Conflict Resolution Usage Example

```typescript
// Example: Resolving conflicts in a sync operation
const resolver = new ConflictResolver();

async function syncData(localData: any, remoteData: any): Promise<any> {
  if (localData.version !== remoteData.version) {
    return resolver.resolve(
      ConflictResolutionStrategy.LAST_WRITE_WINS,
      localData,
      remoteData
    );
  }
  return localData;
}
```

---

## Testing Patterns

### Testing Sagas

```typescript
// src/__tests__/saga/order-processing.saga.test.ts
describe('OrderProcessingSaga', () => {
  let saga: OrderProcessingSaga;
  let orderService: jest.Mocked<OrderService>;
  let inventoryService: jest.Mocked<InventoryService>;
  let paymentService: jest.Mocked<PaymentService>;

  beforeEach(() => {
    orderService = {
      create: jest.fn(),
      cancel: jest.fn(),
    } as any;
    inventoryService = {
      reserve: jest.fn(),
      release: jest.fn(),
    } as any;
    paymentService = {
      charge: jest.fn(),
      refund: jest.fn(),
    } as any;

    saga = new OrderProcessingSaga();
  });

  it('should complete saga successfully', async () => {
    orderService.create.mockResolvedValue({ id: 'order-1', items: [], total: 100 });
    inventoryService.reserve.mockResolvedValue({ id: 'reservation-1' });
    paymentService.charge.mockResolvedValue({ id: 'payment-1' });

    await saga.processOrder(orderData);

    expect(orderService.create).toHaveBeenCalled();
    expect(inventoryService.reserve).toHaveBeenCalled();
    expect(paymentService.charge).toHaveBeenCalled();
  });

  it('should compensate on payment failure', async () => {
    orderService.create.mockResolvedValue({ id: 'order-1', items: [], total: 100 });
    inventoryService.reserve.mockResolvedValue({ id: 'reservation-1' });
    paymentService.charge.mockRejectedValue(new Error('Payment failed'));

    await expect(saga.processOrder(orderData)).rejects.toThrow('Payment failed');

    expect(orderService.cancel).toHaveBeenCalled();
    expect(inventoryService.release).toHaveBeenCalled();
    expect(paymentService.refund).not.toHaveBeenCalled();
  });

  it('should compensate on inventory failure', async () => {
    orderService.create.mockResolvedValue({ id: 'order-1', items: [], total: 100 });
    inventoryService.reserve.mockRejectedValue(new Error('Out of stock'));

    await expect(saga.processOrder(orderData)).rejects.toThrow('Out of stock');

    expect(orderService.cancel).toHaveBeenCalled();
    expect(inventoryService.release).not.toHaveBeenCalled();
  });
});
```

### Testing Idempotency

```typescript
// src/__tests__/idempotency/idempotency-handler.test.ts
describe('IdempotencyHandler', () => {
  let handler: IdempotencyHandler;
  let prisma: PrismaClient;

  beforeEach(() => {
    prisma = new PrismaClient();
    handler = new IdempotencyHandler(prisma);
  });

  it('should execute operation once', async () => {
    const operation = jest.fn().mockResolvedValue({ id: 'result-1' });

    const result1 = await handler.execute('key-1', operation);
    const result2 = await handler.execute('key-1', operation);

    expect(operation).toHaveBeenCalledTimes(1);
    expect(result1).toEqual(result2);
  });

  it('should execute different keys independently', async () => {
    const operation = jest.fn().mockResolvedValue({ id: 'result' });

    await handler.execute('key-1', operation);
    await handler.execute('key-2', operation);

    expect(operation).toHaveBeenCalledTimes(2);
  });
});
```

### Testing Optimistic Locking

```typescript
// src/__tests__/locking/optimistic-lock.test.ts
describe('OptimisticLockService', () => {
  let service: OptimisticLockService;
  let repository: jest.Mocked<Repository>;

  beforeEach(() => {
    repository = {
      findById: jest.fn(),
      update: jest.fn(),
    } as any;
    service = new OptimisticLockService();
  });

  it('should update successfully on first attempt', async () => {
    repository.findById.mockResolvedValue({ id: '1', version: 1, data: 'old' });
    repository.update.mockResolvedValue({ id: '1', version: 2, data: 'new' });

    const result = await service.updateWithLock(
      repository,
      '1',
      (current) => ({ data: 'new' })
    );

    expect(result.version).toBe(2);
    expect(result.data).toBe('new');
  });

  it('should retry on version conflict', async () => {
    repository.findById
      .mockResolvedValueOnce({ id: '1', version: 1, data: 'old' })
      .mockResolvedValueOnce({ id: '1', version: 2, data: 'concurrent' });
    repository.update
      .mockRejectedValueOnce(new Error('Version conflict'))
      .mockResolvedValueOnce({ id: '1', version: 3, data: 'new' });

    const result = await service.updateWithLock(
      repository,
      '1',
      (current) => ({ data: 'new' })
    );

    expect(repository.findById).toHaveBeenCalledTimes(2);
    expect(result.version).toBe(3);
  });

  it('should throw after max retries', async () => {
    repository.findById.mockResolvedValue({ id: '1', version: 1, data: 'old' });
    repository.update.mockRejectedValue(new Error('Version conflict'));

    await expect(
      service.updateWithLock(repository, '1', (current) => ({ data: 'new' }), 3)
    ).rejects.toThrow('Optimistic lock conflict - max retries exceeded');

    expect(repository.findById).toHaveBeenCalledTimes(3);
  });
});
```

---

## Additional Patterns

### Outbox Pattern for Reliable Event Publishing

```typescript
// src/core/outbox/outbox.service.ts
// EN: Outbox pattern for reliable event publishing
// VI: Outbox pattern cho viec publish event dang tin cay
export class OutboxService {
  constructor(private prisma: PrismaClient) {}

  async publishWithOutbox<T>(
    operation: () => Promise<T>,
    event: DomainEvent
  ): Promise<T> {
    return await this.prisma.$transaction(async (tx) => {
      // Execute business operation
      const result = await operation();

      // Store event in outbox (same transaction)
      await tx.outboxEvent.create({
        data: {
          eventType: event.eventType,
          payload: event.data as any,
          status: 'pending',
          createdAt: new Date(),
        },
      });

      return result;
    });
  }
}

// Background processor to publish outbox events
export class OutboxProcessor {
  async processOutbox(): Promise<void> {
    const events = await this.prisma.outboxEvent.findMany({
      where: { status: 'pending' },
      orderBy: { createdAt: 'asc' },
      take: 100,
    });

    for (const event of events) {
      try {
        await eventPublisher.publish(event.eventType, event.payload);
        await this.prisma.outboxEvent.update({
          where: { id: event.id },
          data: { status: 'published', publishedAt: new Date() },
        });
      } catch (error) {
        await this.prisma.outboxEvent.update({
          where: { id: event.id },
          data: {
            status: 'failed',
            retryCount: { increment: 1 },
            lastError: error.message,
          },
        });
      }
    }
  }
}
```

### Dead Letter Queue Pattern

```typescript
// src/core/dlq/dead-letter-queue.ts
// EN: Dead letter queue for failed messages
// VI: Dead letter queue cho cac message that bai
export class DeadLetterQueue {
  constructor(private prisma: PrismaClient) {}

  async moveToDeadLetter(
    originalQueue: string,
    message: any,
    error: Error,
    retryCount: number
  ): Promise<void> {
    await this.prisma.deadLetterMessage.create({
      data: {
        originalQueue,
        payload: message as any,
        errorMessage: error.message,
        errorStack: error.stack,
        retryCount,
        createdAt: new Date(),
      },
    });

    logger.warn('Message moved to dead letter queue', {
      originalQueue,
      retryCount,
      error: error.message,
    });
  }

  async reprocessDeadLetters(queue: string): Promise<number> {
    const messages = await this.prisma.deadLetterMessage.findMany({
      where: { originalQueue: queue, reprocessed: false },
    });

    let processed = 0;
    for (const msg of messages) {
      try {
        await eventPublisher.publish(queue, msg.payload);
        await this.prisma.deadLetterMessage.update({
          where: { id: msg.id },
          data: { reprocessed: true, reprocessedAt: new Date() },
        });
        processed++;
      } catch (error) {
        logger.error('Failed to reprocess dead letter', { id: msg.id, error });
      }
    }

    return processed;
  }
}
```
