| 1 | import type * as Effect from "effect/Effect"; |
| 2 | import type * as Stream from "effect/Stream"; |
| 3 | import { |
| 4 | consumeBusEvents, |
| 5 | type EventRecord, |
| 6 | type EventRouteProps, |
| 7 | } from "../EventBridge/EventSource.ts"; |
| 8 | |
| 9 | /** |
| 10 | * The `detail` payload AWS HealthImaging delivers to EventBridge. Which |
| 11 | * fields are present depends on the event kind — import job events carry |
| 12 | * `jobId`/`jobStatus`, image set events carry |
| 13 | * `imageSetId`/`imageSetState`/`imageSetWorkflowStatus`, data store events |
| 14 | * carry only the data store fields — so everything event-specific is |
| 15 | * optional. |
| 16 | */ |
| 17 | export interface MedicalImagingEventDetail { |
| 18 | /** Id of the data store the event originates from. */ |
| 19 | datastoreId?: string; |
| 20 | /** Data store lifecycle status on `Data Store …` events. */ |
| 21 | datastoreStatus?: string; |
| 22 | /** Import job id on `Import Job …` events. */ |
| 23 | jobId?: string; |
| 24 | /** Import job status (`SUBMITTED`, `IN_PROGRESS`, `COMPLETED`, `FAILED`). */ |
| 25 | jobStatus?: string; |
| 26 | /** Image set id on `Image Set …` events. */ |
| 27 | imageSetId?: string; |
| 28 | /** Image set state (`ACTIVE`, `LOCKED`, `DELETED`). */ |
| 29 | imageSetState?: string; |
| 30 | /** Image set workflow status (`CREATED`, `COPIED`, `UPDATED`, …). */ |
| 31 | imageSetWorkflowStatus?: string; |
| 32 | /** Additional event fields (the schema grows over time). */ |
| 33 | [key: string]: unknown; |
| 34 | } |
| 35 | |
| 36 | /** A HealthImaging EventBridge event delivered to the handler. */ |
| 37 | export type MedicalImagingEvent = EventRecord<MedicalImagingEventDetail>; |
| 38 | |
| 39 | export interface ImagingEventSourceProps extends EventRouteProps { |
| 40 | /** |
| 41 | * Logical id for the backing EventBridge rule. |
| 42 | * @default "MedicalImagingEvents" |
| 43 | */ |
| 44 | id?: string; |
| 45 | /** |
| 46 | * Only deliver events with these `detail-type` values, e.g. |
| 47 | * `["Import Job Completed", "Import Job Failed"]`. HealthImaging publishes |
| 48 | * `Data Store Creating|Created|Creation Failed|Deleting|Deleted`, |
| 49 | * `Import Job Submitted|In Progress|Completed|Failed`, and |
| 50 | * `Image Set Created|Copying|Copying With Read Only Access|Copied|Copy |
| 51 | * Failed|Updating|Updated|Update Failed|Deleting|Deleted`. |
| 52 | * @default all detail types |
| 53 | */ |
| 54 | detailTypes?: readonly string[]; |
| 55 | /** |
| 56 | * Restrict to events from specific data stores (matched against the event |
| 57 | * detail's `datastoreId`). |
| 58 | * @default all data stores |
| 59 | */ |
| 60 | datastoreIds?: readonly string[]; |
| 61 | } |
| 62 | |
| 63 | /** |
| 64 | * Event source connecting AWS HealthImaging state changes to the hosting |
| 65 | * compute. HealthImaging publishes data store lifecycle transitions, DICOM |
| 66 | * import job progress, and image set workflow changes to the account's |
| 67 | * default EventBridge bus (source `aws.medical-imaging`); this subscribes |
| 68 | * the host Function to those events so it can chain post-import processing |
| 69 | * (e.g. read the new image sets with `SearchImageSets` + |
| 70 | * `GetImageSetMetadata`) or alert on failures. |
| 71 | * |
| 72 | * HealthImaging publishes to EventBridge automatically — no additional |
| 73 | * resource is created besides the EventBridge rule targeting the host. |
| 74 | * Provide the host-specific implementation layer (e.g. |
| 75 | * `AWS.Lambda.EventSource`) on the Function effect. |
| 76 | * |
| 77 | * ### Consuming HealthImaging Events |
| 78 | * **Example:** React To Finished Imports |
| 79 | * ```typescript |
| 80 | * import * as AWS from "alchemy/AWS"; |
| 81 | * |
| 82 | * export default ImportReactor.make( |
| 83 | * { main: import.meta.url }, |
| 84 | * Effect.gen(function* () { |
| 85 | * yield* AWS.MedicalImaging.consumeImagingEvents( |
| 86 | * { detailTypes: ["Import Job Completed", "Import Job Failed"] }, |
| 87 | * (events) => |
| 88 | * Stream.runForEach(events, (event) => |
| 89 | * event["detail-type"] === "Import Job Failed" |
| 90 | * ? Effect.logError(`import ${event.detail.jobId} failed`) |
| 91 | * : Effect.log(`import ${event.detail.jobId} completed`), |
| 92 | * ), |
| 93 | * ); |
| 94 | * return {}; |
| 95 | * }).pipe(Effect.provide(AWS.Lambda.EventSource)), |
| 96 | * ); |
| 97 | * ``` |
| 98 | */ |
| 99 | export const consumeImagingEvents = <StreamReq = never, Req = never>( |
| 100 | props: ImagingEventSourceProps, |
| 101 | process: ( |
| 102 | events: Stream.Stream<MedicalImagingEvent, never, StreamReq>, |
| 103 | ) => Effect.Effect<void, never, Req>, |
| 104 | ) => |
| 105 | consumeBusEvents( |
| 106 | props.id ?? "MedicalImagingEvents", |
| 107 | { |
| 108 | source: ["aws.medical-imaging"], |
| 109 | ...(props.detailTypes !== undefined |
| 110 | ? { "detail-type": [...props.detailTypes] } |
| 111 | : {}), |
| 112 | ...(props.datastoreIds !== undefined |
| 113 | ? { detail: { datastoreId: [...props.datastoreIds] } } |
| 114 | : {}), |
| 115 | }, |
| 116 | { description: props.description, state: props.state }, |
| 117 | process, |
| 118 | ); |