diff --git a/packages/infrastructure/src/plugin/plugin.repo.test.ts b/packages/infrastructure/src/plugin/plugin.repo.test.ts index e8cc78bd..a27c5a78 100644 --- a/packages/infrastructure/src/plugin/plugin.repo.test.ts +++ b/packages/infrastructure/src/plugin/plugin.repo.test.ts @@ -7,7 +7,7 @@ import { type PluginType } from '@domain/entities/plugin.entity'; import { PluginStatusEnum } from '@domain/entities/plugin-base.entity'; import type { FileObject } from '@domain/value-objects/file/file-object.vo'; import { type PkgContentFileObjects } from '@domain/value-objects/file/pkg-file.vo'; -import { successResult } from '@domain/value-objects/result.vo'; +import { failureResult, successResult } from '@domain/value-objects/result.vo'; import { PluginRepo, type PluginRepoDeps } from './plugin.repo'; @@ -249,6 +249,8 @@ describe('PluginRepo.createPlugin', () => { it('restores a disabled plugin with the same version and etag to active during direct installation', async () => { (PluginRepo as any)._instance = undefined; + const session = { id: 'session' }; + const sessionRun = vi.fn(async (fn: (session: unknown) => Promise) => fn(session)); const updateOne = vi.fn().mockResolvedValue({ modifiedCount: 1 }); const pluginModel = { findOne: vi.fn().mockReturnValue({ @@ -277,6 +279,7 @@ describe('PluginRepo.createPlugin', () => { const publicSave = vi.fn(); const repo = PluginRepo.getInstance({ mongoClient: { + sessionRun, getModel: vi.fn((modelName: string) => modelName === 'pluginInstallation' ? installationModel : pluginModel ) @@ -296,6 +299,10 @@ describe('PluginRepo.createPlugin', () => { }); expect(err).toBeNull(); + expect(sessionRun).toHaveBeenCalledTimes(1); + expect(privateSave.mock.invocationCallOrder[0]).toBeLessThan( + sessionRun.mock.invocationCallOrder[0] + ); expect(updateOne).toHaveBeenCalledWith( { pluginId: 'plugin-a', @@ -313,6 +320,9 @@ describe('PluginRepo.createPlugin', () => { $unset: { expiredAt: 1 } + }, + { + session } ); expect(privateSave).toHaveBeenCalledWith( @@ -336,6 +346,9 @@ describe('PluginRepo.createPlugin', () => { pluginId: 1, version: 1, etag: 1 + }, + { + session } ); expect(pluginModel.updateMany).toHaveBeenCalledWith( @@ -356,17 +369,25 @@ describe('PluginRepo.createPlugin', () => { $unset: { expiredAt: 1 } + }, + { + session + } + ); + expect(installationModel.deleteMany).toHaveBeenCalledWith( + { + $or: [ + { + pluginId: 'plugin-a', + version: '1.0.0', + etag: 'old-etag' + } + ] + }, + { + session } ); - expect(installationModel.deleteMany).toHaveBeenCalledWith({ - $or: [ - { - pluginId: 'plugin-a', - version: '1.0.0', - etag: 'old-etag' - } - ] - }); expect(installationModel.updateOne).toHaveBeenCalledWith( { source: 'system', @@ -380,11 +401,73 @@ describe('PluginRepo.createPlugin', () => { } }, { - upsert: true + upsert: true, + session } ); }); + it('keeps restored disabled plugin unchanged when direct installation file upload fails', async () => { + (PluginRepo as any)._instance = undefined; + + const sessionRun = vi.fn(); + const pluginModel = { + findOne: vi.fn().mockReturnValue({ + lean: vi.fn().mockResolvedValue({ + _id: 'existing-plugin', + status: PluginStatusEnum.disabled + }) + }), + find: vi.fn(), + updateMany: vi.fn(), + updateOne: vi.fn() + }; + const installationModel = { + deleteMany: vi.fn(), + updateOne: vi.fn() + }; + const privateSave = vi + .fn() + .mockResolvedValue(failureResult({ en: 'save failed', 'zh-CN': '保存失败' })); + const repo = PluginRepo.getInstance({ + mongoClient: { + sessionRun, + getModel: vi.fn((modelName: string) => + modelName === 'pluginInstallation' ? installationModel : pluginModel + ) + }, + privateRemoteFileStorageRepo: { + save: privateSave + }, + publicRemoteFileStorageRepo: { + save: vi.fn() + } + } as unknown as PluginRepoDeps); + + const [, err] = await repo.createPlugin({ + plugin: plugin(), + files: files(), + pending: false + }); + + expect(err?.reason).toEqual({ + en: 'upload temp file error', + 'zh-CN': '上传临时文件错误' + }); + expect(privateSave).toHaveBeenCalledWith( + expect.objectContaining({ + fileKey: 'plugin-a/1.0.0/etag-a/index.js', + fileName: 'index.js' + }) + ); + expect(sessionRun).not.toHaveBeenCalled(); + expect(pluginModel.updateOne).not.toHaveBeenCalled(); + expect(pluginModel.find).not.toHaveBeenCalled(); + expect(pluginModel.updateMany).not.toHaveBeenCalled(); + expect(installationModel.deleteMany).not.toHaveBeenCalled(); + expect(installationModel.updateOne).not.toHaveBeenCalled(); + }); + it('reports same version and etag when uploading an already installed plugin', async () => { (PluginRepo as any)._instance = undefined; diff --git a/packages/infrastructure/src/plugin/plugin.repo.ts b/packages/infrastructure/src/plugin/plugin.repo.ts index 4babe0e1..922ae924 100644 --- a/packages/infrastructure/src/plugin/plugin.repo.ts +++ b/packages/infrastructure/src/plugin/plugin.repo.ts @@ -1,6 +1,7 @@ import path from 'node:path'; import { addMinutes } from 'date-fns'; +import type { ClientSession } from 'mongoose'; import { type PluginType } from '@domain/entities/plugin.entity'; import { PluginStatusEnum } from '@domain/entities/plugin-base.entity'; @@ -42,7 +43,7 @@ import { serializePluginRecordJsonSchemaFields } from './utils/json-schema-storage-codec'; import { Semver } from './utils/semver'; -import { pluginCodecRegistry, PluginRecordSchema } from './codec'; +import { pluginCodecRegistry, type PluginRecordPayloadType, PluginRecordSchema } from './codec'; type PluginListView = 'summary' | 'toolSummary'; @@ -97,6 +98,13 @@ type MongoPluginWithId = MongoPluginSchemaType & { _id: unknown; }; +type ReplaceInstalledPluginInput = { + activateTarget: boolean; + installedPlugin: MongoPluginWithId; + pluginRecord: PluginRecordPayloadType; + uniqueId: PluginUniqueIdType; +}; + type InstalledPluginRecord = { source: PluginSourceType; plugin: ListedMongoPlugin; @@ -163,8 +171,9 @@ export class PluginRepo implements PluginRepoPort { return Object.keys(schema).length > 0; } - private async updateSystemInstallation(plugin: MongoPluginWithId) { + private async updateSystemInstallation(plugin: MongoPluginWithId, session?: ClientSession) { const installationModel = this.deps.mongoClient.getModel('pluginInstallation'); + const options = session ? { upsert: true, session } : { upsert: true }; await installationModel.updateOne( { @@ -178,46 +187,77 @@ export class PluginRepo implements PluginRepoPort { pluginObjectId: plugin._id } }, - { - upsert: true - } + options ); } - private async disableSameVersionActivePlugins(uniqueId: PluginUniqueIdType): Promise { + private async disablePluginIds( + uniqueIds: PluginUniqueIdType[], + session?: ClientSession + ): Promise { + if (uniqueIds.length === 0) { + return; + } + + const pluginModel = this.deps.mongoClient.getModel('plugin'); + const pluginInstallationModel = this.deps.mongoClient.getModel('pluginInstallation'); + const updateFilter = { + $or: uniqueIds + }; + const update = { + $set: { + status: PluginStatusEnum.disabled, + updateAt: new Date() + }, + $unset: { + expiredAt: 1 + } + }; + const installationFilter = { + $or: uniqueIds.map(({ pluginId, version, etag }) => ({ + pluginId, + version, + etag + })) + }; + + if (session) { + await pluginModel.updateMany(updateFilter, update, { session }); + await pluginInstallationModel.deleteMany(installationFilter, { session }); + return; + } + + await pluginModel.updateMany(updateFilter, update); + await pluginInstallationModel.deleteMany(installationFilter); + } + + private async disableSameVersionActivePlugins( + uniqueId: PluginUniqueIdType, + session?: ClientSession + ): Promise { try { - const activePlugins = await this.deps.mongoClient - .getModel('plugin') - .find( - { - pluginId: uniqueId.pluginId, - version: uniqueId.version, - etag: { - $ne: uniqueId.etag - }, - status: PluginStatusEnum.active - }, - { - _id: 0, - pluginId: 1, - version: 1, - etag: 1 - } - ) - .lean(); + const pluginModel = this.deps.mongoClient.getModel('plugin'); + const filter = { + pluginId: uniqueId.pluginId, + version: uniqueId.version, + etag: { + $ne: uniqueId.etag + }, + status: PluginStatusEnum.active + }; + const projection = { + _id: 0, + pluginId: 1, + version: 1, + etag: 1 + }; + const activePlugins = await (session + ? pluginModel.find(filter, projection, { session }) + : pluginModel.find(filter, projection) + ).lean(); const replacedPluginIds = activePlugins.map((plugin) => PluginUniqueIdSchema.parse(plugin)); - const [, disableErr] = await this.disablePlugins(replacedPluginIds); - - if (disableErr) { - return failureResult( - { - en: 'Failed to disable same-version active plugins', - 'zh-CN': '禁用同版本 active 插件失败' - }, - disableErr - ); - } + await this.disablePluginIds(replacedPluginIds, session); return successResult({}); } catch (error) { @@ -231,6 +271,56 @@ export class PluginRepo implements PluginRepoPort { } } + private async replaceInstalledPlugin({ + activateTarget, + installedPlugin, + pluginRecord, + uniqueId + }: ReplaceInstalledPluginInput): Promise { + const pluginModel = this.deps.mongoClient.getModel('plugin'); + + try { + await this.deps.mongoClient.sessionRun(async (session) => { + const [, replaceActiveErr] = await this.disableSameVersionActivePlugins(uniqueId, session); + + if (replaceActiveErr) { + throw replaceActiveErr.error; + } + + if (activateTarget) { + await pluginModel.updateOne( + uniqueId, + { + $set: { + ...pluginRecord, + status: PluginStatusEnum.active, + updateAt: new Date() + }, + $unset: { + expiredAt: 1 + } + }, + { + session + } + ); + } + + await this.updateSystemInstallation(installedPlugin, session); + }, {}); + + return successResult({}); + } catch (error) { + return failureResult( + { + en: 'Failed to replace active plugin installation', + 'zh-CN': '替换 active 插件安装失败' + }, + error + ); + } + } + private getManagedPublicFileNames(fileKeys: string[]): Set { return new Set(fileKeys.map((fileKey) => path.basename(fileKey))); } @@ -957,11 +1047,12 @@ export class PluginRepo implements PluginRepoPort { const pendingExpiresAt = pending ? addMinutes(Date.now(), PluginRepo.ExpiresMinutes) : undefined; + const pluginRecord = this.toPluginRecord(plugin); let installedPlugin: MongoPluginWithId | undefined; + let activateInstalledPlugin = false; try { - const pluginRecord = this.toPluginRecord(plugin); const existingPlugin = await pluginModel .findOne(uniqueId, { _id: true, @@ -999,17 +1090,7 @@ export class PluginRepo implements PluginRepoPort { ...pluginRecord, _id: existingPlugin._id } as MongoPluginWithId; - - await pluginModel.updateOne(uniqueId, { - $set: { - ...pluginRecord, - status: PluginStatusEnum.active, - updateAt: new Date() - }, - $unset: { - expiredAt: 1 - } - }); + activateInstalledPlugin = true; } else { return failureResult({ en: 'Plugin with the same version and etag already exists', @@ -1024,9 +1105,12 @@ export class PluginRepo implements PluginRepoPort { status: PluginStatusEnum.pending, expiredAt: pendingExpiresAt } - : {}) + : { + status: PluginStatusEnum.disabled + }) }); installedPlugin = createdPlugin.toObject(); + activateInstalledPlugin = !pending; } } catch (error) { return failureResult( @@ -1135,20 +1219,13 @@ export class PluginRepo implements PluginRepoPort { } if (!pending && installedPlugin) { - const [, replaceActiveErr] = await this.disableSameVersionActivePlugins(uniqueId); + const [, replaceActiveErr] = await this.replaceInstalledPlugin({ + activateTarget: activateInstalledPlugin, + installedPlugin, + pluginRecord, + uniqueId + }); if (replaceActiveErr) return failureResult(replaceActiveErr); - - try { - await this.updateSystemInstallation(installedPlugin); - } catch (error) { - return failureResult( - { - en: 'Failed to update plugin installation', - 'zh-CN': '更新插件安装记录失败' - }, - error - ); - } } return successResult({});