Distributed, Decentralized Workflow creator for Node, Backed by Redis.
\n\n# Philosophy \nDeFlow attempt to fill the gap between a job scheduler and an ETL. \nIt manage workflow queue with 3 mains elements : \n\n#### Workflow:\nA Workflow defines a set of one or more steps, there created by a node process.\n\n#### Steps:\nSteps define a specific job that compose the workflow.\nThey're treated sequentially and are described by a single file (module).\nThis module file define the step lifecycle with predefined methods (before/after, error handler, task handler).\n\nEach Step contain one or more tasks. \nA step can create one or more other steps, that way they can be multidimensional. \n\n#### Tasks:\nTasks are treated by the step \"handler\" lifecycle method, they are designed to be treated concurrently between node.\nTasks are handled functionnaly, accepting params and returning results. \nTasks are configurable and can have timeout and retry strategy.\n\n# Main features\n- Distributed: Intelligent distribution and parallelization of tasks between multiple nodeJS process.\n- Decentralized: Designed to be crash proof, backed by Redis, pub-sub communication between nodes.\n- Lifecycle: Lifecycle method allow you to manage and evolve the workflow process. \n- Living workflow: Create steps or tasks on the fly, depending on the previous results. \n- Promises based API.\n- Configurable concurrency, retries, error handling and more.\n- TypeScript support.\n\n## Getting started\n\ninstall:\n```\nnpm i deflow\n```\n\ndeclare a step handler:\n```typescript\n// ./steps/string-to-number.ts \n\nimport { StepHandler } from 'deflow';\n \n/**\n * Declare the step handler and types\n * In this one, we convert string to float\n * NOTE: IT MUST BE EXPORTED AS DEFAULT\n */\n export default new StepHandler({\n\n /**\n * Init method allow you to prepare tasks based on anything you want\n * @param step\n */\n async beforeAll(step) {\n const tasks = ['12', '10', '7', '45']; // You can fetch data from external source or db\n await step.addTasks(tasks);\n },\n \n /**\n * This method will run for each task of the step\n * @param task\n */\n async handler(task) {\n return parseFloat(task.data);\n },\n \n /**\n * This method is executed after each tasks done\n * Useful to log progress and stuff\n * @param task\n * @param step\n */\n async afterEach(task, step) {\n const progress = await step.getProgress();\n console.log('Step1: afterEach', progress);\n console.log('Step1: Result', task.result); // Should be a floating number\n },\n \n /**\n * This method is executed after all tasks done\n * Useful to save results in a db or whatever you want\n * @param step\n */\n async afterAll(step) {\n console.log('Step1: afterAll', await step.getProgress());\n console.log('Step1: Result', await step.getResults());\n },\n });\n```\n\ndeclare a workflow:\n```typescript\n// ./index.ts \n\nimport DeFlow, { WorkFlow } from 'deflow';\n\n// Register deflow to your redis backend\nDeFlow.register({ connection: { host: 'localhost', port: 6379 } });\n\n/**\n * Workflow test file\n */\nfunction runWorkflow() {\n WorkFlow\n .create('some-custom-name')\n .addStep({ step: import('./steps/string-to-number') }) // Register the step\n .addStep({ step: import('./steps/anther-process-step') }) // Register the step\n .run(); // Run the workflow\n}\n\n// Run the workflow from somewhere in your code (make sure redis is ready before)\nsetTimeout(() => {\n runWorkflow();\n}, 2000)\n```\n\nThe `step` attribute of `addStep` method can take a module, dynamic import, or a path to the module.\nPrefer the module or dynamic import to take advantage of TS type checking.\n\n###### [Check the complete API Documentation](https://github.com/Fabiencdp/Deflow/tree/main/docs/api.md)\n\n\n## Making things type safe\n\nDeFlow allow you to define type safe stepHandler :\n\n```typescript\n/**\n * Sending invoices for each user having purchased a product today\n */\nexport default new StepHandler<\n { date: Date }, // step data type\n { productName: string; productPrice: number; userEmail: string; userName: string }, // Task data type\n { sentStatus: boolean } // Task result type\n>({\n /**\n * Fetch user data from a database\n */\n async beforeAll(step) {\n const orders = await Order.find({ createdAt: step.data.date });\n\n // Create one task by order\n const tasks = orders.map((order) => ({\n productName: order.name,\n productPrice: order.price,\n userEmail: order.user.email,\n userName: order.user.name,\n }));\n\n return step.addTasks(tasks); // Each task will be processed by \"handler\" method\n },\n\n /**\n * For each task, send the invoice by email\n * Note that you can also access to the step \"global\" data\n * @param task\n * @param step\n */\n async handler(task, step) {\n const sendMailRes = await sendmail({\n from: 'no-reply@yourdomain.com',\n to: task.data.userEmail,\n subject: `Your ${step.data.date} invoice for ${task.data.productName}`,\n html: `\n Hello ${task.data.userName},\n You just spent ${task.data.productPrice} for ${task.data.productName}\n `,\n });\n\n return { sentStatus: sendMailRes.status };\n },\n\n /**\n * After all task done, log some important things and take needed actions based on results\n * @param step\n */\n async afterAll(step) {\n const res = await step.getResults();\n const errors = res.filter((task) => task.result.sentStatus === false);\n\n if (errors.length > 0) {\n console.warn(`WARNING: ${errors.length} sendmail errors! Will try fallback method`);\n\n // Add another step right after this one when we have errors\n await step.addAfter({\n step: sendMailFallBackStep,\n tasks: errors, // If needed, you can directly add tasks while adding step\n options: {\n taskConcurrency: 3, // Allow each node to treat 3 tasks concurrently\n taskTimeout: 2000, // Allow a timeout of 2000ms per task\n taskMaxFailCount: 3, // Allow a maximum of 3 retries\n },\n });\n }\n },\n});\n\n```\n\n## Resources\n\n- [API Documentation](https://github.com/Fabiencdp/Deflow/tree/main/docs/api.md)\n- [Examples](https://github.com/Fabiencdp/Deflow/tree/main/examples)\n\n\n### Coming next\n- More events that you can listen too\n- Advanced concurrency management\n- Reducer\n","readmeFilename":"README.md","homepage":"https://github.com/Fabiencdp/Deflow#readme","repository":{"type":"git","url":"git+https://github.com/Fabiencdp/Deflow.git"},"bugs":{"url":"https://github.com/Fabiencdp/Deflow/issues"}}