task-stream/docs/02-arquitectura.md

2. Arquitectura

TaskStream combina tres ideas:

1. Arquitectura hexagonal (puertos y adaptadores), aplicada igual en la API y en la web.
2. CQRS: las escrituras (commands) y las lecturas (queries) van por caminos y bases de datos distintas.
3. Event sourcing en la API: lo que se guarda no es el estado de los agregados sino sus eventos.

2.1 Capas hexagonales

Cada contexto se divide en cuatro capas, cada una en su propia librería Nx:

flowchart LR
  P[presenters<br/>GraphQL · HTTP · páginas] --> A[application<br/>services · commands · queries · event handlers · stores]
  A --> D[domain<br/>agregados · eventos · reglas · read models · puertos]
  I[infrastructure<br/>Postgres · Mongo · Redis · Pireon · Apollo] -. implementa puertos .-> A
  I -.-> D

Capa	Contiene	Puede depender de
domain	Modelo y reglas de negocio; puertos que necesita el dominio	domain
application	Casos de uso: services, commands/queries + handlers, proyecciones (API) o facades + stores (web), puertos de salida	domain, application
infrastructure	Adaptadores de salida que implementan los puertos (bases de datos, Redis, Pireon, GraphQL)	domain, application, infrastructure
presenters	Adaptadores de entrada (resolvers GraphQL, controllers HTTP, guards, páginas Angular)	domain, application, presenters

Las reglas no son una convención: las comprueba @nx/enforce-module-boundaries en eslint.config.mjs (npx nx run-many -t lint).

• domain solo puede importar @nestjs/common (por @Injectable() de las factories) y @nestjs/cqrs (por AggregateRoot). Nada de TypeORM, Mongoose, Redis, GraphQL, Angular, NgRx ni Apollo.
• application no puede importar adaptadores (TypeORM, Mongoose, Redis, Apollo…).
• presenters nunca importa infrastructure ni usa los buses CQRS: habla con los *.service.ts (API) o con las facades (web).
• infrastructure nunca importa presenters.
• Los contextos también están acotados: tracker puede usar identity y shared; shared no depende de nadie; la web no depende de la API (y viceversa).

Composición

Las apps son la composition root: montan cada contexto de fuera hacia dentro.

// apps/api/src/app/app.module.ts
SharedModule.withInfrastructure(SharedInfrastructureModule.forRootAsync({ … }));
TrackerPresentersModule.withApplication(
  TrackerApplicationModule.withInfrastructure(TrackerInfrastructureModule),
);

// apps/web/src/app/app.config.ts
provideTrackerInfrastructure();                       // puertos → adaptadores GraphQL
{ provide: Notifications, useExisting: ToastService } // puerto de la aplicación → toasts de @pi/ui

2.2 Flujo completo de una escritura

sequenceDiagram
  autonumber
  participant UI as Web (store)
  participant R as Resolver (presenters)
  participant S as IssuesService
  participant H as CommandHandler
  participant AR as AggregateRehydrator
  participant ES as Event store (Postgres)
  participant RL as EventStoreRelay
  participant RS as Redis Stream
  participant EB as EventsBridge
  participant EH as @EventsHandler (proyecciones)
  participant M as MongoDB
  participant PS as Redis pub/sub

  UI->>R: mutation changeIssueStatus
  R->>S: changeStatus(actor, issue, status)
  S->>H: CommandBus.execute(ChangeIssueStatusCommand)
  H->>AR: rehydrate(issueId, Issue)
  AR->>ES: getEventsByStreamId
  ES-->>AR: eventos → loadFromHistory
  H->>H: issue.changeStatus() → apply(IssueStatusChangedEvent)
  H->>ES: commitAndWait() → INSERT (stream_id, position)
  H-->>UI: CommandResult { id, key, version }
  RL->>ES: eventos con published_at IS NULL
  RL->>RS: XADD
  EB->>RS: XREADGROUP (consumer group)
  EB->>EH: handle(evento) en orden, esperando
  EH->>M: UpsertMaterialized*Repository (si version es mayor)
  EH->>PS: ChangeFeed.publish*
  PS-->>UI: subscription issueChanged
  EB->>RS: XACK

Escritura: event sourcing en PostgreSQL

• Los agregados (Project, Issue, Comment, Release) extienden VersionedAggregateRoot, que extiende el AggregateRoot de @nestjs/cqrs y añade la versión.
• Crear: una factory @Injectable() construye el agregado y aplica el evento de creación (skipHandler).
• Modificar: el AggregateRehydrator carga los eventos del stream y los reproduce (loadFromHistory); los métodos del agregado validan y hacen apply(new XEvent(…)); el estado solo cambia en los handlers onXEvent.
• Guardar: el EventStorePublisher sustituye al publisher del EventBus, así que commit() escribe en la tabla events. commitAndWait() espera a que la escritura termine (el commit() de Nest no devuelve la promesa).
• Concurrencia optimista: UNIQUE(stream_id, position). Si dos peticiones modifican el mismo agregado a la vez, la segunda falla con ConcurrencyError (código GraphQL CONFLICT).
• Lookups: lo que no se puede comprobar dentro de un agregado vive en tablas aparte: claves de proyecto únicas (project_keys), numeración de tickets (issue_sequences) y nombres de release únicos (release_names).

De escritura a lectura: Redis

• El EventStoreRelay copia los eventos nuevos (published_at IS NULL) al Redis Stream task-stream:events, en orden y con FOR UPDATE SKIP LOCKED (varias instancias no duplican). La entrega es at-least-once.
• El EventsBridge consume el stream con un consumer group, deserializa cada evento (EventClsRegistry + @AutowiredEvent) y lo entrega a los @EventsHandler registrados para su clase. A diferencia de eventBus.subject$.next(), espera a cada handler y procesa los eventos en orden; solo hace XACK cuando todos terminan. Si uno falla, el evento queda pendiente y se reintenta (y XAUTOCLAIM recupera los de instancias caídas).
• Después de los handlers, el bridge ejecuta también las sagas (ver abajo).

Lectura: MongoDB

• Las proyecciones son @EventsHandler de la capa application/event-handlers, uno por evento (más el de actividad, que escucha todos). Escriben a través de los puertos UpsertMaterialized*Repository.
• Cada documento guarda la version del último evento aplicado; el upsert solo se aplica si la versión es mayor, así que reentregar un evento no tiene efecto.
• Las queries leen por los puertos Find*Repository. Los read models (ProjectReadModel, IssueReadModel…) están en domain/read-models.

Tiempo real

Tras actualizar Mongo, el handler publica en Redis pub/sub (ChangeFeed). Las subscriptions GraphQL escuchan esos canales (uno por proyecto, ticket o usuario), así que funcionan con varias instancias de la API. Como se publica después de proyectar, quien recibe el aviso siempre lee el estado nuevo.

Sagas: reacciones entre agregados

Cuando un evento de un agregado debe cambiar otros agregados, lo hace una saga (application/sagas), con el @Saga() estándar de @nestjs/cqrs: un observable que filtra eventos (ofType) y emite comandos. Como el EventBus en memoria nunca ve los eventos guardados (su publisher es el event store), las sagas las ejecuta el EventsBridge: a cada evento, tras las proyecciones, pasa ese evento solo (of(event)) por cada saga y ejecuta los comandos que emite, uno a uno y esperando. Así heredan las garantías del bridge: orden, reintento y nada se pierde si la API se cae a mitad.

Saga	Evento	Comandos
ProjectDeletionSaga	ProjectDeletedEvent	DeleteIssueCommand por cada ticket, DeleteReleaseCommand por cada release y DeleteSprintCommand por cada sprint del proyecto
	IssueDeletedEvent	PurgeCommentCommand por cada comentario, DropLinksToIssueCommand en los tickets que lo enlazaban y DetachFromParentCommand en los que colgaban de él
ReleaseArchivedSaga	ReleaseArchivedEvent	DetachIssueFromReleaseCommand para sus tickets sin terminar (los terminados se quedan)
SprintCarryOverSaga	SprintClosedEvent / SprintDeletedEvent	CarryOverIssueCommand: lo pendiente pasa al sprint elegido o al backlog; al borrar, todos vuelven al backlog
CommentersFollowSaga	CommentAddedEvent	WatchIssueCommand: quien comenta pasa a seguir el ticket
UserDeactivatedSaga	UserDeactivatedEvent (contexto identity)	UnassignIssueCommand para sus tickets abiertos en todos los proyectos, como actor system («TaskStream»)

Reglas que cumplen:

• A quién afecta sale del event store, no del read model (TrackerStreamsRepository, ReleaseIssuesRepository), salvo la saga de usuarios, que busca tickets asignados ahora mismo en Mongo.
• Comandos idempotentes: un evento puede procesarse dos veces, así que cada comando comprueba si aún hay algo que hacer (WriteAccess.live() devuelve null si el agregado no existe o ya está borrado; desasignar solo si sigue asignado a esa persona…). Son comandos internos: ningún presenter los expone.
• Nada bloquea el stream: si el dominio rechaza un comando (no encontrado, regla de negocio…), el bridge lo registra y sigue; los conflictos de concurrencia se reintentan hasta 5 veces y, si persisten, el evento queda pendiente.
• Autoría: los cambios se atribuyen a quien provocó el evento (quien borró el proyecto o archivó la release).
• Entre contextos: el webhook de Pireon guarda UserDeactivatedEvent en el event store (EventRecorder, stream identity-user-<id>), y la saga del tracker reacciona a él sin que los contextos se llamen entre sí.

Consistencia eventual

Las mutations devuelven CommandResult { id, key, version } en cuanto los eventos están guardados; Mongo se actualiza milisegundos después. Consecuencias:

• El cliente reintenta las queries que devuelven not found justo después de crear algo.
• Si necesitas el dato actualizado, espera un evento de subscription con version >= result.version.
• Las reglas que necesitan datos exactos no leen Mongo: por ejemplo, "no publicar una release con tickets sin terminar" reconstruye esos tickets desde el event store (ReleaseIssuesRepository).

2.3 El cliente web

El cliente replica las capas (ver Cliente web):

flowchart LR
  Page[Página Angular<br/>presenters] -- métodos --> F[Facade<br/>application]
  F -- injectDispatch --> Ev((Eventos))
  Ev --> Red[withReducer<br/>estado del store]
  Ev --> Eff[withEventHandlers<br/>efectos]
  Eff -- puertos --> GW[Gateways GraphQL<br/>infrastructure]
  GW -- respuesta --> Eff
  Eff -- emite eventos api / realtime --> Ev
  Red -- señales --> F
  F -- señales --> Page

2.4 Diagrama de despliegue

flowchart LR
  B[Navegador] -->|:8080| N[nginx<br/>web]
  N -->|/api · /graphql · WS| A[API NestJS]
  A --> PG[(PostgreSQL<br/>event store)]
  A --> R[(Redis<br/>stream · pub/sub · caché)]
  A --> MG[(MongoDB<br/>read model)]
  A -->|/api/user · /oauth/token| P[Pireon]