feat(temporal): add new module for asynchronous workflows with temporal.io

[ochrstn]

Information

Add a new module for asynchronous workflows with temporal.io

Type	Breaking change
Feature	no 🙏

Usage example

for detailed explanation see also readme/tutorial

1. Decorate classes and methods:

import {Temporal, Activity} from "@tsed/temporal";

@Temporal()
export class UserOnboardingActivities {
  constructor(private userService: UserService, private emailService: EmailService) {}

  @Activity()
  async sendVerificationEmail(email: string) {
    return this.emailService.sendVerificationEmail(email);
  }

  @Activity()
  async activateUser(email: string) {
    return this.userService.activateUser(email);
  }

  @Activity()
  async sendWelcomeEmail(email: string) {
    return this.emailService.sendWelcomeEmail(email);
  }

  @Activity()
  async sendFollowUpEmail(email: string) {
    return this.emailService.sendFollowUpEmail(email);
  }

  @Activity()
  async deleteUser(email: string) {
    return this.userService.deleteUser(email);
  }
}

2. Write a workflow:

import {proxyActivities, defineSignal, setHandler, condition, sleep} from "@temporalio/workflow";
import {Activities} from "../activities";

export const isVerifiedSignal = defineSignal("verificationSignal");

export async function onboardUser(email: string): Promise<string> {
  const {sendVerificationEmail, activateUser, sendWelcomeEmail, sendFollowUpEmail, deleteUser} = proxyActivities<Activities>({
    startToCloseTimeout: "1 minute"
  });

  let isVerified = false;
  setHandler(isVerifiedSignal, () => {
    isVerified = true;
  });

  // 1. Send verification email
  await sendVerificationEmail(email);

  // 2. Wait for verification ...
  const verifiedInTime = await condition(() => isVerified, "1w" /* or for a timeout */);
  if (!verifiedInTime) {
    // 3a. If not verified in time, delete user
    await deleteUserAndTenant(email);
    return false;
  }

  // 3b. If verified in time, send welcome email
  await sendWelcomeEmail(email);

  // 4. Send follow up email after one day
  await sleep("1d"); // special sleep function by temporal
  await sendFollowUpEmail(email);
}

3. Inject client:

import {Service, AfterRoutesInit} from "@tsed/common";
import {TemporalClient} from "@tsed/temporal";
import {onboardUser} from "../workflows";

@Service()
export class UsersService implements AfterRoutesInit {
  @Inject()
  private temporalClient: TemporalClient;

  async create(user: User): Promise<User> {
    // ...
    await this.temporalClient.workflow.start(onboardUser, {
      args: [user.email],
      taskQueue: "onboarding",
      workflowId: `onboarding-${user.id}`
    });
  }
}

4. Start a worker:

import {bootstrapWorker} from "@tsed/temporal";
import {Server} from "./app/Server";

const worker = await bootstrapWorker(Server, {
  worker: {
    taskQueue: "onboarding",
    workflowsPath: require.resolve("./temporal") // other example: path.join(process.cwd(), 'dist/apps/api/temporal/index.ts');
  },
  connection: {
    /* optional: see NativeConnectionOptions of @temporalio/worker */
  },
  platform: {
    /* optional: see PlatformBuilderSettings of @tsed/common */
    componentsScan: false,
    logger: {
      level: "info"
    }
  }
});
await worker.run();

Todos

• Tests
• Coverage 🆘
• Example
• Documentation

[Romakita]

Hello @ochrstn
Thanks for the PR ^^
For the coverage don't be afraid. Since the repo use @swc/jest the coverage isn't exact on the branch. It's bug from this package and actually isn't resolved, so just update the coverage branch in the jest.config.js to solve your issue.

See you
Romain

[Romakita]

@ochrstn Just small changes but great PR :)

[ochrstn]

The temporal testing package downloads some stuff to start a local server. Thats why the test probably has a timeout:
https://github.com/tsedio/tsed/actions/runs/6391803694/job/17347786663?pr=2464#step:6:84

how to deal with that?

[Romakita]

Maybe you can add command line before test to install dependencies ? Or add timeout extra option for this test ?

[Romakita]

@ochrstn I'll take time to help you on the integration test issue ;)
Finish your changes on the PR and I'll take the next step!

[ochrstn]

@ochrstn I'll take time to help you on the integration test issue ;) Finish your changes on the PR and I'll take the next step!

Oh thanks, but I just pushed something 😅

[ochrstn]

@ochrstn I'll take time to help you on the integration test issue ;) Finish your changes on the PR and I'll take the next step!

Oh thanks, but I just pushed something 😅

ok, that didn't helped much  🥲

All other changes to the PR are already done, or am I missing something?

[Romakita]

All other changes to the PR are already done, or am I missing something?

I'll check the PR ;)

[Romakita]

You've right all changes are done ! Thanks :)

[Romakita]

Hello @ochrstn
Sorry for the delay but finally fixed :D. I just added timeout option on afterAll, beforeAll (replaced all *Each by *All).

If it's ok for you. Go to merge :D ?
See you
Romain

[ochrstn]

Hello @ochrstn Sorry for the delay but finally fixed :D. I just added timeout option on afterAll, beforeAll (replaced all *Each by *All).

If it's ok for you. Go to merge :D ? See you Romain

Looks good to me. Thanks for your time on the pull request. Feel free to merge it 🚀

[Romakita]

🎉 This PR is included in version 7.37.0 🎉

The release is available on:

• v7.37.0
• GitHub release

Your semantic-release bot 📦🚀