Adding an action to trigger manually

worker/src/index.ts

@@ -1,4 +1,4 @@
-import { from, fromEvent, mergeAll, mergeWith, Observable } from "rxjs";
+import { fromEvent, mergeAll, mergeWith, Observable } from "rxjs";
 import { fromFetch } from 'rxjs/fetch';
 import { Manager } from "./actions";
 import "./services";
@@ -7,6 +7,7 @@ import AbortController from 'abort-controller';
 import { Pipeline, pipelineFromApi, PipelineType } from "./models/pipeline";
 import { EventEmitter } from "events"
 import express from "express";
+import { UtilsService } from "./services/utils";
 
 // @ts-ignore
 global.fetch  = fetch;
@@ -43,6 +44,11 @@ app.delete("/workflow/:id", (req, res) => {
 	res.send()
 });
 
+app.put("/action/:id", (req, res) => {
+	UtilsService.triggerEvents.emit(req.params.id, {});
+	res.send();
+});
+
 app.listen(5000);
 
 

worker/src/models/pipeline.ts

@@ -5,6 +5,7 @@ export enum ServiceType {
 	Spotify,
 	Discord,
 	Anilist,
+	Utils,
 };
 
 
@@ -35,6 +36,8 @@ export enum PipelineType {
 	OnDiscordMention,
 	OnNewDiscordGuildMember,
 	OnDiscordGuildLeave,
+
+	OnTrigger,
 };
 
 export enum ReactionType {

worker/src/services/utils.ts

@@ -0,0 +1,20 @@
+import { fromEvent, Observable } from "rxjs";
+import { Pipeline, PipelineEnv, PipelineType, ServiceType } from "../models/pipeline";
+import { action, BaseService, service } from "../models/base-service";
+import { EventEmitter } from "events"
+
+@service(ServiceType.Utils)
+export class UtilsService extends BaseService {
+	public static triggerEvents = new EventEmitter();
+	private _pipeline: Pipeline;
+
+	constructor(pipeline: Pipeline) {
+		super();
+		this._pipeline = pipeline;
+	}
+
+	@action(PipelineType.OnTrigger, [])
+	manualTrigger(_: any): Observable<PipelineEnv> {
+		return fromEvent(UtilsService.triggerEvents, this._pipeline.id.toString());
+	}
+}