feat: public projects, project deletion and sagas

Public projects:
- Project.isPublic (create/update, owners and admins). Anyone can read a
  public project without a token: project, board, tickets, comments,
  history, releases, activity, tracking and statistics, over GraphQL and
  REST. Writes stay for members (401 anonymous, 403 outsiders).
- @OptionalAuth() / @ApiOptionalPireonAuth(): anonymous callers allowed,
  invalid tokens still rejected. ReadAccess lets anonymous viewers read
  public projects only; linked tickets are filtered by visibility.
- publicProjects (GraphQL) and GET /api/explore/projects (REST).
- Web: «Explorar» page, project pages open without a session (read-only
  notice, sign-in hints), public badge, visibility switch in settings,
  no realtime subscriptions without a session.

Project deletion:
- Owner only, confirming the key (deleteProject / DELETE
  /api/projects/{project}?confirm=KEY). The key is freed; the read model
  drops the project and members are notified (web goes home).

Sagas (run by the EventsBridge after projections, durable and ordered):
- ProjectDeletionSaga: project deleted → delete its tickets and releases;
  ticket deleted → purge its comments and drop links from other tickets.
- ReleaseArchivedSaga: unfinished tickets leave an archived release.
- UserDeactivatedSaga: the Pireon webhook records UserDeactivatedEvent in
  the event store (EventRecorder); the tracker unassigns that user's open
  tickets as the system actor.
- Internal idempotent commands; domain refusals are logged and skipped,
  concurrency conflicts retried. Affected streams come from the event store
  (TrackerStreamsRepository).

Tests: domain lifecycle specs, saga specs run like the bridge, facade
tests, e2e for public access, cascade deletion, archived releases and
deactivation webhook. Docs, READMEs, schema.gql/openapi.json and Word/PDF
updated.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-07 07:04:20 +02:00
parent 0e301ad545
commit e656222b54
164 changed files with 3862 additions and 414 deletions

View File

@@ -9,7 +9,8 @@ Casos de uso de identidad.
- `IdentityService`: punto de entrada para los presenters.
- Queries: `ResolveSessionQuery` (token → usuario, con caché), `FindUserQuery`, `FindUsersByIdsQuery`, `SearchUsersQuery`, `GetAuthorizationConfigQuery`.
- Commands: `IssueTokenCommand` (login / refresh vía Pireon), `HandleIdentityWebhookCommand`.
- Commands: `IssueTokenCommand` (login / refresh vía Pireon), `HandleIdentityWebhookCommand` (al desactivar a un
usuario, guarda `UserDeactivatedEvent` en el event store para que el tracker reaccione).
- Puertos: `IdentityProvider`, `SessionCache`, `UserRegistry`.
Ver [Autenticación](../../../../docs/05-autenticacion.md).

View File

@@ -3,10 +3,7 @@
"$schema": "../../../../node_modules/nx/schemas/project-schema.json",
"sourceRoot": "libs/api/identity/application/src",
"projectType": "library",
"tags": [
"scope:api-identity",
"type:application"
],
"tags": ["scope:api-identity", "type:application"],
"// targets": "to see all targets run: nx show project api-identity-application --web",
"targets": {}
}

View File

@@ -1,5 +1,7 @@
import { Logger } from '@nestjs/common';
import { Command, CommandHandler, ICommandHandler } from '@nestjs/cqrs';
import { EventRecorder } from '@task-stream/api/shared/application';
import { UserDeactivatedEvent } from '@task-stream/api/identity/domain';
import { SessionCache } from '../ports/session-cache.port';
import { UserRegistry } from '../ports/user-registry.port';
@@ -22,6 +24,7 @@ export class HandleIdentityWebhookHandler
constructor(
private readonly cache: SessionCache,
private readonly registry: UserRegistry,
private readonly events: EventRecorder,
) {}
async execute({ event, data }: HandleIdentityWebhookCommand): Promise<void> {
@@ -50,8 +53,14 @@ export class HandleIdentityWebhookHandler
private async deactivate(userId: string): Promise<void> {
const user = await this.registry.findById(userId);
if (!user) return;
if (!user?.isActive) return;
user.deactivate();
await this.registry.upsert(user);
// Stored so other contexts react (the tracker unassigns their open tickets).
for (const event of user.getUncommittedEvents()) {
if (event instanceof UserDeactivatedEvent)
await this.events.record(`identity-user-${userId}`, event);
}
user.uncommit();
}
}

View File

@@ -3,10 +3,7 @@
"$schema": "../../../../node_modules/nx/schemas/project-schema.json",
"sourceRoot": "libs/api/identity/domain/src",
"projectType": "library",
"tags": [
"scope:api-identity",
"type:domain"
],
"tags": ["scope:api-identity", "type:domain"],
"// targets": "to see all targets run: nx show project api-identity-domain --web",
"targets": {}
}

View File

@@ -3,10 +3,7 @@
"$schema": "../../../../node_modules/nx/schemas/project-schema.json",
"sourceRoot": "libs/api/identity/infrastructure/src",
"projectType": "library",
"tags": [
"scope:api-identity",
"type:infrastructure"
],
"tags": ["scope:api-identity", "type:infrastructure"],
"// targets": "to see all targets run: nx show project api-identity-infrastructure --web",
"targets": {}
}

View File

@@ -7,7 +7,8 @@ Adaptadores de entrada de identidad.
## Contenido
- `auth/`: `AuthGuard` global, `@Public()`, `@CurrentUser()`, `@AccessToken()`, `@RequireAppRoles()`, `WsAuthenticator`
- `auth/`: `AuthGuard` global, `@Public()`, `@OptionalAuth()` / `@ApiOptionalPireonAuth()` (lecturas de proyectos
públicos: sin token, el usuario es `undefined`), `@CurrentUser()`, `@AccessToken()`, `@RequireAppRoles()`, `WsAuthenticator`
(subscriptions) y `@ApiPireonAuth()` (esquemas OAuth2 / Bearer de Pireon en OpenAPI).
- `http/`: `AuthController` (`GET /api/auth/config`, `POST /api/auth/token`), `UsersController` (`/api/users/me`,
directorio de Pireon) y `PireonWebhookController` (firma HMAC). Documentados con `@nestjs/swagger`.

View File

@@ -3,10 +3,7 @@
"$schema": "../../../../node_modules/nx/schemas/project-schema.json",
"sourceRoot": "libs/api/identity/presenters/src",
"projectType": "library",
"tags": [
"scope:api-identity",
"type:presenters"
],
"tags": ["scope:api-identity", "type:presenters"],
"// targets": "to see all targets run: nx show project api-identity-presenters --web",
"targets": {}
}

View File

@@ -4,20 +4,27 @@ import {
applyDecorators,
createParamDecorator,
} from '@nestjs/common';
import { ApiBearerAuth, ApiOAuth2 } from '@nestjs/swagger';
import { ApiBearerAuth, ApiOAuth2, ApiSecurity } from '@nestjs/swagger';
import { getAuthRequest } from './auth-request';
export const IS_PUBLIC = 'identity:isPublic';
export const IS_OPTIONAL_AUTH = 'identity:optionalAuth';
export const REQUIRED_APP_ROLES = 'identity:appRoles';
/** Skips authentication (health checks, webhooks...). */
export const Public = () => SetMetadata(IS_PUBLIC, true);
/**
* Anonymous callers are let through (`@CurrentUser()` is undefined); a token,
* if sent, must still be valid. For reads that public projects allow.
*/
export const OptionalAuth = () => SetMetadata(IS_OPTIONAL_AUTH, true);
/** Requires one of these roles in the task-stream OAuth client (Pireon app roles). */
export const RequireAppRoles = (...roles: string[]) =>
SetMetadata(REQUIRED_APP_ROLES, roles);
/** The authenticated caller (User). */
/** The authenticated caller (User); undefined on anonymous `@OptionalAuth()` calls. */
export const CurrentUser = createParamDecorator(
(_: unknown, context: ExecutionContext) => getAuthRequest(context).user,
);
@@ -39,3 +46,10 @@ export const PIREON_BEARER = 'pireon';
*/
export const ApiPireonAuth = () =>
applyDecorators(ApiOAuth2([], PIREON_OAUTH), ApiBearerAuth(PIREON_BEARER));
/**
* `@OptionalAuth()` + OpenAPI: adds "no token" to the security requirements
* the controller declares with `@ApiPireonAuth()`.
*/
export const ApiOptionalPireonAuth = () =>
applyDecorators(OptionalAuth(), ApiSecurity({}));

View File

@@ -11,9 +11,16 @@ import {
InvalidAccessTokenError,
} from '@task-stream/api/identity/application';
import { extractBearerToken, getAuthRequest } from './auth-request';
import { IS_PUBLIC, REQUIRED_APP_ROLES } from './auth.decorators';
import {
IS_OPTIONAL_AUTH,
IS_PUBLIC,
REQUIRED_APP_ROLES,
} from './auth.decorators';
/** Global guard: every HTTP/GraphQL operation needs a valid Pireon token unless @Public(). */
/**
* Global guard: every HTTP/GraphQL operation needs a valid Pireon token unless
* @Public(); with @OptionalAuth() the token may be missing (but not invalid).
*/
@Injectable()
export class AuthGuard implements CanActivate {
constructor(
@@ -30,7 +37,13 @@ export class AuthGuard implements CanActivate {
// WebSocket subscriptions are authenticated once in connection_init.
if (!req.user) {
const token = extractBearerToken(req.headers['authorization']);
if (!token) throw new UnauthorizedException('Missing bearer token');
if (!token) {
if (
this.reflector.getAllAndOverride<boolean>(IS_OPTIONAL_AUTH, targets)
)
return true;
throw new UnauthorizedException('Missing bearer token');
}
try {
req.user = await this.identity.resolveSession(token);
req.accessToken = token;

View File

@@ -9,7 +9,10 @@ Contrato del event store y reconstrucción de agregados.
- `EventStore` (puerto): `persist` y `getEventsByStreamId`.
- `AggregateRehydrator`: carga los eventos de un stream y devuelve el agregado reconstruido, ya enlazado con el `EventPublisher`.
- `SharedModule.withInfrastructure(...)`: módulo global que expone el event store y el rehydrator a todos los contextos.
- `EventRecorder`: guarda un evento en un stream que no es de ningún agregado propio (p. ej. un usuario de Pireon), en
la siguiente posición, para que otros contextos reaccionen a él.
- `SharedModule.withInfrastructure(...)`: módulo global que expone el event store, el rehydrator y el recorder a todos
los contextos.
## Comandos

View File

@@ -3,10 +3,7 @@
"$schema": "../../../../node_modules/nx/schemas/project-schema.json",
"sourceRoot": "libs/api/shared/application/src",
"projectType": "library",
"tags": [
"scope:api-shared",
"type:application"
],
"tags": ["scope:api-shared", "type:application"],
"// targets": "to see all targets run: nx show project api-shared-application --web",
"targets": {}
}

View File

@@ -1,3 +1,4 @@
export * from './lib/ports/event-store';
export * from './lib/aggregate-rehydrator';
export * from './lib/shared.module';
export * from './lib/event-recorder';

View File

@@ -0,0 +1,35 @@
import { Injectable } from '@nestjs/common';
import { ConcurrencyError, DomainEvent } from '@task-stream/api/shared/domain';
import { EventStore } from './ports/event-store';
/**
* Appends a domain event to a stream that no aggregate of ours owns (e.g. a
* user, owned by Pireon), at the next position. Like any stored event, it
* reaches the event handlers and sagas of every context through the events
* bridge.
*/
@Injectable()
export class EventRecorder {
constructor(private readonly eventStore: EventStore) {}
async record(streamId: string, event: DomainEvent): Promise<void> {
for (let attempt = 1; ; attempt++) {
const position =
(await this.eventStore.getEventsByStreamId(streamId)).length + 1;
try {
await this.eventStore.persist({
streamId,
position,
type: event.constructor.name,
data: JSON.parse(
JSON.stringify({ ...event, aggregateVersion: position }),
),
});
return;
} catch (error) {
// Someone appended meanwhile: take the next position.
if (!(error instanceof ConcurrencyError) || attempt === 3) throw error;
}
}
}
}

View File

@@ -1,11 +1,12 @@
import { DynamicModule, Global, Module, Type } from '@nestjs/common';
import { CqrsModule } from '@nestjs/cqrs';
import { AggregateRehydrator } from './aggregate-rehydrator';
import { EventRecorder } from './event-recorder';
/**
* Shared building blocks for every bounded context: the event store (from
* the infrastructure module plugged in by the composition root) and the
* AggregateRehydrator.
* the infrastructure module plugged in by the composition root), the
* AggregateRehydrator and the EventRecorder.
*/
@Global()
@Module({})
@@ -16,8 +17,8 @@ export class SharedModule {
return {
module: SharedModule,
imports: [CqrsModule, infrastructureModule],
providers: [AggregateRehydrator],
exports: [infrastructureModule, AggregateRehydrator],
providers: [AggregateRehydrator, EventRecorder],
exports: [infrastructureModule, AggregateRehydrator, EventRecorder],
};
}
}

View File

@@ -3,10 +3,7 @@
"$schema": "../../../../node_modules/nx/schemas/project-schema.json",
"sourceRoot": "libs/api/shared/domain/src",
"projectType": "library",
"tags": [
"scope:api-shared",
"type:domain"
],
"tags": ["scope:api-shared", "type:domain"],
"// targets": "to see all targets run: nx show project api-shared-domain --web",
"targets": {}
}

View File

@@ -10,7 +10,8 @@ Event store en PostgreSQL y el camino de los eventos hasta la lectura.
- `PostgresEventStore` (tabla `events`, `UNIQUE(stream_id, position)` → `ConcurrencyError`).
- `EventSerializer` / `EventDeserializer` y `EventStorePublisher` (sustituye al publisher del `EventBus`: `commit()` persiste).
- `EventStoreRelay`: copia los eventos nuevos al Redis Stream (`FOR UPDATE SKIP LOCKED`).
- `EventsBridge`: consume el stream y entrega cada evento a sus `@EventsHandler`, en orden y esperando (ACK al final).
- `EventsBridge`: consume el stream y entrega cada evento a sus `@EventsHandler`, en orden y esperando (ACK al final);
después ejecuta los `@Saga()` (el `EventBus` en memoria nunca ve los eventos guardados) y sus comandos, uno a uno.
- Conexiones Redis (`REDIS_CLIENT`, `GRAPHQL_PUB_SUB`) y `SharedInfrastructureModule.forRootAsync`.
Flujo completo: [Arquitectura](../../../../docs/02-arquitectura.md#22-flujo-completo-de-una-escritura).

View File

@@ -3,10 +3,7 @@
"$schema": "../../../../node_modules/nx/schemas/project-schema.json",
"sourceRoot": "libs/api/shared/infrastructure/src",
"projectType": "library",
"tags": [
"scope:api-shared",
"type:infrastructure"
],
"tags": ["scope:api-shared", "type:infrastructure"],
"// targets": "to see all targets run: nx show project api-shared-infrastructure --web",
"targets": {}
}

View File

@@ -7,9 +7,20 @@ import {
Type,
} from '@nestjs/common';
import { DiscoveryService } from '@nestjs/core';
import { IEventHandler } from '@nestjs/cqrs';
import { SerializableEvent } from '@task-stream/api/shared/domain';
import {
CommandBus,
ICommand,
IEvent,
IEventHandler,
ISaga,
} from '@nestjs/cqrs';
import {
ConcurrencyError,
DomainError,
SerializableEvent,
} from '@task-stream/api/shared/domain';
import { hostname } from 'node:os';
import { lastValueFrom, of, toArray } from 'rxjs';
import { EventStreamConsumer } from '../event-stream/event-stream.consumer';
import {
EVENT_STREAM_OPTIONS,
@@ -20,8 +31,12 @@ import { EventDeserializer } from './deserializers/event.deserializer';
export const EVENTS_BRIDGE_ENABLED = Symbol('EVENTS_BRIDGE_ENABLED');
/** Metadata written by @EventsHandler(...events) (@nestjs/cqrs). */
/** Metadata written by @EventsHandler(...events) and @Saga() (@nestjs/cqrs). */
const EVENTS_HANDLER_METADATA = '__eventsHandler__';
const SAGA_METADATA = '__saga__';
/** Attempts of a saga command that keeps hitting concurrent writes. */
const CONCURRENCY_ATTEMPTS = 5;
/**
* Redis stream → @EventsHandler. Plays the role of the event-store change
@@ -32,6 +47,13 @@ const EVENTS_HANDLER_METADATA = '__eventsHandler__';
* time, so events of an aggregate are projected in order, and the entry is
* acknowledged in Redis only once every handler succeeded (otherwise it
* stays pending and is retried).
*
* It also runs the `@Saga()`s (the in-memory EventBus never sees stored
* events, so it would not): after the handlers, each saga gets the event
* alone (`of(event)`), and the commands it emits are executed one by one,
* awaited. Commands must be idempotent (an event may be handled twice).
* A command refused by the domain (not found, business rule...) is logged
* and skipped, so it cannot block the stream; concurrent writes are retried.
*/
@Injectable()
export class EventsBridge
@@ -40,6 +62,7 @@ export class EventsBridge
private readonly logger = new Logger(EventsBridge.name);
private consumer?: EventStreamConsumer;
private readonly handlers = new Map<Type, IEventHandler[]>();
private readonly sagas: { name: string; saga: ISaga }[] = [];
constructor(
private readonly connections: RedisConnectionFactory,
@@ -47,6 +70,7 @@ export class EventsBridge
@Inject(EVENTS_BRIDGE_ENABLED) private readonly enabled: boolean,
private readonly eventDeserializer: EventDeserializer,
private readonly discovery: DiscoveryService,
private readonly commandBus: CommandBus,
) {}
async onApplicationBootstrap(): Promise<void> {
@@ -55,6 +79,7 @@ export class EventsBridge
return;
}
this.registerEventHandlers();
this.registerSagas();
this.consumer = new EventStreamConsumer(
this.connections.create('events-bridge'),
{
@@ -66,7 +91,7 @@ export class EventsBridge
);
await this.consumer.start();
this.logger.log(
`Bridging ${this.stream.streamKey} to ${this.handlers.size} event type(s)`,
`Bridging ${this.stream.streamKey} to ${this.handlers.size} event type(s) and ${this.sagas.length} saga(s)`,
);
}
@@ -90,11 +115,66 @@ export class EventsBridge
}
}
private registerSagas(): void {
for (const wrapper of this.discovery.getProviders()) {
if (!wrapper.metatype || !wrapper.instance) continue;
const keys: string[] =
Reflect.getMetadata(SAGA_METADATA, wrapper.metatype) ?? [];
const instance = wrapper.instance as Record<string, ISaga>;
for (const key of keys) {
this.sagas.push({
name: `${wrapper.metatype.name}.${key}`,
saga: instance[key].bind(instance),
});
}
}
}
private async handleEvent(stored: SerializableEvent): Promise<void> {
const { data: event } = this.eventDeserializer.deserialize(stored);
const eventCls = Object.getPrototypeOf(event).constructor as Type;
// Projections first: sagas may read the read model they just updated.
for (const handler of this.handlers.get(eventCls) ?? []) {
await handler.handle(event);
}
for (const { name, saga } of this.sagas) {
const commands = await lastValueFrom(
saga(of(event as IEvent)).pipe(toArray()),
);
for (const command of commands) {
await this.execute(name, event as IEvent, command);
}
}
}
private async execute(
saga: string,
event: IEvent,
command: ICommand,
): Promise<void> {
for (let attempt = 1; ; attempt++) {
try {
await this.commandBus.execute(command);
return;
} catch (error) {
if (
error instanceof ConcurrencyError &&
attempt < CONCURRENCY_ATTEMPTS
) {
await new Promise((resolve) => setTimeout(resolve, 50 * attempt));
continue;
}
if (
error instanceof DomainError &&
!(error instanceof ConcurrencyError)
) {
this.logger.warn(
`${saga}: ${command.constructor.name} after ${event.constructor.name} skipped: ${error.message}`,
);
return;
}
throw error;
}
}
}
}

View File

@@ -3,10 +3,7 @@
"$schema": "../../../../node_modules/nx/schemas/project-schema.json",
"sourceRoot": "libs/api/shared/presenters/src",
"projectType": "library",
"tags": [
"scope:api-shared",
"type:presenters"
],
"tags": ["scope:api-shared", "type:presenters"],
"// targets": "to see all targets run: nx show project api-shared-presenters --web",
"targets": {}
}

View File

@@ -10,10 +10,14 @@ Casos de uso del tracker (CQRS).
- Services: `ProjectsService`, `IssuesService`, `CommentsService`, `ReleasesService`, `ChangesService` (subscriptions),
`WorkflowsService` (workflow de cada proyecto para presentar tickets, con una caché de 2 s).
- `commands/`: un fichero por command y otro por handler; cargan con `WriteAccess` (rehydrator) y guardan con `commitAndWait()`.
- `queries/`: un fichero por query y otro por handler; leen de los puertos `Find*Repository` y comprueban la membresía (`ReadAccess`);
- `queries/`: un fichero por query y otro por handler; leen de los puertos `Find*Repository` y comprueban el acceso
(`ReadAccess`: miembros, o cualquiera si el proyecto es público; `viewerId` null = anónimo);
`SearchMyIssuesQuery` y `GetUpcomingReleasesQuery` cruzan todos los proyectos del usuario («Mi trabajo»).
- `event-handlers/`: proyecciones (`@EventsHandler`) hacia el read model (incluida la línea temporal de cada ticket),
notificaciones en tiempo real y feed de actividad.
- `sagas/`: `ProjectDeletionSaga` (cascada al borrar un proyecto y sus tickets), `ReleaseArchivedSaga` (los tickets sin
terminar salen de una release archivada) y `UserDeactivatedSaga` (desasigna los tickets abiertos de un usuario que
Pireon desactiva). Las ejecuta el `EventsBridge`; sus comandos son internos e idempotentes. Tests en `sagas.spec.ts`.
- `analytics/`: estadísticas y burnup calculados con funciones puras sobre la línea temporal (con tests).
- `ports/`: lookups de escritura, `Find*`, `UpsertMaterialized*`, `ReleaseIssuesRepository`, `ChangeFeed`, `UserDirectory`.
- `TrackerApplicationModule.withInfrastructure(...)`.

View File

@@ -18,6 +18,13 @@ import { CreateReleaseCommandHandler } from './create-release.command-handler';
import { UpdateReleaseCommandHandler } from './update-release.command-handler';
import { PublishReleaseCommandHandler } from './publish-release.command-handler';
import { ArchiveReleaseCommandHandler } from './archive-release.command-handler';
import { DeleteProjectCommandHandler } from './delete-project.command-handler';
import { DeleteIssueCommandHandler } from './delete-issue.command-handler';
import { DeleteReleaseCommandHandler } from './delete-release.command-handler';
import { PurgeCommentCommandHandler } from './purge-comment.command-handler';
import { DropLinksToIssueCommandHandler } from './drop-links-to-issue.command-handler';
import { DetachIssueFromReleaseCommandHandler } from './detach-issue-from-release.command-handler';
import { UnassignIssueCommandHandler } from './unassign-issue.command-handler';
export const COMMAND_HANDLERS = [
CreateProjectCommandHandler,
@@ -40,4 +47,12 @@ export const COMMAND_HANDLERS = [
UpdateReleaseCommandHandler,
PublishReleaseCommandHandler,
ArchiveReleaseCommandHandler,
DeleteProjectCommandHandler,
// Emitted by sagas only (cascades and reactions to other aggregates).
DeleteIssueCommandHandler,
DeleteReleaseCommandHandler,
PurgeCommentCommandHandler,
DropLinksToIssueCommandHandler,
DetachIssueFromReleaseCommandHandler,
UnassignIssueCommandHandler,
];

View File

@@ -19,6 +19,7 @@ export class CreateProjectCommandHandler
key: command.key,
name: command.name,
description: command.description,
isPublic: command.isPublic,
ownerId: command.actorId,
});
await this.projectKeys.reserve(project.key, project.id);

View File

@@ -7,6 +7,7 @@ export class CreateProjectCommand extends Command<CommandResult> {
public readonly key: string,
public readonly name: string,
public readonly description?: string | null,
public readonly isPublic = false,
) {
super();
}

View File

@@ -0,0 +1,18 @@
import { CommandHandler, ICommandHandler } from '@nestjs/cqrs';
import { Issue } from '@task-stream/api/tracker/domain';
import { DeleteIssueCommand } from './delete-issue.command';
import { WriteAccess } from './write-access';
@CommandHandler(DeleteIssueCommand)
export class DeleteIssueCommandHandler
implements ICommandHandler<DeleteIssueCommand>
{
constructor(private readonly access: WriteAccess) {}
async execute(command: DeleteIssueCommand): Promise<void> {
const issue = await this.access.live(command.issueId, Issue);
if (!issue) return;
issue.delete(command.actorId);
await issue.commitAndWait();
}
}

View File

@@ -0,0 +1,11 @@
import { Command } from '@nestjs/cqrs';
/** Saga only: deletes a ticket of a deleted project. */
export class DeleteIssueCommand extends Command<void> {
constructor(
public readonly actorId: string,
public readonly issueId: string,
) {
super();
}
}

View File

@@ -0,0 +1,25 @@
import { CommandHandler, ICommandHandler } from '@nestjs/cqrs';
import { ProjectKeysRepository } from '../ports/project-keys.repository';
import { CommandResult, ack } from './command-result';
import { DeleteProjectCommand } from './delete-project.command';
import { WriteAccess } from './write-access';
/** Tickets, comments and releases follow through ProjectDeletionSaga. */
@CommandHandler(DeleteProjectCommand)
export class DeleteProjectCommandHandler
implements ICommandHandler<DeleteProjectCommand>
{
constructor(
private readonly access: WriteAccess,
private readonly projectKeys: ProjectKeysRepository,
) {}
async execute(command: DeleteProjectCommand): Promise<CommandResult> {
const project = await this.access.project(command.projectId);
project.delete(command.actorId, command.confirmKey);
await project.commitAndWait();
// The key can be used again by a new project.
await this.projectKeys.release(project.key, project.id);
return ack(project);
}
}

View File

@@ -0,0 +1,13 @@
import { Command } from '@nestjs/cqrs';
import { CommandResult } from './command-result';
/** The owner deletes a project, typing its key to confirm. */
export class DeleteProjectCommand extends Command<CommandResult> {
constructor(
public readonly actorId: string,
public readonly projectId: string,
public readonly confirmKey: string,
) {
super();
}
}

View File

@@ -0,0 +1,18 @@
import { CommandHandler, ICommandHandler } from '@nestjs/cqrs';
import { Release } from '@task-stream/api/tracker/domain';
import { DeleteReleaseCommand } from './delete-release.command';
import { WriteAccess } from './write-access';
@CommandHandler(DeleteReleaseCommand)
export class DeleteReleaseCommandHandler
implements ICommandHandler<DeleteReleaseCommand>
{
constructor(private readonly access: WriteAccess) {}
async execute(command: DeleteReleaseCommand): Promise<void> {
const release = await this.access.live(command.releaseId, Release);
if (!release) return;
release.delete(command.actorId);
await release.commitAndWait();
}
}

View File

@@ -0,0 +1,11 @@
import { Command } from '@nestjs/cqrs';
/** Saga only: deletes a release of a deleted project. */
export class DeleteReleaseCommand extends Command<void> {
constructor(
public readonly actorId: string,
public readonly releaseId: string,
) {
super();
}
}

View File

@@ -0,0 +1,21 @@
import { CommandHandler, ICommandHandler } from '@nestjs/cqrs';
import { Issue, Project } from '@task-stream/api/tracker/domain';
import { DetachIssueFromReleaseCommand } from './detach-issue-from-release.command';
import { WriteAccess } from './write-access';
@CommandHandler(DetachIssueFromReleaseCommand)
export class DetachIssueFromReleaseCommandHandler
implements ICommandHandler<DetachIssueFromReleaseCommand>
{
constructor(private readonly access: WriteAccess) {}
async execute(command: DetachIssueFromReleaseCommand): Promise<void> {
const issue = await this.access.live(command.issueId, Issue);
if (!issue || issue.releaseId !== command.releaseId) return;
const project = await this.access.live(issue.projectId, Project);
// Finished tickets stay: they are what the release shipped.
if (!project || project.workflow.isFinished(issue.status)) return;
issue.setRelease(command.actorId, null);
await issue.commitAndWait();
}
}

View File

@@ -0,0 +1,12 @@
import { Command } from '@nestjs/cqrs';
/** Saga only: takes an unfinished ticket out of an archived release. */
export class DetachIssueFromReleaseCommand extends Command<void> {
constructor(
public readonly actorId: string,
public readonly issueId: string,
public readonly releaseId: string,
) {
super();
}
}

View File

@@ -0,0 +1,19 @@
import { CommandHandler, ICommandHandler } from '@nestjs/cqrs';
import { Issue, Project } from '@task-stream/api/tracker/domain';
import { DropLinksToIssueCommand } from './drop-links-to-issue.command';
import { WriteAccess } from './write-access';
@CommandHandler(DropLinksToIssueCommand)
export class DropLinksToIssueCommandHandler
implements ICommandHandler<DropLinksToIssueCommand>
{
constructor(private readonly access: WriteAccess) {}
async execute(command: DropLinksToIssueCommand): Promise<void> {
const issue = await this.access.live(command.issueId, Issue);
// Gone, or about to be (its project was deleted too): nothing to tidy.
if (!issue || !(await this.access.live(issue.projectId, Project))) return;
issue.unlinkAllTo(command.actorId, command.targetIssueId);
await issue.commitAndWait();
}
}

View File

@@ -0,0 +1,12 @@
import { Command } from '@nestjs/cqrs';
/** Saga only: removes the links of `issueId` that point to a deleted ticket. */
export class DropLinksToIssueCommand extends Command<void> {
constructor(
public readonly actorId: string,
public readonly issueId: string,
public readonly targetIssueId: string,
) {
super();
}
}

View File

@@ -19,3 +19,10 @@ export * from './update-issue.command';
export * from './update-project.command';
export * from './update-project-settings.command';
export * from './update-release.command';
export * from './delete-project.command';
export * from './delete-issue.command';
export * from './delete-release.command';
export * from './purge-comment.command';
export * from './drop-links-to-issue.command';
export * from './detach-issue-from-release.command';
export * from './unassign-issue.command';

View File

@@ -0,0 +1,18 @@
import { CommandHandler, ICommandHandler } from '@nestjs/cqrs';
import { Comment } from '@task-stream/api/tracker/domain';
import { PurgeCommentCommand } from './purge-comment.command';
import { WriteAccess } from './write-access';
@CommandHandler(PurgeCommentCommand)
export class PurgeCommentCommandHandler
implements ICommandHandler<PurgeCommentCommand>
{
constructor(private readonly access: WriteAccess) {}
async execute(command: PurgeCommentCommand): Promise<void> {
const comment = await this.access.live(command.commentId, Comment);
if (!comment || comment.deleted) return;
comment.delete(command.actorId, true);
await comment.commitAndWait();
}
}

View File

@@ -0,0 +1,11 @@
import { Command } from '@nestjs/cqrs';
/** Saga only: deletes a comment of a deleted ticket. */
export class PurgeCommentCommand extends Command<void> {
constructor(
public readonly actorId: string,
public readonly commentId: string,
) {
super();
}
}

View File

@@ -0,0 +1,19 @@
import { CommandHandler, ICommandHandler } from '@nestjs/cqrs';
import { Issue } from '@task-stream/api/tracker/domain';
import { UnassignIssueCommand } from './unassign-issue.command';
import { WriteAccess } from './write-access';
@CommandHandler(UnassignIssueCommand)
export class UnassignIssueCommandHandler
implements ICommandHandler<UnassignIssueCommand>
{
constructor(private readonly access: WriteAccess) {}
async execute(command: UnassignIssueCommand): Promise<void> {
const issue = await this.access.live(command.issueId, Issue);
// Someone may have reassigned it meanwhile: only undo our user's assignment.
if (!issue || issue.assigneeId !== command.userId) return;
issue.assign(command.actorId, null);
await issue.commitAndWait();
}
}

View File

@@ -0,0 +1,12 @@
import { Command } from '@nestjs/cqrs';
/** Saga only: unassigns a ticket if it is still assigned to `userId` (deactivated in Pireon). */
export class UnassignIssueCommand extends Command<void> {
constructor(
public readonly actorId: string,
public readonly issueId: string,
public readonly userId: string,
) {
super();
}
}

View File

@@ -5,7 +5,11 @@ export class UpdateProjectCommand extends Command<CommandResult> {
constructor(
public readonly actorId: string,
public readonly projectId: string,
public readonly changes: { name?: string; description?: string | null },
public readonly changes: {
name?: string;
description?: string | null;
isPublic?: boolean;
},
) {
super();
}

View File

@@ -13,6 +13,11 @@ import {
} from '@task-stream/api/tracker/domain';
import { UserDirectory } from '../ports/user-directory';
type Aggregate = VersionedAggregateRoot & {
createdAt?: string;
deleted?: boolean;
};
/** Loads aggregates from the event store, failing with domain errors. */
@Injectable()
export class WriteAccess {
@@ -78,13 +83,31 @@ export class WriteAccess {
}
}
private async load<T extends VersionedAggregateRoot & { createdAt?: string }>(
/**
* For saga commands, which must be idempotent: the aggregate, or null when
* it does not exist or was deleted (nothing left to do).
*/
async live<T extends Aggregate>(
id: string,
AggregateCls: Type<T>,
): Promise<T | null> {
try {
return await this.load(id, AggregateCls);
} catch (error) {
if (error instanceof NotFoundError) return null;
throw error;
}
}
private async load<T extends Aggregate>(
id: string,
AggregateCls: Type<T>,
): Promise<T> {
const aggregate = await this.rehydrator.rehydrate(id, AggregateCls);
// The stream exists but belongs to another kind of aggregate.
if (!aggregate.createdAt) throw new NotFoundError(AggregateCls.name, id);
// The stream exists but belongs to another kind of aggregate, or the
// aggregate was deleted: same answer, it is not there.
if (!aggregate.createdAt || aggregate.deleted)
throw new NotFoundError(AggregateCls.name, id);
return aggregate;
}
}

View File

@@ -38,7 +38,7 @@ export class CommentsService {
}
list(
viewerId: string,
viewerId: string | null,
issueId: string,
limit?: number,
offset?: number,

View File

@@ -22,6 +22,8 @@ import {
ReleaseUpdatedEvent,
TrackerEvent,
DEFAULT_PROJECT_SETTINGS,
SYSTEM_ACTOR_ID,
SYSTEM_ACTOR_NAME,
ProjectSettingsChangedEvent,
ProjectWorkflow,
} from '@task-stream/api/tracker/domain';
@@ -80,12 +82,16 @@ export class ActivityEventHandler implements IEventHandler<TrackerEvent> {
? await this.findReleases.findById(releaseId)
: null;
const project = await this.findProjects.findById(event.projectId);
// The project was deleted (cascade in progress): no feed to write to.
if (!project) return;
const mentioned = [
event.actorId,
(event as { userId?: string }).userId,
(event as { assigneeId?: string }).assigneeId,
];
const users = await this.users.refs(mentioned);
const actorOf = (id: string) =>
id === SYSTEM_ACTOR_ID ? { id, name: SYSTEM_ACTOR_NAME } : users.of(id);
const activity: ActivityReadModel = {
id: `${event.aggregateId}:${versionOf(event)}`,
@@ -93,9 +99,9 @@ export class ActivityEventHandler implements IEventHandler<TrackerEvent> {
projectId: event.projectId,
issueId,
issueKey,
actor: users.of(event.actorId),
actor: actorOf(event.actorId),
summary: describeEvent(event, {
actor: users.of(event.actorId).name,
actor: actorOf(event.actorId).name,
user: (id) => users.ofNullable(id)?.name ?? '?',
issueKey: issueKey ?? '',
projectKey: project?.key ?? '',

View File

@@ -21,10 +21,16 @@ import { ReleaseUpdatedEventHandler } from './release-updated.event-handler';
import { ReadModelNotifier } from './read-model-notifier';
import { IssueTimelineEventHandler } from './issue-timeline.event-handler';
import { ProjectSettingsChangedEventHandler } from './project-settings-changed.event-handler';
import { ProjectDeletedEventHandler } from './project-deleted.event-handler';
import { IssueDeletedEventHandler } from './issue-deleted.event-handler';
import { ReleaseDeletedEventHandler } from './release-deleted.event-handler';
/** Read-model projections (MongoDB) + realtime, fed by the EventsBridge. */
export const EVENT_HANDLERS = [
ActivityEventHandler,
ProjectDeletedEventHandler,
IssueDeletedEventHandler,
ReleaseDeletedEventHandler,
CommentAddedEventHandler,
CommentDeletedEventHandler,
CommentEditedEventHandler,

View File

@@ -0,0 +1,17 @@
import { EventsHandler, IEventHandler } from '@nestjs/cqrs';
import { IssueDeletedEvent } from '@task-stream/api/tracker/domain';
import { UpsertMaterializedIssueRepository } from '../ports/upsert-materialized-issue.repository';
/** Removes the ticket, its timeline and the links other tickets show to it. */
@EventsHandler(IssueDeletedEvent)
export class IssueDeletedEventHandler
implements IEventHandler<IssueDeletedEvent>
{
constructor(
private readonly upsertMaterializedIssueRepository: UpsertMaterializedIssueRepository,
) {}
async handle(event: IssueDeletedEvent) {
await this.upsertMaterializedIssueRepository.remove(event.aggregateId);
}
}

View File

@@ -25,6 +25,7 @@ export class ProjectCreatedEventHandler
key: project.key,
name: project.name,
description: project.description,
isPublic: project.isPublic ?? false,
owner: users.of(project.ownerId),
members: project.members.map((m) => ({
user: users.of(m.userId),

View File

@@ -0,0 +1,37 @@
import { EventsHandler, IEventHandler } from '@nestjs/cqrs';
import { ProjectDeletedEvent } from '@task-stream/api/tracker/domain';
import { ChangeFeed } from '../ports/change-feed';
import { FindProjectsRepository } from '../ports/find-projects.repository';
import { UpsertMaterializedProjectRepository } from '../ports/upsert-materialized-project.repository';
import { versionOf } from './read-model-notifier';
/**
* Removes the project (and its activity and timeline) from the read model,
* and tells its members, whose project lists drop it. Tickets, comments and
* releases go as the cascade saga deletes them.
*/
@EventsHandler(ProjectDeletedEvent)
export class ProjectDeletedEventHandler
implements IEventHandler<ProjectDeletedEvent>
{
constructor(
private readonly findProjects: FindProjectsRepository,
private readonly upsertMaterializedProjectRepository: UpsertMaterializedProjectRepository,
private readonly changeFeed: ChangeFeed,
) {}
async handle(event: ProjectDeletedEvent) {
const project = await this.findProjects.findById(event.projectId);
await this.upsertMaterializedProjectRepository.remove(event.projectId);
if (!project) return; // Already removed (event handled twice).
await this.changeFeed.publishProject(
{
action: event.constructor.name,
version: versionOf(event),
actorId: event.actorId,
project: { ...project, version: versionOf(event) },
},
event.memberIds,
);
}
}

View File

@@ -0,0 +1,16 @@
import { EventsHandler, IEventHandler } from '@nestjs/cqrs';
import { ReleaseDeletedEvent } from '@task-stream/api/tracker/domain';
import { UpsertMaterializedReleaseRepository } from '../ports/upsert-materialized-release.repository';
@EventsHandler(ReleaseDeletedEvent)
export class ReleaseDeletedEventHandler
implements IEventHandler<ReleaseDeletedEvent>
{
constructor(
private readonly upsertMaterializedReleaseRepository: UpsertMaterializedReleaseRepository,
) {}
async handle(event: ReleaseDeletedEvent) {
await this.upsertMaterializedReleaseRepository.remove(event.aggregateId);
}
}

View File

@@ -126,15 +126,18 @@ export class IssuesService {
);
}
findOne(viewerId: string, idOrKey: string): Promise<IssueReadModel> {
findOne(viewerId: string | null, idOrKey: string): Promise<IssueReadModel> {
return this.queryBus.execute(new GetIssueQuery(viewerId, idOrKey));
}
findMany(viewerId: string, ids: string[]): Promise<IssueReadModel[]> {
findMany(viewerId: string | null, ids: string[]): Promise<IssueReadModel[]> {
return this.queryBus.execute(new GetIssuesByIdsQuery(viewerId, ids));
}
search(viewerId: string, filter: IssueFilter): Promise<Page<IssueReadModel>> {
search(
viewerId: string | null,
filter: IssueFilter,
): Promise<Page<IssueReadModel>> {
return this.queryBus.execute(new SearchIssuesQuery(viewerId, filter));
}
@@ -147,7 +150,7 @@ export class IssuesService {
}
activity(
viewerId: string,
viewerId: string | null,
params: {
projectId: string;
issueId?: string;

View File

@@ -47,6 +47,8 @@ export abstract class FindIssuesRepository {
filter: CrossProjectIssueFilter,
): Promise<Page<IssueReadModel>>;
abstract stats(projectId: string): Promise<ProjectStatsReadModel>;
/** Open tickets (TODO, IN_PROGRESS or QA category) assigned to someone, in any project. */
abstract findOpenAssignedTo(userId: string): Promise<IssueReadModel[]>;
/** Every ticket of a project (statistics). */
abstract findByProject(projectId: string): Promise<IssueReadModel[]>;
abstract releaseProgress(

View File

@@ -8,4 +8,6 @@ export abstract class FindProjectsRepository {
projectId: string,
userId: string,
): Promise<ProjectRole | null>;
/** Projects anyone can read, by name. */
abstract findPublic(): Promise<ProjectReadModel[]>;
}

View File

@@ -16,3 +16,4 @@ export * from './user-directory';
export * from './release-issues.repository';
export * from './upsert-materialized-timeline.repository';
export * from './find-timeline.repository';
export * from './tracker-streams.repository';

View File

@@ -2,4 +2,6 @@
export abstract class ProjectKeysRepository {
/** Throws BusinessRuleError if the key is already taken. */
abstract reserve(key: string, projectId: string): Promise<void>;
/** Frees the key of a deleted project (only if still held by it). */
abstract release(key: string, projectId: string): Promise<void>;
}

View File

@@ -0,0 +1,13 @@
/**
* Write-side lookups over the event store (the source of truth, unlike the
* eventually-consistent read model): which streams belong to a project or a
* ticket. The cascade saga uses them to delete everything that hangs off a
* deleted project.
*/
export abstract class TrackerStreamsRepository {
abstract issueIdsOfProject(projectId: string): Promise<string[]>;
abstract releaseIdsOfProject(projectId: string): Promise<string[]>;
abstract commentIdsOfIssue(issueId: string): Promise<string[]>;
/** Tickets that were ever linked to `issueId` (they may have unlinked since). */
abstract issueIdsLinkingTo(issueId: string): Promise<string[]>;
}

View File

@@ -25,6 +25,8 @@ export abstract class UpsertMaterializedIssueRepository {
projectId: string,
categories: Record<string, StatusCategory>,
): Promise<void>;
/** The ticket was deleted: its view, its timeline and the links shown on other tickets go. */
abstract remove(issueId: string): Promise<void>;
/** Keeps the denormalised release name of every issue in sync. */
abstract renameRelease(releaseId: string, name: string): Promise<void>;
}

View File

@@ -6,4 +6,6 @@ export abstract class UpsertMaterializedProjectRepository {
project: Pick<ProjectReadModel, 'id' | 'version'> &
Partial<ProjectReadModel>,
): Promise<void>;
/** The project was deleted: its view, activity and timeline go. */
abstract remove(projectId: string): Promise<void>;
}

View File

@@ -6,4 +6,5 @@ export abstract class UpsertMaterializedReleaseRepository {
release: Pick<ReleaseReadModel, 'id' | 'version'> &
Partial<ReleaseReadModel>,
): Promise<void>;
abstract remove(releaseId: string): Promise<void>;
}

View File

@@ -12,6 +12,7 @@ import { AddProjectMemberCommand } from './commands/add-project-member.command';
import { ChangeProjectMemberRoleCommand } from './commands/change-project-member-role.command';
import { CommandResult } from './commands/command-result';
import { CreateProjectCommand } from './commands/create-project.command';
import { DeleteProjectCommand } from './commands/delete-project.command';
import { RemoveProjectMemberCommand } from './commands/remove-project-member.command';
import { UpdateProjectCommand } from './commands/update-project.command';
import { UpdateProjectSettingsCommand } from './commands/update-project-settings.command';
@@ -20,6 +21,7 @@ import { BoardColumn, GetBoardQuery } from './queries/get-board.query';
import { GetMyProjectsQuery } from './queries/get-my-projects.query';
import { GetProjectAnalyticsQuery } from './queries/get-project-analytics.query';
import { GetProjectStatsQuery } from './queries/get-project-stats.query';
import { GetPublicProjectsQuery } from './queries/get-public-projects.query';
import { GetProjectQuery } from './queries/get-project.query';
import { GetReleasesQuery } from './queries/get-releases.query';
@@ -32,7 +34,12 @@ export class ProjectsService {
create(
actorId: string,
input: { key: string; name: string; description?: string | null },
input: {
key: string;
name: string;
description?: string | null;
isPublic?: boolean;
},
): Promise<CommandResult> {
return this.commandBus.execute(
new CreateProjectCommand(
@@ -40,6 +47,7 @@ export class ProjectsService {
input.key,
input.name,
input.description,
input.isPublic ?? false,
),
);
}
@@ -47,13 +55,32 @@ export class ProjectsService {
update(
actorId: string,
projectId: string,
changes: { name?: string; description?: string | null },
changes: { name?: string; description?: string | null; isPublic?: boolean },
): Promise<CommandResult> {
return this.commandBus.execute(
new UpdateProjectCommand(actorId, projectId, changes),
);
}
/**
* Owner only, typing the key to confirm. Tickets, comments and releases
* are deleted right after by ProjectDeletionSaga; the key is freed.
*/
delete(
actorId: string,
projectId: string,
confirmKey: string,
): Promise<CommandResult> {
return this.commandBus.execute(
new DeleteProjectCommand(actorId, projectId, confirmKey),
);
}
/** Projects anyone can read, also without signing in. */
findPublic(): Promise<ProjectReadModel[]> {
return this.queryBus.execute(new GetPublicProjectsQuery());
}
/** Statuses, transitions, ticket types and areas (owners and admins). */
updateSettings(
actorId: string,
@@ -109,16 +136,19 @@ export class ProjectsService {
return this.queryBus.execute(new GetMyProjectsQuery(viewerId));
}
findOne(viewerId: string, idOrKey: string): Promise<ProjectReadModel> {
findOne(viewerId: string | null, idOrKey: string): Promise<ProjectReadModel> {
return this.queryBus.execute(new GetProjectQuery(viewerId, idOrKey));
}
stats(viewerId: string, projectId: string): Promise<ProjectStatsReadModel> {
stats(
viewerId: string | null,
projectId: string,
): Promise<ProjectStatsReadModel> {
return this.queryBus.execute(new GetProjectStatsQuery(viewerId, projectId));
}
board(
viewerId: string,
viewerId: string | null,
filter: Omit<IssueFilter, 'statuses' | 'limit' | 'offset'>,
perColumn?: number,
): Promise<BoardColumn[]> {
@@ -127,13 +157,16 @@ export class ProjectsService {
);
}
releases(viewerId: string, projectId: string): Promise<ReleaseReadModel[]> {
releases(
viewerId: string | null,
projectId: string,
): Promise<ReleaseReadModel[]> {
return this.queryBus.execute(new GetReleasesQuery(viewerId, projectId));
}
/** Statistics over the last `days` days (created vs resolved, flow, times, QA). */
analytics(
viewerId: string,
viewerId: string | null,
projectId: string,
days: number,
): Promise<ProjectAnalyticsReadModel> {

View File

@@ -2,7 +2,8 @@ import { Query } from '@nestjs/cqrs';
export class CanViewProjectQuery extends Query<boolean> {
constructor(
public readonly viewerId: string,
/** Null: anonymous (public projects only). */
public readonly viewerId: string | null,
public readonly projectId: string,
) {
super();

View File

@@ -3,7 +3,8 @@ import { ActivityReadModel } from '@task-stream/api/tracker/domain';
export class GetActivityQuery extends Query<ActivityReadModel[]> {
constructor(
public readonly viewerId: string,
/** Null: anonymous (public projects only). */
public readonly viewerId: string | null,
public readonly projectId: string,
public readonly issueId?: string,
public readonly limit = 50,

View File

@@ -12,7 +12,8 @@ export interface BoardColumn {
/** Kanban board: one column per status of the project (cancelled ones aside). */
export class GetBoardQuery extends Query<BoardColumn[]> {
constructor(
public readonly viewerId: string,
/** Null: anonymous (public projects only). */
public readonly viewerId: string | null,
public readonly filter: Omit<IssueFilter, 'statuses' | 'limit' | 'offset'>,
public readonly perColumn = 50,
) {

View File

@@ -4,7 +4,8 @@ import { Page } from '../ports/find-issues.repository';
export class GetCommentsQuery extends Query<Page<CommentReadModel>> {
constructor(
public readonly viewerId: string,
/** Null: anonymous (public projects only). */
public readonly viewerId: string | null,
public readonly issueId: string,
public readonly limit = 50,
public readonly offset = 0,

View File

@@ -3,7 +3,8 @@ import { IssueReadModel } from '@task-stream/api/tracker/domain';
export class GetIssueQuery extends Query<IssueReadModel> {
constructor(
public readonly viewerId: string,
/** Null: anonymous (public projects only). */
public readonly viewerId: string | null,
/** Issue id or key ("TS-42"). */
public readonly idOrKey: string,
) {

View File

@@ -1,7 +1,7 @@
import { IQueryHandler, QueryHandler } from '@nestjs/cqrs';
import { IssueReadModel } from '@task-stream/api/tracker/domain';
import { FindIssuesRepository } from '../ports/find-issues.repository';
import { FindProjectsRepository } from '../ports/find-projects.repository';
import { ReadAccess } from './read-access';
import { GetIssuesByIdsQuery } from './get-issues-by-ids.query';
@QueryHandler(GetIssuesByIdsQuery)
@@ -10,14 +10,19 @@ export class GetIssuesByIdsQueryHandler
{
constructor(
private readonly findIssues: FindIssuesRepository,
private readonly findProjects: FindProjectsRepository,
private readonly access: ReadAccess,
) {}
async execute(query: GetIssuesByIdsQuery): Promise<IssueReadModel[]> {
const issues = await this.findIssues.findByIds([...new Set(query.ids)]);
const visible = new Set(
(await this.findProjects.findByMember(query.viewerId)).map((p) => p.id),
);
return issues.filter((issue) => visible.has(issue.projectId));
// Linked tickets may sit in other projects: keep the ones the viewer can read.
const visible = new Map<string, boolean>();
for (const projectId of new Set(issues.map((issue) => issue.projectId))) {
visible.set(
projectId,
await this.access.canView(projectId, query.viewerId),
);
}
return issues.filter((issue) => visible.get(issue.projectId));
}
}

View File

@@ -3,7 +3,8 @@ import { IssueReadModel } from '@task-stream/api/tracker/domain';
export class GetIssuesByIdsQuery extends Query<IssueReadModel[]> {
constructor(
public readonly viewerId: string,
/** Null: anonymous (public projects only). */
public readonly viewerId: string | null,
public readonly ids: string[],
) {
super();

View File

@@ -4,7 +4,8 @@ import { ProjectAnalyticsReadModel } from '@task-stream/api/tracker/domain';
/** Statistics of a project over the last `days` days. */
export class GetProjectAnalyticsQuery extends Query<ProjectAnalyticsReadModel> {
constructor(
public readonly viewerId: string,
/** Null: anonymous (public projects only). */
public readonly viewerId: string | null,
public readonly projectId: string,
public readonly days: number,
) {

View File

@@ -3,7 +3,8 @@ import { ProjectStatsReadModel } from '@task-stream/api/tracker/domain';
export class GetProjectStatsQuery extends Query<ProjectStatsReadModel> {
constructor(
public readonly viewerId: string,
/** Null: anonymous (public projects only). */
public readonly viewerId: string | null,
public readonly projectId: string,
) {
super();

View File

@@ -3,7 +3,8 @@ import { ProjectReadModel } from '@task-stream/api/tracker/domain';
export class GetProjectQuery extends Query<ProjectReadModel> {
constructor(
public readonly viewerId: string,
/** Null: anonymous (public projects only). */
public readonly viewerId: string | null,
/** Project id or key ("TS"). */
public readonly idOrKey: string,
) {

View File

@@ -0,0 +1,15 @@
import { IQueryHandler, QueryHandler } from '@nestjs/cqrs';
import { ProjectReadModel } from '@task-stream/api/tracker/domain';
import { FindProjectsRepository } from '../ports/find-projects.repository';
import { GetPublicProjectsQuery } from './get-public-projects.query';
@QueryHandler(GetPublicProjectsQuery)
export class GetPublicProjectsQueryHandler
implements IQueryHandler<GetPublicProjectsQuery, ProjectReadModel[]>
{
constructor(private readonly findProjects: FindProjectsRepository) {}
execute(): Promise<ProjectReadModel[]> {
return this.findProjects.findPublic();
}
}

View File

@@ -0,0 +1,5 @@
import { Query } from '@nestjs/cqrs';
import { ProjectReadModel } from '@task-stream/api/tracker/domain';
/** Projects anyone can read (no sign-in needed). */
export class GetPublicProjectsQuery extends Query<ProjectReadModel[]> {}

View File

@@ -4,7 +4,8 @@ import { ReleaseBurnupReadModel } from '@task-stream/api/tracker/domain';
/** Scope vs finished tickets of a release, day by day. */
export class GetReleaseBurnupQuery extends Query<ReleaseBurnupReadModel> {
constructor(
public readonly viewerId: string,
/** Null: anonymous (public projects only). */
public readonly viewerId: string | null,
public readonly releaseId: string,
) {
super();

View File

@@ -3,7 +3,8 @@ import { ReleaseProgressReadModel } from '@task-stream/api/tracker/domain';
export class GetReleaseProgressQuery extends Query<ReleaseProgressReadModel> {
constructor(
public readonly viewerId: string,
/** Null: anonymous (public projects only). */
public readonly viewerId: string | null,
public readonly releaseId: string,
) {
super();

View File

@@ -3,7 +3,8 @@ import { ReleaseReadModel } from '@task-stream/api/tracker/domain';
export class GetReleaseQuery extends Query<ReleaseReadModel> {
constructor(
public readonly viewerId: string,
/** Null: anonymous (public projects only). */
public readonly viewerId: string | null,
public readonly releaseId: string,
) {
super();

View File

@@ -3,7 +3,8 @@ import { ReleaseReadModel } from '@task-stream/api/tracker/domain';
export class GetReleasesQuery extends Query<ReleaseReadModel[]> {
constructor(
public readonly viewerId: string,
/** Null: anonymous (public projects only). */
public readonly viewerId: string | null,
public readonly projectId: string,
) {
super();

View File

@@ -15,3 +15,4 @@ export * from './get-releases.query';
export * from './get-upcoming-releases.query';
export * from './search-issues.query';
export * from './search-my-issues.query';
export * from './get-public-projects.query';

View File

@@ -1,3 +1,4 @@
import { GetPublicProjectsQueryHandler } from './get-public-projects.query-handler';
import { CanViewProjectQueryHandler } from './can-view-project.query-handler';
import { GetActivityQueryHandler } from './get-activity.query-handler';
import { GetBoardQueryHandler } from './get-board.query-handler';
@@ -17,6 +18,7 @@ import { SearchIssuesQueryHandler } from './search-issues.query-handler';
import { SearchMyIssuesQueryHandler } from './search-my-issues.query-handler';
export const QUERY_HANDLERS = [
GetPublicProjectsQueryHandler,
CanViewProjectQueryHandler,
GetActivityQueryHandler,
GetBoardQueryHandler,

View File

@@ -3,23 +3,38 @@ import { NotFoundError } from '@task-stream/api/shared/domain';
import { ProjectRole } from '@task-stream/api/tracker/domain';
import { FindProjectsRepository } from '../ports/find-projects.repository';
/** Authorisation for the read side, based on the projected memberships. */
/**
* Authorisation for the read side, based on the projected memberships:
* members read their projects; anyone (also anonymous, `viewerId` null)
* reads public ones.
*/
@Injectable()
export class ReadAccess {
constructor(private readonly findProjects: FindProjectsRepository) {}
/** The viewer's role, or null when they read a public project as an outsider. */
async assertCanView(
projectId: string,
viewerId: string,
): Promise<ProjectRole> {
const role = await this.findProjects.findRole(projectId, viewerId);
viewerId: string | null,
): Promise<ProjectRole | null> {
const role = viewerId
? await this.findProjects.findRole(projectId, viewerId)
: null;
if (role) return role;
const project = await this.findProjects.findById(projectId);
// Same answer for "doesn't exist" and "not yours": don't leak projects.
if (!role) throw new NotFoundError('Project', projectId);
return role;
if (!project?.isPublic) throw new NotFoundError('Project', projectId);
return null;
}
async canView(projectId: string, viewerId: string): Promise<boolean> {
return (await this.findProjects.findRole(projectId, viewerId)) !== null;
async canView(projectId: string, viewerId: string | null): Promise<boolean> {
try {
await this.assertCanView(projectId, viewerId);
return true;
} catch (error) {
if (error instanceof NotFoundError) return false;
throw error;
}
}
assertFound<T>(value: T | null, entity: string, id: string): T {

View File

@@ -4,7 +4,8 @@ import { IssueFilter, Page } from '../ports/find-issues.repository';
export class SearchIssuesQuery extends Query<Page<IssueReadModel>> {
constructor(
public readonly viewerId: string,
/** Null: anonymous (public projects only). */
public readonly viewerId: string | null,
public readonly filter: IssueFilter,
) {
super();

View File

@@ -70,7 +70,10 @@ export class ReleasesService {
);
}
findOne(viewerId: string, releaseId: string): Promise<ReleaseReadModel> {
findOne(
viewerId: string | null,
releaseId: string,
): Promise<ReleaseReadModel> {
return this.queryBus.execute(new GetReleaseQuery(viewerId, releaseId));
}
@@ -80,7 +83,7 @@ export class ReleasesService {
}
progress(
viewerId: string,
viewerId: string | null,
releaseId: string,
): Promise<ReleaseProgressReadModel> {
return this.queryBus.execute(
@@ -89,7 +92,10 @@ export class ReleasesService {
}
/** Scope vs finished tickets of the release, day by day. */
burnup(viewerId: string, releaseId: string): Promise<ReleaseBurnupReadModel> {
burnup(
viewerId: string | null,
releaseId: string,
): Promise<ReleaseBurnupReadModel> {
return this.queryBus.execute(
new GetReleaseBurnupQuery(viewerId, releaseId),
);

View File

@@ -0,0 +1,86 @@
import { Injectable } from '@nestjs/common';
import { ICommand, IEvent, ofType, Saga } from '@nestjs/cqrs';
import {
IssueDeletedEvent,
ProjectDeletedEvent,
} from '@task-stream/api/tracker/domain';
import { from, mergeMap, Observable } from 'rxjs';
import { DeleteIssueCommand } from '../commands/delete-issue.command';
import { DeleteReleaseCommand } from '../commands/delete-release.command';
import { DropLinksToIssueCommand } from '../commands/drop-links-to-issue.command';
import { PurgeCommentCommand } from '../commands/purge-comment.command';
import { TrackerStreamsRepository } from '../ports/tracker-streams.repository';
/**
* Deleting a project cascades, one aggregate per command (each stream ends
* with its own "deleted" event):
*
* ProjectDeleted → DeleteIssue (each ticket) + DeleteRelease (each release)
* IssueDeleted → PurgeComment (each comment) + DropLinksToIssue (tickets
* of other projects that pointed to it)
*
* Who to delete comes from the event store, not the read model. The changes
* are attributed to whoever deleted the project.
*/
@Injectable()
export class ProjectDeletionSaga {
constructor(private readonly streams: TrackerStreamsRepository) {}
@Saga()
projectDeleted = (events$: Observable<IEvent>): Observable<ICommand> =>
events$.pipe(
ofType(ProjectDeletedEvent),
mergeMap((event) =>
from(this.contentsOf(event.projectId)).pipe(
mergeMap(({ issueIds, releaseIds }) => [
...issueIds.map((id) => new DeleteIssueCommand(event.actorId, id)),
...releaseIds.map(
(id) => new DeleteReleaseCommand(event.actorId, id),
),
]),
),
),
);
@Saga()
issueDeleted = (events$: Observable<IEvent>): Observable<ICommand> =>
events$.pipe(
ofType(IssueDeletedEvent),
mergeMap((event) =>
from(this.dependentsOf(event.aggregateId)).pipe(
mergeMap(({ commentIds, linkingIds }) => [
...commentIds.map(
(id) => new PurgeCommentCommand(event.actorId, id),
),
...linkingIds.map(
(id) =>
new DropLinksToIssueCommand(
event.actorId,
id,
event.aggregateId,
),
),
]),
),
),
);
private async contentsOf(projectId: string) {
const [issueIds, releaseIds] = await Promise.all([
this.streams.issueIdsOfProject(projectId),
this.streams.releaseIdsOfProject(projectId),
]);
return { issueIds, releaseIds };
}
private async dependentsOf(issueId: string) {
const [commentIds, linkingIds] = await Promise.all([
this.streams.commentIdsOfIssue(issueId),
this.streams.issueIdsLinkingTo(issueId),
]);
return {
commentIds,
linkingIds: linkingIds.filter((id) => id !== issueId),
};
}
}

View File

@@ -0,0 +1,36 @@
import { Injectable } from '@nestjs/common';
import { ICommand, IEvent, ofType, Saga } from '@nestjs/cqrs';
import { ReleaseArchivedEvent } from '@task-stream/api/tracker/domain';
import { from, mergeMap, Observable } from 'rxjs';
import { DetachIssueFromReleaseCommand } from '../commands/detach-issue-from-release.command';
import { ReleaseIssuesRepository } from '../ports/release-issues.repository';
/**
* An archived release no longer accepts changes, so its unfinished tickets
* would be stuck in it: they leave it (back to "no release"), attributed to
* whoever archived it. Finished tickets stay, as the record of what shipped.
*/
@Injectable()
export class ReleaseArchivedSaga {
constructor(private readonly releaseIssues: ReleaseIssuesRepository) {}
@Saga()
releaseArchived = (events$: Observable<IEvent>): Observable<ICommand> =>
events$.pipe(
ofType(ReleaseArchivedEvent),
mergeMap((event) =>
from(this.releaseIssues.findIssueIds(event.aggregateId)).pipe(
mergeMap((issueIds) =>
issueIds.map(
(issueId) =>
new DetachIssueFromReleaseCommand(
event.actorId,
issueId,
event.aggregateId,
),
),
),
),
),
);
}

View File

@@ -0,0 +1,237 @@
import {
CommandBus,
CqrsModule,
EventBus,
ICommand,
ISaga,
} from '@nestjs/cqrs';
import { Test } from '@nestjs/testing';
import {
AggregateRehydrator,
EventStore,
} from '@task-stream/api/shared/application';
import { ForbiddenError, NotFoundError } from '@task-stream/api/shared/domain';
import { UserDeactivatedEvent } from '@task-stream/api/identity/domain';
import {
Comment,
Issue,
IssueLinkType,
ProjectRole,
Release,
TRACKER_FACTORIES,
} from '@task-stream/api/tracker/domain';
import { lastValueFrom, of, toArray } from 'rxjs';
import {
FakeUserDirectory,
InMemoryEventStore,
InMemoryEventStorePublisher,
InMemoryIssueNumbers,
InMemoryProjectKeys,
InMemoryReleaseIssues,
InMemoryReleaseNames,
InMemoryTrackerStreams,
} from '../../testing/in-memory';
import { AddCommentCommand } from '../commands/add-comment.command';
import { AddProjectMemberCommand } from '../commands/add-project-member.command';
import { ArchiveReleaseCommand } from '../commands/archive-release.command';
import { ChangeIssueStatusCommand } from '../commands/change-issue-status.command';
import { COMMAND_HANDLERS } from '../commands/command-handlers';
import { CommandResult } from '../commands/command-result';
import { CreateIssueCommand } from '../commands/create-issue.command';
import { CreateProjectCommand } from '../commands/create-project.command';
import { CreateReleaseCommand } from '../commands/create-release.command';
import { DeleteProjectCommand } from '../commands/delete-project.command';
import { LinkIssuesCommand } from '../commands/link-issues.command';
import { AssignIssueCommand } from '../commands/assign-issue.command';
import { WriteAccess } from '../commands/write-access';
import { FindIssuesRepository } from '../ports/find-issues.repository';
import { IssueNumbersRepository } from '../ports/issue-numbers.repository';
import { ProjectKeysRepository } from '../ports/project-keys.repository';
import { ReleaseIssuesRepository } from '../ports/release-issues.repository';
import { ReleaseNamesRepository } from '../ports/release-names.repository';
import { TrackerStreamsRepository } from '../ports/tracker-streams.repository';
import { UserDirectory } from '../ports/user-directory';
import { ProjectDeletionSaga } from './project-deletion.saga';
import { ReleaseArchivedSaga } from './release-archived.saga';
import { UserDeactivatedSaga } from './user-deactivated.saga';
async function setup() {
const store = new InMemoryEventStore();
const users = new FakeUserDirectory().add('owner').add('dev').add('qa');
// Read model double for the user saga: open tickets are tracked by the test.
const openAssigned = new Map<string, string[]>();
const moduleRef = await Test.createTestingModule({
imports: [CqrsModule.forRoot()],
providers: [
...COMMAND_HANDLERS,
...TRACKER_FACTORIES,
ProjectDeletionSaga,
ReleaseArchivedSaga,
UserDeactivatedSaga,
WriteAccess,
AggregateRehydrator,
{ provide: EventStore, useValue: store },
{ provide: UserDirectory, useValue: users },
{ provide: ProjectKeysRepository, useClass: InMemoryProjectKeys },
{ provide: IssueNumbersRepository, useClass: InMemoryIssueNumbers },
{ provide: ReleaseNamesRepository, useClass: InMemoryReleaseNames },
{
provide: ReleaseIssuesRepository,
useValue: new InMemoryReleaseIssues(store),
},
{
provide: TrackerStreamsRepository,
useValue: new InMemoryTrackerStreams(store),
},
{
provide: FindIssuesRepository,
useValue: {
findOpenAssignedTo: async (userId: string) =>
(openAssigned.get(userId) ?? []).map((id) => ({ id })),
},
},
],
}).compile();
await moduleRef.init();
InMemoryEventStorePublisher.install(moduleRef.get(EventBus), store);
const commandBus = moduleRef.get(CommandBus);
const run = (command: object): Promise<CommandResult> =>
commandBus.execute(command as never);
const sagas: ISaga[] = [
...Object.values(moduleRef.get(ProjectDeletionSaga)),
...Object.values(moduleRef.get(ReleaseArchivedSaga)),
...Object.values(moduleRef.get(UserDeactivatedSaga)),
].filter((value): value is ISaga => typeof value === 'function');
/** What the events bridge does: every stored event through every saga, until quiet. */
let cursor = 0;
const react = async (event: object) => {
for (const saga of sagas) {
const commands = await lastValueFrom(saga(of(event)).pipe(toArray()));
for (const command of commands)
await commandBus.execute(command as ICommand);
}
};
const drain = async () => {
while (cursor < store.events.length) {
const stored = store.events[cursor++];
const [event] = (await store.getEventsByStreamId(stored.streamId))
.filter((e) => e.position === stored.position)
.map((e) => e.data);
await react(event);
}
};
const access = moduleRef.get(WriteAccess);
const rehydrator = moduleRef.get(AggregateRehydrator);
return { run, drain, react, access, rehydrator, openAssigned };
}
const ticket = (projectId: string, title: string) =>
new CreateIssueCommand('owner', projectId, 'TASK', title, {});
describe('sagas', () => {
it('deleting a project cascades to its tickets, comments, releases and links', async () => {
const { run, drain, access, rehydrator } = await setup();
const web = (await run(new CreateProjectCommand('owner', 'WEB', 'Web'))).id;
const api = (await run(new CreateProjectCommand('owner', 'API', 'Api'))).id;
const a = (await run(ticket(web, 'A'))).id;
const b = (await run(ticket(web, 'B'))).id;
const c = (await run(ticket(api, 'C'))).id;
const comments = [
(await run(new AddCommentCommand('owner', a, 'one'))).id,
(await run(new AddCommentCommand('owner', a, 'two'))).id,
];
await run(new LinkIssuesCommand('owner', a, IssueLinkType.BLOCKS, b));
await run(new LinkIssuesCommand('owner', c, IssueLinkType.DEPENDS_ON, a));
const release = (
await run(new CreateReleaseCommand('owner', web, 'v1', null, null))
).id;
await expect(
run(new DeleteProjectCommand('dev', web, 'WEB')),
).rejects.toThrow(ForbiddenError);
await run(new DeleteProjectCommand('owner', web, 'web'));
await drain();
for (const [id, Cls] of [
[a, Issue],
[b, Issue],
[release, Release],
[comments[0], Comment],
[comments[1], Comment],
] as const) {
expect(await access.live(id, Cls as never)).toBeNull();
}
// The ticket of the other project no longer points to the deleted one.
expect((await rehydrator.rehydrate(c, Issue)).links).toEqual([]);
// Commands on the deleted project fail as if it never existed.
await expect(run(ticket(web, 'late'))).rejects.toThrow(NotFoundError);
// And its key is free again.
await expect(
run(new CreateProjectCommand('owner', 'WEB', 'Web again')),
).resolves.toMatchObject({ key: 'WEB' });
});
it('only the owner deletes a project', async () => {
const { run } = await setup();
const web = (await run(new CreateProjectCommand('owner', 'WEB', 'Web'))).id;
await expect(
run(new DeleteProjectCommand('owner', web, 'NOPE')),
).rejects.toThrow(/WEB/);
await run(
new AddProjectMemberCommand('owner', web, 'dev', ProjectRole.ADMIN),
);
await expect(
run(new DeleteProjectCommand('dev', web, 'WEB')),
).rejects.toThrow(ForbiddenError);
});
it('archiving a release lets its unfinished tickets go', async () => {
const { run, drain, rehydrator } = await setup();
const web = (await run(new CreateProjectCommand('owner', 'WEB', 'Web'))).id;
const release = (
await run(new CreateReleaseCommand('owner', web, 'v1', null, null))
).id;
const open = (
await run(
new CreateIssueCommand('owner', web, 'TASK', 'open', {}, null, release),
)
).id;
const done = (
await run(
new CreateIssueCommand('owner', web, 'TASK', 'done', {}, null, release),
)
).id;
for (const status of ['TODO', 'IN_PROGRESS', 'IN_REVIEW', 'DONE'])
await run(new ChangeIssueStatusCommand('owner', done, status));
await run(new ArchiveReleaseCommand('owner', release));
await drain();
expect((await rehydrator.rehydrate(open, Issue)).releaseId).toBeNull();
expect((await rehydrator.rehydrate(done, Issue)).releaseId).toBe(release);
});
it('a user deactivated in Pireon leaves their open tickets unassigned', async () => {
const { run, react, rehydrator, openAssigned } = await setup();
const web = (await run(new CreateProjectCommand('owner', 'WEB', 'Web'))).id;
for (const user of ['dev', 'qa'])
await run(
new AddProjectMemberCommand('owner', web, user, ProjectRole.MEMBER),
);
const mine = (await run(ticket(web, 'mine'))).id;
const moved = (await run(ticket(web, 'moved'))).id;
await run(new AssignIssueCommand('owner', mine, 'dev'));
// The read model still lists it as dev's, but it was reassigned meanwhile.
await run(new AssignIssueCommand('owner', moved, 'qa'));
openAssigned.set('dev', [mine, moved]);
await react(new UserDeactivatedEvent('dev'));
const unassigned = await rehydrator.rehydrate(mine, Issue);
expect(unassigned.assigneeId).toBeNull();
expect((await rehydrator.rehydrate(moved, Issue)).assigneeId).toBe('qa');
});
});

View File

@@ -0,0 +1,10 @@
import { ProjectDeletionSaga } from './project-deletion.saga';
import { ReleaseArchivedSaga } from './release-archived.saga';
import { UserDeactivatedSaga } from './user-deactivated.saga';
/** Reactions to events that change other aggregates (run by the events bridge). */
export const SAGAS = [
ProjectDeletionSaga,
ReleaseArchivedSaga,
UserDeactivatedSaga,
];

View File

@@ -0,0 +1,37 @@
import { Injectable } from '@nestjs/common';
import { ICommand, IEvent, ofType, Saga } from '@nestjs/cqrs';
import { UserDeactivatedEvent } from '@task-stream/api/identity/domain';
import { SYSTEM_ACTOR_ID } from '@task-stream/api/tracker/domain';
import { from, mergeMap, Observable } from 'rxjs';
import { UnassignIssueCommand } from '../commands/unassign-issue.command';
import { FindIssuesRepository } from '../ports/find-issues.repository';
/**
* Pireon deactivated a user (identity context, `user.deactivated` webhook):
* their open tickets in every project become unassigned, so nobody waits on
* someone who is gone. Done by the system actor ("TaskStream").
*/
@Injectable()
export class UserDeactivatedSaga {
constructor(private readonly findIssues: FindIssuesRepository) {}
@Saga()
userDeactivated = (events$: Observable<IEvent>): Observable<ICommand> =>
events$.pipe(
ofType(UserDeactivatedEvent),
mergeMap((event) =>
from(this.findIssues.findOpenAssignedTo(event.userId)).pipe(
mergeMap((issues) =>
issues.map(
(issue) =>
new UnassignIssueCommand(
SYSTEM_ACTOR_ID,
issue.id,
event.userId,
),
),
),
),
),
);
}

View File

@@ -14,6 +14,7 @@ import { ProjectsService } from './projects.service';
import { QUERY_HANDLERS } from './queries/query-handlers';
import { ReadAccess } from './queries/read-access';
import { ReleasesService } from './releases.service';
import { SAGAS } from './sagas/sagas';
import { WorkflowsService } from './workflows.service';
const SERVICES = [
@@ -41,6 +42,7 @@ const SERVICES = [
...COMMAND_HANDLERS,
...QUERY_HANDLERS,
...EVENT_HANDLERS,
...SAGAS,
],
exports: SERVICES,
})

View File

@@ -11,6 +11,7 @@ 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,
@@ -128,6 +129,9 @@ export class InMemoryProjectKeys extends ProjectKeysRepository {
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 {
@@ -189,3 +193,54 @@ export class InMemoryReleaseIssues extends ReleaseIssuesRepository {
];
}
}
/** 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,
);
}
}

View File

@@ -9,6 +9,8 @@ El modelo del producto: proyectos, tickets, comentarios y releases.
- Agregados event-sourced: `Project`, `Issue`, `Comment`, `Release`. Validan en sus métodos y cambian de estado en `on<Evento>`.
- `events/`: un fichero por evento (`IssueCreatedEvent`, `IssueStatusChangedEvent`…), todos con `@AutowiredEvent`.
- Proyectos públicos (`Project.isPublic`) y borrado: `Project.delete` (propietario, confirmando la clave),
`Issue.delete`, `Release.delete` (los usa la cascada); `SYSTEM_ACTOR_ID` para cambios que no hace una persona.
- `factories/`: `ProjectFactory`, `IssueFactory`, `CommentFactory`, `ReleaseFactory` (`@Injectable()`).
- `project/project-settings.ts`: configuración por proyecto (estados con categoría, transiciones, tipos, áreas), sus
valores por defecto, `ProjectWorkflow` (lo que consultan tickets y queries) y `validateProjectSettings` (archivar,

View File

@@ -9,3 +9,4 @@ export * from './lib/release/release-status';
export * from './lib/events';
export * from './lib/factories';
export * from './lib/read-models';
export * from './lib/system-actor';

View File

@@ -19,3 +19,6 @@ export * from './release-created.event';
export * from './release-updated.event';
export * from './release-shipped.event';
export * from './release-archived.event';
export * from './project-deleted.event';
export * from './issue-deleted.event';
export * from './release-deleted.event';

View File

@@ -0,0 +1,15 @@
import { AutowiredEvent } from '@task-stream/api/shared/domain';
import { TrackerEvent } from './tracker.event';
/** The ticket was deleted (its project was). Its comments and the links to it go next. */
@AutowiredEvent
export class IssueDeletedEvent extends TrackerEvent {
constructor(
actorId: string,
issueId: string,
projectId: string,
readonly key: string,
) {
super(actorId, issueId, projectId);
}
}

View File

@@ -0,0 +1,19 @@
import { AutowiredEvent } from '@task-stream/api/shared/domain';
import { TrackerEvent } from './tracker.event';
/**
* The owner deleted the project. Its tickets and releases are deleted next by
* the cascade saga; the key becomes free again.
*/
@AutowiredEvent
export class ProjectDeletedEvent extends TrackerEvent {
constructor(
actorId: string,
projectId: string,
readonly key: string,
/** Members at the time, so their project lists drop it. */
readonly memberIds: string[],
) {
super(actorId, projectId, projectId);
}
}

View File

@@ -2,13 +2,15 @@ import { AutowiredEvent } from '@task-stream/api/shared/domain';
import type { ProjectProps } from '../project/project';
import { TrackerEvent } from './tracker.event';
/** Name or description changed. */
/** Name, description or visibility changed. */
@AutowiredEvent
export class ProjectUpdatedEvent extends TrackerEvent {
constructor(
actorId: string,
projectId: string,
readonly changes: Partial<Pick<ProjectProps, 'name' | 'description'>>,
readonly changes: Partial<
Pick<ProjectProps, 'name' | 'description' | 'isPublic'>
>,
) {
super(actorId, projectId, projectId);
}

View File

@@ -0,0 +1,10 @@
import { AutowiredEvent } from '@task-stream/api/shared/domain';
import { TrackerEvent } from './tracker.event';
/** The release was deleted (its project was). */
@AutowiredEvent
export class ReleaseDeletedEvent extends TrackerEvent {
constructor(actorId: string, releaseId: string, projectId: string) {
super(actorId, releaseId, projectId);
}
}

View File

@@ -13,6 +13,7 @@ export class ProjectFactory {
key: string;
name: string;
description?: string | null;
isPublic?: boolean;
ownerId: string;
}): Project {
const key = props.key.trim().toUpperCase();
@@ -29,6 +30,7 @@ export class ProjectFactory {
key,
name,
description: props.description?.trim() || null,
isPublic: props.isPublic ?? false,
ownerId: props.ownerId,
members: [{ userId: props.ownerId, role: ProjectRole.OWNER }],
settings: defaultProjectSettings(),

View File

@@ -5,6 +5,7 @@ import {
} from '@task-stream/api/shared/domain';
import { IssueAssignedEvent } from '../events/issue-assigned.event';
import { IssueCreatedEvent } from '../events/issue-created.event';
import { IssueDeletedEvent } from '../events/issue-deleted.event';
import { IssueLinkedEvent } from '../events/issue-linked.event';
import { IssueReleaseChangedEvent } from '../events/issue-release-changed.event';
import { IssueStatusChangedEvent } from '../events/issue-status-changed.event';
@@ -83,6 +84,7 @@ export class Issue extends VersionedAggregateRoot implements IssueProps {
public links: IssueLink[] = [];
public createdAt!: string;
public updatedAt!: string;
public deleted = false;
constructor(id: string) {
super();
@@ -214,6 +216,25 @@ export class Issue extends VersionedAggregateRoot implements IssueProps {
);
}
/** Deletes the ticket (its project was deleted). Idempotent. */
delete(actorId: string): void {
if (this.deleted) return;
this.apply(
new IssueDeletedEvent(actorId, this.id, this.projectId, this.key),
);
}
/** Drops every link to `targetIssueId` (that ticket was deleted). */
unlinkAllTo(actorId: string, targetIssueId: string): void {
for (const link of this.links.filter(
(l) => l.targetIssueId === targetIssueId,
)) {
this.apply(
new IssueUnlinkedEvent(actorId, this.id, this.projectId, { ...link }),
);
}
}
/* ---- State transitions (also replayed by loadFromHistory) ---- */
onIssueCreatedEvent({ issue }: IssueCreatedEvent): void {
@@ -255,6 +276,10 @@ export class Issue extends VersionedAggregateRoot implements IssueProps {
this.updatedAt = event.occurredAt;
}
onIssueDeletedEvent(): void {
this.deleted = true;
}
onIssueReleaseChangedEvent(event: IssueReleaseChangedEvent): void {
this.releaseId = event.releaseId;
this.updatedAt = event.occurredAt;

View File

@@ -0,0 +1,109 @@
import {
BusinessRuleError,
ForbiddenError,
} from '@task-stream/api/shared/domain';
import { IssueDeletedEvent } from '../events/issue-deleted.event';
import { IssueUnlinkedEvent } from '../events/issue-unlinked.event';
import { ProjectDeletedEvent } from '../events/project-deleted.event';
import { ProjectUpdatedEvent } from '../events/project-updated.event';
import { ReleaseDeletedEvent } from '../events/release-deleted.event';
import { IssueLinkType } from '../issue/issue.enums';
import { IssueFactory } from '../factories/issue.factory';
import { ProjectFactory } from '../factories/project.factory';
import { ReleaseFactory } from '../factories/release.factory';
import { replay } from '../testing';
import { Project } from './project';
import { ProjectRole } from './project-role';
const newProject = (isPublic = false) => {
const project = new ProjectFactory().create({
key: 'web',
name: 'Web',
ownerId: 'owner',
isPublic,
});
project.addMember('owner', 'admin', ProjectRole.ADMIN);
return project;
};
const newIssue = (number: number) =>
new IssueFactory().create({
projectId: 'p',
projectKey: 'WEB',
number,
title: `Ticket ${number}`,
type: 'BUG',
reporterId: 'owner',
});
describe('Public projects', () => {
it('are private unless created or switched public, also when replayed', () => {
const project = newProject();
expect(() => project.assertCanView('stranger')).toThrow(ForbiddenError);
project.updateDetails('admin', { isPublic: true });
expect(project.getUncommittedEvents().at(-1)).toBeInstanceOf(
ProjectUpdatedEvent,
);
expect(() => project.assertCanView('stranger')).not.toThrow();
// Reading is open; changing tickets still needs membership.
expect(() => project.assertCanContribute('stranger')).toThrow(
ForbiddenError,
);
expect(replay(project, Project).rebuilt.isPublic).toBe(true);
expect(newProject(true).isPublic).toBe(true);
});
});
describe('Deleting a project', () => {
it('is for the owner only, confirming the key', () => {
const project = newProject();
expect(() => project.delete('admin', 'WEB')).toThrow(ForbiddenError);
expect(() => project.delete('owner', 'APP')).toThrow(BusinessRuleError);
project.delete('owner', ' web ');
const event = project.getUncommittedEvents().at(-1) as ProjectDeletedEvent;
expect(event).toBeInstanceOf(ProjectDeletedEvent);
expect(event.memberIds).toEqual(['owner', 'admin']);
expect(replay(project, Project).rebuilt.deleted).toBe(true);
});
it('cascades through tickets and releases, idempotently', () => {
const issue = newIssue(1);
issue.delete('owner');
issue.delete('owner');
expect(
issue
.getUncommittedEvents()
.filter((e) => e instanceof IssueDeletedEvent),
).toHaveLength(1);
expect(issue.deleted).toBe(true);
const release = new ReleaseFactory().create('owner', {
projectId: 'p',
name: 'v1',
});
release.delete('owner');
release.delete('owner');
expect(
release
.getUncommittedEvents()
.filter((e) => e instanceof ReleaseDeletedEvent),
).toHaveLength(1);
});
it('drops every link to a deleted ticket', () => {
const source = newIssue(1);
source.linkTo('owner', IssueLinkType.BLOCKS, { id: 't', key: 'WEB-2' });
source.linkTo('owner', IssueLinkType.RELATES_TO, { id: 't', key: 'WEB-2' });
source.linkTo('owner', IssueLinkType.BLOCKS, { id: 'o', key: 'WEB-3' });
source.unlinkAllTo('owner', 't');
expect(
source
.getUncommittedEvents()
.filter((e) => e instanceof IssueUnlinkedEvent),
).toHaveLength(2);
expect(source.links.map((l) => l.targetIssueId)).toEqual(['o']);
});
});

View File

@@ -5,6 +5,7 @@ import {
assertThat,
} from '@task-stream/api/shared/domain';
import { ProjectCreatedEvent } from '../events/project-created.event';
import { ProjectDeletedEvent } from '../events/project-deleted.event';
import { ProjectMemberAddedEvent } from '../events/project-member-added.event';
import { ProjectMemberRemovedEvent } from '../events/project-member-removed.event';
import { ProjectMemberRoleChangedEvent } from '../events/project-member-role-changed.event';
@@ -29,6 +30,8 @@ export interface ProjectProps {
key: string;
name: string;
description: string | null;
/** Anyone (even without signing in) can read it. Absent in old projects: private. */
isPublic?: boolean;
ownerId: string;
members: ProjectMember[];
/** Absent in projects created before settings existed: defaults apply. */
@@ -45,11 +48,14 @@ export class Project extends VersionedAggregateRoot implements ProjectProps {
public key!: string;
public name!: string;
public description: string | null = null;
public isPublic = false;
public ownerId!: string;
public members: ProjectMember[] = [];
public settings: ProjectSettings = defaultProjectSettings();
public createdAt!: string;
public updatedAt!: string;
/** Set once deleted: the project can no longer be loaded for commands. */
public deleted = false;
constructor(id: string) {
super();
@@ -69,16 +75,18 @@ export class Project extends VersionedAggregateRoot implements ProjectProps {
return this.roleOf(userId) !== null;
}
/** Can read the project. */
/** Can read the project: members, or anyone if it is public. */
assertCanView(userId: string): void {
if (!this.isMember(userId)) {
if (!this.isPublic && !this.isMember(userId)) {
throw new ForbiddenError(`You are not a member of project ${this.key}`);
}
}
/** Can create and modify tickets, comments, links... */
/** Can create and modify tickets, comments, links... (members only, even if public). */
assertCanContribute(userId: string): void {
this.assertCanView(userId);
if (!this.isMember(userId)) {
throw new ForbiddenError(`You are not a member of project ${this.key}`);
}
if (this.roleOf(userId) === ProjectRole.VIEWER) {
throw new ForbiddenError(`Viewers cannot modify project ${this.key}`);
}
@@ -96,10 +104,14 @@ export class Project extends VersionedAggregateRoot implements ProjectProps {
updateDetails(
actorId: string,
changes: { name?: string; description?: string | null },
changes: { name?: string; description?: string | null; isPublic?: boolean },
): void {
this.assertCanManage(actorId);
const patch: { name?: string; description?: string | null } = {};
const patch: {
name?: string;
description?: string | null;
isPublic?: boolean;
} = {};
if (changes.name !== undefined) {
const name = changes.name.trim();
assertThat(name.length > 0, 'Project name is required');
@@ -109,10 +121,35 @@ export class Project extends VersionedAggregateRoot implements ProjectProps {
const description = changes.description?.trim() || null;
if (description !== this.description) patch.description = description;
}
if (changes.isPublic !== undefined && changes.isPublic !== this.isPublic)
patch.isPublic = changes.isPublic;
if (Object.keys(patch).length === 0) return;
this.apply(new ProjectUpdatedEvent(actorId, this.id, patch));
}
/**
* Only the owner deletes a project, confirming its key. Tickets, comments
* and releases follow (cascade saga); the events stay in the store.
*/
delete(actorId: string, confirmKey: string): void {
if (actorId !== this.ownerId) {
throw new ForbiddenError(`Only the owner can delete project ${this.key}`);
}
if (confirmKey.trim().toUpperCase() !== this.key) {
throw new BusinessRuleError(
`Type the project key (${this.key}) to confirm the deletion`,
);
}
this.apply(
new ProjectDeletedEvent(
actorId,
this.id,
this.key,
this.members.map((m) => m.userId),
),
);
}
/** Owners and admins customise statuses, transitions, types and areas. */
updateSettings(actorId: string, next: ProjectSettings): void {
this.assertCanManage(actorId);
@@ -175,6 +212,7 @@ export class Project extends VersionedAggregateRoot implements ProjectProps {
this.key = project.key;
this.name = project.name;
this.description = project.description;
this.isPublic = project.isPublic ?? false;
this.ownerId = project.ownerId;
this.members = project.members.map((m) => ({ ...m }));
this.settings = structuredClone(
@@ -189,6 +227,10 @@ export class Project extends VersionedAggregateRoot implements ProjectProps {
this.updatedAt = event.occurredAt;
}
onProjectDeletedEvent(): void {
this.deleted = true;
}
onProjectSettingsChangedEvent(event: ProjectSettingsChangedEvent): void {
this.settings = structuredClone(event.settings);
this.updatedAt = event.occurredAt;

View File

@@ -12,6 +12,8 @@ export class ProjectReadModel {
key!: string;
name!: string;
description!: string | null;
/** Readable by anyone, also without signing in. */
isPublic!: boolean;
owner!: UserRefReadModel;
members!: ProjectMemberReadModel[];
/** Statuses, transitions, ticket types and areas. */

View File

@@ -5,6 +5,7 @@ import {
} from '@task-stream/api/shared/domain';
import { ReleaseArchivedEvent } from '../events/release-archived.event';
import { ReleaseCreatedEvent } from '../events/release-created.event';
import { ReleaseDeletedEvent } from '../events/release-deleted.event';
import { ReleaseShippedEvent } from '../events/release-shipped.event';
import { ReleaseUpdatedEvent } from '../events/release-updated.event';
import { ReleaseStatus } from './release-status';
@@ -30,6 +31,7 @@ export class Release extends VersionedAggregateRoot implements ReleaseProps {
public status: ReleaseStatus = ReleaseStatus.PLANNED;
public releasedAt: string | null = null;
public createdAt!: string;
public deleted = false;
constructor(id: string) {
super();
@@ -112,6 +114,12 @@ export class Release extends VersionedAggregateRoot implements ReleaseProps {
this.apply(new ReleaseArchivedEvent(actorId, this.id, this.projectId));
}
/** Deletes the release (its project was deleted). Idempotent. */
delete(actorId: string): void {
if (this.deleted) return;
this.apply(new ReleaseDeletedEvent(actorId, this.id, this.projectId));
}
/* ---- State transitions (also replayed by loadFromHistory) ---- */
onReleaseCreatedEvent({ release }: ReleaseCreatedEvent): void {
@@ -130,4 +138,8 @@ export class Release extends VersionedAggregateRoot implements ReleaseProps {
onReleaseArchivedEvent(): void {
this.status = ReleaseStatus.ARCHIVED;
}
onReleaseDeletedEvent(): void {
this.deleted = true;
}
}

View File

@@ -0,0 +1,6 @@
/**
* Actor of changes nobody made by hand (e.g. unassigning the tickets of a
* user Pireon deactivated). Shown as "TaskStream" in the activity feed.
*/
export const SYSTEM_ACTOR_ID = 'system';
export const SYSTEM_ACTOR_NAME = 'TaskStream';

View File

@@ -7,7 +7,9 @@ Adaptadores de salida del tracker.
## Contenido
- `postgres/`: lookups de escritura (`project_keys`, `issue_sequences`, `release_names`) y `ReleaseIssuesRepository` (consulta el event store).
- `postgres/`: lookups de escritura (`project_keys`, `issue_sequences`, `release_names`), `ReleaseIssuesRepository` y
`TrackerStreamsRepository` (consultan el event store: tickets y releases de un proyecto, comentarios y enlaces de un
ticket).
- `mongo/`: esquemas del read model (incluida la línea temporal `issue_timeline`), repositorios `Find*` y
`UpsertMaterialized*` (upsert idempotente por versión) y `StatusCategoryBackfill` (al arrancar, rellena la categoría
del estado en los tickets proyectados antes de existir).

Some files were not shown because too many files have changed in this diff Show More