# 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, temporizadores, 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…). - Los buses CQRS solo los usan los `*.service.ts` de `application` (los comandos, a través del `CommandRunner`, que los reintenta si chocan; las consultas, con el `QueryBus`); todo lo demás pasa por un servicio: resolvers, controllers, temporizadores (recordatorios, papelera; por eso viven en `presenters`), el handler de eventos de las automatizaciones y el directorio de usuarios del tracker (`IdentityService`). La única excepción es el `EventsBridge`, que ejecuta los comandos que devuelven las sagas. - Los comandos son atómicos: un handler no ejecuta otros comandos. Lo que encadena varios (duplicar un ticket, la edición en bloque, importar un CSV, las propuestas, vaciar la papelera) es un método del servicio, que ejecuta los comandos uno a uno con su bus. Así no hay ciclos entre servicios y handlers, ni `forwardRef`. - Sin módulos globales entre contextos: el tracker recibe el módulo de aplicación de identidad en la raíz de composición (`TrackerInfrastructureModule.withIdentity(...)`), el mismo con el que se monta ese contexto. - `presenters` nunca importa `infrastructure`: 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 provideTrackerShellInfrastructure(); // puertos de la cabecera → adaptadores GraphQL { provide: Notifications, useExisting: ToastService } // puerto de la aplicación → toasts de @pi/ui // apps/web/src/app/tracker.routes.ts (en diferido) { path: '', providers: [provideTrackerPagesInfrastructure()], children: trackerRoutes } ``` ## 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`. El `CommandRunner` (`shared/application`) la repite sobre el estado nuevo hasta 5 veces, con una pequeña espera creciente: cada comando carga su agregado de cero y guarda sus eventos en una sola transacción, así que un choque significa que no se guardó nada. Solo si sigue chocando llega al cliente (código GraphQL `CONFLICT`, HTTP `409`). Las sagas tienen su propio reintento en el `EventsBridge`. - **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`). - **Ids**: todas las tablas tienen `id uuid` (`gen_random_uuid()`); lo que hace única una fila de lookup es una restricción aparte (la clave, el proyecto, proyecto + nombre). En `events`, el orden global con el que se reenvían los eventos va en `seq` (`bigserial`), porque un uuid no se ordena; `stream_id` es el id del agregado. ### De escritura a lectura: Redis - La tabla `events` es la **bandeja de salida (outbox)**: cada evento entra en la misma transacción que la escritura, con `published_at` vacío. Si Postgres falla, la escritura falla entera y no queda nada a medias. - 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_. Si Redis falla, los eventos esperan en la tabla y salen al volver. - 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. El grupo de consumidores hace de **bandeja de entrada (inbox)**, y la idempotencia la ponen las proyecciones (versión, abajo). - **Si un evento falla, se reintenta en su sitio** (`EventStreamConsumer`), esperando cada vez más (hasta 30 s), y el siguiente espera: saltárselo aplicaría antes un cambio posterior y la comprobación de versión descartaría después el anterior para siempre. Si el fallo es de la infraestructura (Mongo, Redis o la red caídos, ver `isTransientError`), espera lo que haga falta y se pone al día solo al volver, sin reiniciar. Si es del propio evento (un fallo del código), tras 5 intentos va al stream `task-stream:events:dead` con el error, para no bloquear el resto. Lo que queda pendiente al parar se reintenta al arrancar (y `XAUTOCLAIM` recupera lo de instancias caídas). - `GET /api/health` dice en `events` cuántos eventos esperan en la bandeja de salida, cuántos están pendientes de proyectar y cuántos se apartaron; con alguno de más, `status: degraded` (pero 200: un retraso de la lectura no reinicia el contenedor). El stream se recorta a `EVENT_STREAM_MAX_LENGTH` (un millón): solo un corte larguísimo podría recortar algo pendiente, y Postgres siempre permite reconstruir la lectura. - 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`. - **Esquemas**: clases con `@Schema()` y `@Prop()` de `@nestjs/mongoose` (`read-model.schemas.ts`, `user.schema.ts`), en `strict: false`: cada documento guarda el read model entero y las props declaradas son las que tienen índice o una forma que conviene fijar. - **Ids**: cada documento tiene `_id` (el `ObjectId` de Mongo, que no sale de la base de datos) e `id` con índice único, que es el id del dominio: el mismo uuid que el `stream_id` del agregado en Postgres. Todas las consultas van por `id`. En usuarios es el id de Pireon; en actividad, historial y avisos, uno compuesto que hace idempotente la proyección. ### 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, a la release elegida al archivar o a ninguna (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 | | `PeopleFollowSaga` | `IssueCreatedEvent`, `IssueAssignedEvent` | `WatchIssueCommand`: quien abre el ticket y quien lo tiene asignado pasan a seguirlo | | `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] ```