task-stream/libs/api/tracker/application/src/testing/in-memory.ts

import { Type } from '@nestjs/common';
import { EventBus, IEvent, IEventPublisher } from '@nestjs/cqrs';
import { EventStore } from '@task-stream/api/shared/application';
import {
  BusinessRuleError,
  ConcurrencyError,
  SerializableEvent,
  VersionedAggregateRoot,
} from '@task-stream/api/shared/domain';
import { Readable } from 'node:stream';
import { NotFoundError } from '@task-stream/api/shared/domain';
import { AttachmentStorage } from '../lib/ports/attachment-storage';
import { IssueNumbersRepository } from '../lib/ports/issue-numbers.repository';
import { ProjectKeysRepository } from '../lib/ports/project-keys.repository';
import { ReleaseIssuesRepository } from '../lib/ports/release-issues.repository';
import { ReleaseNamesRepository } from '../lib/ports/release-names.repository';
import { TrackerStreamsRepository } from '../lib/ports/tracker-streams.repository';
import {
  DirectoryUser,
  UserDirectory,
  UserRefs,
} from '../lib/ports/user-directory';

/** Event store double: JSON round-trip + UNIQUE(streamId, position) like Postgres. */
export class InMemoryEventStore extends EventStore {
  readonly events: SerializableEvent[] = [];

  async persist(
    eventOrEvents: SerializableEvent | SerializableEvent[],
  ): Promise<void> {
    const events = Array.isArray(eventOrEvents)
      ? eventOrEvents
      : [eventOrEvents];
    for (const e of events) {
      if (
        this.events.some(
          (x) => x.streamId === e.streamId && x.position === e.position,
        )
      ) {
        throw new ConcurrencyError('Aggregate', e.streamId);
      }
    }
    this.events.push(
      ...events.map((e) => ({
        ...e,
        data: JSON.parse(JSON.stringify(e.data)),
      })),
    );
  }

  async getEventsByStreamId(streamId: string): Promise<SerializableEvent[]> {
    return this.events
      .filter((e) => e.streamId === streamId)
      .map((e) => ({ ...e, data: this.instantiate(e) }));
  }

  /** Rebuilds every aggregate of a type (test assertions only). */
  async all<T extends VersionedAggregateRoot>(
    AggregateCls: Type<T>,
    prefix: string,
  ): Promise<T[]> {
    const ids = [
      ...new Set(
        this.events
          .filter((e) => e.type.startsWith(prefix))
          .map((e) => e.streamId),
      ),
    ];
    return Promise.all(
      ids.map(async (id) => {
        const aggregate = new AggregateCls(id);
        aggregate.loadFromHistory(await this.getEventsByStreamId(id));
        return aggregate;
      }),
    );
  }

  types(): string[] {
    return this.events.map((e) => e.type);
  }

  private instantiate(e: SerializableEvent) {
    const cls = this.registry.get(e.type);
    return cls ? Object.assign(Object.create(cls.prototype), e.data) : e.data;
  }

  private readonly registry = new Map<string, Type>();
  remember(event: object): void {
    this.registry.set(event.constructor.name, event.constructor as Type);
  }
}

/** Same contract as the real EventStorePublisher (serialize → persist → trackCommit). */
export class InMemoryEventStorePublisher implements IEventPublisher {
  constructor(private readonly store: InMemoryEventStore) {}

  static install(eventBus: EventBus, store: InMemoryEventStore): void {
    eventBus.publisher = new InMemoryEventStorePublisher(store);
  }

  publish<T extends IEvent>(event: T, dispatcher: VersionedAggregateRoot) {
    return this.publishAll([event], dispatcher);
  }

  publishAll<T extends IEvent>(
    events: T[],
    dispatcher: VersionedAggregateRoot,
  ) {
    const serialized = events.map((event, index) => {
      this.store.remember(event);
      const position = dispatcher.version.value + index + 1;
      return {
        streamId: dispatcher.id,
        type: event.constructor.name,
        position,
        data: {
          ...JSON.parse(JSON.stringify(event)),
          aggregateVersion: position,
        },
      };
    });
    const persisted = this.store.persist(serialized);
    dispatcher.trackCommit(persisted, serialized.length);
    return persisted;
  }
}

export class InMemoryProjectKeys extends ProjectKeysRepository {
  private readonly keys = new Set<string>();
  async reserve(key: string) {
    if (this.keys.has(key))
      throw new BusinessRuleError(`Project key ${key} is already taken`);
    this.keys.add(key);
  }
  async release(key: string) {
    this.keys.delete(key);
  }
}

export class InMemoryIssueNumbers extends IssueNumbersRepository {
  private readonly counters = new Map<string, number>();
  async next(projectId: string) {
    const next = (this.counters.get(projectId) ?? 0) + 1;
    this.counters.set(projectId, next);
    return next;
  }
}

export class InMemoryReleaseNames extends ReleaseNamesRepository {
  private readonly names = new Set<string>();
  async reserve(projectId: string, name: string) {
    if (this.names.has(`${projectId}/${name}`))
      throw new BusinessRuleError(`Release ${name} already exists`);
    this.names.add(`${projectId}/${name}`);
  }
  async free(projectId: string, name: string) {
    this.names.delete(`${projectId}/${name}`);
  }
}

export class FakeUserDirectory extends UserDirectory {
  readonly users = new Map<string, DirectoryUser>();

  add(id: string, name = `user ${id}`, isActive = true): this {
    this.users.set(id, { id, name, isActive });
    return this;
  }

  async findById(userId: string) {
    return this.users.get(userId) ?? null;
  }

  async refs(): Promise<UserRefs> {
    const of = (id: string) => ({ id, name: this.users.get(id)?.name ?? id });
    return { of, ofNullable: (id) => (id ? of(id) : null) };
  }
}

/** Same lookup as the Postgres adapter, over the in-memory events. */
export class InMemoryReleaseIssues extends ReleaseIssuesRepository {
  constructor(private readonly store: InMemoryEventStore) {
    super();
  }

  async findIssueIds(releaseId: string): Promise<string[]> {
    return [
      ...new Set(
        this.store.events
          .filter(
            (e) =>
              e.type === 'IssueReleaseChangedEvent' &&
              (e.data as { releaseId?: string }).releaseId === releaseId,
          )
          .map((e) => e.streamId),
      ),
    ];
  }
}

/** Same lookups as the Postgres adapter, over the in-memory events. */
export class InMemoryTrackerStreams extends TrackerStreamsRepository {
  constructor(private readonly store: InMemoryEventStore) {
    super();
  }

  private streams(
    type: string,
    matches: (data: Record<string, unknown>) => boolean,
  ): string[] {
    return [
      ...new Set(
        this.store.events
          .filter(
            (e) =>
              e.type === type && matches(e.data as Record<string, unknown>),
          )
          .map((e) => e.streamId),
      ),
    ];
  }

  async issueIdsOfProject(projectId: string) {
    return this.streams(
      'IssueCreatedEvent',
      (d) => d['projectId'] === projectId,
    );
  }

  async releaseIdsOfProject(projectId: string) {
    return this.streams(
      'ReleaseCreatedEvent',
      (d) => d['projectId'] === projectId,
    );
  }

  async commentIdsOfIssue(issueId: string) {
    return this.streams(
      'CommentAddedEvent',
      (d) => (d['comment'] as { issueId: string }).issueId === issueId,
    );
  }

  async issueIdsLinkingTo(issueId: string) {
    return this.streams(
      'IssueLinkedEvent',
      (d) => (d['link'] as { targetIssueId: string }).targetIssueId === issueId,
    );
  }
}

/** Attachment storage double: files in a map, keyed like on disk. */
export class InMemoryAttachmentStorage extends AttachmentStorage {
  readonly files = new Map<string, Buffer>();

  async put(key: string, content: Buffer) {
    this.files.set(key, content);
  }

  async read(key: string) {
    const content = this.files.get(key);
    if (!content) throw new NotFoundError('Attachment', key);
    return Readable.from(content);
  }

  async remove(key: string) {
    this.files.delete(key);
  }

  async removePrefix(prefix: string) {
    for (const key of [...this.files.keys()]) {
      if (key.startsWith(prefix)) this.files.delete(key);
    }
  }
}