`JobAttributes` not typed correctly

[kostysh]

I just tried Pulse and faced an issue with a job processor definition type. Here is a short example:

import { JobAttributes, JobAttributesData } from '@pulsecron/pulse';

interface ProcessJobData extends JobAttributesData {
  myId: string;
}

pulse.define<JobAttributes<ProcessJobData>>(
    'processJob',
    async (job) => {
      const { myId } = job.attrs.data; // `data` here is `any` of type, but must be `ProcessJobData` of type
      // ...
    },
    {
      concurrency: 3,
      priority: 'normal',
    },
  );

The issue is, possibly, here:

export interface JobAttributes<T extends JobAttributesData = JobAttributesData> {
    _id: mongodb.ObjectId;
    pulse: Pulse;
    type: string;
    name: string;
    disabled?: boolean;
    nextRunAt?: Date | null;
    lockedAt?: Date | null;
    priority: number | string;
    data: any; // <-- should be T, not any
    uniqueQuery?: any;
    uniqueOpts?: {
        insertOnly: boolean;
    };
    repeatInterval?: string;
    repeatTimezone?: string | null;
    repeatAt?: string;
    lastRunAt?: Date;
    lastFinishedAt?: Date;
    startDate?: Date | number | null;
    endDate?: Date | number | null;
    skipDays?: string | null;
    failReason?: string;
    failCount?: number;
    failedAt?: Date;
    lastModifiedBy?: string;
    shouldSaveResult?: boolean;
    result?: unknown;
}

[code-xhyun]

Thank you for the issue! I will solve these and give you an answer as soon as possible.

[code-xhyun]

@kostysh This was fixed in v1.1.9 🚀

Use it like this!

interface ProcessJobData extends JobAttributesData {
  myId: string;
}

pulse.define<ProcessJobData>(
  'processJob',
  async (job) => {
    const { myId } = job.attrs.data;
    // ...
  },
  {
    concurrency: 3,
    priority: 'normal',
  }
);

[cheeselemon]

Cool! Thanks!

[kostysh]

Thank you, it works now. What is about a done callback argument? In Agenda it was possible to provide done with an error for saving the finished job state as failed with a specific error.

[code-xhyun]

Thank you, it works now. What is about a done callback argument? In Agenda it was possible to provide done with an error for saving the finished job state as failed with a specific error.

@kostysh The done() callback argument is just a void. Is there an example code for the case you're talking about?

Maybe this is what you want?

pulse.define(
  'send email report',
  async (job, done) => {
    const { to } = job.attrs.data;

    console.log(`Sending email report to ${to}`);
    // this!
    try {
      done();
    } catch (error) {
      console.log('error:', error);
    }
  },
  { lockLifetime: 5 * 1000, priority: 'high', concurrency: 10 }
);

[tiberioalunni]

Thank you, it works now. What is about a done callback argument? In Agenda it was possible to provide done with an error for saving the finished job state as failed with a specific error.

@kostysh The done() callback argument is just a void. Is there an example code for the case you're talking about?

As I can see done function should match the function const jobCallback = async (error?: Error, result?: unknown) of https://github.com/pulsecron/pulse/blob/main/src/job/run.ts

as it is called here:

if (definition.fn.length === 2) {
  debug('[%s:%s] process function being called', this.attrs.name, this.attrs._id);
  await definition.fn(this, jobCallback); // <---- definition function with done callback param
} else {
  debug('[%s:%s] process function being called', this.attrs.name, this.attrs._id);
  const result = await definition.fn(this);
  await jobCallback(undefined, result);
}

Processor definition should be something like:

export type Processor<T extends JobAttributesData> = (job: Job<T>, done?: (error?: Error, result?: unknown) => void) => unknown | Promise<unknown>;

Usage with done param:

pulse.define('sendEmail', async (job, done) => {
 //....
  if (error) {
    done(new Error('Error message'));
    return;
  }

  // if some result has to be returned:
  done(undefined, 'Result data');
  // else 
  done();
})

Usage without done param:

pulse.define('sendEmail', async (job) => {
 //....

  if (error) {
    throw new Error('Error message')
  }

  // if some result has to be returned:
  return 'Result data';
  // else just return void
})

[code-xhyun]

감사합니다. 이제 작동합니다. 콜백 인수 는 무엇입니까 done? 특정 오류로 인해 완료된 작업 상태를 실패로 저장하는 데 오류를 Agenda제공할 수 있었습니다 .done

@kostysh콜백 done()인수는 단지 무효입니다. 당신이 말하는 사례에 대한 예제 코드가 있습니까?

내가 볼 수 있듯이 함수는 https://github.com/pulsecron/pulse/blob/main/src/job/run.ts 의 done함수와 일치해야 합니다 .const jobCallback = async (error?: Error, result?: unknown)

여기에서는 다음과 같이 부릅니다.

if (definition.fn.length === 2) {
  debug('[%s:%s] process function being called', this.attrs.name, this.attrs._id);
  await definition.fn(this, jobCallback); // <---- definition function with done callback param
} else {
  debug('[%s:%s] process function being called', this.attrs.name, this.attrs._id);
  const result = await definition.fn(this);
  await jobCallback(undefined, result);
}

프로세서 정의는 다음과 같아야 합니다.

export type Processor<T extends JobAttributesData> = (job: Job<T>, done?: (error?: Error, result?: unknown) => void) => unknown | Promise<unknown>;

done 매개변수를 사용한 사용법:

pulse.define('sendEmail', async (job, done) => {
 //....
  if (error) {
    done(new Error('Error message'));
    return;
  }

  // if some result has to be returned:
  done(undefined, 'Result data');
  // else 
  done();
})

done 매개변수가 없는 사용법:

pulse.define('sendEmail', async (job) => {
 //....

  if (error) {
    throw new Error('Error message')
  }

  // if some result has to be returned:
  return 'Result data';
  // else just return void
})

Thank you for your dedication. @tiberioalunni @kostysh We released a new version reflecting your opinion > v1.1.11 🚀

Also we updated docs