feat(server): asset added to album workflow trigger

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",

packages/plugin-core/src/index.ts

@@ -66,6 +66,8 @@ const methods = wrapper<Manifest>({
     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<Manifest>({
 
 const {
   assetAddToAlbums,
+  assetAlbumFilter,
   assetArchive,
   assetFavorite,
   assetFileFilter,
@@ -227,6 +230,7 @@ const {
 
 export {
   assetAddToAlbums,
+  assetAlbumFilter,
   assetArchive,
   assetFavorite,
   assetFileFilter,

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",

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 {

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<Asset, 'id' | 'ownerId'>; file?: UploadFile }];

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<T>(qb: SelectQueryBuilder<DB, 'asset', T>) {
+    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();
+      ]);
   }
 }

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;
   }

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<T extends WorkflowType> = {
 
 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<T extends WorkflowType>(assetId: string) {
     const assetService = BaseService.create(AssetService, this);
 
-    return async (auth: AuthDto, changes: WorkflowChanges<typeof type>) => {
+    return async (auth: AuthDto, changes: WorkflowChanges<T>) => {
       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<typeof type>(assetId),
           } satisfies ExecuteOptions<typeof type>;
         }
+        default: {
+          return;
+        }
       }
     });
   }
 
   @OnJob({ name: JobName.WorkflowAlbumAssetTrigger, queue: QueueName.Workflow })
-  handleAlbumAssetTrigger({ workflowId, assetId, albumId }: JobOf<JobName.WorkflowAlbumAssetTrigger>) {
+  handleAlbumAssetTrigger({ workflowId, userId, albumAsset }: JobOf<JobName.WorkflowAlbumAssetTrigger>) {
     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<typeof type>(albumAsset.asset.id)(auth, changes);
               }
             },
           } satisfies ExecuteOptions<typeof type>;
         }
+        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;

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<SystemMetadataKey, Record<string,
   [SystemMetadataKey.VersionCheckState]: VersionCheckMetadata;
   [SystemMetadataKey.MemoriesState]: MemoriesState;
   [SystemMetadataKey.IntegrityChecksumCheckpoint]: { date?: string };
+  [SystemMetadataKey.AlbumAssetWorkflowCheckpoint]: { lastUuid: string };
 }
 
 export type UserPreferences = {

server/src/utils/workflow.ts

@@ -19,7 +19,7 @@ const inferredMap: Record<WorkflowType, WorkflowType[]> = {
   // [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) {