Skip to content

Commit 2ec4877

Browse files
sogehigesogehige
committed
feat(plugins): add cron node
1 parent 75d292d commit 2ec4877

16 files changed

Lines changed: 129 additions & 36 deletions

‎locales/en/ui/registry/plugins.json‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,9 @@
77
"name": "Permission filter"
88
}
99
},
10+
"cron": {
11+
"name": "Cron"
12+
},
1013
"listener": {
1114
"name": "Event listener",
1215
"type": {

‎src/plugins.ts‎

Lines changed: 88 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,23 @@
1+
import { MINUTE, SECOND } from '@sogebot/ui-helpers/constants';
12
import { validateOrReject } from 'class-validator';
3+
import * as cronparser from 'cron-parser';
24
import { cloneDeep } from 'lodash';
35
import merge from 'lodash/merge';
46

57
import type { Node } from '../d.ts/src/plugins';
68
import { Plugin, PluginVariable } from './database/entity/plugins';
79
import { isValidationError } from './helpers/errors';
810
import { eventEmitter } from './helpers/events';
11+
import { error } from './helpers/log';
912
import { adminEndpoint } from './helpers/socket';
1013
import { processes, processNode } from './plugins/index';
1114

1215
import Core from '~/_interface';
1316
import { onStartup } from '~/decorators/on';
1417

18+
const cronTriggers = new Map<string, Node>();
19+
const plugins: Plugin[] = [];
20+
1521
const twitchChatMessage = {
1622
sender: {
1723
userName: 'string',
@@ -72,6 +78,14 @@ class Plugins extends Core {
7278
category: 'registry', name: 'plugins', id: 'registry/plugins', this: null,
7379
});
7480

81+
this.updateCache();
82+
setInterval(() => {
83+
this.updateAllCrons();
84+
}, MINUTE);
85+
setInterval(() => {
86+
this.triggerCrons();
87+
}, SECOND);
88+
7589
eventEmitter.on('tip', async (data) => {
7690
const users = (await import('./users')).default;
7791
const user = {
@@ -88,16 +102,75 @@ class Plugins extends Core {
88102
});
89103
}
90104

105+
async updateCache () {
106+
const _plugins = await Plugin.find();
107+
while (plugins.length > 0) {
108+
plugins.shift();
109+
}
110+
for (const plugin of _plugins) {
111+
plugins.push(plugin);
112+
}
113+
await this.updateAllCrons();
114+
}
115+
116+
async updateAllCrons() {
117+
// we will generate at least 2 minutes of span of crons
118+
// e.g. if we have cron every 1s -> 120 crons
119+
// 10s -> 12 crons
120+
// 10m -> 1 cron
121+
const cron = await this.process('cron', '', null, {});
122+
123+
cronTriggers.clear();
124+
for (const { plugin, listeners } of cron) {
125+
for (const node of listeners) {
126+
try {
127+
const cronParsed = cronparser.parseExpression(node.data.value);
128+
129+
const currentTime = Date.now();
130+
let lastTime = new Date().toISOString();
131+
const intervals: string[] = [];
132+
while (currentTime + (2 * MINUTE) > new Date(lastTime).getTime()) {
133+
lastTime = cronParsed.next().toISOString();
134+
intervals.push(lastTime);
135+
}
136+
137+
for (const interval of intervals) {
138+
cronTriggers.set(`${plugin.id}|${interval}`, node);
139+
}
140+
} catch (e) {
141+
error(e);
142+
}
143+
}
144+
}
145+
}
146+
147+
async triggerCrons() {
148+
for (const [pluginId, timestamp] of [...cronTriggers.keys()].map(o => o.split('|'))) {
149+
if (new Date(timestamp).getTime() < Date.now()) {
150+
const plugin = plugins.find(o => o.id === pluginId);
151+
const node = cronTriggers.get(`${pluginId}|${timestamp}`);
152+
if (plugin && node) {
153+
const workflow = Object.values(
154+
JSON.parse(plugin.workflow).drawflow.Home.data
155+
) as Node[];
156+
this.processPath(pluginId, workflow, node, {}, {}, null);
157+
}
158+
cronTriggers.delete(`${pluginId}|${timestamp}`);
159+
}
160+
}
161+
}
162+
91163
sockets() {
92164
adminEndpoint('/core/plugins', 'generic::getAll', async (cb) => {
93-
cb(null, await Plugin.find());
165+
cb(null, plugins);
94166
});
95167
adminEndpoint('/core/plugins', 'generic::getOne', async (id, cb) => {
96-
cb(null, await Plugin.findOne(id));
168+
cb(null, plugins.find(o => o.id === id));
97169
});
98170
adminEndpoint('/core/plugins', 'generic::deleteById', async (id, cb) => {
99171
await Plugin.delete({ id });
100172
await PluginVariable.delete({ pluginId: id });
173+
await this.updateCache();
101174
cb(null);
102175
});
103176
adminEndpoint('/core/plugins', 'generic::validate', async (data, cb) => {
@@ -121,6 +194,7 @@ class Plugins extends Core {
121194
merge(itemToSave, item);
122195
await validateOrReject(itemToSave);
123196
await itemToSave.save();
197+
await this.updateCache();
124198
cb(null, itemToSave);
125199
} catch (e) {
126200
if (e instanceof Error) {
@@ -136,7 +210,7 @@ class Plugins extends Core {
136210
});
137211
}
138212

139-
async processPath(pluginId: string, workflow: Node[], currentNode: Node, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string } ) {
213+
async processPath(pluginId: string, workflow: Node[], currentNode: Node, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string } | null ) {
140214
parameters = cloneDeep(parameters);
141215
variables = cloneDeep(variables);
142216

@@ -174,22 +248,27 @@ class Plugins extends Core {
174248
}
175249
}
176250

177-
async process(type: keyof typeof this.listeners, message: string, userstate: { userName: string, userId: string }, params?: Record<string, any>) {
178-
const plugins = await Plugin.find({ enabled: true });
179-
const pluginsWithListener: Plugin[] = [];
180-
for (const plugin of plugins) {
251+
async process(type: keyof typeof this.listeners | 'cron', message: string, userstate: { userName: string, userId: string } | null, params?: Record<string, any>) {
252+
const pluginsEnabled = plugins.filter(o => o.enabled);
253+
const pluginsWithListener: { plugin: Plugin, listeners: Node[] }[] = [];
254+
for (const plugin of pluginsEnabled) {
181255
// explore drawflow
182256
const workflow = Object.values(
183257
JSON.parse(plugin.workflow).drawflow.Home.data
184258
) as Node[];
185259

186-
const listeners = workflow.filter((o: any) => {
260+
const listeners = workflow.filter((o: Node) => {
187261
params ??= {};
188262
const isListener = o.name === 'listener';
263+
const isCron = o.name === 'cron' && type === 'cron';
189264
const isType = o.data.value === type;
190265

191266
params.message = message;
192267

268+
if (isCron) {
269+
return true;
270+
}
271+
193272
if (isListener && isType) {
194273
switch(type) {
195274
case 'twitchCommand': {
@@ -239,7 +318,7 @@ class Plugins extends Core {
239318
});
240319

241320
if (listeners.length > 0) {
242-
pluginsWithListener.push(plugin);
321+
pluginsWithListener.push({ plugin, listeners });
243322
}
244323
}
245324
return pluginsWithListener;

‎src/plugins/index.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ import { warning } from '~/helpers/log';
1717

1818
export const processes = {
1919
listener,
20+
cron: listener, // no-op
2021
othersIdle,
2122
outputLog,
2223
gateCounter,
@@ -29,13 +30,13 @@ export const processes = {
2930
twitchTimeoutUser,
3031
twitchBanUser,
3132
twitchSendMessage,
32-
default: (pluginId: string, currentNode: Node, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; }) => {
33+
default: (pluginId: string, currentNode: Node, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; } | null) => {
3334
warning(`PLUGINS: no idea what should I do with ${currentNode.name}, stopping`);
3435
return false;
3536
},
3637
};
3738

38-
function processNode (type: keyof typeof processes, pluginId: string, currentNode: Node, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; }): Promise<boolean> | boolean {
39+
function processNode (type: keyof typeof processes, pluginId: string, currentNode: Node, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; } | null): Promise<boolean> | boolean {
3940
return (processes[processes[type] ? type : 'default'](pluginId, currentNode as any, parameters, variables, userstate));
4041
}
4142

‎src/plugins/nodes/clearCounter.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ import { clearCounter } from './gateCounter';
33
import type { Node } from '~/../d.ts/src/plugins';
44
import { Plugin } from '~/database/entity/plugins';
55

6-
export default async function(pluginId: string, currentNode: Node<string>, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; }) {
6+
export default async function(pluginId: string, currentNode: Node<string>, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; } | null) {
77
try {
88
const plugin = await Plugin.findOneOrFail({ id: pluginId });
99
const workflow = Object.values(

‎src/plugins/nodes/debounce.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ setInterval(() => {
1616
}
1717
}, 10000);
1818

19-
export default async function(pluginId: string, currentNode: Node<string>, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; }) {
19+
export default async function(pluginId: string, currentNode: Node<string>, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; } | null) {
2020
const perUser = JSON.parse(currentNode.data.data).perUser ?? false;
2121
const uuid = JSON.parse(currentNode.data.data).uuid;
2222

@@ -31,7 +31,7 @@ export default async function(pluginId: string, currentNode: Node<string>, param
3131
miliseconds = 1000;
3232
}
3333

34-
const key = perUser ? `${userstate.userId}|${uuid}` : uuid;
34+
const key = perUser ? `${userstate!.userId}|${uuid}` : uuid;
3535
intervals.set(uuid, miliseconds);
3636

3737
const timestamp = debounce.get(key);

‎src/plugins/nodes/filter.ts‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ import { VM } from 'vm2';
44
import type { Node } from '~/../d.ts/src/plugins';
55
import { error } from '~/helpers/log';
66

7-
export default async function(pluginId: string, currentNode: Node<string>, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; }) {
7+
export default async function(pluginId: string, currentNode: Node<string>, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; } | null) {
88
const advancedMode = JSON.parse(currentNode.data.data).advancedMode ?? false;
99

1010
let script = null;
@@ -24,10 +24,10 @@ export default async function(pluginId: string, currentNode: Node<string>, param
2424
}
2525
try {
2626
const sandbox = {
27-
sender: {
27+
sender: userstate ? {
2828
userName: userstate.userName,
2929
userId: userstate.userId,
30-
},
30+
} : null,
3131
parameters,
3232
...variables,
3333
};

‎src/plugins/nodes/filterPermission.ts‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,17 @@
11
import type { Node } from '~/../d.ts/src/plugins';
22
import { check } from '~/helpers/permissions/index';
33

4-
export default async function(pluginId: string, currentNode: Node<string[]>, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; }) {
4+
export default async function(pluginId: string, currentNode: Node<string[]>, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; } | null) {
55
const permissionsAccessList = currentNode.data.value;
66
let haveAccess = false;
77
for (const permId of permissionsAccessList) {
88
if (haveAccess) {
99
break;
1010
}
11-
const status = await check(userstate.userId, permId);
12-
haveAccess = status.access;
11+
if (userstate) {
12+
const status = await check(userstate.userId, permId);
13+
haveAccess = status.access;
14+
}
1315
}
1416
return haveAccess;
1517
}

‎src/plugins/nodes/gateCounter.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ export const clearCounter = (uuid: string) => {
1313
}
1414
};
1515

16-
export default async function(pluginId: string, currentNode: Node<string>, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; }) {
16+
export default async function(pluginId: string, currentNode: Node<string>, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; } | null) {
1717
const perUser = JSON.parse(currentNode.data.data).perUser ?? false;
1818
const resetAfterTrigger = JSON.parse(currentNode.data.data).resetAfterTrigger ?? false;
1919
const uuid = JSON.parse(currentNode.data.data).uuid;
@@ -29,7 +29,7 @@ export default async function(pluginId: string, currentNode: Node<string>, param
2929
expectedCount = 10;
3030
}
3131

32-
const key = perUser ? `${userstate.userId}|${uuid}` : uuid;
32+
const key = perUser ? `${userstate!.userId}|${uuid}` : uuid;
3333

3434
const count = (counter.get(key) ?? 0) + 1;
3535
counter.set(key, count);

‎src/plugins/nodes/othersIdle.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ import { template } from '../../plugins/template';
55
import type { Node } from '~/../d.ts/src/plugins';
66
import { warning } from '~/helpers/log';
77

8-
export default async function(pluginId: string, currentNode: Node, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; }) {
8+
export default async function(pluginId: string, currentNode: Node, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; } | null) {
99
let miliseconds = await template(currentNode.data.value, { parameters, ...variables });
1010
if (isNaN(Number(miliseconds))) {
1111
warning(`PLUGINS#${pluginId}: Idling value is not a number! Got: ${miliseconds}, defaulting to 1000`);

‎src/plugins/nodes/outputLog.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ import { template } from '../template';
33
import type { Node } from '~/../d.ts/src/plugins';
44
import { info } from '~/helpers/log';
55

6-
export default async function(pluginId: string, currentNode: Node, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; }) {
6+
export default async function(pluginId: string, currentNode: Node, parameters: Record<string, any>, variables: Record<string, any>, userstate: { userName: string; userId: string; } | null) {
77
info(`PLUGINS#${pluginId}: ${await template(currentNode.data.value, { parameters, ...variables }, userstate)}`);
88
return true;
99
}

0 commit comments

Comments
 (0)