Merge 3469183699 into 66d010381e

i18n/en.json

@@ -2138,6 +2138,8 @@
   "trash_page_info": "Trashed items will be permanently deleted after {days} days",
   "trashed_items_will_be_permanently_deleted_after": "Trashed items will be permanently deleted after {days, plural, one {# day} other {# days}}.",
   "trigger": "Trigger",
+  "trigger_album_asset_added": "Asset Added to Album",
+  "trigger_album_asset_added_description": "Triggered when an asset is added to an album",
   "trigger_asset_metadata_extraction": "Asset Metadata Extraction",
   "trigger_asset_metadata_extraction_description": "Triggered when the EXIF metadata of an asset is extracted",
   "trigger_asset_tagged": "Asset Tagged",

open-api/immich-openapi-specs.json

@@ -19436,6 +19436,8 @@
           "VersionCheck",
           "OcrQueueAll",
           "Ocr",
+          "WorkflowScan",
+          "WorkflowRun",
           "WorkflowAssetTrigger",
           "IntegrityUntrackedFilesQueueAll",
           "IntegrityUntrackedFiles",
@@ -28237,7 +28239,8 @@
         "enum": [
           "AssetCreate",
           "AssetMetadataExtraction",
-          "AssetTagged"
+          "AssetTagged",
+          "AlbumAssetAdded"
         ],
         "type": "string"
       },
@@ -28264,7 +28267,8 @@
       "WorkflowType": {
         "description": "Workflow type",
         "enum": [
-          "AssetV1"
+          "AssetV1",
+          "AlbumAssetV1"
         ],
         "type": "string"
       },

packages/plugin-core/manifest.json

@@ -519,6 +519,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

@@ -71,6 +71,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 } } };
@@ -233,6 +235,7 @@ const methods = wrapper<Manifest>({
 const {
   assetAddTags,
   assetAddToAlbums,
+  assetAlbumFilter,
   assetArchive,
   assetFavorite,
   assetFileFilter,
@@ -253,6 +256,7 @@ const {
 export {
   assetAddTags,
   assetAddToAlbums,
+  assetAlbumFilter,
   assetArchive,
   assetFavorite,
   assetFileFilter,

packages/plugin-sdk/src/types.ts

@@ -15,6 +15,7 @@ type DeepPartial<T> = T extends Date
 
 export type WorkflowEventMap = {
   [WorkflowType.AssetV1]: AssetV1;
+  [WorkflowType.AlbumAssetV1]: AlbumAssetV1;
   // [WorkflowType.AssetPersonV1]: AssetPersonV1;
 } & { [K in WorkflowType]: unknown };
 
@@ -24,6 +25,7 @@ export enum WorkflowTrigger {
   AssetCreate = 'AssetCreate',
   AssetMetadataExtraction = 'AssetMetadataExtraction',
   AssetTagged = 'AssetTagged',
+  AlbumAssetAdded = 'AlbumAssetAdded',
   // PersonRecognized = 'PersonRecognized',
 }
 
@@ -130,6 +132,12 @@ export type AssetV1 = {
   };
 };
 
+export type AlbumAssetV1 = AssetV1 & {
+  album: {
+    id: string;
+  };
+};
+
 // export type AssetPersonV1 = AssetV1 & {
 //   person: {
 //     id: string;

packages/sdk/src/fetch-client.ts

@@ -7464,12 +7464,14 @@ export enum PartnerDirection {
     SharedWith = "shared-with"
 }
 export enum WorkflowType {
-    AssetV1 = "AssetV1"
+    AssetV1 = "AssetV1",
+    AlbumAssetV1 = "AlbumAssetV1"
 }
 export enum WorkflowTrigger {
     AssetCreate = "AssetCreate",
     AssetMetadataExtraction = "AssetMetadataExtraction",
-    AssetTagged = "AssetTagged"
+    AssetTagged = "AssetTagged",
+    AlbumAssetAdded = "AlbumAssetAdded"
 }
 export enum QueueJobStatus {
     Active = "active",
@@ -7535,6 +7537,8 @@ export enum JobName {
     VersionCheck = "VersionCheck",
     OcrQueueAll = "OcrQueueAll",
     Ocr = "Ocr",
+    WorkflowScan = "WorkflowScan",
+    WorkflowRun = "WorkflowRun",
     WorkflowAssetTrigger = "WorkflowAssetTrigger",
     IntegrityUntrackedFilesQueueAll = "IntegrityUntrackedFilesQueueAll",
     IntegrityUntrackedFiles = "IntegrityUntrackedFiles",

server/src/enum.ts

@@ -345,6 +345,7 @@ export enum SystemMetadataKey {
   VersionCheckState = 'version-check-state',
   License = 'license',
   IntegrityChecksumCheckpoint = 'integrity-checksum-checkpoint',
+  WorkflowCheckpoint = 'workflow-checkpoint',
 }
 
 export enum UserMetadataKey {
@@ -903,6 +904,8 @@ export enum JobName {
   Ocr = 'Ocr',
 
   // Workflow
+  WorkflowScan = 'WorkflowScan',
+  WorkflowRun = 'WorkflowRun',
   WorkflowAssetTrigger = 'WorkflowAssetTrigger',
 
   // Integrity
@@ -1226,11 +1229,21 @@ export const WorkflowTriggerSchema = z
 
 export enum WorkflowType {
   AssetV1 = 'AssetV1',
+  AlbumAssetV1 = 'AlbumAssetV1',
   // AssetPersonV1 = 'AssetPersonV1',
 }
 
 export const WorkflowTypeSchema = z.enum(WorkflowType).describe('Workflow type').meta({ id: 'WorkflowType' });
 
+export enum WorkflowScanType {
+  AlbumAsset = 'AlbumAsset',
+}
+
+export const WorkflowScanTypeSchema = z
+  .enum(WorkflowScanType)
+  .describe('Workflow scan type')
+  .meta({ id: 'WorkflowScanType' });
+
 export enum CalendarHeatmapType {
   Upload = 'Upload',
   Taken = 'Taken',

server/src/queries/workflow.repository.sql

@@ -9,6 +9,7 @@ select
   "workflow"."enabled",
   "workflow"."createdAt",
   "workflow"."updatedAt",
+  "workflow"."ownerId",
   "workflow"."logging",
   (
     select
@@ -44,6 +45,7 @@ select
   "workflow"."enabled",
   "workflow"."createdAt",
   "workflow"."updatedAt",
+  "workflow"."ownerId",
   "workflow"."logging",
   (
     select
@@ -75,6 +77,7 @@ select
   "workflow"."id",
   "workflow"."name",
   "workflow"."trigger",
+  "workflow"."ownerId",
   "workflow"."logging",
   (
     select

server/src/repositories/event.repository.ts

@@ -40,6 +40,7 @@ type EventMap = {
   // album events
   AlbumUpdate: [{ id: string; userIds: string[]; recipientIds: string[] }];
   AlbumInvite: [{ id: string; userId: string; senderName: 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';
@@ -7,6 +7,7 @@ import { DummyValue, GenerateSql } from 'src/decorators';
 import { WorkflowGetLogsDto, WorkflowSearchDto } from 'src/dtos/workflow.dto';
 import { DB } from 'src/schema';
 import { WorkflowLogTable } from 'src/schema/tables/workflow-log.table';
+import { WorkflowQueueTable } from 'src/schema/tables/workflow-queue';
 import { WorkflowStepTable } from 'src/schema/tables/workflow-step.table';
 import { WorkflowTable } from 'src/schema/tables/workflow.table';
 import { withTags } from 'src/utils/database';
@@ -28,6 +29,7 @@ export class WorkflowRepository {
         'workflow.enabled',
         'workflow.createdAt',
         'workflow.updatedAt',
+        'workflow.ownerId',
         'workflow.logging',
       ])
       .select((eb) => [
@@ -68,7 +70,7 @@ export class WorkflowRepository {
   getForWorkflowRun(id: string) {
     return this.db
       .selectFrom('workflow')
-      .select(['workflow.id', 'workflow.name', 'workflow.trigger', 'workflow.logging'])
+      .select(['workflow.id', 'workflow.name', 'workflow.trigger', 'workflow.ownerId', 'workflow.logging'])
       .select((eb) => [
         jsonArrayFrom(
           eb
@@ -177,8 +179,38 @@ 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();
+  }
+
+  addToQueue(dto: Insertable<WorkflowQueueTable>[]) {
+    return this.db.insertInto('workflow_queue').values(dto).returning(['id']).execute();
+  }
+
+  getQueue(id: string) {
+    return this.db.selectFrom('workflow_queue').selectAll().where('id', '=', id).executeTakeFirstOrThrow();
+  }
+
+  removeFromQueue(workflowId: string) {
+    return this.db.deleteFrom('workflow_queue').where('id', '=', workflowId).execute();
+  }
+
+  private assetV1Query<T>(qb: SelectQueryBuilder<DB, 'asset', T>) {
+    return qb
       .leftJoin('asset_exif', 'asset_exif.assetId', 'asset.id')
       .select((eb) => [
         ...columns.workflowAssetV1,
@@ -220,8 +252,6 @@ export class WorkflowRepository {
             ])
             .whereRef('asset_exif.assetId', '=', 'asset.id'),
         ).as('exifInfo'),
-      ])
-      .where('id', '=', assetId)
-      .executeTakeFirstOrThrow();
+      ]);
   }
 }

server/src/schema/index.ts

@@ -88,6 +88,7 @@ import {
   VideoStreamVariantTable,
 } from 'src/schema/tables/video-stream.table';
 import { WorkflowLogTable } from 'src/schema/tables/workflow-log.table';
+import { WorkflowQueueTable } from 'src/schema/tables/workflow-queue';
 import { WorkflowStepTable } from 'src/schema/tables/workflow-step.table';
 import { WorkflowTable } from 'src/schema/tables/workflow.table';
 
@@ -280,5 +281,6 @@ export interface DB {
 
   workflow: WorkflowTable;
   workflow_step: WorkflowStepTable;
+  workflow_queue: WorkflowQueueTable;
   workflow_log: WorkflowLogTable;
 }

server/src/schema/migrations/1786346220385-AddWorkflowQueue.ts

@@ -0,0 +1,16 @@
+import { Kysely, sql } from 'kysely';
+
+export async function up(db: Kysely<any>): Promise<void> {
+  await sql`CREATE TABLE "workflow_queue" (
+  "id" uuid NOT NULL DEFAULT uuid_generate_v4(),
+  "workflowId" uuid NOT NULL,
+  "data" jsonb NOT NULL,
+  CONSTRAINT "workflow_queue_workflowId_fkey" FOREIGN KEY ("workflowId") REFERENCES "workflow" ("id") ON UPDATE CASCADE ON DELETE CASCADE,
+  CONSTRAINT "workflow_queue_pkey" PRIMARY KEY ("id")
+);`.execute(db);
+  await sql`CREATE INDEX "workflow_queue_workflowId_idx" ON "workflow_queue" ("workflowId");`.execute(db);
+}
+
+export async function down(db: Kysely<any>): Promise<void> {
+  await sql`DROP TABLE "workflow_queue";`.execute(db);
+}

server/src/schema/tables/workflow-queue.ts

@@ -0,0 +1,14 @@
+import { Column, ForeignKeyColumn, Generated, PrimaryGeneratedColumn, Table } from '@immich/sql-tools';
+import { WorkflowTable } from 'src/schema/tables/workflow.table';
+
+@Table('workflow_queue')
+export class WorkflowQueueTable {
+  @PrimaryGeneratedColumn('uuid')
+  id!: Generated<string>;
+
+  @ForeignKeyColumn(() => WorkflowTable, { onDelete: 'CASCADE', onUpdate: 'CASCADE' })
+  workflowId!: string;
+
+  @Column({ type: 'jsonb' })
+  data!: unknown[];
+}

server/src/services/album.service.ts

@@ -194,6 +194,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');
     }
 
     return results;
@@ -222,7 +223,7 @@ export class AlbumService extends BaseService {
     }
 
     const albumAssetValues: { albumId: string; assetId: string }[] = [];
-    const events: { id: string; userIds: string[]; recipientIds: string[] }[] = [];
+    const updateEvents: { id: string; userIds: string[]; recipientIds: string[] }[] = [];
     for (const albumId of allowedAlbumIds) {
       const existingAssetIds = await this.albumRepository.getAssetIds(albumId, [...allowedAssetIds]);
       const notPresentAssetIds = [...allowedAssetIds.difference(existingAssetIds)];
@@ -247,13 +248,14 @@ export class AlbumService extends BaseService {
       );
       const userIds = album.albumUsers.map(({ user }) => user.id);
       const recipientIds = userIds.filter((userId) => userId !== auth.user.id);
-      events.push({ id: albumId, userIds, recipientIds });
+      updateEvents.push({ id: albumId, userIds, recipientIds });
     }
 
     await this.albumRepository.addAssetIdsToAlbums(albumAssetValues);
-    for (const event of events) {
+    for (const event of updateEvents) {
       await this.eventRepository.emit('AlbumUpdate', event);
     }
+    await this.eventRepository.emit('AlbumAssetsAdded');
 
     return results;
   }

server/src/services/workflow-execution.service.ts

@@ -1,5 +1,6 @@
 import { CurrentPlugin } from '@extism/extism';
 import {
+  AlbumAssetV1,
   WorkflowChanges,
   WorkflowEventData,
   WorkflowEventPayload,
@@ -22,7 +23,9 @@ import {
   JobName,
   JobStatus,
   QueueName,
+  SystemMetadataKey,
   WorkflowResult,
+  WorkflowScanType,
   WorkflowType,
 } from 'src/enum';
 import { ArgOf } from 'src/repositories/event.repository';
@@ -31,6 +34,7 @@ import { AssetService } from 'src/services/asset.service';
 import { BaseService } from 'src/services/base.service';
 import { TagService } from 'src/services/tag.service';
 import { JobOf } from 'src/types';
+import { withImpliedItems } from 'src/utils/workflow';
 
 const dummy = () => {
   throw new Error(
@@ -51,6 +55,7 @@ type HostContext = {
 
 export class WorkflowExecutionService extends BaseService {
   private jwtSecret!: string;
+  private scanning = false;
 
   @OnEvent({ name: 'AppBootstrap', priority: BootstrapEventPriority.PluginSync, workers: [ImmichWorker.Microservices] })
   async onPluginSync() {
@@ -318,6 +323,11 @@ export class WorkflowExecutionService extends BaseService {
     return this.onAssetTrigger({ userId, assetId, trigger: WorkflowTrigger.AssetMetadataExtraction });
   }
 
+  @OnEvent({ name: 'AlbumAssetsAdded' })
+  onAlbumAssetsAdded() {
+    return this.jobRepository.queue({ name: JobName.WorkflowScan, data: { type: WorkflowScanType.AlbumAsset } });
+  }
+
   @OnEvent({ name: 'AssetTag' })
   onAssetTagged({ assetId, userId }: ArgOf<'AssetTag'>) {
     return this.onAssetTrigger({ userId, assetId, trigger: WorkflowTrigger.AssetTagged });
@@ -333,11 +343,108 @@ export class WorkflowExecutionService extends BaseService {
     );
   }
 
+  @OnJob({ name: JobName.WorkflowScan, queue: QueueName.Workflow })
+  private async scan({ type }: JobOf<JobName.WorkflowScan>) {
+    if (this.scanning) {
+      return JobStatus.Skipped;
+    }
+
+    this.scanning = true;
+
+    if (type !== WorkflowScanType.AlbumAsset) {
+      return;
+    }
+
+    let checkpoint = await this.systemMetadataRepository.get(SystemMetadataKey.WorkflowCheckpoint);
+    const now = await this.syncCheckpointRepository.getNow();
+
+    if (!checkpoint) {
+      checkpoint = { albumAssetUuid: now.nowId };
+      await this.systemMetadataRepository.set(SystemMetadataKey.WorkflowCheckpoint, checkpoint);
+    }
+
+    const workflows = new Map();
+
+    while (checkpoint.albumAssetUuid < now.nowId) {
+      const albumAssets = await this.workflowRepository.getForAlbumAssetV1(checkpoint.albumAssetUuid);
+      if (albumAssets.length === 0) {
+        break;
+      }
+
+      const jobs = new Map<string, AlbumAssetV1[]>();
+      for (const albumAsset of albumAssets) {
+        const userId = albumAsset.asset?.ownerId;
+
+        if (!workflows.has(userId)) {
+          workflows.set(
+            userId,
+            await this.workflowRepository.search({ userId, trigger: WorkflowTrigger.AlbumAssetAdded }),
+          );
+        }
+
+        for (const workflow of workflows.get(userId)) {
+          if (!jobs.has(workflow.id)) {
+            jobs.set(workflow.id, []);
+          }
+
+          jobs.get(workflow.id)!.push({ asset: albumAsset.asset as any, album: { id: albumAsset.albumId } });
+        }
+      }
+
+      const queues = await this.workflowRepository.addToQueue(
+        jobs
+          .entries()
+          .map(([workflowId, data]) => ({ workflowId, data }))
+          .toArray(),
+      );
+      await this.jobRepository.queueAll(queues.map(({ id }) => ({ name: JobName.WorkflowRun, data: { queueId: id } })));
+
+      checkpoint!.albumAssetUuid = albumAssets[0].updateId;
+      await this.systemMetadataRepository.set(SystemMetadataKey.WorkflowCheckpoint, checkpoint);
+    }
+
+    this.scanning = false;
+    return JobStatus.Success;
+  }
+
+  private writeAssetV1<T extends WorkflowType>(assetId: string) {
+    const assetService = BaseService.create(AssetService, this);
+
+    return async (auth: AuthDto, changes: WorkflowChanges<T>) => {
+      const asset = changes.asset;
+      if (!asset) {
+        return;
+      }
+
+      await assetService.update(auth, assetId, {
+        isFavorite: asset.isFavorite,
+        visibility: asset.visibility,
+        dateTimeOriginal: asset.exifInfo?.dateTimeOriginal ?? undefined,
+        // TODO allow setting to null
+        longitude: asset.exifInfo?.longitude ?? undefined,
+        // TODO allow setting to null
+        latitude: asset.exifInfo?.latitude ?? undefined,
+        // TODO allow setting to null
+        description: asset.exifInfo?.description ?? undefined,
+        rating: asset.exifInfo?.rating,
+
+        // TODO add to update dto
+        // make: asset.exifInfo?.make,
+        // model: asset.exifInfo?.model,
+        // city: asset.exifInfo?.city,
+        // state: asset.exifInfo?.state,
+        // country: asset.exifInfo?.country,
+        // lensModel: asset.exifInfo?.lensModel,
+        // fNumber: asset.exifInfo?.fNumber,
+        // fps: asset.exifInfo?.fps,
+        // iso: asset.exifInfo?.iso,
+      });
+    };
+  }
+
   @OnJob({ name: JobName.WorkflowAssetTrigger, queue: QueueName.Workflow })
   handleAssetTrigger({ workflowId, assetId }: JobOf<JobName.WorkflowAssetTrigger>) {
     return this.execute(workflowId, (type) => {
-      const assetService = BaseService.create(AssetService, this);
-
       switch (type) {
         case WorkflowType.AssetV1: {
           return {
@@ -349,42 +456,49 @@ export class WorkflowExecutionService extends BaseService {
                 entityId: asset.id,
               };
             },
-            write: async (auth, changes) => {
-              const asset = changes.asset;
-              if (!asset) {
-                return;
-              }
-
-              await assetService.update(auth, assetId, {
-                isFavorite: asset.isFavorite,
-                visibility: asset.visibility,
-                dateTimeOriginal: asset.exifInfo?.dateTimeOriginal ?? undefined,
-                // TODO allow setting to null
-                longitude: asset.exifInfo?.longitude ?? undefined,
-                // TODO allow setting to null
-                latitude: asset.exifInfo?.latitude ?? undefined,
-                // TODO allow setting to null
-                description: asset.exifInfo?.description ?? undefined,
-                rating: asset.exifInfo?.rating,
-
-                // TODO add to update dto
-                // make: asset.exifInfo?.make,
-                // model: asset.exifInfo?.model,
-                // city: asset.exifInfo?.city,
-                // state: asset.exifInfo?.state,
-                // country: asset.exifInfo?.country,
-                // lensModel: asset.exifInfo?.lensModel,
-                // fNumber: asset.exifInfo?.fNumber,
-                // fps: asset.exifInfo?.fps,
-                // iso: asset.exifInfo?.iso,
-              });
-            },
+            write: this.writeAssetV1<typeof type>(assetId),
           } satisfies ExecuteOptions<typeof type>;
         }
+        default: {
+          return;
+        }
       }
     });
   }
 
+  @OnJob({ name: JobName.WorkflowRun, queue: QueueName.Workflow })
+  async runQueue({ queueId }: JobOf<JobName.WorkflowRun>) {
+    const queue = await this.workflowRepository.getQueue(queueId);
+
+    for (const item of queue.data) {
+      await this.execute(queue.workflowId, (type) => {
+        switch (type) {
+          case WorkflowType.AssetV1:
+          case WorkflowType.AlbumAssetV1: {
+            return {
+              read: async () => {
+                const workflow = await this.workflowRepository.getForWorkflowRun(queue.workflowId);
+                return {
+                  data: item as any,
+                  authUserId: workflow!.ownerId,
+                };
+              },
+              write: async (auth, changes) => {
+                const workflow = await this.workflowRepository.getForWorkflowRun(queue.workflowId);
+                if ((item as AlbumAssetV1).asset.ownerId === workflow?.ownerId) {
+                  await this.writeAssetV1<typeof type>((item as AlbumAssetV1).asset.id)(auth, changes);
+                }
+              },
+            } satisfies ExecuteOptions<typeof type>;
+          }
+          default: {
+            return;
+          }
+        }
+      });
+    }
+  }
+
   private async execute<T extends WorkflowType>(
     workflowId: string,
     getHandler: (type: T) => ExecuteOptions<T> | undefined,
@@ -397,7 +511,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

@@ -30,6 +30,7 @@ import {
   SystemMetadataKey,
   TranscodeTarget,
   UserMetadataKey,
+  WorkflowScanType,
   WorkflowType,
 } from 'src/enum';
 import { Mocked } from 'vitest';
@@ -459,6 +460,8 @@ export type JobItem =
 
   // Workflow
   | { name: JobName.WorkflowAssetTrigger; data: { workflowId: string; assetId: string } }
+  | { name: JobName.WorkflowRun; data: { queueId: string } }
+  | { name: JobName.WorkflowScan; data: { type: WorkflowScanType } }
 
   // Integrity
   | { name: JobName.IntegrityUntrackedFilesQueueAll; data?: IIntegrityJob }
@@ -572,6 +575,7 @@ export interface SystemMetadata extends Record<SystemMetadataKey, Record<string,
   [SystemMetadataKey.VersionCheckState]: VersionCheckMetadata;
   [SystemMetadataKey.MemoriesState]: MemoriesState;
   [SystemMetadataKey.IntegrityChecksumCheckpoint]: { date?: string };
+  [SystemMetadataKey.WorkflowCheckpoint]: { albumAssetUuid: string };
 }
 
 export type UserPreferences = {

server/src/utils/workflow.ts

@@ -7,6 +7,7 @@ export const triggerMap: Record<WorkflowTrigger, WorkflowType[]> = {
   // [WorkflowTrigger.PersonRecognized]: [WorkflowType.AssetPersonV1],
   [WorkflowTrigger.AssetMetadataExtraction]: [WorkflowType.AssetV1],
   [WorkflowTrigger.AssetTagged]: [WorkflowType.AssetV1],
+  [WorkflowTrigger.AlbumAssetAdded]: [WorkflowType.AlbumAssetV1],
 };
 
 export const getWorkflowTriggers = () =>
@@ -15,10 +16,11 @@ export const getWorkflowTriggers = () =>
 /** some types extend other types and have implied compatibility */
 const inferredMap: Record<WorkflowType, WorkflowType[]> = {
   [WorkflowType.AssetV1]: [],
+  [WorkflowType.AlbumAssetV1]: [WorkflowType.AssetV1],
   // [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) {
@@ -30,8 +32,8 @@ const withImpliedItems = (type: WorkflowType): WorkflowType[] => {
 
 export const isMethodCompatible = (pluginMethod: { types: WorkflowType[] }, trigger: WorkflowTrigger) => {
   const validTypes = triggerMap[trigger];
-  const pluginCompatibility = pluginMethod.types.map((type) => withImpliedItems(type));
-  for (const requested of validTypes) {
+  const pluginCompatibility = validTypes.map((type) => withImpliedItems(type));
+  for (const requested of pluginMethod.types) {
     for (const pluginCompatibilityGroup of pluginCompatibility) {
       if (pluginCompatibilityGroup.includes(requested)) {
         return true;

web/src/lib/utils/workflow.ts

@@ -16,6 +16,9 @@ export const getTriggerName = ($t: MessageFormatter, type: WorkflowTrigger) => {
     case WorkflowTrigger.AssetTagged: {
       return $t('trigger_asset_tagged');
     }
+    case WorkflowTrigger.AlbumAssetAdded: {
+      return $t('trigger_album_asset_added');
+    }
     default: {
       return type;
     }
@@ -36,6 +39,9 @@ export const getTriggerDescription = ($t: MessageFormatter, type: WorkflowTrigge
     case WorkflowTrigger.AssetTagged: {
       return $t('trigger_asset_tagged_description');
     }
+    case WorkflowTrigger.AlbumAssetAdded: {
+      return $t('trigger_album_asset_added_description');
+    }
     default: {
       return type;
     }