1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
|
import { model, Schema, Model } from 'mongoose';
import cron from 'cron';
import schema, { EventDocument } from './event.schema';
import { LogDocument } from './log.schema';
import { LogModel } from './log.model';
import Connection from './connection';
export interface Event<Context = any> extends EventDocument<Context> {
log(message: string): Promise<LogDocument>;
start(): void;
complete(): void;
fail(error: Error | string): void;
computeNextRunAt(): Date;
getLogs(): Promise<LogDocument[]>;
}
export interface EventModel<Context = any> extends Model<Event<Context>> {
findNextEvents(): Event<Context>[];
findMissedEvents(): Event<Context>[];
}
const { CronJob } = cron;
const createEventModel = (connection: Connection): EventModel => {
const LogModel: LogModel = connection.models['Log'];
// Schema methods
schema.method('log', function (message: string) {
const timestamp = new Date().toLocaleString('en');
console.log(`[${timestamp}] ${this.type}: ${message}`);
return LogModel.create({ eventId: this._id, message });
});
schema.method('start', function () {
this.log('Event started');
this.lastRunAt = new Date();
this.status = 'running';
return this.save();
});
schema.method('complete', function () {
this.log('Event complete');
this.status = 'complete';
return this.save();
});
schema.method('fail', function (error: Error) {
this.log(error);
this.log('Event failed');
this.error = error;
this.status = 'failed';
return this.save();
});
schema.method('computeNextRunAt', function () {
const job = new CronJob(this.schedule);
const nextRunAt = job.nextDates();
return nextRunAt.toDate();
});
schema.method('getLogs', function () {
return LogModel.find({ eventId: this._id });
});
// Statics
schema.static('findMissedEvents', async function () {
return this.find({
nextRunAt: {
// TODO: skip single-fire events
$lt: new Date()
}
});
});
schema.static('findNextEvents', function (limit = 10) {
return this.find(
{
nextRunAt: {
$gt: new Date()
}
},
null,
{
sort: {
nextRunAt: 1
},
limit
}
);
});
// Hooks
schema.pre<Event>('save', async function () {
this.nextRunAt = this.computeNextRunAt();
});
return connection.model<Event, EventModel>('Event', schema);
};
export default createEventModel;
|
