diff --git a/packages/@n8n/db/src/repositories/activity-event.repository.ts b/packages/@n8n/db/src/repositories/activity-event.repository.ts index 715fcebb5ec..2261a29a86c 100644 --- a/packages/@n8n/db/src/repositories/activity-event.repository.ts +++ b/packages/@n8n/db/src/repositories/activity-event.repository.ts @@ -1,10 +1,21 @@ import { Service } from '@n8n/di'; -import { And, DataSource, In, LessThan, LessThanOrEqual, MoreThan, Repository } from '@n8n/typeorm'; +import { + And, + DataSource, + In, + IsNull, + LessThan, + LessThanOrEqual, + MoreThan, + Not, + Repository, +} from '@n8n/typeorm'; import type { FindOperator, FindOptionsWhere } from '@n8n/typeorm'; import type { IDataObject } from 'n8n-workflow'; import { ActivityEvent } from '../entities'; import type { ActivityResourceType } from '../entities'; +import { chunkIds } from '../utils/chunk-ids'; /** Long enough for any name a list needs to show, short enough that a row stays a pointer. */ export const activityResourceNameMaxLength = 128; @@ -78,6 +89,15 @@ function projectScopeWhere(scope: ActivityProjectScope): FindOptionsWhere { constructor(dataSource: DataSource) { @@ -181,6 +201,51 @@ export class ActivityEventRepository extends Repository { }); } + /** + * The newest entries per resource that name a user, newest first and at most + * `perResource` each, keyed by resource id. Reduced here, not in SQL: `DISTINCT ON` + * is Postgres-only. Pages by id so an uncapped table is never loaded whole; the + * page stops as soon as every resource of the chunk has its share. Resources + * without such an entry are absent from the result. + */ + async findRecentAttributedByResource( + resourceType: ActivityResourceType, + resourceIds: string[], + perResource: number, + ): Promise> { + const recent = new Map(); + if (resourceIds.length === 0 || perResource <= 0) return recent; + + for (const chunk of chunkIds([...new Set(resourceIds)])) { + const pending = new Set(chunk); + let beforeId: number | undefined; + while (pending.size > 0) { + const rows = await this.find({ + select: ['id', 'resourceId', 'userId', 'createdAt'], + where: { + resourceType, + resourceId: In([...pending]), + userId: Not(IsNull()), + ...(beforeId !== undefined ? { id: LessThan(beforeId) } : {}), + }, + order: { id: 'DESC' }, + take: ATTRIBUTED_PAGE_SIZE, + }); + for (const row of rows) { + if (row.resourceId === null || row.userId === null) continue; + const entries = recent.get(row.resourceId) ?? []; + if (entries.length >= perResource) continue; + entries.push({ userId: row.userId, at: row.createdAt }); + recent.set(row.resourceId, entries); + if (entries.length >= perResource) pending.delete(row.resourceId); + } + if (rows.length < ATTRIBUTED_PAGE_SIZE) break; + beforeId = rows[rows.length - 1].id; + } + } + return recent; + } + /** Retention by age. Returns how many entries went, so a caller can log a sweep worth noticing. */ async deleteOlderThan(cutoff: Date, signal?: AbortSignal): Promise { return await this.deleteInBatches({ createdAt: LessThan(cutoff) }, undefined, signal); diff --git a/packages/@n8n/db/src/repositories/project-relation.repository.ts b/packages/@n8n/db/src/repositories/project-relation.repository.ts index adbbca87c45..a3342356639 100644 --- a/packages/@n8n/db/src/repositories/project-relation.repository.ts +++ b/packages/@n8n/db/src/repositories/project-relation.repository.ts @@ -1,5 +1,9 @@ import { Service } from '@n8n/di'; -import { PROJECT_OWNER_ROLE_SLUG, type ProjectRole } from '@n8n/permissions'; +import { + PROJECT_ADMIN_ROLE_SLUG, + PROJECT_OWNER_ROLE_SLUG, + type ProjectRole, +} from '@n8n/permissions'; import { DataSource, In, Repository } from '@n8n/typeorm'; import { ProjectRelation, Role } from '../entities'; @@ -40,6 +44,26 @@ export class ProjectRelationRepository extends Repository { }); } + /** Admin user ids per project, sorted, keyed by project id. Projects without an admin are absent. */ + async findAdminUserIdsByProjectIds(projectIds: string[]): Promise> { + const admins = new Map(); + if (projectIds.length === 0) return admins; + + for (const chunk of chunkIds([...new Set(projectIds)])) { + const rows = await this.find({ + select: ['projectId', 'userId'], + where: { projectId: In(chunk), role: { slug: PROJECT_ADMIN_ROLE_SLUG } }, + order: { projectId: 'ASC', userId: 'ASC' }, + }); + for (const row of rows) { + const ids = admins.get(row.projectId) ?? []; + ids.push(row.userId); + admins.set(row.projectId, ids); + } + } + return admins; + } + async getPersonalProjectsForUsers(userIds: string[]) { const projectRelations = await this.find({ where: { diff --git a/packages/@n8n/db/src/repositories/user.repository.ts b/packages/@n8n/db/src/repositories/user.repository.ts index 05235781f21..b1a614b82f2 100644 --- a/packages/@n8n/db/src/repositories/user.repository.ts +++ b/packages/@n8n/db/src/repositories/user.repository.ts @@ -48,6 +48,15 @@ export class UserRepository extends BaseRepository { }); } + /** + * Every user with the relations `isPending` needs: the role and the auth + * identities. Without the identities, a user who signs in through SSO or LDAP + * and so has no password would read as pending. + */ + async findAllWithRoleAndAuthIdentities(): Promise { + return await this.find({ relations: ['role', 'authIdentities'], order: { id: 'ASC' } }); + } + async findByIdWithRole(id: string, ctx: OperationContext = {}): Promise { return await this.managerFor(ctx).findOne(User, { where: { id }, diff --git a/packages/@n8n/db/src/repositories/workflow-history.repository.ts b/packages/@n8n/db/src/repositories/workflow-history.repository.ts index 3bba3598019..b51aaa06d0d 100644 --- a/packages/@n8n/db/src/repositories/workflow-history.repository.ts +++ b/packages/@n8n/db/src/repositories/workflow-history.repository.ts @@ -9,6 +9,7 @@ import { WorkflowPublishHistory, } from '../entities'; import { BaseRepository } from './base-repository'; +import { chunkIds } from '../utils/chunk-ids'; import { WorkflowReviewRequestWorkflow } from '../entities/workflow-review-request-workflow.ee'; import { WorkflowReviewRequest } from '../entities/workflow-review-request.ee'; import type { OperationContext } from '../services/transaction'; @@ -30,6 +31,34 @@ export class WorkflowHistoryRepository extends BaseRepository { await this.managerFor(ctx).insert(WorkflowHistory, version); } + /** + * The authors of the newest versions per workflow, newest first and at most + * `perWorkflow` each, keyed by workflow id. `authors` is the display name + * recorded at save time, not a user id. Workflows without a version are absent. + */ + async findRecentAuthorsByWorkflowIds( + workflowIds: string[], + perWorkflow: number, + ): Promise>> { + const recent = new Map>(); + if (workflowIds.length === 0 || perWorkflow <= 0) return recent; + + for (const chunk of chunkIds([...new Set(workflowIds)])) { + const rows = await this.find({ + select: ['versionId', 'workflowId', 'authors', 'createdAt'], + where: { workflowId: In(chunk) }, + order: { createdAt: 'DESC' }, + }); + for (const row of rows) { + const entries = recent.get(row.workflowId) ?? []; + if (entries.length >= perWorkflow) continue; + entries.push({ authors: row.authors, at: row.createdAt }); + recent.set(row.workflowId, entries); + } + } + return recent; + } + async deleteEarlierThan(date: Date) { return await this.delete({ createdAt: LessThan(date) }); } diff --git a/packages/@n8n/db/src/repositories/workflow-publish-history.repository.ts b/packages/@n8n/db/src/repositories/workflow-publish-history.repository.ts index 022fa05e2ab..148f39b3f72 100644 --- a/packages/@n8n/db/src/repositories/workflow-publish-history.repository.ts +++ b/packages/@n8n/db/src/repositories/workflow-publish-history.repository.ts @@ -1,5 +1,5 @@ import { Service } from '@n8n/di'; -import { DataSource } from '@n8n/typeorm'; +import { DataSource, In, IsNull, Not } from '@n8n/typeorm'; import type { EntityManager } from '@n8n/typeorm'; import { BaseRepository } from './base-repository'; @@ -62,6 +62,35 @@ export class WorkflowPublishHistoryRepository extends BaseRepository>> { + const recent = new Map>(); + if (workflowIds.length === 0 || perWorkflow <= 0) return recent; + + for (const chunk of chunkIds([...new Set(workflowIds)])) { + const rows = await this.find({ + select: ['id', 'workflowId', 'userId', 'createdAt'], + where: { workflowId: In(chunk), userId: Not(IsNull()) }, + order: { createdAt: 'DESC', id: 'DESC' }, + }); + for (const row of rows) { + if (row.userId === null) continue; + const entries = recent.get(row.workflowId) ?? []; + if (entries.length >= perWorkflow) continue; + entries.push({ userId: row.userId, at: row.createdAt }); + recent.set(row.workflowId, entries); + } + } + return recent; + } + async findLatestActivations(workflowId: string, versionIds: string[]) { if (versionIds.length === 0) return []; diff --git a/packages/cli/src/modules/breaking-changes/README.md b/packages/cli/src/modules/breaking-changes/README.md index 74477e44366..3f90d1a0c85 100644 --- a/packages/cli/src/modules/breaking-changes/README.md +++ b/packages/cli/src/modules/breaking-changes/README.md @@ -24,6 +24,8 @@ breaking-changes/ database/ entities/ # migration_finding, migration_finding_sync, migration_workflow_owner tables repositories/ # Use-case-named DB access (BaseRepository + OperationContext) + owners/ + migration-owner-suggestion.service.ts # Proposes who should fix a workflow's findings sync/ migration-finding-diff.ts # Pure diff of scan hits against stored findings migration-finding-sync.service.ts # Runs a scan and writes the diff, one transaction per batch diff --git a/packages/cli/src/modules/breaking-changes/owners/__tests__/migration-owner-suggestion.service.test.ts b/packages/cli/src/modules/breaking-changes/owners/__tests__/migration-owner-suggestion.service.test.ts new file mode 100644 index 00000000000..27e16e4fd01 --- /dev/null +++ b/packages/cli/src/modules/breaking-changes/owners/__tests__/migration-owner-suggestion.service.test.ts @@ -0,0 +1,285 @@ +import type { + ActivityEventRepository, + Project, + ProjectRelation, + ProjectRelationRepository, + SharedWorkflowRepository, + User, + UserRepository, + WorkflowHistoryRepository, + WorkflowPublishHistoryRepository, +} from '@n8n/db'; +import type { MockProxy } from 'vitest-mock-extended'; +import { mock } from 'vitest-mock-extended'; + +import { + MigrationOwnerSuggestionService, + RECENT_ACTIONS_PER_WORKFLOW, +} from '../migration-owner-suggestion.service'; + +const at = (iso: string) => new Date(iso); + +function user(id: string, firstName: string, lastName: string, flags: Partial = {}): User { + return { id, firstName, lastName, disabled: false, isPending: false, ...flags } as User; +} + +describe('MigrationOwnerSuggestionService', () => { + let activityEventRepository: MockProxy; + let workflowPublishHistoryRepository: MockProxy; + let workflowHistoryRepository: MockProxy; + let sharedWorkflowRepository: MockProxy; + let projectRelationRepository: MockProxy; + let userRepository: MockProxy; + let service: MigrationOwnerSuggestionService; + + const alice = user('alice', 'Alice', 'Adams'); + const bob = user('bob', 'Bob', 'Brown'); + const carol = user('carol', 'Carol', 'Clark'); + + function givenUsers(...users: User[]) { + userRepository.findAllWithRoleAndAuthIdentities.mockResolvedValue(users); + } + + /** Each workflow's recent entries, newest first, as the repositories return them. */ + type Recent = Record; + const asLists = (entries: Recent) => + new Map( + Object.entries(entries).map(([id, value]) => [id, Array.isArray(value) ? value : [value]]), + ); + + function givenActivity(entries: Recent<{ userId: string; at: Date }>) { + activityEventRepository.findRecentAttributedByResource.mockResolvedValue(asLists(entries)); + } + + function givenPublishes(entries: Recent<{ userId: string; at: Date }>) { + workflowPublishHistoryRepository.findRecentAttributedByWorkflowIds.mockResolvedValue( + asLists(entries), + ); + } + + function givenVersions(entries: Recent<{ authors: string; at: Date }>) { + workflowHistoryRepository.findRecentAuthorsByWorkflowIds.mockResolvedValue(asLists(entries)); + } + + function givenProjects( + byWorkflow: Record, + personalOwners: Record = {}, + admins: Record = {}, + ) { + sharedWorkflowRepository.findOwnerProjectsByWorkflowIds.mockResolvedValue( + new Map(Object.entries(byWorkflow).map(([id, project]) => [id, project as Project])), + ); + projectRelationRepository.getPersonalProjectOwners.mockResolvedValue( + Object.entries(personalOwners).map( + ([projectId, userId]) => ({ projectId, userId }) as ProjectRelation, + ), + ); + projectRelationRepository.findAdminUserIdsByProjectIds.mockResolvedValue( + new Map(Object.entries(admins)), + ); + } + + beforeEach(() => { + activityEventRepository = mock(); + workflowPublishHistoryRepository = mock(); + workflowHistoryRepository = mock(); + sharedWorkflowRepository = mock(); + projectRelationRepository = mock(); + userRepository = mock(); + + givenUsers(alice, bob, carol); + givenActivity({}); + givenPublishes({}); + givenVersions({}); + givenProjects({}); + + service = new MigrationOwnerSuggestionService( + activityEventRepository, + workflowPublishHistoryRepository, + workflowHistoryRepository, + sharedWorkflowRepository, + projectRelationRepository, + userRepository, + ); + }); + + it('returns nothing for no workflows without touching the database', async () => { + expect(await service.suggestOwners([])).toEqual([]); + + expect(userRepository.findAllWithRoleAndAuthIdentities).not.toHaveBeenCalled(); + }); + + it('asks each source for the same bounded number of recent actions per workflow', async () => { + await service.suggestOwners(['wf-1']); + + expect(activityEventRepository.findRecentAttributedByResource).toHaveBeenCalledWith( + 'workflow', + ['wf-1'], + RECENT_ACTIONS_PER_WORKFLOW, + ); + expect(workflowPublishHistoryRepository.findRecentAttributedByWorkflowIds).toHaveBeenCalledWith( + ['wf-1'], + RECENT_ACTIONS_PER_WORKFLOW, + ); + expect(workflowHistoryRepository.findRecentAuthorsByWorkflowIds).toHaveBeenCalledWith( + ['wf-1'], + RECENT_ACTIONS_PER_WORKFLOW, + ); + expect(RECENT_ACTIONS_PER_WORKFLOW).toBe(10); + }); + + describe('most recent attributable activity', () => { + it('picks the user of the newest action across the activity feed and publish history', async () => { + givenActivity({ 'wf-1': { userId: 'alice', at: at('2026-09-01T10:00:00Z') } }); + givenPublishes({ 'wf-1': { userId: 'bob', at: at('2026-09-02T10:00:00Z') } }); + + expect(await service.suggestOwners(['wf-1'])).toEqual([ + { workflowId: 'wf-1', userId: 'bob' }, + ]); + }); + + it('prefers a newer activity entry over an older publish', async () => { + givenActivity({ 'wf-1': { userId: 'alice', at: at('2026-09-03T10:00:00Z') } }); + givenPublishes({ 'wf-1': { userId: 'bob', at: at('2026-09-02T10:00:00Z') } }); + + expect(await service.suggestOwners(['wf-1'])).toEqual([ + { workflowId: 'wf-1', userId: 'alice' }, + ]); + }); + + it('resolves the author name of the newest version to the one user with that name', async () => { + givenVersions({ 'wf-1': { authors: 'Carol Clark', at: at('2026-09-05T10:00:00Z') } }); + givenPublishes({ 'wf-1': { userId: 'bob', at: at('2026-09-02T10:00:00Z') } }); + + expect(await service.suggestOwners(['wf-1'])).toEqual([ + { workflowId: 'wf-1', userId: 'carol' }, + ]); + }); + + it('strips the MCP marker from an author name before matching', async () => { + givenVersions({ + 'wf-1': { authors: 'Carol Clark (via MCP)', at: at('2026-09-05T10:00:00Z') }, + }); + + expect(await service.suggestOwners(['wf-1'])).toEqual([ + { workflowId: 'wf-1', userId: 'carol' }, + ]); + }); + + it('ignores an author name that matches two users', async () => { + givenUsers(alice, bob, user('bob-2', 'Bob', 'Brown')); + givenVersions({ 'wf-1': { authors: 'Bob Brown', at: at('2026-09-05T10:00:00Z') } }); + givenPublishes({ 'wf-1': { userId: 'alice', at: at('2026-09-01T10:00:00Z') } }); + + expect(await service.suggestOwners(['wf-1'])).toEqual([ + { workflowId: 'wf-1', userId: 'alice' }, + ]); + }); + + it('ignores an author name that matches no user', async () => { + givenVersions({ 'wf-1': { authors: 'eval-snapshot', at: at('2026-09-05T10:00:00Z') } }); + + expect(await service.suggestOwners(['wf-1'])).toEqual([]); + }); + + it('skips a disabled or pending user and falls through to the next signal', async () => { + givenUsers( + user('dan', 'Dan', 'Dole', { disabled: true }), + user('eve', 'Eve', 'Evans', { isPending: true }), + bob, + ); + givenActivity({ 'wf-1': { userId: 'dan', at: at('2026-09-09T10:00:00Z') } }); + givenPublishes({ 'wf-1': { userId: 'eve', at: at('2026-09-08T10:00:00Z') } }); + givenVersions({ 'wf-1': { authors: 'Bob Brown', at: at('2026-09-01T10:00:00Z') } }); + + expect(await service.suggestOwners(['wf-1'])).toEqual([ + { workflowId: 'wf-1', userId: 'bob' }, + ]); + }); + + it('falls back to an older attributable action when the newest one is by a user who cannot own', async () => { + givenUsers( + user('dan', 'Dan', 'Dole', { disabled: true }), + alice, + bob, + user('bob-2', 'Bob', 'Brown'), + ); + givenActivity({ + 'wf-1': [ + { userId: 'dan', at: at('2026-09-09T10:00:00Z') }, + { userId: 'alice', at: at('2026-09-07T10:00:00Z') }, + ], + }); + givenVersions({ + 'wf-1': [ + { authors: 'Bob Brown', at: at('2026-09-08T10:00:00Z') }, + { authors: 'Bob Brown', at: at('2026-09-01T10:00:00Z') }, + ], + }); + + expect(await service.suggestOwners(['wf-1'])).toEqual([ + { workflowId: 'wf-1', userId: 'alice' }, + ]); + }); + + it('skips a user that no longer exists', async () => { + givenActivity({ 'wf-1': { userId: 'gone', at: at('2026-09-09T10:00:00Z') } }); + givenProjects({ 'wf-1': { id: 'p-1', type: 'personal' } }, { 'p-1': 'alice' }); + + expect(await service.suggestOwners(['wf-1'])).toEqual([ + { workflowId: 'wf-1', userId: 'alice' }, + ]); + }); + }); + + describe('project owner fallback', () => { + it('uses the personal project owner when no activity is attributable', async () => { + givenProjects({ 'wf-1': { id: 'p-1', type: 'personal' } }, { 'p-1': 'alice' }); + + expect(await service.suggestOwners(['wf-1'])).toEqual([ + { workflowId: 'wf-1', userId: 'alice' }, + ]); + }); + + it('uses one admin of a team project, the same one each time', async () => { + givenProjects({ 'wf-1': { id: 'p-team', type: 'team' } }, {}, { 'p-team': ['bob', 'carol'] }); + + expect(await service.suggestOwners(['wf-1'])).toEqual([ + { workflowId: 'wf-1', userId: 'bob' }, + ]); + expect(await service.suggestOwners(['wf-1'])).toEqual([ + { workflowId: 'wf-1', userId: 'bob' }, + ]); + }); + + it('skips an admin who cannot own and takes the next one', async () => { + givenUsers(user('bob', 'Bob', 'Brown', { disabled: true }), carol); + givenProjects({ 'wf-1': { id: 'p-team', type: 'team' } }, {}, { 'p-team': ['bob', 'carol'] }); + + expect(await service.suggestOwners(['wf-1'])).toEqual([ + { workflowId: 'wf-1', userId: 'carol' }, + ]); + }); + + it('suggests nobody for a team project without an eligible admin', async () => { + givenProjects({ 'wf-1': { id: 'p-team', type: 'team' } }, {}, {}); + + expect(await service.suggestOwners(['wf-1'])).toEqual([]); + }); + + it('only looks up projects for the workflows that activity did not resolve', async () => { + givenActivity({ 'wf-1': { userId: 'alice', at: at('2026-09-01T10:00:00Z') } }); + givenProjects({ 'wf-2': { id: 'p-2', type: 'personal' } }, { 'p-2': 'bob' }); + + const suggestions = await service.suggestOwners(['wf-1', 'wf-2']); + + expect(sharedWorkflowRepository.findOwnerProjectsByWorkflowIds).toHaveBeenCalledWith([ + 'wf-2', + ]); + expect(suggestions).toEqual([ + { workflowId: 'wf-1', userId: 'alice' }, + { workflowId: 'wf-2', userId: 'bob' }, + ]); + }); + }); +}); diff --git a/packages/cli/src/modules/breaking-changes/owners/migration-owner-suggestion.service.ts b/packages/cli/src/modules/breaking-changes/owners/migration-owner-suggestion.service.ts new file mode 100644 index 00000000000..5cc3cf1bad8 --- /dev/null +++ b/packages/cli/src/modules/breaking-changes/owners/migration-owner-suggestion.service.ts @@ -0,0 +1,167 @@ +import { + ActivityEventRepository, + ProjectRelationRepository, + SharedWorkflowRepository, + UserRepository, + WorkflowHistoryRepository, + WorkflowPublishHistoryRepository, + type User, +} from '@n8n/db'; +import { Service } from '@n8n/di'; + +import type { OwnerSuggestion } from '../database/repositories/migration-workflow-owner.repository'; + +/** A user and when they acted on the workflow. */ +interface Candidate { + userId: string; + at: Date; +} + +/** Workflow history stores the author's display name, with this suffix for saves through MCP. */ +const MCP_AUTHOR_SUFFIX = ' (via MCP)'; + +/** + * How many recent actions per source and workflow are considered. The newest + * action may belong to a user who cannot own, so a few older ones are kept to + * fall back on before the project owner is used. + */ +export const RECENT_ACTIONS_PER_WORKFLOW = 10; + +/** + * Proposes who should fix a workflow's migration findings: the user behind the + * most recent activity on the workflow that can be attributed to a single user, + * else one of the owners of the workflow's project. A workflow with neither gets + * no suggestion. + * + * Three records carry attributable activity, each with a gap the others cover: + * the activity feed (user ids, but capped and recent), publish history (user + * ids, but only for published workflows) and workflow history (every save, but + * a display name instead of an id). A disabled, pending or deleted user is + * never suggested. + */ +@Service() +export class MigrationOwnerSuggestionService { + constructor( + private readonly activityEventRepository: ActivityEventRepository, + private readonly workflowPublishHistoryRepository: WorkflowPublishHistoryRepository, + private readonly workflowHistoryRepository: WorkflowHistoryRepository, + private readonly sharedWorkflowRepository: SharedWorkflowRepository, + private readonly projectRelationRepository: ProjectRelationRepository, + private readonly userRepository: UserRepository, + ) {} + + async suggestOwners(workflowIds: string[]): Promise { + if (workflowIds.length === 0) return []; + + const users = await this.loadEligibleUsers(); + const latestByWorkflow = await this.findLatestAttributedActivity(workflowIds, users); + + const unresolved = workflowIds.filter((workflowId) => !latestByWorkflow.has(workflowId)); + const fallbackByWorkflow = await this.findProjectOwners(unresolved, users); + + const suggestions: OwnerSuggestion[] = []; + for (const workflowId of workflowIds) { + const userId = latestByWorkflow.get(workflowId) ?? fallbackByWorkflow.get(workflowId); + if (userId) suggestions.push({ workflowId, userId }); + } + return suggestions; + } + + /** Every user who can be an owner, by id and by the display name workflow history records. */ + private async loadEligibleUsers(): Promise { + const all = await this.userRepository.findAllWithRoleAndAuthIdentities(); + const byId = new Map(); + const idsByName = new Map(); + for (const user of all) { + if (user.disabled || user.isPending) continue; + byId.set(user.id, user); + const name = `${user.firstName} ${user.lastName}`; + idsByName.set(name, [...(idsByName.get(name) ?? []), user.id]); + } + return { byId, idsByName }; + } + + /** + * Per workflow, the user behind the most recent action that one eligible user + * can be held to. An action by a user who cannot own, or a version whose author + * name is ambiguous, is skipped in favour of the next older one. + */ + private async findLatestAttributedActivity( + workflowIds: string[], + users: EligibleUsers, + ): Promise> { + const limit = RECENT_ACTIONS_PER_WORKFLOW; + const [activity, publishes, versions] = await Promise.all([ + this.activityEventRepository.findRecentAttributedByResource('workflow', workflowIds, limit), + this.workflowPublishHistoryRepository.findRecentAttributedByWorkflowIds(workflowIds, limit), + this.workflowHistoryRepository.findRecentAuthorsByWorkflowIds(workflowIds, limit), + ]); + + const latest = new Map(); + for (const workflowId of workflowIds) { + const candidates: Candidate[] = [ + ...(activity.get(workflowId) ?? []), + ...(publishes.get(workflowId) ?? []), + ...(versions.get(workflowId) ?? []).flatMap((version) => { + const userId = resolveAuthor(version.authors, users); + return userId ? [{ userId, at: version.at }] : []; + }), + ]; + const newest = candidates + .filter((candidate) => users.byId.has(candidate.userId)) + .sort((a, b) => b.at.getTime() - a.at.getTime())[0]; + if (newest) latest.set(workflowId, newest.userId); + } + return latest; + } + + /** + * One owner of each workflow's project: the owner of a personal project, or + * any admin of a team project. The pick among admins is stable, not meaningful. + */ + private async findProjectOwners( + workflowIds: string[], + users: EligibleUsers, + ): Promise> { + const owners = new Map(); + if (workflowIds.length === 0) return owners; + + const projectByWorkflow = + await this.sharedWorkflowRepository.findOwnerProjectsByWorkflowIds(workflowIds); + const projectIds = [...new Set([...projectByWorkflow.values()].map((project) => project.id))]; + const [personalOwners, adminsByProject] = await Promise.all([ + this.projectRelationRepository.getPersonalProjectOwners(projectIds), + this.projectRelationRepository.findAdminUserIdsByProjectIds(projectIds), + ]); + const personalOwnerByProject = new Map( + personalOwners.map((relation) => [relation.projectId, relation.userId]), + ); + + for (const [workflowId, project] of projectByWorkflow) { + const candidates = + project.type === 'personal' + ? [personalOwnerByProject.get(project.id)] + : (adminsByProject.get(project.id) ?? []); + const userId = candidates.find((id): id is string => !!id && users.byId.has(id)); + if (userId) owners.set(workflowId, userId); + } + return owners; + } +} + +interface EligibleUsers { + byId: Map; + idsByName: Map; +} + +/** + * A version names its author, not a user. The name counts only when exactly one + * eligible user carries it. + */ +function resolveAuthor(authors: string, users: EligibleUsers): string | undefined { + const name = authors.endsWith(MCP_AUTHOR_SUFFIX) + ? authors.slice(0, -MCP_AUTHOR_SUFFIX.length) + : authors; + const ids = users.idsByName.get(name); + return ids?.length === 1 ? ids[0] : undefined; +} diff --git a/packages/cli/test/integration/activity-event.repository.test.ts b/packages/cli/test/integration/activity-event.repository.test.ts index 7fe68cffb21..ad3eec11dab 100644 --- a/packages/cli/test/integration/activity-event.repository.test.ts +++ b/packages/cli/test/integration/activity-event.repository.test.ts @@ -3,6 +3,8 @@ import type { Project } from '@n8n/db'; import { ActivityEventRepository, activityDataMaxLength } from '@n8n/db'; import { Container } from '@n8n/di'; +import { createMember } from './shared/db/users'; + describe('ActivityEventRepository', () => { let repository: ActivityEventRepository; let project: Project; @@ -305,4 +307,64 @@ describe('ActivityEventRepository', () => { expect(remaining.map((entry) => entry.action)).toEqual(['third', 'second']); }); }); + + describe('findRecentAttributedByResource', () => { + // Real users: the column is a foreign key to the user table. + let alice: string; + let bob: string; + let carol: string; + beforeAll(async () => { + [alice, bob, carol] = ( + await Promise.all([createMember(), createMember(), createMember()]) + ).map((user) => user.id); + }); + const entry = async (resourceId: string, userId: string | null) => + await repository.record({ + category: 'workflow', + action: 'saved', + projectId: project.id, + resourceType: 'workflow', + resourceId, + userId, + }); + + it('returns the newest attributed entries per resource first, newest entry first', async () => { + await entry('wf-1', alice); + await entry('wf-1', bob); + await entry('wf-1', null); + await entry('wf-2', carol); + await entry('wf-3', alice); + + const recent = await repository.findRecentAttributedByResource( + 'workflow', + ['wf-1', 'wf-2'], + 10, + ); + + expect([...recent.keys()].sort()).toEqual(['wf-1', 'wf-2']); + expect(recent.get('wf-1')?.map((e) => e.userId)).toEqual([bob, alice]); + expect(recent.get('wf-2')?.map((e) => e.userId)).toEqual([carol]); + }); + + it('keeps at most the requested number of entries per resource', async () => { + await entry('wf-1', alice); + await entry('wf-1', bob); + await entry('wf-1', carol); + + const recent = await repository.findRecentAttributedByResource('workflow', ['wf-1'], 2); + + expect(recent.get('wf-1')?.map((e) => e.userId)).toEqual([carol, bob]); + }); + + it('omits a resource with no attributed entry and returns nothing for no ids', async () => { + await entry('wf-1', null); + + expect(await repository.findRecentAttributedByResource('workflow', ['wf-1'], 10)).toEqual( + new Map(), + ); + expect(await repository.findRecentAttributedByResource('workflow', [], 10)).toEqual( + new Map(), + ); + }); + }); }); diff --git a/packages/cli/test/integration/workflow-history.repository.test.ts b/packages/cli/test/integration/workflow-history.repository.test.ts new file mode 100644 index 00000000000..06143ac86f5 --- /dev/null +++ b/packages/cli/test/integration/workflow-history.repository.test.ts @@ -0,0 +1,73 @@ +import { createWorkflow, createWorkflowHistory, testDb } from '@n8n/backend-test-utils'; +import { WorkflowHistoryRepository } from '@n8n/db'; +import type { WorkflowEntity } from '@n8n/db'; +import { Container } from '@n8n/di'; +import { randomUUID } from 'node:crypto'; + +describe('WorkflowHistoryRepository', () => { + let repository: WorkflowHistoryRepository; + + beforeAll(async () => { + await testDb.init(); + repository = Container.get(WorkflowHistoryRepository); + }); + + beforeEach(async () => { + await testDb.truncate(['WorkflowHistory', 'WorkflowEntity']); + }); + + afterAll(async () => await testDb.terminate()); + + async function version(workflow: WorkflowEntity, authors: string, createdAt: string) { + await createWorkflowHistory(workflow, undefined, undefined, { + versionId: randomUUID(), + authors, + createdAt: new Date(createdAt), + }); + } + + describe('findRecentAuthorsByWorkflowIds', () => { + test('returns the newest versions per requested workflow first', async () => { + const [first, second, other] = await Promise.all([ + createWorkflow(), + createWorkflow(), + createWorkflow(), + ]); + await version(first, 'Alice Adams', '2026-01-01T00:00:00.000Z'); + await version(first, 'Bob Brown', '2026-01-03T00:00:00.000Z'); + await version(second, 'Carol Clark', '2026-01-02T00:00:00.000Z'); + await version(other, 'Dan Dole', '2026-01-04T00:00:00.000Z'); + + const recent = await repository.findRecentAuthorsByWorkflowIds([first.id, second.id], 10); + + expect([...recent.keys()].sort()).toEqual([first.id, second.id].sort()); + expect(recent.get(first.id)?.map((entry) => entry.authors)).toEqual([ + 'Bob Brown', + 'Alice Adams', + ]); + expect(recent.get(first.id)?.[0].at.getTime()).toBe(Date.parse('2026-01-03T00:00:00.000Z')); + expect(recent.get(second.id)?.map((entry) => entry.authors)).toEqual(['Carol Clark']); + }); + + test('keeps at most the requested number of versions per workflow', async () => { + const workflow = await createWorkflow(); + await version(workflow, 'Alice Adams', '2026-01-01T00:00:00.000Z'); + await version(workflow, 'Bob Brown', '2026-01-02T00:00:00.000Z'); + await version(workflow, 'Carol Clark', '2026-01-03T00:00:00.000Z'); + + const recent = await repository.findRecentAuthorsByWorkflowIds([workflow.id], 2); + + expect(recent.get(workflow.id)?.map((entry) => entry.authors)).toEqual([ + 'Carol Clark', + 'Bob Brown', + ]); + }); + + test('omits a workflow without versions and returns nothing for no ids', async () => { + const workflow = await createWorkflow(); + + expect(await repository.findRecentAuthorsByWorkflowIds([workflow.id], 10)).toEqual(new Map()); + expect(await repository.findRecentAuthorsByWorkflowIds([], 10)).toEqual(new Map()); + }); + }); +}); diff --git a/packages/cli/test/integration/workflow-publish-history.repository.test.ts b/packages/cli/test/integration/workflow-publish-history.repository.test.ts new file mode 100644 index 00000000000..1bc628b9ab4 --- /dev/null +++ b/packages/cli/test/integration/workflow-publish-history.repository.test.ts @@ -0,0 +1,82 @@ +import { createWorkflow, testDb } from '@n8n/backend-test-utils'; +import { WorkflowPublishHistoryRepository } from '@n8n/db'; +import type { User, WorkflowEntity } from '@n8n/db'; +import { Container } from '@n8n/di'; + +import { createMember } from './shared/db/users'; + +describe('WorkflowPublishHistoryRepository', () => { + let repository: WorkflowPublishHistoryRepository; + let alice: User; + let bob: User; + + beforeAll(async () => { + await testDb.init(); + repository = Container.get(WorkflowPublishHistoryRepository); + }); + + beforeEach(async () => { + await testDb.truncate(['WorkflowPublishHistory', 'WorkflowEntity', 'User']); + [alice, bob] = await Promise.all([createMember(), createMember()]); + }); + + afterAll(async () => await testDb.terminate()); + + async function publishEvent( + workflow: WorkflowEntity, + userId: string | null, + createdAt: string, + event: 'activated' | 'deactivated' = 'activated', + ) { + await repository.insert({ + workflowId: workflow.id, + versionId: null, + event, + userId, + createdAt: new Date(createdAt), + }); + } + + describe('findRecentAttributedByWorkflowIds', () => { + test('returns the newest events per workflow first, without those whose user is gone', async () => { + const [first, second, other] = await Promise.all([ + createWorkflow(), + createWorkflow(), + createWorkflow(), + ]); + await publishEvent(first, alice.id, '2026-01-01T00:00:00.000Z'); + await publishEvent(first, bob.id, '2026-01-03T00:00:00.000Z', 'deactivated'); + await publishEvent(first, null, '2026-01-05T00:00:00.000Z'); + await publishEvent(second, bob.id, '2026-01-02T00:00:00.000Z'); + await publishEvent(other, alice.id, '2026-01-04T00:00:00.000Z'); + + const recent = await repository.findRecentAttributedByWorkflowIds([first.id, second.id], 10); + + expect([...recent.keys()].sort()).toEqual([first.id, second.id].sort()); + expect(recent.get(first.id)?.map((entry) => entry.userId)).toEqual([bob.id, alice.id]); + expect(recent.get(first.id)?.[0].at.getTime()).toBe(Date.parse('2026-01-03T00:00:00.000Z')); + expect(recent.get(second.id)?.map((entry) => entry.userId)).toEqual([bob.id]); + }); + + test('keeps at most the requested number of events per workflow', async () => { + const workflow = await createWorkflow(); + await publishEvent(workflow, alice.id, '2026-01-01T00:00:00.000Z'); + await publishEvent(workflow, bob.id, '2026-01-02T00:00:00.000Z'); + await publishEvent(workflow, alice.id, '2026-01-03T00:00:00.000Z'); + + const recent = await repository.findRecentAttributedByWorkflowIds([workflow.id], 2); + + expect(recent.get(workflow.id)?.map((entry) => entry.userId)).toEqual([alice.id, bob.id]); + }); + + test('omits a workflow with no attributable event and returns nothing for no ids', async () => { + const workflow = await createWorkflow(); + await publishEvent(workflow, null, '2026-01-01T00:00:00.000Z'); + + expect(await repository.findRecentAttributedByWorkflowIds([workflow.id], 10)).toEqual( + new Map(), + ); + expect(await repository.findRecentAttributedByWorkflowIds([], 10)).toEqual(new Map()); + }); + }); +});