Services run commands through a CommandRunner that runs one again (up to 5 times, waiting a little longer each time) when its aggregate changed underneath, e.g. a saga writing to the same ticket. Each command loads its aggregate fresh and stores its events in one transaction, so a ConcurrencyError means nothing was stored and running it again is safe. People no longer get a 409 for it, and the e2e suite stops flaking. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
114 lines
3.4 KiB
TypeScript
114 lines
3.4 KiB
TypeScript
import { CommandRunner } from '@task-stream/api/shared/application';
|
|
import { Injectable } from '@nestjs/common';
|
|
import { QueryBus } from '@nestjs/cqrs';
|
|
import { ForbiddenError } from '@task-stream/api/shared/domain';
|
|
import {
|
|
AddGitRepositoryCommand,
|
|
GetGitIntegrationQuery,
|
|
GitIntegration,
|
|
GitRepository,
|
|
RecordGitMentionsCommand,
|
|
RemoveGitRepositoryCommand,
|
|
RotateGitRepositorySecretCommand,
|
|
} from './git/git';
|
|
import {
|
|
gitMentionsOf,
|
|
gitProviderOf,
|
|
verifyGitSignature,
|
|
} from './git/git-payloads';
|
|
import { FindProjectsRepository } from './ports/find-projects.repository';
|
|
import { GitIntegrationsRepository } from './ports/git-integrations.repository';
|
|
|
|
export interface GitDelivery {
|
|
/** Project id or key, from the webhook URL. */
|
|
project: string;
|
|
headers: Record<string, string | undefined>;
|
|
rawBody: Buffer | undefined;
|
|
body: Record<string, unknown>;
|
|
}
|
|
|
|
/** Links commits, branches and pull requests to the tickets they name. */
|
|
@Injectable()
|
|
export class GitService {
|
|
constructor(
|
|
private readonly commands: CommandRunner,
|
|
private readonly queryBus: QueryBus,
|
|
private readonly findProjects: FindProjectsRepository,
|
|
private readonly integrations: GitIntegrationsRepository,
|
|
) {}
|
|
|
|
integration(viewerId: string, projectId: string): Promise<GitIntegration> {
|
|
return this.queryBus.execute(
|
|
new GetGitIntegrationQuery(viewerId, projectId),
|
|
);
|
|
}
|
|
|
|
addRepository(
|
|
actorId: string,
|
|
projectId: string,
|
|
name: string,
|
|
): Promise<GitRepository> {
|
|
return this.commands.execute(
|
|
new AddGitRepositoryCommand(actorId, projectId, name),
|
|
);
|
|
}
|
|
|
|
rotateSecret(
|
|
actorId: string,
|
|
projectId: string,
|
|
repositoryId: string,
|
|
): Promise<GitRepository> {
|
|
return this.commands.execute(
|
|
new RotateGitRepositorySecretCommand(actorId, projectId, repositoryId),
|
|
);
|
|
}
|
|
|
|
removeRepository(
|
|
actorId: string,
|
|
projectId: string,
|
|
repositoryId: string,
|
|
): Promise<void> {
|
|
return this.commands.execute(
|
|
new RemoveGitRepositoryCommand(actorId, projectId, repositoryId),
|
|
);
|
|
}
|
|
|
|
/**
|
|
* A webhook delivery. Every repository of the project shares the URL; the
|
|
* secret that verifies the signature says which one it came from. Unknown
|
|
* project, integration off or a bad signature all look the same from
|
|
* outside (403): nothing to learn by probing.
|
|
*/
|
|
async receive(delivery: GitDelivery): Promise<{ recorded: number }> {
|
|
const project =
|
|
(await this.findProjects.findById(delivery.project)) ??
|
|
(await this.findProjects.findByKey(delivery.project.toUpperCase()));
|
|
const source = gitProviderOf(delivery.headers);
|
|
const repository =
|
|
project && source
|
|
? (await this.integrations.repositoriesOf(project.id)).find((r) =>
|
|
verifyGitSignature(
|
|
source.provider,
|
|
delivery.headers,
|
|
delivery.rawBody,
|
|
r.secret,
|
|
),
|
|
)
|
|
: undefined;
|
|
if (!project || !source || !repository)
|
|
throw new ForbiddenError(
|
|
'Unknown project, integration off or bad signature',
|
|
);
|
|
const mentions = gitMentionsOf(
|
|
source.provider,
|
|
source.event,
|
|
delivery.body,
|
|
);
|
|
if (!mentions.length) return { recorded: 0 };
|
|
const recorded: number = await this.commands.execute(
|
|
new RecordGitMentionsCommand(project.id, repository.name, mentions),
|
|
);
|
|
return { recorded };
|
|
}
|
|
}
|