sdk: add BaseMultiCallStreamUseCase rucio#226

.github/workflows/sdk.yml

@@ -0,0 +1,26 @@
+name: SDK Tests
+on:
+  push:
+    branches:
+      - master
+  pull_request:
+    branches:
+      - master
+jobs:
+  test:
+    name: Test WebUI SDK
+    strategy:
+      matrix:
+        node-version: [16.14.0, 18.5.0]
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Use Node.js 18.x
+        uses: actions/setup-node@v3
+        with:
+          node-version: ${{ matrix.node-version }}
+      - name: Install dependencies
+        run: npm install
+      - name: Run SDK Tests
+        run: npm run test:sdk
+

jest.config.js

@@ -3,5 +3,6 @@ module.exports = {
       '<rootDir>/test/api/jest.api.config.js',
       '<rootDir>/test/component/jest.component.config.js',
       '<rootDir>/test/gateway/jest.gateway.config.js',
+      '<rootDir>/test/sdk/jest.sdk.config.js',
     ],
   }

package.json

@@ -12,6 +12,7 @@
     "test:api": "jest --testPathPattern=test/api --projects test/api/jest.api.config.js",
     "test:component": "jest --testPathPattern=test/component --projects test/component/jest.component.config.js",
     "test:gateway": "jest --testPathPattern=test/gateway --projects test/gateway/jest.gateway.config.js",
+    "test:sdk": "jest --testPathPattern=test/sdk --projects test/sdk/jest.sdk.config.js",
     "storybook": "storybook dev -p 6006",
     "build-storybook": "storybook build",
     "build-tailwind": "tailwindcss -i src/component-library/tailwind.css -o src/component-library/outputtailwind.css"

src/lib/core/use-case/list-dids-usecase-pipeline-element.ts

@@ -0,0 +1,29 @@
+import { BaseMultiCallUseCasePipelineElement } from "@/lib/sdk/usecase-stream-element";
+import { inject } from "inversify";
+import { ListDIDDTO } from "../dto/did-dto";
+import { ListDIDsError, ListDIDsRequest, ListDIDsResponse } from "../usecase-models/list-dids-usecase-models";
+
+class ListDIDsGetListOfDIDs extends BaseMultiCallUseCasePipelineElement<
+    ListDIDsRequest,
+    ListDIDsResponse,
+    ListDIDsError,
+    ListDIDDTO
+>{
+    constructor(
+        @inject('didGateway') private didGateway: DIDGatewayOutputPort,
+    ) {
+        super(requestModel, responseModel)
+    }
+    makeGatewayRequest(requestModel: ListDIDsRequest): Promise<ListDIDDTO> {
+        throw new Error("Method not implemented.");
+    }
+    processDTO(dto: ListDIDDTO): { data: ListDIDsResponse | ListDIDsError; status: "success" | "error"; } {
+        throw new Error("Method not implemented.");
+    }
+    handleGatewayError(error: ListDIDDTO): ListDIDsError {
+        throw new Error("Method not implemented.");
+    }
+    transformResponseModel(responseModel: ListDIDsResponse, dto: ListDIDDTO): ListDIDsResponse {
+        throw new Error("Method not implemented.");
+    }
+}

src/lib/core/use-case/list-dids-usecase.ts

@@ -64,6 +64,7 @@ class ListDIDsUseCase extends BaseStreamingUseCase<ListDIDsRequest, ListDIDsResp
             length: 0,
             bytes: 0,
         }
+        this.didGateway.getDID()
         return {
             data: responseModel,
             status: 'success',

src/lib/core/use-case/list-dids-usecase2.ts

@@ -0,0 +1,77 @@
+import { injectable } from "inversify";
+import type { ListDIDsInputPort, ListDIDsOutputPort } from "@/lib/core/port/primary/list-dids-ports";
+import type DIDGatewayOutputPort from "@/lib/core/port/secondary/did-gateway-output-port";
+import { ListDIDDTO } from "../dto/did-dto";
+import { ListDIDsError, ListDIDsRequest, ListDIDsResponse } from "../usecase-models/list-dids-usecase-models";
+import { parseDIDString } from "@/lib/common/did-utils";
+import { DID } from "../entity/rucio";
+import { BaseMultiCallStreamableUseCase, BaseStreamingUseCase } from "@/lib/sdk/usecase";
+import { AuthenticatedRequestModel } from "@/lib/sdk/usecase-models";
+
+@injectable()
+class ListDIDsUseCase2 extends BaseMultiCallStreamableUseCase<ListDIDsRequest, ListDIDsResponse, ListDIDsError, DID> implements ListDIDsInputPort {
+    validateFinalResponseModel(responseModel: ListDIDsResponse): { isValid: boolean; errorModel?: ListDIDsError | undefined; } {
+        throw new Error("Method not implemented.");
+    }
+    constructor(
+        protected presenter: ListDIDsOutputPort,
+        private didGateway: DIDGatewayOutputPort,
+    ) {
+        super(presenter)
+        this.didGateway = didGateway;
+    }
+
+    validateRequestModel(requestModel: AuthenticatedRequestModel<ListDIDsRequest>): ListDIDsError | undefined {
+        let scope: string;
+        let name: string;
+        try{
+            let didComponents = parseDIDString(requestModel.query);
+            scope = didComponents.scope;
+            name = didComponents.name;
+        } catch (error: any) {
+            return {
+                status: 'error',
+                error: 'Invalid DID Query',
+                message: (error as Error).message,
+            } as ListDIDsError
+        }
+    }
+
+    // async makeGatewayRequest(requestModel: AuthenticatedRequestModel<ListDIDsRequest>): Promise<ListDIDDTO> {
+    //     const { scope, name } = parseDIDString(requestModel.query);
+    //     const listDIDDTO: ListDIDDTO = await this.didGateway.listDIDs(requestModel.rucioAuthToken, scope, name, requestModel.type);
+    //     return listDIDDTO;
+    // }
+
+    // handleGatewayError(error: ListDIDDTO): ListDIDsError {
+    //     let errorType = 'Unknown Error'
+    //     if(error.error === 'Invalid Auth Token') {
+    //         errorType = 'Invalid Request'
+    //     }
+    //     else if(error.error !== 'Unknown Error') {
+    //         errorType = 'Invalid DID Query'
+    //     }
+
+    //     return {
+    //         error: errorType,
+    //         message: `${error.error}: ${error.message}`,
+    //     } as ListDIDsError
+    // }
+
+    // processStreamedData(dto: DID): { data: ListDIDsResponse | ListDIDsError; status: "success" | "error"; } {
+    //     const responseModel: ListDIDsResponse = {
+    //         status: 'success',
+    //         name: dto.name,
+    //         scope: dto.scope,
+    //         did_type: dto.did_type,
+    //         length: 0,
+    //         bytes: 0,
+    //     }
+    //     return {
+    //         data: responseModel,
+    //         status: 'success',
+    //     }
+    // }
+}
+
+export default ListDIDsUseCase2;

src/lib/sdk/primary-ports.ts

@@ -1,5 +1,7 @@
 import { PassThrough, Transform } from 'stream'
-import { AuthenticatedRequestModel } from './usecase-models'
+import { BaseDTO } from './dto'
+import { AuthenticatedRequestModel, BaseErrorResponseModel, BaseResponseModel } from './usecase-models'
+import { BaseMultiCallUseCasePipelineElement } from './usecase-stream-element'
 import { TWebResponse } from './web'
 
 /**
@@ -27,6 +29,25 @@ export interface BaseStreamableInputPort<AuthenticatedRequestModel>
     execute(requestModel: AuthenticatedRequestModel): Promise<void>
 }
 
+/**
+ * A base interface for multi-call streamable input ports. A streamable input port provides a pipeline of {@link BaseMultiCallUseCasePipelineElement} elements that are used to process the request model.
+ * These pipeline elements recieve the request model and the latest response model and return a new response model.
+ * The pipeline elements are executed in the order they are provided.
+ * @typeparam AuthenticatedRequestModel The type of the authenticated request model for the input port.
+ * @typeparam TResponseModel The type of the response model for the input port.
+ * @typeparam TErrorModel The type of the error model for the input port.
+ */
+export interface BaseMultiCallStreamableInputPort<AuthenticatedRequestModel, TResponseModel extends BaseResponseModel, TErrorModel extends BaseErrorResponseModel>{
+    /**
+     * Validates the final response model.
+     * @param responseModel The response model to validate.
+     */
+    validateFinalResponseModel(responseModel: TResponseModel): {
+        isValid: boolean,
+        errorModel?: TErrorModel
+    }
+}
+
 /**
  * A base interface for output ports.
  * @typeparam TResponseModel The type of the response model for the output port.

src/lib/sdk/usecase-stream-element.ts

@@ -0,0 +1,85 @@
+import { Transform } from 'stream';
+import { BaseDTO } from './dto';
+import { BaseErrorResponseModel, BaseResponseModel } from './usecase-models';
+
+export abstract class BaseMultiCallUseCasePipelineElement<TRequestModel, TResponseModel extends BaseResponseModel, TErrorModel extends BaseErrorResponseModel, TDTO extends BaseDTO> extends Transform {
+    protected requestModel: TRequestModel
+    protected responseModel: TResponseModel
+
+    constructor(requestModel: TRequestModel, responseModel: TResponseModel) {
+        super({ objectMode: true });
+        this.requestModel = requestModel
+        this.responseModel = responseModel
+    }
+
+    /**
+     * Makes a gateway request with the given request model.
+     * @param requestModel The request model to send to the gateway.
+     * @returns A promise that resolves with the DTO returned by the gateway.
+     */
+    abstract makeGatewayRequest(requestModel: TRequestModel): Promise<TDTO>
+
+    /**
+     * Handles the DTO returned by the gateway.
+     * This method is called when the gateway returns a DTO with a status of `success`.
+     * @param dto The DTO returned by the gateway.
+     * @returns An object that contains the response or error model and the status of processing the DTO.
+     */
+    abstract processDTO(dto: TDTO): {
+        data: TResponseModel | TErrorModel
+        status: 'success' | 'error'
+    }
+
+    /**
+     * Handles a gateway error by converting it to an error model.
+     * This method is called when the gateway returns a DTO with a status of `error`.
+     * @param error The DTO returned by the gateway.
+     * @returns An error model that represents the gateway error.
+     */
+    abstract handleGatewayError(error: TDTO): TErrorModel
+
+    /**
+     * Modifies the response model of the use case with the data returned by the gateway.
+     * @param responseModel The response model to transform.
+     * @param dto The valid DTO returned by the gateway.
+     */
+    abstract transformResponseModel(responseModel: TResponseModel, dto: TDTO): TResponseModel
+
+    /**
+     * Processes the response and DTO returned by the gateway.
+     * @param response The DTO returned by the gateway.
+     * @returns The ResponseModel or ErrorModel returned, based on processing of the DTO.
+     */
+    processGatewayResponse(response: TDTO): TResponseModel | TErrorModel {
+        if (response.status === 'success') {
+            const { status, data } = this.processDTO(response)
+            if( status === 'success') {
+                return data as TResponseModel
+            }
+            return data as TErrorModel
+        }
+        const errorModel: TErrorModel = this.handleGatewayError(response)
+        return errorModel
+    }
+
+    async exceute(): Promise<TResponseModel | TErrorModel> {
+        const dto = await this.makeGatewayRequest(this.requestModel)
+        const data: TResponseModel | TErrorModel = this.processGatewayResponse(dto)
+        if(data.status === 'error') {
+            return Promise.reject(data)
+        }
+        const transformedResponseModel = this.transformResponseModel(this.responseModel, dto)
+        return Promise.resolve(transformedResponseModel)
+    }
+
+    _transform(chunk: TResponseModel, encoding: BufferEncoding, callback: (error?: Error, data?: TResponseModel | TErrorModel) => void): void {
+        this.responseModel = chunk
+        this.exceute().then((data) => {
+            this.push(data)
+            callback(undefined, data) // TODO check if data is not pushed twice
+        }).catch((error) => {
+            this.emit('error', error)
+            callback(error)
+        })
+    }
+}

src/lib/sdk/usecase.ts

@@ -1,13 +1,16 @@
 import {
     BaseAuthenticatedInputPort,
     BaseInputPort,
+    BaseMultiCallStreamableInputPort,
     BaseOutputPort,
     BaseStreamableInputPort,
     BaseStreamingOutputPort,
 } from './primary-ports'
 import { AuthenticatedRequestModel, BaseErrorResponseModel, BaseResponseModel } from './usecase-models'
 import { Transform, TransformCallback } from 'stream'
 import { BaseDTO, BaseStreamableDTO } from './dto'
+import { BaseMultiCallUseCasePipelineElement } from './usecase-stream-element'
+import { BaseStreamingPresenter } from './presenter'
 
 /**
  * A type that represents a simple use case that does not require authentication.
@@ -235,4 +238,70 @@ export abstract class BaseStreamingUseCase<TRequestModel,
         }
         callback()
     }
+}
+
+
+export abstract class BaseMultiCallStreamableUseCase<TRequestModel, 
+TResponseModel extends BaseResponseModel, 
+TErrorModel extends BaseErrorResponseModel,
+TStreamData> 
+extends Transform implements BaseMultiCallStreamableInputPort<AuthenticatedRequestModel<TRequestModel>, TResponseModel, TErrorModel> {
+    protected presenter: BaseStreamingPresenter<TResponseModel, TStreamData, TErrorModel>
+    protected pipelineElements: BaseMultiCallUseCasePipelineElement<AuthenticatedRequestModel<TRequestModel>, TResponseModel, TErrorModel, any>[] = []
+    constructor(
+        presenter: BaseStreamingPresenter<TResponseModel, TStreamData, TErrorModel>,
+        pipelineElements: BaseMultiCallUseCasePipelineElement<AuthenticatedRequestModel<TRequestModel>, TResponseModel, TErrorModel, any>[]
+    ) {
+        super({ objectMode: true })
+        this.presenter = presenter
+        this.pipelineElements = pipelineElements
+    }
+
+    abstract validateRequestModel(requestModel: TRequestModel): TErrorModel | undefined;
+
+    setupPipeline(pipelineElements: BaseMultiCallUseCasePipelineElement<AuthenticatedRequestModel<TRequestModel>, TResponseModel, TErrorModel, any>[]): void {
+        // loop over pipeline elements and pipe them together. Pipe the last element to this object
+        // for validation and pipe this to presenter
+        for (let i = 0; i < pipelineElements.length; i++) {
+            const pipelineElement = pipelineElements[i]
+            if(i === pipelineElements.length - 1) {
+                pipelineElement.on('error', (error) => this.handleError(error))
+                .pipe(this.presenter)
+            }
+            else {
+                const previousPipelineElement = pipelineElements[i - 1]
+                previousPipelineElement.pipe(pipelineElement)
+            }
+        }
+    }
+    abstract validateFinalResponseModel(responseModel: TResponseModel): { isValid: boolean; errorModel?: TErrorModel | undefined }
+    abstract handleError(error: Error): void
+
+    async execute(requestModel: AuthenticatedRequestModel<TRequestModel>): Promise<void> {
+        const validationError = this.validateRequestModel(requestModel)
+        if (validationError) {
+            this.presenter.presentError(validationError)
+        }
+        try {
+            this.setupPipeline(this.pipelineElements)
+        } catch (error: any) {
+            // TODO here we catch any critical errors that occur during pipeline setup or execution
+            const errorModel = error as TErrorModel
+
+            this.presenter.presentError(error)
+        }
+    }
+
+    _transform(chunk: any, encoding: BufferEncoding, callback: TransformCallback): void {
+        const data = JSON.parse(chunk)
+        const {requestModel, responseModel, errorModel} = data
+        const { isValid, errorModel: finalErrorModel } = this.validateFinalResponseModel(responseModel)
+        if(isValid) {
+            this.push(JSON.stringify(responseModel))
+        }
+        else {
+            this.emit('error', JSON.stringify(finalErrorModel))
+        }
+        callback()
+    }
 }