Concerns About Nesting Effect Executors in BullMQ Workers

Hi, I recall reading that you should never nest Effect executors within other executors, and I'm concerned about the use of Effect.runPromise within my

@effect/plaform-bun

@effect/plaform-bun

@effect/plaform-bun

@effect/plaform-bun app that uses BullMQ.

BullMQ workers need a 'processor' method, so I'm wrapping my processJob function with a runPromise when setting up the worker. It works, but I can't shake the feeling I'm committing some cardinal sin.

I've been diving into runtimes, and it SEEMS like what I'm doing is okay since I'm using a managed runtime? (BunRuntime)

Any insights would be super helpful

...
const make = Effect.gen(function* (_) {
    const workerService = yield* BullMQWorkerService;
    const userService = yield* UserService;

    const processNotifyUserJob = (job: Job<AddNotifyUserJobData, void, UserJobName>) => {
        return Effect.gen(function* () {
            const users = yield* userService.getUsers();
            yield* Effect.log(`This works! Found ${users.length} users`);
        }).pipe(Effect.withSpan("UserJobConsumer.processNotifyUserJob"));
    };

        ...

    const jobProcessors: Record<UserJobName, ...> = {
        [UserJobName.NOTIFY_USER]: processNotifyUserJob,
        ...
    };

    const processJob = (job: Job<UserJobData, UserJobResult, UserJobName>) => {
        return Effect.gen(function* () {
            const jobProcessor = jobProcessors[job.name];

            if (!jobProcessor) {
                yield* Effect.fail( ... );
            }

            yield* jobProcessor(job);
        }).pipe(
            Effect.catchAll((error) =>
                ...
            )
        );
    };

    yield* workerService.createWorker<UserJobData, UserJobResult, UserJobName>(
        QueueName.USER,
        (job) => Effect.runPromise(processJob(job))
    );
});

export class UserJobConsumer extends ... {
    static readonly Live = Layer.scoped(this, make).pipe(
        Layer.provide(BullMQWorkerService.Live),
        Layer.provide(UserService.Live)
    );
}

...
const make = Effect.gen(function* (_) {
    const workerService = yield* BullMQWorkerService;
    const userService = yield* UserService;

    const processNotifyUserJob = (job: Job<AddNotifyUserJobData, void, UserJobName>) => {
        return Effect.gen(function* () {
            const users = yield* userService.getUsers();
            yield* Effect.log(`This works! Found ${users.length} users`);
        }).pipe(Effect.withSpan("UserJobConsumer.processNotifyUserJob"));
    };

        ...

    const jobProcessors: Record<UserJobName, ...> = {
        [UserJobName.NOTIFY_USER]: processNotifyUserJob,
        ...
    };

    const processJob = (job: Job<UserJobData, UserJobResult, UserJobName>) => {
        return Effect.gen(function* () {
            const jobProcessor = jobProcessors[job.name];

            if (!jobProcessor) {
                yield* Effect.fail( ... );
            }

            yield* jobProcessor(job);
        }).pipe(
            Effect.catchAll((error) =>
                ...
            )
        );
    };

    yield* workerService.createWorker<UserJobData, UserJobResult, UserJobName>(
        QueueName.USER,
        (job) => Effect.runPromise(processJob(job))
    );
});

export class UserJobConsumer extends ... {
    static readonly Live = Layer.scoped(this, make).pipe(
        Layer.provide(BullMQWorkerService.Live),
        Layer.provide(UserService.Live)
    );
}

...
const make = Effect.gen(function* (_) {
    const workerService = yield* BullMQWorkerService;
    const userService = yield* UserService;

    const processNotifyUserJob = (job: Job<AddNotifyUserJobData, void, UserJobName>) => {
        return Effect.gen(function* () {
            const users = yield* userService.getUsers();
            yield* Effect.log(`This works! Found ${users.length} users`);
        }).pipe(Effect.withSpan("UserJobConsumer.processNotifyUserJob"));
    };

        ...

    const jobProcessors: Record<UserJobName, ...> = {
        [UserJobName.NOTIFY_USER]: processNotifyUserJob,
        ...
    };

    const processJob = (job: Job<UserJobData, UserJobResult, UserJobName>) => {
        return Effect.gen(function* () {
            const jobProcessor = jobProcessors[job.name];

            if (!jobProcessor) {
                yield* Effect.fail( ... );
            }

            yield* jobProcessor(job);
        }).pipe(
            Effect.catchAll((error) =>
                ...
            )
        );
    };

    yield* workerService.createWorker<UserJobData, UserJobResult, UserJobName>(
        QueueName.USER,
        (job) => Effect.runPromise(processJob(job))
    );
});

export class UserJobConsumer extends ... {
    static readonly Live = Layer.scoped(this, make).pipe(
        Layer.provide(BullMQWorkerService.Live),
        Layer.provide(UserService.Live)
    );
}

...
const make = Effect.gen(function* (_) {
    const workerService = yield* BullMQWorkerService;
    const userService = yield* UserService;

    const processNotifyUserJob = (job: Job<AddNotifyUserJobData, void, UserJobName>) => {
        return Effect.gen(function* () {
            const users = yield* userService.getUsers();
            yield* Effect.log(`This works! Found ${users.length} users`);
        }).pipe(Effect.withSpan("UserJobConsumer.processNotifyUserJob"));
    };

        ...

    const jobProcessors: Record<UserJobName, ...> = {
        [UserJobName.NOTIFY_USER]: processNotifyUserJob,
        ...
    };

    const processJob = (job: Job<UserJobData, UserJobResult, UserJobName>) => {
        return Effect.gen(function* () {
            const jobProcessor = jobProcessors[job.name];

            if (!jobProcessor) {
                yield* Effect.fail( ... );
            }

            yield* jobProcessor(job);
        }).pipe(
            Effect.catchAll((error) =>
                ...
            )
        );
    };

    yield* workerService.createWorker<UserJobData, UserJobResult, UserJobName>(
        QueueName.USER,
        (job) => Effect.runPromise(processJob(job))
    );
});

export class UserJobConsumer extends ... {
    static readonly Live = Layer.scoped(this, make).pipe(
        Layer.provide(BullMQWorkerService.Live),
        Layer.provide(UserService.Live)
    );
}

No replies yetBe the first to reply to this messageJoin

Concerns About Nesting Effect Executors in BullMQ Workers

Similar Threads

Similar Threads

Similar Threads