8.5 KiB
2. Arquitectura
TaskStream combina tres ideas:
- Arquitectura hexagonal (puertos y adaptadores), aplicada igual en la API y en la web.
- CQRS: las escrituras (commands) y las lecturas (queries) van por caminos y bases de datos distintas.
- 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).
domainsolo puede importar@nestjs/common(por@Injectable()de las factories) y@nestjs/cqrs(porAggregateRoot). Nada de TypeORM, Mongoose, Redis, GraphQL, Angular, NgRx ni Apollo.applicationno puede importar adaptadores (TypeORM, Mongoose, Redis, Apollo…).presentersnunca importainfrastructureni usa los buses CQRS: habla con los*.service.ts(API) o con las facades (web).infrastructurenunca importapresenters.- Los contextos también están acotados:
trackerpuede usaridentityyshared;sharedno 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) extiendenVersionedAggregateRoot, que extiende elAggregateRootde@nestjs/cqrsy añade la versión. - Crear: una factory
@Injectable()construye el agregado y aplica el evento de creación (skipHandler). - Modificar: el
AggregateRehydratorcarga los eventos del stream y los reproduce (loadFromHistory); los métodos del agregado validan y hacenapply(new XEvent(…)); el estado solo cambia en los handlersonXEvent. - Guardar: el
EventStorePublishersustituye al publisher delEventBus, así quecommit()escribe en la tablaevents.commitAndWait()espera a que la escritura termine (elcommit()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 conConcurrencyError(código GraphQLCONFLICT). - 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
EventStoreRelaycopia los eventos nuevos (published_at IS NULL) al Redis Streamtask-stream:events, en orden y conFOR UPDATE SKIP LOCKED(varias instancias no duplican). La entrega es at-least-once. - El
EventsBridgeconsume el stream con un consumer group, deserializa cada evento (EventClsRegistry+@AutowiredEvent) y lo entrega a los@EventsHandlerregistrados para su clase. A diferencia deeventBus.subject$.next(), espera a cada handler y procesa los eventos en orden; solo haceXACKcuando todos terminan. Si uno falla, el evento queda pendiente y se reintenta (yXAUTOCLAIMrecupera los de instancias caídas).
Lectura: MongoDB
- Las proyecciones son
@EventsHandlerde la capaapplication/event-handlers, uno por evento (más el de actividad, que escucha todos). Escriben a través de los puertosUpsertMaterialized*Repository. - Cada documento guarda la
versiondel ú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 endomain/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.
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]