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, InMemoryAttachmentStorage, 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 { DeleteIssueCommand } from '../commands/delete-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 { IssueTree } from '../commands/issue-tree'; 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 { AttachmentStorage } from '../ports/attachment-storage'; 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 files = new InMemoryAttachmentStorage(); 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(); const moduleRef = await Test.createTestingModule({ imports: [CqrsModule.forRoot()], providers: [ ...COMMAND_HANDLERS, ...TRACKER_FACTORIES, ProjectDeletionSaga, ReleaseArchivedSaga, UserDeactivatedSaga, WriteAccess, IssueTree, AggregateRehydrator, { provide: EventStore, useValue: store }, { provide: AttachmentStorage, useValue: files }, { 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 => 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('a deleted ticket lets the tickets under it go back to the top', async () => { const { run, drain, rehydrator } = await setup(); const web = (await run(new CreateProjectCommand('owner', 'WEB', 'Web'))).id; const epic = (await run(ticket(web, 'Epic'))).id; const child = ( await run( new CreateIssueCommand( 'owner', web, 'TASK', 'child', {}, null, null, epic, ), ) ).id; await run(new DeleteIssueCommand('owner', epic)); await drain(); expect((await rehydrator.rehydrate(child, Issue)).parentId).toBeNull(); }); 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'); }); });