diff --git a/packages/plugin-core/manifest.json b/packages/plugin-core/manifest.json index a72d60c04d..6a2559362b 100644 --- a/packages/plugin-core/manifest.json +++ b/packages/plugin-core/manifest.json @@ -469,6 +469,28 @@ "required": ["albumIds"] } }, + { + "name": "assetAlbumFilter", + "title": "Filter by album(s)", + "description": "Filter by which album(s) triggered the workflow", + "types": ["AlbumAssetV1"], + "schema": { + "type": "object", + "properties": { + "albumIds": { + "type": "string", + "title": "Album IDs", + "array": true, + "description": "Allowed album IDs", + "uiHint": { + "type": "AlbumId" + } + } + }, + "required": ["albumIds"] + }, + "uiHints": ["Filter"] + }, { "name": "webhook", "title": "Trigger Webhook", diff --git a/packages/plugin-core/src/index.ts b/packages/plugin-core/src/index.ts index 7b91ed7111..9c4d81670a 100644 --- a/packages/plugin-core/src/index.ts +++ b/packages/plugin-core/src/index.ts @@ -66,6 +66,8 @@ const methods = wrapper({ return {}; }, + assetAlbumFilter: ({ config, data }) => ({ workflow: { continue: config.albumIds.includes(data.album.id) } }), + assetArchive: ({ config, data }) => { if (!config.inverse && data.asset.visibility !== AssetVisibility.Archive) { return { changes: { asset: { visibility: AssetVisibility.Archive } } }; @@ -209,6 +211,7 @@ const methods = wrapper({ const { assetAddToAlbums, + assetAlbumFilter, assetArchive, assetFavorite, assetFileFilter, @@ -227,6 +230,7 @@ const { export { assetAddToAlbums, + assetAlbumFilter, assetArchive, assetFavorite, assetFileFilter, diff --git a/packages/sdk/src/fetch-client.ts b/packages/sdk/src/fetch-client.ts index 81c2090f43..9f8764f9e5 100644 --- a/packages/sdk/src/fetch-client.ts +++ b/packages/sdk/src/fetch-client.ts @@ -7492,6 +7492,7 @@ export enum JobName { OcrQueueAll = "OcrQueueAll", Ocr = "Ocr", WorkflowAssetTrigger = "WorkflowAssetTrigger", + WorkflowAlbumAssetTrigger = "WorkflowAlbumAssetTrigger", IntegrityUntrackedFilesQueueAll = "IntegrityUntrackedFilesQueueAll", IntegrityUntrackedFiles = "IntegrityUntrackedFiles", IntegrityUntrackedRefresh = "IntegrityUntrackedRefresh", diff --git a/server/src/enum.ts b/server/src/enum.ts index 66e36b7121..803cad4410 100644 --- a/server/src/enum.ts +++ b/server/src/enum.ts @@ -344,6 +344,7 @@ export enum SystemMetadataKey { VersionCheckState = 'version-check-state', License = 'license', IntegrityChecksumCheckpoint = 'integrity-checksum-checkpoint', + AlbumAssetWorkflowCheckpoint = 'album-asset-workflow-checkpoint', } export enum UserMetadataKey { diff --git a/server/src/repositories/event.repository.ts b/server/src/repositories/event.repository.ts index 7a92446c77..cbd7d9ed62 100644 --- a/server/src/repositories/event.repository.ts +++ b/server/src/repositories/event.repository.ts @@ -40,7 +40,7 @@ type EventMap = { // album events AlbumUpdate: [{ id: string; userIds: string[]; recipientIds: string[] }]; AlbumInvite: [{ id: string; userId: string; senderName: string }]; - AlbumAssetsAdded: [{ albumId: string; userIds: string[]; recipientIds: string[]; assetIds: string[] }]; + AlbumAssetsAdded: []; // asset events AssetCreate: [{ asset: Pick; file?: UploadFile }]; diff --git a/server/src/repositories/workflow.repository.ts b/server/src/repositories/workflow.repository.ts index 9c4e435d00..62cb1f4aa4 100644 --- a/server/src/repositories/workflow.repository.ts +++ b/server/src/repositories/workflow.repository.ts @@ -1,5 +1,5 @@ import { Injectable } from '@nestjs/common'; -import { Insertable, Kysely, Updateable } from 'kysely'; +import { Insertable, Kysely, SelectQueryBuilder, Updateable } from 'kysely'; import { jsonArrayFrom, jsonObjectFrom } from 'kysely/helpers/postgres'; import { InjectKysely } from 'nestjs-kysely'; import { columns } from 'src/database'; @@ -139,8 +139,26 @@ export class WorkflowRepository { } getForAssetV1(assetId: string) { + return this.assetV1Query(this.db.selectFrom('asset').where('id', '=', assetId)).executeTakeFirstOrThrow(); + } + + getForAlbumAssetV1(after: string) { return this.db - .selectFrom('asset') + .selectFrom('album_asset') + .select(['albumId', 'assetId', 'album_asset.updateId']) + .orderBy('album_asset.updateId', 'asc') + .where('album_asset.updateId', '>', after) + .limit(2000) + .select((eb) => + jsonObjectFrom(this.assetV1Query(eb.selectFrom('asset').whereRef('asset.id', '=', 'album_asset.assetId'))).as( + 'asset', + ), + ) + .execute(); + } + + private assetV1Query(qb: SelectQueryBuilder) { + return qb .leftJoin('asset_exif', 'asset_exif.assetId', 'asset.id') .select((eb) => [ ...columns.workflowAssetV1, @@ -181,8 +199,6 @@ export class WorkflowRepository { ]) .whereRef('asset_exif.assetId', '=', 'asset.id'), ).as('exifInfo'), - ]) - .where('id', '=', assetId) - .executeTakeFirstOrThrow(); + ]); } } diff --git a/server/src/services/album.service.ts b/server/src/services/album.service.ts index 313c8affb1..55e3682e38 100644 --- a/server/src/services/album.service.ts +++ b/server/src/services/album.service.ts @@ -193,12 +193,7 @@ export class AlbumService extends BaseService { const userIds = album.albumUsers.map(({ user }) => user.id); const recipientIds = userIds.filter((userId) => userId !== auth.user.id); await this.eventRepository.emit('AlbumUpdate', { id, userIds, recipientIds }); - await this.eventRepository.emit('AlbumAssetsAdded', { - albumId: id, - userIds, - recipientIds, - assetIds: results.filter((a) => a.success).map((a) => a.id), - }); + await this.eventRepository.emit('AlbumAssetsAdded'); } return results; @@ -228,7 +223,6 @@ export class AlbumService extends BaseService { const albumAssetValues: { albumId: string; assetId: string }[] = []; const updateEvents: { id: string; userIds: string[]; recipientIds: string[] }[] = []; - const addedEvents: { albumId: string; userIds: string[]; recipientIds: string[]; assetIds: string[] }[] = []; for (const albumId of allowedAlbumIds) { const existingAssetIds = await this.albumRepository.getAssetIds(albumId, [...allowedAssetIds]); const notPresentAssetIds = [...allowedAssetIds.difference(existingAssetIds)]; @@ -254,16 +248,13 @@ export class AlbumService extends BaseService { const userIds = album.albumUsers.map(({ user }) => user.id); const recipientIds = userIds.filter((userId) => userId !== auth.user.id); updateEvents.push({ id: albumId, userIds, recipientIds }); - addedEvents.push({ albumId, userIds, recipientIds, assetIds: notPresentAssetIds }); } await this.albumRepository.addAssetIdsToAlbums(albumAssetValues); for (const event of updateEvents) { await this.eventRepository.emit('AlbumUpdate', event); } - for (const event of addedEvents) { - await this.eventRepository.emit('AlbumAssetsAdded', event); - } + await this.eventRepository.emit('AlbumAssetsAdded'); return results; } diff --git a/server/src/services/workflow-execution.service.ts b/server/src/services/workflow-execution.service.ts index ebe12523a2..1ba7dd32de 100644 --- a/server/src/services/workflow-execution.service.ts +++ b/server/src/services/workflow-execution.service.ts @@ -21,6 +21,7 @@ import { JobName, JobStatus, QueueName, + SystemMetadataKey, WorkflowType, } from 'src/enum'; import { ArgOf } from 'src/repositories/event.repository'; @@ -28,6 +29,7 @@ import { AlbumService } from 'src/services/album.service'; import { AssetService } from 'src/services/asset.service'; import { BaseService } from 'src/services/base.service'; import { JobItem, JobOf } from 'src/types'; +import { withImpliedItems } from 'src/utils/workflow'; const dummy = () => { throw new Error( @@ -42,8 +44,6 @@ type ExecuteOptions = { type AssetTrigger = { userId: string; assetId: string; trigger: WorkflowTrigger }; -type AlbumAssetTrigger = { userIds: string[]; albumId: string; assetIds: string[]; trigger: WorkflowTrigger }; - type HostContext = { allowedHosts: string[]; }; @@ -312,13 +312,8 @@ export class WorkflowExecutionService extends BaseService { } @OnEvent({ name: 'AlbumAssetsAdded' }) - onAlbumAssetsAdded({ albumId, userIds, recipientIds, assetIds }: ArgOf<'AlbumAssetsAdded'>) { - return this.onAlbumAssetTrigger({ - albumId, - userIds: [...userIds, ...recipientIds], - assetIds, - trigger: WorkflowTrigger.AlbumAssetAdded, - }); + onAlbumAssetsAdded() { + return this.onAlbumAssetTrigger(WorkflowTrigger.AlbumAssetAdded); } private async onAssetTrigger({ userId, assetId, trigger }: AssetTrigger) { @@ -331,30 +326,53 @@ export class WorkflowExecutionService extends BaseService { ); } - private async onAlbumAssetTrigger({ albumId, userIds, assetIds, trigger }: AlbumAssetTrigger) { - let jobs: JobItem[] = []; - for (const userId of userIds) { - const items = await this.workflowRepository.search({ userId, trigger }); - if (!items.length) { - continue; - } + private async onAlbumAssetTrigger(trigger: WorkflowTrigger) { + let checkpoint = await this.systemMetadataRepository.get(SystemMetadataKey.AlbumAssetWorkflowCheckpoint); + const now = await this.syncCheckpointRepository.getNow(); - let batch: JobItem[] = items.flatMap(({ id: workflowId }) => - assetIds.map((assetId) => ({ - name: JobName.WorkflowAlbumAssetTrigger, - data: { workflowId, assetId, albumId, trigger }, - })), - ); - jobs.push(...batch); + if (!checkpoint) { + checkpoint = { lastUuid: now.nowId }; + await this.systemMetadataRepository.set(SystemMetadataKey.AlbumAssetWorkflowCheckpoint, checkpoint); } - await this.jobRepository.queueAll(jobs); + const workflows = new Map(); + + while (checkpoint.lastUuid < now.nowId) { + const albumAssets = await this.workflowRepository.getForAlbumAssetV1(checkpoint.lastUuid); + if (albumAssets.length === 0) { + break; + } + + const jobs: JobItem[] = []; + for (const albumAsset of albumAssets) { + const userId = albumAsset.asset?.ownerId; + + if (!workflows.has(userId)) { + workflows.set(userId, await this.workflowRepository.search({ userId, trigger })); + } + + for (const workflow of workflows.get(userId)) { + jobs.push({ + name: JobName.WorkflowAlbumAssetTrigger, + data: { + workflowId: workflow.id, + albumAsset: { asset: albumAsset.asset as any, album: { id: albumAsset.albumId } }, + userId: workflow.ownerId, + }, + }); + } + } + + await this.jobRepository.queueAll(jobs); + checkpoint!.lastUuid = albumAssets[0].updateId; + await this.systemMetadataRepository.set(SystemMetadataKey.AlbumAssetWorkflowCheckpoint, checkpoint); + } } - private writeAssetV1(assetId: string, type: WorkflowType) { + private writeAssetV1(assetId: string) { const assetService = BaseService.create(AssetService, this); - return async (auth: AuthDto, changes: WorkflowChanges) => { + return async (auth: AuthDto, changes: WorkflowChanges) => { const asset = changes.asset; if (!asset) { return; @@ -399,38 +417,37 @@ export class WorkflowExecutionService extends BaseService { authUserId: asset.ownerId, }; }, - write: this.writeAssetV1(assetId, type), + write: this.writeAssetV1(assetId), } satisfies ExecuteOptions; } + default: { + return; + } } }); } @OnJob({ name: JobName.WorkflowAlbumAssetTrigger, queue: QueueName.Workflow }) - handleAlbumAssetTrigger({ workflowId, assetId, albumId }: JobOf) { + handleAlbumAssetTrigger({ workflowId, userId, albumAsset }: JobOf) { return this.execute(workflowId, (type) => { switch (type) { - case WorkflowType.AssetV1: { + case WorkflowType.AlbumAssetV1: { return { - read: async () => { - const asset = await this.workflowRepository.getForAssetV1(assetId); - const workflow = await this.workflowRepository.get(workflowId); - - return { - data: { asset, album: { id: albumId } } as any, - authUserId: workflow!.ownerId, - }; - }, + read: () => + Promise.resolve({ + data: albumAsset, + authUserId: userId, + }), write: async (auth, changes) => { - const asset = await this.workflowRepository.getForAssetV1(assetId); - const workflow = await this.workflowRepository.get(workflowId); - - if (asset.ownerId === workflow!.ownerId) { - await this.writeAssetV1(assetId, type)(auth, changes); + if (albumAsset.asset.ownerId === userId) { + await this.writeAssetV1(albumAsset.asset.id)(auth, changes); } }, } satisfies ExecuteOptions; } + default: { + return; + } } }); } @@ -447,7 +464,8 @@ export class WorkflowExecutionService extends BaseService { // TODO infer from steps let type: T | undefined; for (const targetType of Object.values(WorkflowType)) { - const isMissing = workflow.steps.some((step) => !step.types.includes(targetType)); + const implied = withImpliedItems(targetType); + const isMissing = workflow.steps.some((step) => step.types.every((type) => !implied.includes(type))); if (!isMissing) { type = targetType as unknown as T; break; diff --git a/server/src/types.ts b/server/src/types.ts index c738fd2559..521aba6a05 100644 --- a/server/src/types.ts +++ b/server/src/types.ts @@ -1,4 +1,4 @@ -import { WorkflowTrigger } from '@immich/plugin-sdk'; +import { AlbumAssetV1, WorkflowTrigger } from '@immich/plugin-sdk'; import { ShallowDehydrateObject } from 'kysely'; import { SystemConfig } from 'src/config'; import { VECTOR_EXTENSIONS } from 'src/constants'; @@ -458,7 +458,7 @@ export type JobItem = // Workflow | { name: JobName.WorkflowAssetTrigger; data: { workflowId: string; assetId: string } } - | { name: JobName.WorkflowAlbumAssetTrigger; data: { workflowId: string; assetId: string; albumId: string } } + | { name: JobName.WorkflowAlbumAssetTrigger; data: { workflowId: string; userId: string; albumAsset: AlbumAssetV1 } } // Integrity | { name: JobName.IntegrityUntrackedFilesQueueAll; data?: IIntegrityJob } @@ -572,6 +572,7 @@ export interface SystemMetadata extends Record = { // [WorkflowType.AssetPersonV1]: [WorkflowType.AssetV1], }; -const withImpliedItems = (type: WorkflowType): WorkflowType[] => { +export const withImpliedItems = (type: WorkflowType): WorkflowType[] => { const childTypes = inferredMap[type]; const results = [type]; for (const child of childTypes) {