🌊 Add Data Flows to Directus 🌊 (#12522)

* Replace attachment with correct icon * Use standardized options formatting * Improve preview styling, fix names * Format IDs of DB read operation as csv * Remove flow active state from header * Don't return null for unknown flows * Fix webhook trigger not showing id * Fix alignment of attachment * Make heading secondary if it's the reject handler * Use flow name in subtitle of operation drawer * Rename "Create new Operation" -> "Create Operation" * Make name/key required * Give name autofocus * Add "uncaught exception" log message * Various improvements on operations * default status to "active" * Add status dropdown at the bottom of trigger * put status dot to the right of flow name header * add toggle status option in context menu * fix trigger options staging * fix flow deletion * show configured operation key on name hover * prevent block pushing status toggle down * ensure key is unique between operations in a flow * allow add new panel when previous one is deleted * fix staged panels temporarily disappear The deletion of newTree.id causes it to "disconnect" when saving, thus causing it seemingly disappear. Using a cloneDeep prevents it from mutating the current stagedPanels * hide key input when in query mode * add write operation * undo previous route props change * include staged panel keys in key validation * fix key validation logic * add color to flow & insights * ensure trigger does not have reject button * prevent operation key error showing up when saving * change context menu to Delete Operation * fix add operation when removed operation is staged * Hide ID in read preview when in query mode * fix reject button showing without edit mode * fix status toggle in flow overview * simplify request operation methods & allow other * fix preview function type * simplify slot syntax * add manual trigger * simplify manual trigger handler * prevent drawer closing on esc * allow filter config without selecting collection * improve affordance of add operation button * fix loner reject button color * Added emitEvents option to write operation (#13121) * added emitEvent option toggle to write operation * Re-gen package-lock Co-authored-by: rijkvanzanten <rijkvanzanten@me.com> * Clean up active/inactive toggle * Align arrow to grid * Visually align border radius of glow * Tweak padding of footer in panels * Remove init event * Combine event triggers into single "Hook" trigger * Fix mail operation * Cleanup imports, return undefined for webhook executino flows * Add border to panel footer * Upgrade preview of hook trigger * Don't warn on uncaught flow executions * Clean up types * Fix typing of reload * Default to correct icon * Add migrations to remove webhooks table * Remove webhooks * Update icons for triggers * Reorganize triggers * Merge flows and webhooks migrations * Add permissions option to database read operatoin * Add permissions configuration to database write * Remove flow logs in favor of using directus_activity * Upgrade webhook configuration, fix create operatoin * Rename validation->condition * Subdue everything when inactive * Tweak tests * Fix the test for real * Remove circular FK trigger, please MSSQL * Make things worse to please MSFT * Add input * Drop input scope from condition operation * naming and description changes * Default flow overview icon color to primary * add danger styling to delete flow button * fix hint buttons subdued style * Hide trigger unlinked resolve btn when not editing Don't show "check mark icon" or "operation arrow" on empty trigger when NOT in edit mode * show email "to" value as CSV * remove unused webhook.preview translation * Default sort order of overview table to `name` * Track activity / revisions in flows * Extract w/ the intent to reuse revisions fetching * Move Action type to shared to facilitate app use * [WIP] Start rendering logs drawer from sidebar * Fix type error (sorry Eron) * add update operation * add delete operation * use parseJSON util in operations * Add missing fields to flows system data * Await promise in sleep operation * fix e2e test missing flows & operations tables * Add fallback title to flow logs drawer * Add default value to flow prop for flow-dialog * Hydrate flows store before moving to details page when creating flow * Rename CRUD operations to item-* * Change trigger options subtitle to Trigger Options * Remove trigger name option * Fix typescript complaining * Remove two lines * Fix notification operation * Log error when executing a schedule flow * Fix schedule flow activity tracking * Fix notification operation when there is no user * Make permissions for notifications configurable * Do not drop non null constraint from column that is nullable * Remove invalid option from activity seed * Show resolve/reject dot when operation has successor * Improve flow arrow placement * Prevent arrow color from flickering * Fix arrow being stuck when hovered while saving * Fix arrows not being subdued on lone leaf operations * Add tooltips to operation handles * Remove option to trigger flow on init * Move operation handle tooltip to icon * Disconnect duplicated operation * Fix deleting connected operations * Make delete action name generic in v-workspace-panel * Use flow-specific wording in flow edit tooltip * Simplify hint handle check * Fix deleting first operation * Use useEditsGuard composable in flow component * Add asynchronous option to webhook trigger * Add option to make preview elements copyable * Add hover transition to panels * Register operation preview components as operation-preview-* * Remove selectability of panel header and operation body * Add return option to filter and operation triggers * Add missing key * Remove unnecessary ampersand from URLs in filter tests (#13523) * Remove unused prop * New translations en-US.yaml (Polish) (#13524) * My favorite * v9.11.1 * v9.11.1 * New translations en-US.yaml (Polish) (#13528) * New translations en-US.yaml (Czech) (#13541) * New translations en-US.yaml (Czech) (#13545) * fix metadata for directus_folders (#13527) * add `meta` to list responses for OAS (#13531) * remove existing pasting check on slug values (#13532) * Add copy button to user id on user info page (#13540) * New translations en-US.yaml (Czech) (#13547) * New translations en-US.yaml (Czech) (#13548) * Added missing "DB_SSL_*_FILE" to the "_FILE" allow list. (#13539) * Remove workaround in release flow (#13455) This forces the release workflow to use `node@16.15` which includes `npm@8.5`. * Remove npmrc files which prevent lockfile creation in workspaces (#13444) * Remove npmrc files which prevent lockfile creation in workspaces Since `v8.5.0` npm will detect that it is running inside a workspace and issue commands at the root package. * Require a minimum npm version of 8.5.0 * Package-lock 🖤 * Don't consider SIGN_OUT an SSO error (#13389) * Don't consider SIGN_OUT an SSO error * Add SESSION_EXPIRED as valid reason * Improve translation for require_value_to_be_set (#13363) English + Dutch Co-authored-by: Rijk van Zanten <rijkvanzanten@me.com> * Fix field conditions optionDefaults computed property (#13563) * fix: remove .value from options * additional ref fix & type/null errors Co-authored-by: Azri Kahar <42867097+azrikahar@users.noreply.github.com> * `last_page` type was `date` instead of `str` (#13577) * `last_page` type was `date` instead of `str` * Update docs/reference/system/users.md Co-authored-by: Rijk van Zanten <rijkvanzanten@me.com> * New Crowdin updates (#13557) * New translations en-US.yaml (Romanian) * New translations en-US.yaml (Indonesian) * New translations en-US.yaml (Spanish, Chile) * New translations en-US.yaml (Thai) * New translations en-US.yaml (Spanish, Latin America) * New translations en-US.yaml (Russian) * New translations en-US.yaml (Polish) * New translations en-US.yaml (Swedish) * New translations en-US.yaml (Turkish) * New translations en-US.yaml (Portuguese, Brazilian) * New translations en-US.yaml (French) * New translations en-US.yaml (Spanish) * New translations en-US.yaml (Bulgarian) * New translations en-US.yaml (Catalan) * New translations en-US.yaml (Danish) * New translations en-US.yaml (German) * New translations en-US.yaml (Finnish) * New translations en-US.yaml (Hungarian) * New translations en-US.yaml (Chinese Simplified) * New translations en-US.yaml (Italian) * New translations en-US.yaml (Slovenian) * New translations en-US.yaml (Ukrainian) * New translations en-US.yaml (English, United Kingdom) * New translations en-US.yaml (English, Canada) * New translations en-US.yaml (French, Canada) * New translations en-US.yaml (Croatian) * New translations en-US.yaml (Czech) * New translations en-US.yaml (Czech) * New translations en-US.yaml (Czech) * New translations en-US.yaml (Czech) * New translations en-US.yaml (Czech) * New translations en-US.yaml (Czech) * New translations en-US.yaml (Czech) * New translations en-US.yaml (Czech) * New translations en-US.yaml (Czech) * New translations en-US.yaml (Czech) * New translations en-US.yaml (Czech) * New translations en-US.yaml (Czech) * New translations en-US.yaml (Czech) * New translations en-US.yaml (Czech) * New translations en-US.yaml (Czech) * Fix validate query number comparison Ref https://github.com/directus/directus/pull/13492#issuecomment-1138770254 * New translations en-US.yaml (Polish) (#13580) * add to project (#13581) * Allow authentication using MSSQL azure-active-directory-service-principal-secret (#11141) * Extract ignored settings requires by azure authentication * Change the way to extract initial database settings * Fix invalid names after extracting from env util * Replace missing var after solving conflicts * Add default value to poolconfig * This should unbreak it Co-authored-by: Rijk van Zanten <rijkvanzanten@me.com> * Create pull_request_template.md * Update pull_request_template.md * Add System token interface (#13549) * Add system token interface * use system token interface in users token field * Update placeholder * move notice below input * fix clear value interaction * update placeholder and notice text * remove unused translation key * rename class to match current naming * fix bugs in disabled state and it's UX Co-authored-by: Rijk van Zanten <rijkvanzanten@me.com> Co-authored-by: jaycammarano <jay.cammarano@gmail.com> * New Crowdin updates (#13586) * New translations en-US.yaml (Romanian) * New translations en-US.yaml (Indonesian) * New translations en-US.yaml (Spanish, Chile) * New translations en-US.yaml (Thai) * New translations en-US.yaml (Serbian (Latin)) * New translations en-US.yaml (Spanish, Latin America) * New translations en-US.yaml (Russian) * New translations en-US.yaml (Polish) * New translations en-US.yaml (Portuguese) * New translations en-US.yaml (Swedish) * New translations en-US.yaml (Turkish) * New translations en-US.yaml (Estonian) * New translations en-US.yaml (Portuguese, Brazilian) * New translations en-US.yaml (French) * New translations en-US.yaml (Spanish) * New translations en-US.yaml (Bulgarian) * New translations en-US.yaml (Catalan) * New translations en-US.yaml (Czech) * New translations en-US.yaml (Danish) * New translations en-US.yaml (German) * New translations en-US.yaml (Finnish) * New translations en-US.yaml (Hungarian) * New translations en-US.yaml (Chinese Simplified) * New translations en-US.yaml (Italian) * New translations en-US.yaml (Japanese) * New translations en-US.yaml (Dutch) * New translations en-US.yaml (Slovenian) * New translations en-US.yaml (Ukrainian) * New translations en-US.yaml (English, United Kingdom) * New translations en-US.yaml (English, Canada) * New translations en-US.yaml (French, Canada) * New translations en-US.yaml (Croatian) * Add project_url to defaultTemplateData (#12033) Might be useful in template footers. * Update items.md * Rename panel to tile * Rename preview->overview * Style flow log detail * Log all parsed options * Show used options in revision * Finish log detail drawer * new create flow flow * fix firstOpen for new create flow flow * update field layout for create flow form * Fix TS typing * Fix missing import * Append random hash to key when duplicating operations * Revert "Remove webhooks" This reverts commit 044d3d8b66. * Don't delete webhooks * Make option preview selectable * Prevent invalid linking when duplicating operations after creating operations * Prevent sending of malformed query filter when deleting flow * implement new manual trigger * simplify payload for manual trigger * use buttons instead of dropdown + run button * add async option & loading state * add collection check to manual trigger * emit refresh after running flow in sidebar * Add cross-instance messenger for reloading * Use flow drawer for both create and edit * Add manual trigger flow permissions to app recommended Ensures that non-admin users can actually see the flows sidebar detail * Add basic logs redaction * Remove endpoint to trigger an operation * Allow configuring location for manual trigger * Rename "hook" trigger to "event" * Tweak icon size * Fix create flow button in info notice * Make activity tracking full width * Tweak descriptions * Too long for comfort * Remove mode option from item-* operations * fix manual trigger empty collections option * Add no-logs-yet message in sidebar detail * Reset trigger options on change of trigger * Rename `data`->`payload` * Remove mode from preview of item-* operations * Return operation options with "{{key}}" as raw value * Show flow name in delete confirmation * Add default generated name/key to new operations * shorten arrows WIP still needs icons moved * rename note to description * fix hint button icons * update event hook type labels * Animate resolve/reject arrow hints * reorder event types * Use x+4 instead of x+6 for new operation panels * compress options to fit 6 lines in operation * update hook labels * animate trigger box shadow sorry, rijk! * update (global) disabled button color 1 shade * Format times nicer * Add placeholder for query * add a note * Fix formatting for curly brackets in translations * Add item Create/update payload placeholder * Add placeholder to user uuid * Accept either null or undefined for nullable operation options * Allow any string as request body * Add more placeholders * Consolidate filterScope and actionScope, filterCollections and actionCollections * Rename flow note to description in types Co-authored-by: rijkvanzanten <rijkvanzanten@me.com> Co-authored-by: Azri Kahar <42867097+azrikahar@users.noreply.github.com> Co-authored-by: Brainslug <br41nslug@users.noreply.github.com> Co-authored-by: Ben Haynes <ben@rngr.org> Co-authored-by: Nicola Krumschmidt <nicola.krumschmidt@freenet.de> Co-authored-by: Aiden Foxx <aiden.foxx.mail@gmail.com> Co-authored-by: Jan-Willem <jan-willem@qdentity.nl> Co-authored-by: Yasser Lahbibi <yasser.lahbibi@apenhet.com> Co-authored-by: Louis <32719150+louisgavalda@users.noreply.github.com> Co-authored-by: Jay Cammarano <67079013+jaycammarano@users.noreply.github.com> Co-authored-by: Erick Torres <ericktorresh@gmail.com> Co-authored-by: jaycammarano <jay.cammarano@gmail.com> Co-authored-by: Yuriy Belenko <yura-bely@mail.ru> Co-authored-by: ian <licitdev@gmail.com>
2026-04-25 03:00:53 -04:00 · 2022-06-03 00:18:49 +02:00
parent d175290657
commit 857eae54de
133 changed files with 9498 additions and 2448 deletions
--- a/api/package.json
+++ b/api/package.json
@@ -111,8 +111,10 @@
 		"execa": "^5.1.1",
 		"exifr": "^7.1.2",
 		"express": "^4.17.1",
+		"fast-redact": "^3.1.1",
 		"flat": "^5.0.2",
 		"fs-extra": "^10.0.0",
+		"globby": "^11.0.4",
 		"graphql": "^15.5.0",
 		"graphql-compose": "^9.0.1",
 		"helmet": "^4.6.0",
@@ -130,6 +132,7 @@
 		"lodash": "^4.17.21",
 		"macos-release": "^2.4.1",
 		"marked": "^4.0.3",
+		"micromustache": "^8.0.3",
 		"mime-types": "^2.1.31",
 		"ms": "^2.1.3",
 		"nanoid": "^3.1.23",
@@ -182,6 +185,7 @@
 		"@types/express": "4.17.13",
 		"@types/express-pino-logger": "4.0.3",
 		"@types/express-session": "1.17.4",
+		"@types/fast-redact": "^3.0.1",
 		"@types/flat": "5.0.2",
 		"@types/fs-extra": "9.0.13",
 		"@types/inquirer": "8.1.3",
--- a/api/src/app.ts
+++ b/api/src/app.ts
@@ -13,12 +13,14 @@ import dashboardsRouter from './controllers/dashboards';
 import extensionsRouter from './controllers/extensions';
 import fieldsRouter from './controllers/fields';
 import filesRouter from './controllers/files';
+import flowsRouter from './controllers/flows';
 import foldersRouter from './controllers/folders';
 import graphqlRouter from './controllers/graphql';
 import itemsRouter from './controllers/items';
 import notFoundHandler from './controllers/not-found';
 import panelsRouter from './controllers/panels';
 import notificationsRouter from './controllers/notifications';
+import operationsRouter from './controllers/operations';
 import permissionsRouter from './controllers/permissions';
 import presetsRouter from './controllers/presets';
 import relationsRouter from './controllers/relations';
@@ -35,6 +37,7 @@ import emitter from './emitter';
 import env from './env';
 import { InvalidPayloadException } from './exceptions';
 import { getExtensionManager } from './extensions';
+import { getFlowManager } from './flows';
 import logger, { expressLogger } from './logger';
 import authenticate from './middleware/authenticate';
 import getPermissions from './middleware/get-permissions';
@@ -83,8 +86,10 @@ export default async function createApp(): Promise<express.Application> {
 	await registerAuthProviders();

 	const extensionManager = getExtensionManager();
+	const flowManager = getFlowManager();

 	await extensionManager.initialize();
+	await flowManager.initialize();

 	const app = express();

@@ -215,9 +220,11 @@ export default async function createApp(): Promise<express.Application> {
 	app.use('/extensions', extensionsRouter);
 	app.use('/fields', fieldsRouter);
 	app.use('/files', filesRouter);
+	app.use('/flows', flowsRouter);
 	app.use('/folders', foldersRouter);
 	app.use('/items', itemsRouter);
 	app.use('/notifications', notificationsRouter);
+	app.use('/operations', operationsRouter);
 	app.use('/panels', panelsRouter);
 	app.use('/permissions', permissionsRouter);
 	app.use('/presets', presetsRouter);
--- a/api/src/controllers/activity.ts
+++ b/api/src/controllers/activity.ts
@@ -1,3 +1,4 @@
+import { Action } from '@directus/shared/types';
 import express from 'express';
 import Joi from 'joi';
 import { ForbiddenException, InvalidPayloadException } from '../exceptions';
@@ -5,7 +6,6 @@ import { respond } from '../middleware/respond';
 import useCollection from '../middleware/use-collection';
 import { validateBatch } from '../middleware/validate-batch';
 import { ActivityService, MetaService } from '../services';
-import { Action } from '../types';
 import asyncHandler from '../utils/async-handler';
 import { getIPFromReq } from '../utils/get-ip-from-req';

--- a/api/src/controllers/flows.ts
+++ b/api/src/controllers/flows.ts
@@ -0,0 +1,211 @@
+import express from 'express';
+import { UUID_REGEX } from '../constants';
+import { ForbiddenException } from '../exceptions';
+import { getFlowManager } from '../flows';
+import { respond } from '../middleware/respond';
+import useCollection from '../middleware/use-collection';
+import { validateBatch } from '../middleware/validate-batch';
+import { MetaService, FlowsService } from '../services';
+import { PrimaryKey } from '../types';
+import asyncHandler from '../utils/async-handler';
+
+const router = express.Router();
+
+router.use(useCollection('directus_flows'));
+
+const webhookFlowHandler = asyncHandler(async (req, res, next) => {
+	const flowManager = getFlowManager();
+
+	const result = await flowManager.runWebhookFlow(
+		`${req.method}-${req.params.pk}`,
+		{
+			path: req.path,
+			query: req.query,
+			body: req.body,
+			method: req.method,
+			headers: req.headers,
+		},
+		{
+			accountability: req.accountability,
+			schema: req.schema,
+		}
+	);
+
+	res.locals.payload = result;
+	return next();
+});
+
+router.get(`/trigger/:pk(${UUID_REGEX})`, webhookFlowHandler, respond);
+router.post(`/trigger/:pk(${UUID_REGEX})`, webhookFlowHandler, respond);
+
+router.post(
+	'/',
+	asyncHandler(async (req, res, next) => {
+		const service = new FlowsService({
+			accountability: req.accountability,
+			schema: req.schema,
+		});
+
+		const savedKeys: PrimaryKey[] = [];
+
+		if (Array.isArray(req.body)) {
+			const keys = await service.createMany(req.body);
+			savedKeys.push(...keys);
+		} else {
+			const key = await service.createOne(req.body);
+			savedKeys.push(key);
+		}
+
+		try {
+			if (Array.isArray(req.body)) {
+				const items = await service.readMany(savedKeys, req.sanitizedQuery);
+				res.locals.payload = { data: items };
+			} else {
+				const item = await service.readOne(savedKeys[0], req.sanitizedQuery);
+				res.locals.payload = { data: item };
+			}
+		} catch (error) {
+			if (error instanceof ForbiddenException) {
+				return next();
+			}
+
+			throw error;
+		}
+
+		return next();
+	}),
+	respond
+);
+
+const readHandler = asyncHandler(async (req, res, next) => {
+	const service = new FlowsService({
+		accountability: req.accountability,
+		schema: req.schema,
+	});
+	const metaService = new MetaService({
+		accountability: req.accountability,
+		schema: req.schema,
+	});
+
+	const records = await service.readByQuery(req.sanitizedQuery);
+	const meta = await metaService.getMetaForQuery(req.collection, req.sanitizedQuery);
+
+	res.locals.payload = { data: records || null, meta };
+	return next();
+});
+
+router.get('/', validateBatch('read'), readHandler, respond);
+router.search('/', validateBatch('read'), readHandler, respond);
+
+router.get(
+	'/:pk',
+	asyncHandler(async (req, res, next) => {
+		const service = new FlowsService({
+			accountability: req.accountability,
+			schema: req.schema,
+		});
+
+		const record = await service.readOne(req.params.pk, req.sanitizedQuery);
+
+		res.locals.payload = { data: record || null };
+		return next();
+	}),
+	respond
+);
+
+router.patch(
+	'/',
+	validateBatch('update'),
+	asyncHandler(async (req, res, next) => {
+		const service = new FlowsService({
+			accountability: req.accountability,
+			schema: req.schema,
+		});
+
+		let keys: PrimaryKey[] = [];
+
+		if (req.body.keys) {
+			keys = await service.updateMany(req.body.keys, req.body.data);
+		} else {
+			keys = await service.updateByQuery(req.body.query, req.body.data);
+		}
+
+		try {
+			const result = await service.readMany(keys, req.sanitizedQuery);
+			res.locals.payload = { data: result };
+		} catch (error) {
+			if (error instanceof ForbiddenException) {
+				return next();
+			}
+
+			throw error;
+		}
+
+		return next();
+	}),
+	respond
+);
+
+router.patch(
+	'/:pk',
+	asyncHandler(async (req, res, next) => {
+		const service = new FlowsService({
+			accountability: req.accountability,
+			schema: req.schema,
+		});
+
+		const primaryKey = await service.updateOne(req.params.pk, req.body);
+
+		try {
+			const item = await service.readOne(primaryKey, req.sanitizedQuery);
+			res.locals.payload = { data: item || null };
+		} catch (error) {
+			if (error instanceof ForbiddenException) {
+				return next();
+			}
+
+			throw error;
+		}
+
+		return next();
+	}),
+	respond
+);
+
+router.delete(
+	'/',
+	asyncHandler(async (req, res, next) => {
+		const service = new FlowsService({
+			accountability: req.accountability,
+			schema: req.schema,
+		});
+
+		if (Array.isArray(req.body)) {
+			await service.deleteMany(req.body);
+		} else if (req.body.keys) {
+			await service.deleteMany(req.body.keys);
+		} else {
+			await service.deleteByQuery(req.body.query);
+		}
+
+		return next();
+	}),
+	respond
+);
+
+router.delete(
+	'/:pk',
+	asyncHandler(async (req, res, next) => {
+		const service = new FlowsService({
+			accountability: req.accountability,
+			schema: req.schema,
+		});
+
+		await service.deleteOne(req.params.pk);
+
+		return next();
+	}),
+	respond
+);
+
+export default router;
--- a/api/src/controllers/operations.ts
+++ b/api/src/controllers/operations.ts
@@ -0,0 +1,184 @@
+import express from 'express';
+import { ForbiddenException } from '../exceptions';
+import { respond } from '../middleware/respond';
+import useCollection from '../middleware/use-collection';
+import { validateBatch } from '../middleware/validate-batch';
+import { MetaService, OperationsService } from '../services';
+import { PrimaryKey } from '../types';
+import asyncHandler from '../utils/async-handler';
+
+const router = express.Router();
+
+router.use(useCollection('directus_operations'));
+
+router.post(
+	'/',
+	asyncHandler(async (req, res, next) => {
+		const service = new OperationsService({
+			accountability: req.accountability,
+			schema: req.schema,
+		});
+
+		const savedKeys: PrimaryKey[] = [];
+
+		if (Array.isArray(req.body)) {
+			const keys = await service.createMany(req.body);
+			savedKeys.push(...keys);
+		} else {
+			const key = await service.createOne(req.body);
+			savedKeys.push(key);
+		}
+
+		try {
+			if (Array.isArray(req.body)) {
+				const items = await service.readMany(savedKeys, req.sanitizedQuery);
+				res.locals.payload = { data: items };
+			} else {
+				const item = await service.readOne(savedKeys[0], req.sanitizedQuery);
+				res.locals.payload = { data: item };
+			}
+		} catch (error) {
+			if (error instanceof ForbiddenException) {
+				return next();
+			}
+
+			throw error;
+		}
+
+		return next();
+	}),
+	respond
+);
+
+const readHandler = asyncHandler(async (req, res, next) => {
+	const service = new OperationsService({
+		accountability: req.accountability,
+		schema: req.schema,
+	});
+	const metaService = new MetaService({
+		accountability: req.accountability,
+		schema: req.schema,
+	});
+
+	const records = await service.readByQuery(req.sanitizedQuery);
+	const meta = await metaService.getMetaForQuery(req.collection, req.sanitizedQuery);
+
+	res.locals.payload = { data: records || null, meta };
+	return next();
+});
+
+router.get('/', validateBatch('read'), readHandler, respond);
+router.search('/', validateBatch('read'), readHandler, respond);
+
+router.get(
+	'/:pk',
+	asyncHandler(async (req, res, next) => {
+		const service = new OperationsService({
+			accountability: req.accountability,
+			schema: req.schema,
+		});
+
+		const record = await service.readOne(req.params.pk, req.sanitizedQuery);
+
+		res.locals.payload = { data: record || null };
+		return next();
+	}),
+	respond
+);
+
+router.patch(
+	'/',
+	validateBatch('update'),
+	asyncHandler(async (req, res, next) => {
+		const service = new OperationsService({
+			accountability: req.accountability,
+			schema: req.schema,
+		});
+
+		let keys: PrimaryKey[] = [];
+
+		if (req.body.keys) {
+			keys = await service.updateMany(req.body.keys, req.body.data);
+		} else {
+			keys = await service.updateByQuery(req.body.query, req.body.data);
+		}
+
+		try {
+			const result = await service.readMany(keys, req.sanitizedQuery);
+			res.locals.payload = { data: result };
+		} catch (error) {
+			if (error instanceof ForbiddenException) {
+				return next();
+			}
+
+			throw error;
+		}
+
+		return next();
+	}),
+	respond
+);
+
+router.patch(
+	'/:pk',
+	asyncHandler(async (req, res, next) => {
+		const service = new OperationsService({
+			accountability: req.accountability,
+			schema: req.schema,
+		});
+
+		const primaryKey = await service.updateOne(req.params.pk, req.body);
+
+		try {
+			const item = await service.readOne(primaryKey, req.sanitizedQuery);
+			res.locals.payload = { data: item || null };
+		} catch (error) {
+			if (error instanceof ForbiddenException) {
+				return next();
+			}
+
+			throw error;
+		}
+
+		return next();
+	}),
+	respond
+);
+
+router.delete(
+	'/',
+	asyncHandler(async (req, res, next) => {
+		const service = new OperationsService({
+			accountability: req.accountability,
+			schema: req.schema,
+		});
+
+		if (Array.isArray(req.body)) {
+			await service.deleteMany(req.body);
+		} else if (req.body.keys) {
+			await service.deleteMany(req.body.keys);
+		} else {
+			await service.deleteByQuery(req.body.query);
+		}
+
+		return next();
+	}),
+	respond
+);
+
+router.delete(
+	'/:pk',
+	asyncHandler(async (req, res, next) => {
+		const service = new OperationsService({
+			accountability: req.accountability,
+			schema: req.schema,
+		});
+
+		await service.deleteOne(req.params.pk);
+
+		return next();
+	}),
+	respond
+);
+
+export default router;
--- a/api/src/database/migrations/20220429A-add-flows.ts
+++ b/api/src/database/migrations/20220429A-add-flows.ts
@@ -0,0 +1,89 @@
+import { Knex } from 'knex';
+import { toArray } from '@directus/shared/utils';
+import { v4 as uuidv4 } from 'uuid';
+import { parseJSON } from '../../utils/parse-json';
+
+export async function up(knex: Knex): Promise<void> {
+	await knex.schema.createTable('directus_flows', (table) => {
+		table.uuid('id').primary().notNullable();
+		table.string('name').notNullable();
+		table.string('icon', 30);
+		table.string('color').nullable();
+		table.text('description');
+		table.string('status').notNullable().defaultTo('active');
+		table.string('trigger');
+		table.string('accountability').defaultTo('all');
+		table.json('options');
+		table.uuid('operation').unique();
+		table.timestamp('date_created').defaultTo(knex.fn.now());
+		table.uuid('user_created').references('id').inTable('directus_users').onDelete('SET NULL');
+	});
+
+	await knex.schema.createTable('directus_operations', (table) => {
+		table.uuid('id').primary().notNullable();
+		table.string('name');
+		table.string('key').notNullable();
+		table.string('type').notNullable();
+		table.integer('position_x').notNullable();
+		table.integer('position_y').notNullable();
+		table.json('options');
+		table.uuid('resolve').unique().references('id').inTable('directus_operations');
+		table.uuid('reject').unique().references('id').inTable('directus_operations');
+		table.uuid('flow').notNullable().references('id').inTable('directus_flows').onDelete('CASCADE');
+		table.timestamp('date_created').defaultTo(knex.fn.now());
+		table.uuid('user_created').references('id').inTable('directus_users').onDelete('SET NULL');
+	});
+
+	const webhooks = await knex.select('*').from('directus_webhooks');
+
+	const flows = [];
+	const operations = [];
+
+	for (const webhook of webhooks) {
+		const flowID = uuidv4();
+
+		flows.push({
+			id: flowID,
+			name: webhook.name,
+			status: 'inactive',
+			trigger: 'hook',
+			options: JSON.stringify({
+				name: webhook.name,
+				type: 'action',
+				scope: toArray(webhook.actions).map((scope) => `items.${scope}`),
+				collections: toArray(webhook.collections),
+			}),
+		});
+
+		operations.push({
+			id: uuidv4(),
+			name: 'Request',
+			key: 'request',
+			type: 'request',
+			position_x: 21,
+			position_y: 1,
+			options: JSON.stringify({
+				url: webhook.url,
+				headers: typeof webhook.headers === 'string' ? parseJSON(webhook.headers) : webhook.headers,
+				data: webhook.data ? '{{$trigger}}' : null,
+				method: webhook.method,
+			}),
+			date_created: new Date(),
+			flow: flowID,
+		});
+	}
+
+	if (flows.length && operations.length) {
+		await knex.insert(flows).into('directus_flows');
+		await knex.insert(operations).into('directus_operations');
+
+		for (const operation of operations) {
+			await knex('directus_flows').update({ operation: operation.id }).where({ id: operation.flow });
+		}
+	}
+}
+
+export async function down(knex: Knex): Promise<void> {
+	await knex.schema.dropTable('directus_operations');
+	await knex.schema.dropTable('directus_flows');
+}
--- a/api/src/database/migrations/20220429B-add-color-to-insights-icon.ts
+++ b/api/src/database/migrations/20220429B-add-color-to-insights-icon.ts
@@ -0,0 +1,13 @@
+import { Knex } from 'knex';
+
+export async function up(knex: Knex): Promise<void> {
+	await knex.schema.alterTable('directus_dashboards', (table) => {
+		table.string('color').nullable();
+	});
+}
+
+export async function down(knex: Knex): Promise<void> {
+	await knex.schema.alterTable('directus_dashboards', (table) => {
+		table.dropColumn('color');
+	});
+}
--- a/api/src/database/migrations/20220429C-drop-non-null-from-ip-of-activity.ts
+++ b/api/src/database/migrations/20220429C-drop-non-null-from-ip-of-activity.ts
@@ -0,0 +1,13 @@
+import { Knex } from 'knex';
+
+export async function up(knex: Knex): Promise<void> {
+	await knex.schema.alterTable('directus_activity', (table) => {
+		table.setNullable('ip');
+	});
+}
+
+export async function down(knex: Knex): Promise<void> {
+	await knex.schema.alterTable('directus_activity', (table) => {
+		table.dropNullable('ip');
+	});
+}
--- a/api/src/database/migrations/20220429D-drop-non-null-from-sender-of-notifications.ts
+++ b/api/src/database/migrations/20220429D-drop-non-null-from-sender-of-notifications.ts
@@ -0,0 +1,13 @@
+import { Knex } from 'knex';
+
+export async function up(knex: Knex): Promise<void> {
+	await knex.schema.alterTable('directus_notifications', (table) => {
+		table.setNullable('sender');
+	});
+}
+
+export async function down(knex: Knex): Promise<void> {
+	await knex.schema.alterTable('directus_notifications', (table) => {
+		table.dropNullable('sender');
+	});
+}
--- a/api/src/database/seeds/05-activity.yaml
+++ b/api/src/database/seeds/05-activity.yaml
@@ -20,7 +20,6 @@ columns:
  user_agent:
    type: string
    length: 255
-    nullabel: false
  collection:
    type: string
    length: 64
--- a/api/src/database/system-data/collections/collections.yaml
+++ b/api/src/database/system-data/collections/collections.yaml
@@ -68,3 +68,7 @@ data:
  - collection: directus_shares
    icon: share
    note: $t:directus_collection.directus_shares
+  - collection: directus_flows
+    note: $t:directus_collection.directus_flows
+  - collection: directus_operations
+    note: $t:directus_collection.directus_operations
--- a/api/src/database/system-data/fields/flows.yaml
+++ b/api/src/database/system-data/fields/flows.yaml
@@ -0,0 +1,21 @@
+table: directus_flows
+
+fields:
+  - field: id
+    special: uuid
+  - field: name
+  - field: icon
+  - field: color
+  - field: note
+  - field: status
+  - field: trigger
+  - field: accountability
+  - field: options
+    special: cast-json
+  - field: operation
+  - field: operations
+    special: o2m
+  - field: date_created
+    special: date-created
+  - field: user_created
+    special: user-created
--- a/api/src/database/system-data/fields/operations.yaml
+++ b/api/src/database/system-data/fields/operations.yaml
@@ -0,0 +1,19 @@
+table: directus_operations
+
+fields:
+  - field: id
+    special: uuid
+  - field: name
+  - field: key
+  - field: type
+  - field: position_x
+  - field: position_y
+  - field: options
+    special: cast-json
+  - field: resolve
+  - field: reject
+  - field: flow
+  - field: date_created
+    special: date-created
+  - field: user_created
+    special: user-created
--- a/api/src/database/system-data/relations/relations.yaml
+++ b/api/src/database/system-data/relations/relations.yaml
@@ -61,6 +61,20 @@ data:
    many_field: dashboard
    one_collection: directus_dashboards
    one_field: panels
+  - many_collection: directus_flows
+    many_field: operation
+    one_collection: directus_operations
+  - many_collection: directus_operations
+    many_field: flow
+    one_collection: directus_flows
+    one_field: operations
+    one_deselect_action: delete
+  - many_collection: directus_operations
+    many_field: resolve
+    one_collection: directus_operations
+  - many_collection: directus_operations
+    many_field: reject
+    one_collection: directus_operations
  - many_collection: directus_files
    many_field: modified_by
    one_collection: directus_users
@@ -88,6 +102,12 @@ data:
  - many_collection: directus_panels
    many_field: user_created
    one_collection: directus_users
+  - many_collection: directus_flows
+    many_field: user_created
+    one_collection: directus_users
+  - many_collection: directus_operations
+    many_field: user_created
+    one_collection: directus_users
  - many_collection: directus_notifications
    many_field: recipient
    one_collection: directus_users
--- a/api/src/emitter.ts
+++ b/api/src/emitter.ts
@@ -29,7 +29,7 @@ export class Emitter {
 		context: EventContext
 	): Promise<T> {
 		const events = Array.isArray(event) ? event : [event];
-		const listeners: FilterHandler[] = events.flatMap((event) => this.filterEmitter.listeners(event));
+		const listeners = events.flatMap((event) => this.filterEmitter.listeners(event) as FilterHandler<T>[]);

 		let updatedPayload = payload;
 		for (const listener of listeners) {
--- a/api/src/extensions.ts
+++ b/api/src/extensions.ts
@@ -2,13 +2,16 @@ import express, { Router } from 'express';
 import path from 'path';
 import {
 	ActionHandler,
+	ApiExtension,
 	AppExtensionType,
 	EndpointConfig,
 	Extension,
 	ExtensionType,
 	FilterHandler,
 	HookConfig,
+	HybridExtension,
 	InitHandler,
+	OperationApiConfig,
 	ScheduleHandler,
 } from '@directus/shared/types';
 import {
@@ -25,6 +28,8 @@ import {
 	APP_SHARED_DEPS,
 	EXTENSION_PACKAGE_TYPES,
 	EXTENSION_TYPES,
+	HYBRID_EXTENSION_TYPES,
+	PACK_EXTENSION_TYPE,
 } from '@directus/shared/constants';
 import getDatabase from './database';
 import emitter, { Emitter } from './emitter';
@@ -36,7 +41,7 @@ import fse from 'fs-extra';
 import { getSchema } from './utils/get-schema';

 import * as services from './services';
-import { schedule, ScheduledTask, validate } from 'node-cron';
+import { schedule, validate } from 'node-cron';
 import { rollup } from 'rollup';
 import virtual from '@rollup/plugin-virtual';
 import alias from '@rollup/plugin-alias';
@@ -44,7 +49,10 @@ import { Url } from './utils/url';
 import getModuleDefault from './utils/get-module-default';
 import { clone, escapeRegExp } from 'lodash';
 import chokidar, { FSWatcher } from 'chokidar';
-import { pluralize } from '@directus/shared/utils';
+import { isExtensionObject, isHybridExtension, pluralize } from '@directus/shared/utils';
+import { getFlowManager } from './flows';
+import globby from 'globby';
+import { EventHandler } from './types';

 let extensionManager: ExtensionManager | undefined;

@@ -58,16 +66,11 @@ export function getExtensionManager(): ExtensionManager {
 	return extensionManager;
 }

-type EventHandler =
-	| { type: 'filter'; name: string; handler: FilterHandler }
-	| { type: 'action'; name: string; handler: ActionHandler }
-	| { type: 'init'; name: string; handler: InitHandler }
-	| { type: 'schedule'; task: ScheduledTask };
-
 type AppExtensions = Partial<Record<AppExtensionType, string>>;
 type ApiExtensions = {
 	hooks: { path: string; events: EventHandler[] }[];
 	endpoints: { path: string }[];
+	operations: { path: string }[];
 };

 type Options = {
@@ -87,7 +90,7 @@ class ExtensionManager {
 	private extensions: Extension[] = [];

 	private appExtensions: AppExtensions = {};
-	private apiExtensions: ApiExtensions = { hooks: [], endpoints: [] };
+	private apiExtensions: ApiExtensions = { hooks: [], endpoints: [], operations: [] };

 	private apiEmitter: Emitter;
 	private endpointRouter: Router;
@@ -180,6 +183,7 @@ class ExtensionManager {

 		this.registerHooks();
 		this.registerEndpoints();
+		await this.registerOperations();

 		if (env.SERVE_APP) {
 			this.appExtensions = await this.generateExtensionBundles();
@@ -191,6 +195,7 @@ class ExtensionManager {
 	private async unload(): Promise<void> {
 		this.unregisterHooks();
 		this.unregisterEndpoints();
+		this.unregisterOperations();

 		this.apiEmitter.offAll();

@@ -205,16 +210,18 @@ class ExtensionManager {
 		if (this.options.watch && !this.watcher) {
 			logger.info('Watching extensions for changes...');

-			const localExtensionPaths = (env.SERVE_APP ? EXTENSION_TYPES : API_EXTENSION_TYPES).map((type) =>
-				path.posix.join(
+			const localExtensionPaths = (env.SERVE_APP ? EXTENSION_TYPES : API_EXTENSION_TYPES).flatMap((type) => {
+				const typeDir = path.posix.join(
 					path.relative('.', env.EXTENSIONS_PATH).split(path.sep).join(path.posix.sep),
-					pluralize(type),
-					'*',
-					'index.js'
-				)
-			);
+					pluralize(type)
+				);

-			this.watcher = chokidar.watch([path.resolve('.', 'package.json'), ...localExtensionPaths], {
+				return isHybridExtension(type)
+					? [path.posix.join(typeDir, '*', 'app.js'), path.posix.join(typeDir, '*', 'api.js')]
+					: path.posix.join(typeDir, '*', 'index.js');
+			});
+
+			this.watcher = chokidar.watch([path.resolve('package.json'), ...localExtensionPaths], {
 				ignoreInitial: true,
 			});

@@ -230,10 +237,15 @@ class ExtensionManager {
 			const toPackageExtensionPaths = (extensions: Extension[]) =>
 				extensions
 					.filter((extension) => !extension.local)
-					.map((extension) =>
-						extension.type !== 'pack'
-							? path.resolve(extension.path, extension.entrypoint || '')
-							: path.resolve(extension.path, 'package.json')
+					.flatMap((extension) =>
+						extension.type === PACK_EXTENSION_TYPE
+							? path.resolve(extension.path, 'package.json')
+							: isExtensionObject(extension, HYBRID_EXTENSION_TYPES)
+							? [
+									path.resolve(extension.path, extension.entrypoint.app),
+									path.resolve(extension.path, extension.entrypoint.api),
+							  ]
+							: path.resolve(extension.path, extension.entrypoint)
 					);

 			const addedPackageExtensionPaths = toPackageExtensionPaths(added);
@@ -311,11 +323,16 @@ class ExtensionManager {
 	}

 	private registerHooks(): void {
-		const hooks = this.extensions.filter((extension) => extension.type === 'hook');
+		const hooks = this.extensions.filter((extension): extension is ApiExtension => extension.type === 'hook');

 		for (const hook of hooks) {
 			try {
-				this.registerHook(hook);
+				const hookPath = path.resolve(hook.path, hook.entrypoint);
+				const hookInstance: HookConfig | { default: HookConfig } = require(hookPath);
+
+				const config = getModuleDefault(hookInstance);
+
+				this.registerHook(config, hookPath);
 			} catch (error: any) {
 				logger.warn(`Couldn't register hook "${hook.name}"`);
 				logger.warn(error);
@@ -324,11 +341,16 @@ class ExtensionManager {
 	}

 	private registerEndpoints(): void {
-		const endpoints = this.extensions.filter((extension) => extension.type === 'endpoint');
+		const endpoints = this.extensions.filter((extension): extension is ApiExtension => extension.type === 'endpoint');

 		for (const endpoint of endpoints) {
 			try {
-				this.registerEndpoint(endpoint, this.endpointRouter);
+				const endpointPath = path.resolve(endpoint.path, endpoint.entrypoint);
+				const endpointInstance: EndpointConfig | { default: EndpointConfig } = require(endpointPath);
+
+				const config = getModuleDefault(endpointInstance);
+
+				this.registerEndpoint(config, endpointPath, this.endpointRouter);
 			} catch (error: any) {
 				logger.warn(`Couldn't register endpoint "${endpoint.name}"`);
 				logger.warn(error);
@@ -336,14 +358,43 @@ class ExtensionManager {
 		}
 	}

-	private registerHook(hook: Extension) {
-		const hookPath = path.resolve(hook.path, hook.entrypoint || '');
-		const hookInstance: HookConfig | { default: HookConfig } = require(hookPath);
+	private async registerOperations(): Promise<void> {
+		const internalPaths = await globby(
+			path.posix.join(path.relative('.', __dirname).split(path.sep).join(path.posix.sep), 'operations/*/index.(js|ts)')
+		);

-		const register = getModuleDefault(hookInstance);
+		const internalOperations = internalPaths.map((internalPath) => {
+			const dirs = internalPath.split(path.sep);

+			return {
+				name: dirs[dirs.length - 2],
+				path: dirs.slice(0, -1).join(path.sep),
+				entrypoint: { api: dirs[dirs.length - 1] },
+			};
+		});
+
+		const operations = this.extensions.filter(
+			(extension): extension is HybridExtension => extension.type === 'operation'
+		);
+
+		for (const operation of [...internalOperations, ...operations]) {
+			try {
+				const operationPath = path.resolve(operation.path, operation.entrypoint.api);
+				const operationInstance: OperationApiConfig | { default: OperationApiConfig } = require(operationPath);
+
+				const config = getModuleDefault(operationInstance);
+
+				this.registerOperation(config, operationPath);
+			} catch (error: any) {
+				logger.warn(`Couldn't register operation "${operation.name}"`);
+				logger.warn(error);
+			}
+		}
+	}
+
+	private registerHook(register: HookConfig, path: string) {
 		const hookHandler: { path: string; events: EventHandler[] } = {
-			path: hookPath,
+			path,
 			events: [],
 		};

@@ -410,14 +461,9 @@ class ExtensionManager {
 		this.apiExtensions.hooks.push(hookHandler);
 	}

-	private registerEndpoint(endpoint: Extension, router: Router) {
-		const endpointPath = path.resolve(endpoint.path, endpoint.entrypoint || '');
-		const endpointInstance: EndpointConfig | { default: EndpointConfig } = require(endpointPath);
-
-		const mod = getModuleDefault(endpointInstance);
-
-		const register = typeof mod === 'function' ? mod : mod.handler;
-		const routeName = typeof mod === 'function' ? endpoint.name : mod.id;
+	private registerEndpoint(config: EndpointConfig, path: string, router: Router) {
+		const register = typeof config === 'function' ? config : config.handler;
+		const routeName = typeof config === 'function' ? config.name : config.id;

 		const scopedRouter = express.Router();
 		router.use(`/${routeName}`, scopedRouter);
@@ -433,7 +479,17 @@ class ExtensionManager {
 		});

 		this.apiExtensions.endpoints.push({
-			path: endpointPath,
+			path,
+		});
+	}
+
+	private registerOperation(config: OperationApiConfig, path: string) {
+		const flowManager = getFlowManager();
+
+		flowManager.addOperation(config.id, config.handler);
+
+		this.apiExtensions.operations.push({
+			path,
 		});
 	}

@@ -471,4 +527,16 @@ class ExtensionManager {

 		this.apiExtensions.endpoints = [];
 	}
+
+	private unregisterOperations(): void {
+		for (const operation of this.apiExtensions.operations) {
+			delete require.cache[require.resolve(operation.path)];
+		}
+
+		const flowManager = getFlowManager();
+
+		flowManager.clearOperations();
+
+		this.apiExtensions.operations = [];
+	}
 }
--- a/api/src/flows.ts
+++ b/api/src/flows.ts
@@ -0,0 +1,367 @@
+import * as sharedExceptions from '@directus/shared/exceptions';
+import {
+	ActionHandler,
+	FilterHandler,
+	Flow,
+	Operation,
+	OperationHandler,
+	SchemaOverview,
+	Accountability,
+	Action,
+} from '@directus/shared/types';
+import { get } from 'micromustache';
+import { schedule, validate } from 'node-cron';
+import getDatabase from './database';
+import emitter from './emitter';
+import env from './env';
+import * as exceptions from './exceptions';
+import logger from './logger';
+import * as services from './services';
+import { FlowsService } from './services';
+import { EventHandler } from './types';
+import { constructFlowTree } from './utils/construct-flow-tree';
+import { getSchema } from './utils/get-schema';
+import { ActivityService } from './services/activity';
+import { RevisionsService } from './services/revisions';
+import { Knex } from 'knex';
+import { omit } from 'lodash';
+import { getMessenger } from './messenger';
+import fastRedact from 'fast-redact';
+import { applyOperationOptions } from './utils/operation-options';
+
+let flowManager: FlowManager | undefined;
+
+const redactLogs = fastRedact({
+	censor: '--redacted--',
+	paths: ['*.headers.authorization', '*.access_token', '*.headers.cookie'],
+	serialize: false,
+});
+
+export function getFlowManager(): FlowManager {
+	if (flowManager) {
+		return flowManager;
+	}
+
+	flowManager = new FlowManager();
+
+	return flowManager;
+}
+
+type TriggerHandler = {
+	id: string;
+	events: EventHandler[];
+};
+
+const TRIGGER_KEY = '$trigger';
+const ACCOUNTABILITY_KEY = '$accountability';
+const LAST_KEY = '$last';
+
+class FlowManager {
+	private operations: Record<string, OperationHandler> = {};
+
+	private triggerHandlers: TriggerHandler[] = [];
+	private operationFlowHandlers: Record<string, any> = {};
+	private webhookFlowHandlers: Record<string, any> = {};
+
+	public async initialize(): Promise<void> {
+		const flowsService = new FlowsService({ knex: getDatabase(), schema: await getSchema() });
+
+		const flows = await flowsService.readByQuery({
+			filter: { status: { _eq: 'active' } },
+			fields: ['*', 'operations.*'],
+		});
+
+		const flowTrees = flows.map((flow) => constructFlowTree(flow));
+
+		for (const flow of flowTrees) {
+			if (flow.trigger === 'event') {
+				const events: string[] =
+					flow.options?.scope
+						?.map((scope: string) => {
+							if (['items.create', 'items.update', 'items.delete'].includes(scope)) {
+								return (
+									flow.options?.collections?.map((collection: string) => {
+										if (collection.startsWith('directus_')) {
+											const action = scope.split('.')[1];
+											return collection.substring(9) + '.' + action;
+										}
+
+										return `${collection}.${scope}`;
+									}) ?? []
+								);
+							}
+
+							return scope;
+						})
+						?.flat() ?? [];
+
+				if (flow.options.type === 'filter') {
+					const handler: FilterHandler = (payload, meta, context) =>
+						this.executeFlow(
+							flow,
+							{ payload, ...meta },
+							{
+								accountability: context.accountability,
+								database: context.database,
+								getSchema: context.schema ? () => context.schema : getSchema,
+							}
+						);
+
+					events.forEach((event) => emitter.onFilter(event, handler));
+					this.triggerHandlers.push({
+						id: flow.id,
+						events: events.map((event) => ({ type: 'filter', name: event, handler })),
+					});
+				} else if (flow.options.type === 'action') {
+					const handler: ActionHandler = (meta, context) =>
+						this.executeFlow(flow, meta, {
+							accountability: context.accountability,
+							database: context.database,
+							getSchema: context.schema ? () => context.schema : getSchema,
+						});
+
+					events.forEach((event) => emitter.onAction(event, handler));
+					this.triggerHandlers.push({
+						id: flow.id,
+						events: events.map((event) => ({ type: 'action', name: event, handler })),
+					});
+				}
+			} else if (flow.trigger === 'schedule') {
+				if (validate(flow.options.cron)) {
+					const task = schedule(flow.options.cron, async () => {
+						try {
+							await this.executeFlow(flow);
+						} catch (error: any) {
+							logger.error(error);
+						}
+					});
+
+					this.triggerHandlers.push({ id: flow.id, events: [{ type: flow.trigger, task }] });
+				} else {
+					logger.warn(`Couldn't register cron trigger. Provided cron is invalid: ${flow.options.cron}`);
+				}
+			} else if (flow.trigger === 'operation') {
+				const handler = (data: unknown, context: Record<string, unknown>) => this.executeFlow(flow, data, context);
+
+				this.operationFlowHandlers[flow.id] = handler;
+			} else if (flow.trigger === 'webhook') {
+				const handler = (data: unknown, context: Record<string, unknown>) => {
+					if (flow.options.async) {
+						this.executeFlow(flow, data, context);
+					} else {
+						return this.executeFlow(flow, data, context);
+					}
+				};
+
+				const method = flow.options?.method ?? 'GET';
+
+				// Default return to $last for webhooks
+				flow.options.return = flow.options.return ?? '$last';
+
+				this.webhookFlowHandlers[`${method}-${flow.id}`] = handler;
+			} else if (flow.trigger === 'manual') {
+				const handler = (data: unknown, context: Record<string, unknown>) => {
+					const enabledCollections = flow.options?.collections ?? [];
+					const targetCollection = (data as Record<string, any>)?.body.collection;
+
+					if (!targetCollection) {
+						logger.warn(`Manual trigger requires "collection" to be specified in the payload`);
+						throw new exceptions.ForbiddenException();
+					}
+
+					if (enabledCollections.length === 0) {
+						logger.warn(`There is no collections configured for this manual trigger`);
+						throw new exceptions.ForbiddenException();
+					}
+
+					if (!enabledCollections.includes(targetCollection)) {
+						logger.warn(`Specified collection must be one of: ${enabledCollections.join(', ')}.`);
+						throw new exceptions.ForbiddenException();
+					}
+
+					if (flow.options.async) {
+						this.executeFlow(flow, data, context);
+					} else {
+						return this.executeFlow(flow, data, context);
+					}
+				};
+
+				// Default return to $last for manual
+				flow.options.return = '$last';
+
+				this.webhookFlowHandlers[`POST-${flow.id}`] = handler;
+			}
+		}
+
+		getMessenger().subscribe('flows', (event) => {
+			if (event.type === 'reload') {
+				this.reload();
+			}
+		});
+	}
+
+	public async reload(): Promise<void> {
+		for (const trigger of this.triggerHandlers) {
+			trigger.events.forEach((event) => {
+				switch (event.type) {
+					case 'filter':
+						emitter.offFilter(event.name, event.handler);
+						break;
+					case 'action':
+						emitter.offAction(event.name, event.handler);
+						break;
+					case 'schedule':
+						event.task.stop();
+						break;
+				}
+			});
+		}
+
+		this.triggerHandlers = [];
+		this.operationFlowHandlers = {};
+		this.webhookFlowHandlers = {};
+
+		await this.initialize();
+	}
+
+	public addOperation(id: string, operation: OperationHandler): void {
+		this.operations[id] = operation;
+	}
+
+	public clearOperations(): void {
+		this.operations = {};
+	}
+
+	public async runOperationFlow(id: string, data: unknown, context: Record<string, unknown>): Promise<unknown> {
+		if (!(id in this.operationFlowHandlers)) {
+			logger.warn(`Couldn't find operation triggered flow with id "${id}"`);
+			return null;
+		}
+
+		const handler = this.operationFlowHandlers[id];
+
+		return handler(data, context);
+	}
+
+	public async runWebhookFlow(id: string, data: unknown, context: Record<string, unknown>): Promise<unknown> {
+		if (!(id in this.webhookFlowHandlers)) {
+			logger.warn(`Couldn't find webhook or manual triggered flow with id "${id}"`);
+			throw new exceptions.ForbiddenException();
+		}
+
+		const handler = this.webhookFlowHandlers[id];
+
+		return handler(data, context);
+	}
+
+	private async executeFlow(flow: Flow, data: unknown = null, context: Record<string, unknown> = {}): Promise<unknown> {
+		const database = (context.database as Knex) ?? getDatabase();
+		const schema = (context.schema as SchemaOverview) ?? (await getSchema({ database }));
+
+		const keyedData: Record<string, unknown> = {
+			[TRIGGER_KEY]: data,
+			[LAST_KEY]: data,
+			[ACCOUNTABILITY_KEY]: context?.accountability ?? null,
+		};
+
+		let nextOperation = flow.operation;
+
+		const steps: {
+			operation: string;
+			key: string;
+			status: 'resolve' | 'reject' | 'unknown';
+			options: Record<string, any> | null;
+		}[] = [];
+
+		while (nextOperation !== null) {
+			const { successor, data, status, options } = await this.executeOperation(nextOperation, keyedData, context);
+
+			keyedData[nextOperation.key] = data;
+			keyedData[LAST_KEY] = data;
+			steps.push({ operation: nextOperation!.id, key: nextOperation.key, status, options });
+
+			nextOperation = successor;
+		}
+
+		if (flow.accountability !== null) {
+			const activityService = new ActivityService({
+				knex: database,
+				schema: schema,
+			});
+
+			const accountability = context?.accountability as Accountability | undefined;
+
+			const activity = await activityService.createOne({
+				action: Action.RUN,
+				user: accountability?.user ?? null,
+				collection: 'directus_flows',
+				ip: accountability?.ip ?? null,
+				user_agent: accountability?.userAgent ?? null,
+				item: flow.id,
+			});
+
+			if (flow.accountability === 'all') {
+				const revisionsService = new RevisionsService({
+					knex: database,
+					schema: schema,
+				});
+
+				await revisionsService.createOne({
+					activity: activity,
+					collection: 'directus_flows',
+					item: flow.id,
+					data: {
+						steps: steps,
+						data: redactLogs(omit(keyedData, '$accountability.permissions')), // Permissions is a ton of data, and is just a copy of what's in the directus_permissions table
+					},
+				});
+			}
+		}
+
+		if (flow.options.return === '$all') {
+			return keyedData;
+		} else if (flow.options.return) {
+			return get(keyedData, flow.options.return);
+		}
+
+		return undefined;
+	}
+
+	private async executeOperation(
+		operation: Operation,
+		keyedData: Record<string, unknown>,
+		context: Record<string, unknown> = {}
+	): Promise<{
+		successor: Operation | null;
+		status: 'resolve' | 'reject' | 'unknown';
+		data: unknown;
+		options: Record<string, any> | null;
+	}> {
+		if (!(operation.type in this.operations)) {
+			logger.warn(`Couldn't find operation ${operation.type}`);
+			return { successor: null, status: 'unknown', data: null, options: null };
+		}
+
+		const handler = this.operations[operation.type];
+
+		const options = applyOperationOptions(operation.options, keyedData);
+
+		try {
+			const result = await handler(options, {
+				services,
+				exceptions: { ...exceptions, ...sharedExceptions },
+				env,
+				database: getDatabase(),
+				logger,
+				getSchema,
+				data: keyedData,
+				accountability: null,
+				...context,
+			});
+
+			return { successor: operation.resolve, status: 'resolve', data: result ?? null, options };
+		} catch (error: unknown) {
+			return { successor: operation.reject, status: 'reject', data: error ?? null, options };
+		}
+	}
+}
--- a/api/src/operations/condition/index.ts
+++ b/api/src/operations/condition/index.ts
@@ -0,0 +1,20 @@
+import { Filter } from '@directus/shared/types';
+import { defineOperationApi, validatePayload } from '@directus/shared/utils';
+
+type Options = {
+	filter: Filter;
+};
+
+export default defineOperationApi<Options>({
+	id: 'condition',
+
+	handler: ({ filter }, { data }) => {
+		const errors = validatePayload(filter, data);
+
+		if (errors.length > 0) {
+			throw errors;
+		} else {
+			return null;
+		}
+	},
+});
--- a/api/src/operations/item-create/index.ts
+++ b/api/src/operations/item-create/index.ts
@@ -0,0 +1,51 @@
+import { Accountability, PrimaryKey } from '@directus/shared/types';
+import { defineOperationApi, toArray } from '@directus/shared/utils';
+import { ItemsService } from '../../services';
+import { Item } from '../../types';
+import { optionToObject } from '../../utils/operation-options';
+import { getAccountabilityForRole } from '../../utils/get-accountability-for-role';
+
+type Options = {
+	collection: string;
+	payload?: Record<string, any> | string | null;
+	emitEvents: boolean;
+	permissions: string; // $public, $trigger, $full, or UUID of a role
+};
+
+export default defineOperationApi<Options>({
+	id: 'item-create',
+
+	handler: async ({ collection, payload, emitEvents, permissions }, { accountability, database, getSchema }) => {
+		const schema = await getSchema({ database });
+
+		let customAccountability: Accountability | null;
+
+		if (!permissions || permissions === '$trigger') {
+			customAccountability = accountability;
+		} else if (permissions === '$full') {
+			customAccountability = null;
+		} else if (permissions === '$public') {
+			customAccountability = await getAccountabilityForRole(null, { database, schema, accountability });
+		} else {
+			customAccountability = await getAccountabilityForRole(permissions, { database, schema, accountability });
+		}
+
+		const itemsService = new ItemsService(collection, {
+			schema: await getSchema({ database }),
+			accountability: customAccountability,
+			knex: database,
+		});
+
+		const payloadObject: Partial<Item> | Partial<Item>[] | null = optionToObject(payload) ?? null;
+
+		let result: PrimaryKey[] | null;
+
+		if (!payloadObject) {
+			result = null;
+		} else {
+			result = await itemsService.createMany(toArray(payloadObject), { emitEvents });
+		}
+
+		return result;
+	},
+});
--- a/api/src/operations/item-delete/index.ts
+++ b/api/src/operations/item-delete/index.ts
@@ -0,0 +1,56 @@
+import { Accountability, PrimaryKey } from '@directus/shared/types';
+import { defineOperationApi, toArray } from '@directus/shared/utils';
+import { ItemsService } from '../../services';
+import { optionToObject } from '../../utils/operation-options';
+import { getAccountabilityForRole } from '../../utils/get-accountability-for-role';
+
+type Options = {
+	collection: string;
+	key?: PrimaryKey | PrimaryKey[] | null;
+	query?: Record<string, any> | string | null;
+	permissions: string; // $public, $trigger, $full, or UUID of a role
+};
+
+export default defineOperationApi<Options>({
+	id: 'item-delete',
+
+	handler: async ({ collection, key, query, permissions }, { accountability, database, getSchema }) => {
+		const schema = await getSchema({ database });
+
+		let customAccountability: Accountability | null;
+
+		if (!permissions || permissions === '$trigger') {
+			customAccountability = accountability;
+		} else if (permissions === '$full') {
+			customAccountability = null;
+		} else if (permissions === '$public') {
+			customAccountability = await getAccountabilityForRole(null, { database, schema, accountability });
+		} else {
+			customAccountability = await getAccountabilityForRole(permissions, { database, schema, accountability });
+		}
+
+		const itemsService: ItemsService = new ItemsService(collection, {
+			schema: await getSchema({ database }),
+			accountability: customAccountability,
+			knex: database,
+		});
+
+		const queryObject = query ? optionToObject(query) : {};
+
+		let result: PrimaryKey | PrimaryKey[] | null;
+
+		if (!key) {
+			result = await itemsService.deleteByQuery(queryObject);
+		} else {
+			const keys = toArray(key);
+
+			if (keys.length === 1) {
+				result = await itemsService.deleteOne(keys[0]);
+			} else {
+				result = await itemsService.deleteMany(keys);
+			}
+		}
+
+		return result;
+	},
+});
--- a/api/src/operations/item-read/index.ts
+++ b/api/src/operations/item-read/index.ts
@@ -0,0 +1,57 @@
+import { Accountability, PrimaryKey } from '@directus/shared/types';
+import { defineOperationApi, toArray } from '@directus/shared/utils';
+import { ItemsService } from '../../services';
+import { Item } from '../../types';
+import { optionToObject } from '../../utils/operation-options';
+import { getAccountabilityForRole } from '../../utils/get-accountability-for-role';
+
+type Options = {
+	collection: string;
+	key?: PrimaryKey | PrimaryKey[] | null;
+	query?: Record<string, any> | string | null;
+	permissions: string; // $public, $trigger, $full, or UUID of a role
+};
+
+export default defineOperationApi<Options>({
+	id: 'item-read',
+
+	handler: async ({ collection, key, query, permissions }, { accountability, database, getSchema }) => {
+		const schema = await getSchema({ database });
+
+		let customAccountability: Accountability | null;
+
+		if (!permissions || permissions === '$trigger') {
+			customAccountability = accountability;
+		} else if (permissions === '$full') {
+			customAccountability = null;
+		} else if (permissions === '$public') {
+			customAccountability = await getAccountabilityForRole(null, { database, schema, accountability });
+		} else {
+			customAccountability = await getAccountabilityForRole(permissions, { database, schema, accountability });
+		}
+
+		const itemsService = new ItemsService(collection, {
+			schema,
+			accountability: customAccountability,
+			knex: database,
+		});
+
+		const queryObject = query ? optionToObject(query) : {};
+
+		let result: Item | Item[] | null;
+
+		if (!key) {
+			result = await itemsService.readByQuery(queryObject);
+		} else {
+			const keys = toArray(key);
+
+			if (keys.length === 1) {
+				result = await itemsService.readOne(keys[0], queryObject);
+			} else {
+				result = await itemsService.readMany(keys, queryObject);
+			}
+		}
+
+		return result;
+	},
+});
--- a/api/src/operations/item-update/index.ts
+++ b/api/src/operations/item-update/index.ts
@@ -0,0 +1,64 @@
+import { Accountability, PrimaryKey } from '@directus/shared/types';
+import { defineOperationApi, toArray } from '@directus/shared/utils';
+import { ItemsService } from '../../services';
+import { Item } from '../../types';
+import { optionToObject } from '../../utils/operation-options';
+import { getAccountabilityForRole } from '../../utils/get-accountability-for-role';
+
+type Options = {
+	collection: string;
+	key?: PrimaryKey | PrimaryKey[] | null;
+	payload?: Record<string, any> | string | null;
+	query?: Record<string, any> | string | null;
+	permissions: string; // $public, $trigger, $full, or UUID of a role
+};
+
+export default defineOperationApi<Options>({
+	id: 'item-update',
+
+	handler: async ({ collection, key, payload, query, permissions }, { accountability, database, getSchema }) => {
+		const schema = await getSchema({ database });
+
+		let customAccountability: Accountability | null;
+
+		if (!permissions || permissions === '$trigger') {
+			customAccountability = accountability;
+		} else if (permissions === '$full') {
+			customAccountability = null;
+		} else if (permissions === '$public') {
+			customAccountability = await getAccountabilityForRole(null, { database, schema, accountability });
+		} else {
+			customAccountability = await getAccountabilityForRole(permissions, { database, schema, accountability });
+		}
+
+		const itemsService: ItemsService = new ItemsService(collection, {
+			schema: await getSchema({ database }),
+			accountability: customAccountability,
+			knex: database,
+		});
+
+		const payloadObject: Partial<Item> | Partial<Item>[] | null = optionToObject(payload) ?? null;
+
+		const queryObject = query ? optionToObject(query) : {};
+
+		if (!payloadObject) {
+			return null;
+		}
+
+		let result: PrimaryKey | PrimaryKey[] | null;
+
+		if (!key) {
+			result = await itemsService.updateByQuery(queryObject, payloadObject);
+		} else {
+			const keys = toArray(key);
+
+			if (keys.length === 1) {
+				result = await itemsService.updateOne(keys[0], payloadObject);
+			} else {
+				result = await itemsService.updateMany(keys, payloadObject);
+			}
+		}
+
+		return result;
+	},
+});
--- a/api/src/operations/log/index.ts
+++ b/api/src/operations/log/index.ts
@@ -0,0 +1,15 @@
+import { defineOperationApi } from '@directus/shared/utils';
+import logger from '../../logger';
+import { optionToString } from '../../utils/operation-options';
+
+type Options = {
+	message: unknown;
+};
+
+export default defineOperationApi<Options>({
+	id: 'log',
+
+	handler: ({ message }) => {
+		logger.info(optionToString(message));
+	},
+});
--- a/api/src/operations/mail/index.ts
+++ b/api/src/operations/mail/index.ts
@@ -0,0 +1,23 @@
+import { defineOperationApi } from '@directus/shared/utils';
+import { MailService } from '../../services';
+import { md } from '../../utils/md';
+
+type Options = {
+	body: string;
+	to: string;
+	subject: string;
+};
+
+export default defineOperationApi<Options>({
+	id: 'mail',
+
+	handler: async ({ body, to, subject }, { accountability, database, getSchema }) => {
+		const mailService = new MailService({ schema: await getSchema({ database }), accountability, knex: database });
+
+		await mailService.send({
+			html: md(body),
+			to,
+			subject,
+		});
+	},
+});
--- a/api/src/operations/notification/index.ts
+++ b/api/src/operations/notification/index.ts
@@ -0,0 +1,49 @@
+import { Accountability } from '@directus/shared/types';
+import { defineOperationApi } from '@directus/shared/utils';
+import { NotificationsService } from '../../services';
+import { optionToString } from '../../utils/operation-options';
+import { getAccountabilityForRole } from '../../utils/get-accountability-for-role';
+
+type Options = {
+	recipient: string;
+	subject: string;
+	message?: unknown | null;
+	permissions: string; // $public, $trigger, $full, or UUID of a role
+};
+
+export default defineOperationApi<Options>({
+	id: 'notification',
+
+	handler: async ({ recipient, subject, message, permissions }, { accountability, database, getSchema }) => {
+		const schema = await getSchema({ database });
+
+		let customAccountability: Accountability | null;
+
+		if (!permissions || permissions === '$trigger') {
+			customAccountability = accountability;
+		} else if (permissions === '$full') {
+			customAccountability = null;
+		} else if (permissions === '$public') {
+			customAccountability = await getAccountabilityForRole(null, { database, schema, accountability });
+		} else {
+			customAccountability = await getAccountabilityForRole(permissions, { database, schema, accountability });
+		}
+
+		const notificationsService = new NotificationsService({
+			schema: await getSchema({ database }),
+			accountability: customAccountability,
+			knex: database,
+		});
+
+		const messageString = message ? optionToString(message) : null;
+
+		const result = await notificationsService.createOne({
+			recipient,
+			sender: customAccountability?.user ?? null,
+			subject,
+			message: messageString,
+		});
+
+		return result;
+	},
+});
--- a/api/src/operations/request/index.ts
+++ b/api/src/operations/request/index.ts
@@ -0,0 +1,19 @@
+import { defineOperationApi } from '@directus/shared/utils';
+import axios, { Method } from 'axios';
+
+type Options = {
+	url: string;
+	method: Method;
+	body: Record<string, any> | string | null;
+	headers: Record<string, string>;
+};
+
+export default defineOperationApi<Options>({
+	id: 'request',
+
+	handler: async ({ url, method, body, headers }) => {
+		const result = await axios({ url, method, data: body, headers });
+
+		return { status: result.status, statusText: result.statusText, headers: result.headers, data: result.data };
+	},
+});
--- a/api/src/operations/sleep/index.ts
+++ b/api/src/operations/sleep/index.ts
@@ -0,0 +1,13 @@
+import { defineOperationApi } from '@directus/shared/utils';
+
+type Options = {
+	milliseconds: string | number;
+};
+
+export default defineOperationApi<Options>({
+	id: 'sleep',
+
+	handler: async ({ milliseconds }) => {
+		await new Promise((resolve) => setTimeout(resolve, Number(milliseconds)));
+	},
+});
--- a/api/src/operations/transform/index.ts
+++ b/api/src/operations/transform/index.ts
@@ -0,0 +1,14 @@
+import { defineOperationApi } from '@directus/shared/utils';
+import { parseJSON } from '../../utils/parse-json';
+
+type Options = {
+	json: string;
+};
+
+export default defineOperationApi<Options>({
+	id: 'transform',
+
+	handler: ({ json }) => {
+		return parseJSON(json);
+	},
+});
--- a/api/src/operations/trigger/index.ts
+++ b/api/src/operations/trigger/index.ts
@@ -0,0 +1,28 @@
+import { defineOperationApi } from '@directus/shared/utils';
+import { getFlowManager } from '../../flows';
+import { optionToObject } from '../../utils/operation-options';
+
+type Options = {
+	flow: string;
+	payload?: Record<string, any> | Record<string, any>[] | string | null;
+};
+
+export default defineOperationApi<Options>({
+	id: 'trigger',
+
+	handler: async ({ flow, payload }, context) => {
+		const flowManager = getFlowManager();
+
+		const payloadObject = optionToObject(payload) ?? null;
+
+		let result: unknown | unknown[];
+
+		if (Array.isArray(payloadObject)) {
+			result = await Promise.all(payloadObject.map((payload) => flowManager.runOperationFlow(flow, payload, context)));
+		} else {
+			result = await flowManager.runOperationFlow(flow, payloadObject, context);
+		}
+
+		return result;
+	},
+});
--- a/api/src/services/activity.ts
+++ b/api/src/services/activity.ts
@@ -1,18 +1,17 @@
-import { AbstractServiceOptions, PrimaryKey, Item, Action } from '../types';
-import { ItemsService } from './items';
-import { MutationOptions } from '../types';
-import { NotificationsService } from './notifications';
-import { UsersService } from './users';
-import { AuthorizationService } from './authorization';
-import { Accountability } from '@directus/shared/types';
-import { getPermissions } from '../utils/get-permissions';
+import { Accountability, Action } from '@directus/shared/types';
+import { uniq } from 'lodash';
+import validateUUID from 'uuid-validate';
+import env from '../env';
 import { ForbiddenException } from '../exceptions/forbidden';
 import logger from '../logger';
-import { userName } from '../utils/user-name';
-import { uniq } from 'lodash';
-import env from '../env';
-import validateUUID from 'uuid-validate';
+import { AbstractServiceOptions, Item, MutationOptions, PrimaryKey } from '../types';
+import { getPermissions } from '../utils/get-permissions';
 import { Url } from '../utils/url';
+import { userName } from '../utils/user-name';
+import { AuthorizationService } from './authorization';
+import { ItemsService } from './items';
+import { NotificationsService } from './notifications';
+import { UsersService } from './users';

 export class ActivityService extends ItemsService {
 	notificationsService: NotificationsService;
--- a/api/src/services/authentication.ts
+++ b/api/src/services/authentication.ts
@@ -1,27 +1,27 @@
+import { Accountability, Action, SchemaOverview } from '@directus/shared/types';
 import jwt from 'jsonwebtoken';
 import { Knex } from 'knex';
+import { clone, cloneDeep } from 'lodash';
 import ms from 'ms';
 import { nanoid } from 'nanoid';
+import { performance } from 'perf_hooks';
+import { getAuthProvider } from '../auth';
+import { DEFAULT_AUTH_PROVIDER } from '../constants';
 import getDatabase from '../database';
 import emitter from '../emitter';
 import env from '../env';
-import { getAuthProvider } from '../auth';
-import { DEFAULT_AUTH_PROVIDER } from '../constants';
 import {
 	InvalidCredentialsException,
 	InvalidOTPException,
-	UserSuspendedException,
 	InvalidProviderException,
+	UserSuspendedException,
 } from '../exceptions';
 import { createRateLimiter } from '../rate-limiter';
-import { ActivityService } from './activity';
-import { TFAService } from './tfa';
-import { AbstractServiceOptions, Action, Session, User, DirectusTokenPayload, LoginResult } from '../types';
-import { Accountability, SchemaOverview } from '@directus/shared/types';
-import { SettingsService } from './settings';
-import { clone, cloneDeep } from 'lodash';
-import { performance } from 'perf_hooks';
+import { AbstractServiceOptions, DirectusTokenPayload, LoginResult, Session, User } from '../types';
 import { stall } from '../utils/stall';
+import { ActivityService } from './activity';
+import { SettingsService } from './settings';
+import { TFAService } from './tfa';

 const loginAttemptsLimiter = createRateLimiter({ duration: 0 });

--- a/api/src/services/flows.ts
+++ b/api/src/services/flows.ts
@@ -0,0 +1,49 @@
+import { FlowRaw } from '@directus/shared/types';
+import { getMessenger, Messenger } from '../messenger';
+import { AbstractServiceOptions, Item, MutationOptions, PrimaryKey } from '../types';
+import { ItemsService } from './items';
+
+export class FlowsService extends ItemsService<FlowRaw> {
+	messenger: Messenger;
+
+	constructor(options: AbstractServiceOptions) {
+		super('directus_flows', options);
+		this.messenger = getMessenger();
+	}
+
+	async createOne(data: Partial<Item>, opts?: MutationOptions): Promise<PrimaryKey> {
+		const result = await super.createOne(data, opts);
+		this.messenger.publish('flows', { type: 'reload' });
+		return result;
+	}
+
+	async createMany(data: Partial<Item>[], opts?: MutationOptions): Promise<PrimaryKey[]> {
+		const result = await super.createMany(data, opts);
+		this.messenger.publish('flows', { type: 'reload' });
+		return result;
+	}
+
+	async updateOne(key: PrimaryKey, data: Partial<Item>, opts?: MutationOptions): Promise<PrimaryKey> {
+		const result = await super.updateOne(key, data, opts);
+		this.messenger.publish('flows', { type: 'reload' });
+		return result;
+	}
+
+	async updateMany(keys: PrimaryKey[], data: Partial<Item>, opts?: MutationOptions): Promise<PrimaryKey[]> {
+		const result = await super.updateMany(keys, data, opts);
+		this.messenger.publish('flows', { type: 'reload' });
+		return result;
+	}
+
+	async deleteOne(key: PrimaryKey, opts?: MutationOptions): Promise<PrimaryKey> {
+		const result = await super.deleteOne(key, opts);
+		this.messenger.publish('flows', { type: 'reload' });
+		return result;
+	}
+
+	async deleteMany(keys: PrimaryKey[], opts?: MutationOptions): Promise<PrimaryKey[]> {
+		const result = await super.deleteMany(keys, opts);
+		this.messenger.publish('flows', { type: 'reload' });
+		return result;
+	}
+}
--- a/api/src/services/graphql.ts
+++ b/api/src/services/graphql.ts
@@ -1,5 +1,5 @@
 import { BaseException } from '@directus/shared/exceptions';
-import { Accountability, Aggregate, Filter, Query, SchemaOverview } from '@directus/shared/types';
+import { Accountability, Action, Aggregate, Filter, Query, SchemaOverview } from '@directus/shared/types';
 import argon2 from 'argon2';
 import {
 	ArgumentNode,
@@ -52,7 +52,7 @@ import getDatabase from '../database';
 import env from '../env';
 import { ForbiddenException, GraphQLValidationException, InvalidPayloadException } from '../exceptions';
 import { getExtensionManager } from '../extensions';
-import { AbstractServiceOptions, Action, GraphQLParams, Item } from '../types';
+import { AbstractServiceOptions, GraphQLParams, Item } from '../types';
 import { generateHash } from '../utils/generate-hash';
 import { getGraphQLType } from '../utils/get-graphql-type';
 import { reduceSchema } from '../utils/reduce-schema';
@@ -63,9 +63,11 @@ import { AuthenticationService } from './authentication';
 import { CollectionsService } from './collections';
 import { FieldsService } from './fields';
 import { FilesService } from './files';
+import { FlowsService } from './flows';
 import { FoldersService } from './folders';
 import { ItemsService } from './items';
 import { NotificationsService } from './notifications';
+import { OperationsService } from './operations';
 import { PermissionsService } from './permissions';
 import { PresetsService } from './presets';
 import { RelationsService } from './relations';
@@ -1635,6 +1637,10 @@ export class GraphQLService {
 				return new WebhooksService(opts);
 			case 'directus_shares':
 				return new SharesService(opts);
+			case 'directus_flows':
+				return new FlowsService(opts);
+			case 'directus_operations':
+				return new OperationsService(opts);
 			default:
 				return new ItemsService(collection, opts);
 		}
--- a/api/src/services/index.ts
+++ b/api/src/services/index.ts
@@ -8,12 +8,14 @@ export * from './collections';
 export * from './dashboards';
 export * from './fields';
 export * from './files';
+export * from './flows';
 export * from './folders';
 export * from './graphql';
 export * from './import-export';
 export * from './mail';
 export * from './meta';
 export * from './notifications';
+export * from './operations';
 export * from './panels';
 export * from './payload';
 export * from './permissions';
--- a/api/src/services/items.ts
+++ b/api/src/services/items.ts
@@ -1,4 +1,4 @@
-import { Accountability, PermissionsAction, Query, SchemaOverview } from '@directus/shared/types';
+import { Accountability, Action, PermissionsAction, Query, SchemaOverview } from '@directus/shared/types';
 import Keyv from 'keyv';
 import { Knex } from 'knex';
 import { assign, clone, cloneDeep, pick, without } from 'lodash';
@@ -9,14 +9,7 @@ import emitter from '../emitter';
 import env from '../env';
 import { ForbiddenException } from '../exceptions';
 import { translateDatabaseError } from '../exceptions/database/translate';
-import {
-	AbstractService,
-	AbstractServiceOptions,
-	Action,
-	Item as AnyItem,
-	MutationOptions,
-	PrimaryKey,
-} from '../types';
+import { AbstractService, AbstractServiceOptions, Item as AnyItem, MutationOptions, PrimaryKey } from '../types';
 import getASTFromQuery from '../utils/get-ast-from-query';
 import { AuthorizationService } from './authorization';
 import { ActivityService, RevisionsService } from './index';
--- a/api/src/services/notifications.ts
+++ b/api/src/services/notifications.ts
@@ -1,8 +1,9 @@
-import { UsersService, MailService } from '.';
 import { AbstractServiceOptions, PrimaryKey, MutationOptions } from '../types';
 import { ItemsService } from './items';
 import { Notification } from '@directus/shared/types';
 import { md } from '../utils/md';
+import { UsersService } from './users';
+import { MailService } from './mail';

 export class NotificationsService extends ItemsService {
 	usersService: UsersService;
--- a/api/src/services/operations.ts
+++ b/api/src/services/operations.ts
@@ -0,0 +1,49 @@
+import { OperationRaw } from '@directus/shared/types';
+import { getMessenger, Messenger } from '../messenger';
+import { AbstractServiceOptions, Item, MutationOptions, PrimaryKey } from '../types';
+import { ItemsService } from './items';
+
+export class OperationsService extends ItemsService<OperationRaw> {
+	messenger: Messenger;
+
+	constructor(options: AbstractServiceOptions) {
+		super('directus_operations', options);
+		this.messenger = getMessenger();
+	}
+
+	async createOne(data: Partial<Item>, opts?: MutationOptions): Promise<PrimaryKey> {
+		const result = await super.createOne(data, opts);
+		this.messenger.publish('flows', { type: 'reload' });
+		return result;
+	}
+
+	async createMany(data: Partial<Item>[], opts?: MutationOptions): Promise<PrimaryKey[]> {
+		const result = await super.createMany(data, opts);
+		this.messenger.publish('flows', { type: 'reload' });
+		return result;
+	}
+
+	async updateOne(key: PrimaryKey, data: Partial<Item>, opts?: MutationOptions): Promise<PrimaryKey> {
+		const result = await super.updateOne(key, data, opts);
+		this.messenger.publish('flows', { type: 'reload' });
+		return result;
+	}
+
+	async updateMany(keys: PrimaryKey[], data: Partial<Item>, opts?: MutationOptions): Promise<PrimaryKey[]> {
+		const result = await super.updateMany(keys, data, opts);
+		this.messenger.publish('flows', { type: 'reload' });
+		return result;
+	}
+
+	async deleteOne(key: PrimaryKey, opts?: MutationOptions): Promise<PrimaryKey> {
+		const result = await super.deleteOne(key, opts);
+		this.messenger.publish('flows', { type: 'reload' });
+		return result;
+	}
+
+	async deleteMany(keys: PrimaryKey[], opts?: MutationOptions): Promise<PrimaryKey[]> {
+		const result = await super.deleteMany(keys, opts);
+		this.messenger.publish('flows', { type: 'reload' });
+		return result;
+	}
+}
--- a/api/src/types/activity.ts
+++ b/api/src/types/activity.ts
@@ -1,9 +0,0 @@
-export enum Action {
-	CREATE = 'create',
-	UPDATE = 'update',
-	DELETE = 'delete',
-	REVERT = 'revert',
-	COMMENT = 'comment',
-	UPLOAD = 'upload',
-	LOGIN = 'login',
-}
--- a/api/src/types/events.ts
+++ b/api/src/types/events.ts
@@ -0,0 +1,8 @@
+import { ActionHandler, FilterHandler, InitHandler } from '@directus/shared/types';
+import { ScheduledTask } from 'node-cron';
+
+export type EventHandler =
+	| { type: 'filter'; name: string; handler: FilterHandler }
+	| { type: 'action'; name: string; handler: ActionHandler }
+	| { type: 'init'; name: string; handler: InitHandler }
+	| { type: 'schedule'; task: ScheduledTask };
--- a/api/src/types/express.d.ts
+++ b/api/src/types/express.d.ts
@@ -2,8 +2,7 @@
 * Custom properties on the req object in express
 */

-import { Accountability } from '@directus/shared/types';
-import { Query } from '@directus/shared/types';
+import { Accountability, Query } from '@directus/shared/types';
 import { SchemaOverview } from './schema';

 export {};
--- a/api/src/types/index.ts
+++ b/api/src/types/index.ts
@@ -1,8 +1,8 @@
-export * from './activity';
 export * from './assets';
 export * from './ast';
 export * from './auth';
 export * from './collection';
+export * from './events';
 export * from './files';
 export * from './graphql';
 export * from './items';
--- a/api/src/utils/construct-flow-tree.ts
+++ b/api/src/utils/construct-flow-tree.ts
@@ -0,0 +1,36 @@
+import { Flow, FlowRaw, Operation, OperationRaw } from '@directus/shared/types';
+import { omit } from 'lodash';
+
+export function constructFlowTree(flow: FlowRaw): Flow {
+	const rootOperation = flow.operations.find((operation) => operation.id === flow.operation) ?? null;
+
+	const operationTree = constructOperationTree(rootOperation, flow.operations);
+
+	const flowTree: Flow = {
+		...omit(flow, 'operations'),
+		operation: operationTree,
+	};
+
+	return flowTree;
+}
+
+function constructOperationTree(root: OperationRaw | null, operations: OperationRaw[]): Operation | null {
+	if (root === null) {
+		return null;
+	}
+
+	const resolveOperation = root.resolve !== null ? operations.find((operation) => operation.id === root.resolve) : null;
+	const rejectOperation = root.reject !== null ? operations.find((operation) => operation.id === root.reject) : null;
+
+	if (resolveOperation === undefined || rejectOperation === undefined) {
+		throw new Error('Undefined reference in operations');
+	}
+
+	const operationTree: Operation = {
+		...omit(root, 'flow'),
+		resolve: constructOperationTree(resolveOperation, operations),
+		reject: constructOperationTree(rejectOperation, operations),
+	};
+
+	return operationTree;
+}
--- a/api/src/utils/get-accountability-for-role.ts
+++ b/api/src/utils/get-accountability-for-role.ts
@@ -0,0 +1,47 @@
+import { Accountability, SchemaOverview } from '@directus/shared/types';
+import { getPermissions } from './get-permissions';
+import { InvalidConfigException } from '../exceptions';
+import { Knex } from 'knex';
+
+export async function getAccountabilityForRole(
+	role: null | string,
+	context: {
+		accountability: null | Accountability;
+		schema: SchemaOverview;
+		database: Knex;
+	}
+): Promise<Accountability> {
+	let generatedAccountability: Accountability | null = context.accountability;
+
+	if (role === null) {
+		generatedAccountability = {
+			role: null,
+			user: null,
+			admin: false,
+			app: false,
+		};
+
+		generatedAccountability.permissions = await getPermissions(generatedAccountability, context.schema);
+	} else {
+		const roleInfo = await context.database
+			.select(['app_access', 'admin_access'])
+			.from('directus_roles')
+			.where({ id: role })
+			.first();
+
+		if (!roleInfo) {
+			throw new InvalidConfigException(`Configured role "${role}" isn't a valid role ID or doesn't exist.`);
+		}
+
+		generatedAccountability = {
+			role,
+			user: null,
+			admin: roleInfo.admin_access === 1 || roleInfo.admin_access === '1' || roleInfo.admin_access === true,
+			app: roleInfo.app_access === 1 || roleInfo.app_access === '1' || roleInfo.app_access === true,
+		};
+
+		generatedAccountability.permissions = await getPermissions(generatedAccountability, context.schema);
+	}
+
+	return generatedAccountability;
+}
--- a/api/src/utils/operation-options.ts
+++ b/api/src/utils/operation-options.ts
@@ -0,0 +1,51 @@
+import { renderFn, get, Scope } from 'micromustache';
+import { parseJSON } from './parse-json';
+
+type Mustacheable = string | number | boolean | null | Mustacheable[] | { [key: string]: Mustacheable };
+type GenericString<T> = T extends string ? string : T;
+
+function resolveFn(path: string, scope?: Scope): unknown {
+	if (!scope) return undefined;
+
+	const value = get(scope, path);
+
+	return typeof value === 'object' ? JSON.stringify(value) : value;
+}
+
+function renderMustache<T extends Mustacheable>(item: T, scope: Scope): GenericString<T> {
+	if (typeof item === 'string') {
+		return renderFn(item, resolveFn, scope, { explicit: true }) as GenericString<T>;
+	} else if (Array.isArray(item)) {
+		return item.map((element) => renderMustache(element, scope)) as GenericString<T>;
+	} else if (typeof item === 'object' && item !== null) {
+		return Object.fromEntries(
+			Object.entries(item).map(([key, value]) => [key, renderMustache(value, scope)])
+		) as GenericString<T>;
+	} else {
+		return item as GenericString<T>;
+	}
+}
+
+export function applyOperationOptions(options: Record<string, any>, data: Record<string, any>): Record<string, any> {
+	return Object.fromEntries(
+		Object.entries(options).map(([key, value]) => {
+			if (typeof value === 'string') {
+				const single = value.match(/^\{\{\s*([^}\s]+)\s*\}\}$/);
+
+				if (single !== null) {
+					return [key, get(data, single[1])];
+				}
+			}
+
+			return [key, renderMustache(value, data)];
+		})
+	);
+}
+
+export function optionToObject<T>(option: T): Exclude<T, string> {
+	return typeof option === 'string' ? parseJSON(option) : option;
+}
+
+export function optionToString(option: unknown): string {
+	return typeof option === 'object' ? JSON.stringify(option) : String(option);
+}