# 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: ```mermaid flowchart LR P[presenters
GraphQL · HTTP · páginas] --> A[application
services · commands · queries · event handlers · stores] A --> D[domain
agregados · eventos · reglas · read models · puertos] I[infrastructure
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. ```ts // apps/api/src/app/app.module.ts SharedModule.withInfrastructure(SharedInfrastructureModule.forRootAsync({ … })); TrackerPresentersModule.withApplication( TrackerApplicationModule.withInfrastructure(TrackerInfrastructureModule), ); ``` ```ts // 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 ```mermaid 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 y `DeleteReleaseCommand` por cada release del proyecto | | | `IssueDeletedEvent` | `PurgeCommentCommand` por cada comentario y `DropLinksToIssueCommand` en los tickets que lo enlazaban | | `ReleaseArchivedSaga` | `ReleaseArchivedEvent` | `DetachIssueFromReleaseCommand` para sus tickets sin terminar (los terminados se quedan) | | `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-`), 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](04-cliente-web.md)): ```mermaid flowchart LR Page[Página Angular
presenters] -- métodos --> F[Facade
application] F -- injectDispatch --> Ev((Eventos)) Ev --> Red[withReducer
estado del store] Ev --> Eff[withEventHandlers
efectos] Eff -- puertos --> GW[Gateways GraphQL
infrastructure] GW -- respuesta --> Eff Eff -- emite eventos api / realtime --> Ev Red -- señales --> F F -- señales --> Page ``` ## 2.4 Diagrama de despliegue ```mermaid flowchart LR B[Navegador] -->|:8080| N[nginx
web] N -->|/api · /graphql · WS| A[API NestJS] A --> PG[(PostgreSQL
event store)] A --> R[(Redis
stream · pub/sub · caché)] A --> MG[(MongoDB
read model)] A -->|/api/user · /oauth/token| P[Pireon] ```