directus

1
import { parseJSON, toArray } from '@directus/utils';
2
import type { Knex } from 'knex';
3
import { randomUUID } from 'node:crypto';
4
import type { Webhook } from '../../types/webhooks.js';
5

6
// To avoid typos
7
const TABLE_WEBHOOKS = 'directus_webhooks';
8
const TABLE_FLOWS = 'directus_flows';
9
const TABLE_OPERATIONS = 'directus_operations';
10
const NEW_COLUMN_WAS_ACTIVE = 'was_active_before_deprecation';
11
const NEW_COLUMN_FLOW = 'migrated_flow';
12

13
/**
14
 * 0. Identify and persist which webhooks were active before deprecation
15
 * 1. Migrate existing webhooks over to identically behaving Flows
16
 * 2. Disable existing webhooks
17
 * 3. Dont drop webhooks yet
18
 */
19
export async function up(knex: Knex): Promise<void> {
20
	// 0. Identify and persist which webhooks were active before deprecation
21
	await knex.schema.alterTable(TABLE_WEBHOOKS, (table) => {
22
		table.boolean(NEW_COLUMN_WAS_ACTIVE).notNullable().defaultTo(false);
23
		table.uuid(NEW_COLUMN_FLOW).nullable();
24
		table.foreign(NEW_COLUMN_FLOW).references(`${TABLE_FLOWS}.id`).onDelete('SET NULL');
25
	});
26

27
	await knex(TABLE_WEBHOOKS)
28
		.update({ [NEW_COLUMN_WAS_ACTIVE]: true })
29
		.where({ status: 'active' });
30

31
	const webhooks: Webhook[] = await knex.select('*').from(TABLE_WEBHOOKS);
32

33
	// 1. Migrate existing webhooks over to identically behaving Flows
34
	await knex.transaction(async (trx) => {
35
		for (const webhook of webhooks) {
36
			const flowId = randomUUID();
37
			const operationIdRunScript = randomUUID();
38
			const operationIdWebhook = randomUUID();
39

40
			const newFlow = {
41
				id: flowId,
42
				name: webhook.name,
43
				icon: 'webhook',
44
				// color: null,
45
				description:
46
					'Auto-generated by Directus Migration\n\nWebhooks were deprecated and have been moved into Flows for you automatically.',
47
				status: webhook.status, // Only activate already active webhooks!
48
				trigger: 'event',
49
				// accountability: "all",
50
				options: {
51
					type: 'action',
52
					scope: toArray(webhook.actions)
53
						.filter((action) => action.trim() !== '')
54
						.map((scope) => `items.${scope}`),
55
					collections: toArray(webhook.collections).filter((collection) => collection.trim() !== ''),
56
				},
57
				operation: null, // Fill this in later --> `operationIdRunScript`
58
				// date_created: Default Date,
59
				// user_created: null,
60
			};
61

62
			const operationRunScript = {
63
				id: operationIdRunScript,
64
				name: 'Transforming Payload for Webhook',
65
				key: 'payload-transform',
66
				type: 'exec',
67
				position_x: 19,
68
				position_y: 1,
69
				options: {
70
					code: '/**\n * Webhook data slightly differs from Flow trigger data.\n * This flow converts the new Flow format into the older Webhook shape\n * in order to not break existing logic of consumers.\n */ \nmodule.exports = async function(data) {\n    const crudOperation = data.$trigger.event.split(".").at(-1);\n    const keyOrKeys = crudOperation === "create" ? "key" : "keys";\n    return {\n        event: `items.${crudOperation}`,\n        accountability: { user: data.$accountability.user, role: data.$accountability.role },\n        payload: data.$trigger.payload,\n        [keyOrKeys]: data.$trigger?.[keyOrKeys],\n        collection: data.$trigger.collection,\n    };\n}',
71
				},
72
				resolve: operationIdWebhook,
73
				reject: null,
74
				flow: flowId,
75
				// date_created: "",
76
				// user_created: "",
77
			};
78

79
			const operationWebhook = {
80
				id: operationIdWebhook,
81
				name: 'Webhook',
82
				key: 'webhook',
83
				type: 'request',
84
				position_x: 37,
85
				position_y: 1,
86
				options: {
87
					url: webhook.url,
88
					body: webhook.data ? '{{$last}}' : undefined,
89
					method: webhook.method,
90
					headers: typeof webhook.headers === 'string' ? parseJSON(webhook.headers) : webhook.headers,
91
				},
92
				resolve: null,
93
				reject: null,
94
				flow: flowId,
95
				// date_created: "",
96
				// user_created: "",
97
			};
98

99
			await trx(TABLE_FLOWS).insert(newFlow);
100

101
			// Only need to transform the payload if the webhook enabled transmitting it
102
			if (webhook.data && webhook.method !== 'GET') {
103
				// Order is important due to IDs
104
				await trx(TABLE_OPERATIONS).insert(operationWebhook);
105
				await trx(TABLE_OPERATIONS).insert(operationRunScript);
106
				await trx(TABLE_FLOWS).update({ operation: operationIdRunScript }).where({ id: flowId });
107
			} else {
108
				operationWebhook.position_x = 19; // Reset it because the Run-Script will be missing
109
				await trx(TABLE_OPERATIONS).insert(operationWebhook);
110
				await trx(TABLE_FLOWS).update({ operation: operationIdWebhook }).where({ id: flowId });
111
			}
112

113
			// Persist new Flow/s so that we can retroactively remove them on potential down-migrations
114
			await trx(TABLE_WEBHOOKS)
115
				.update({ [NEW_COLUMN_FLOW]: flowId })
116
				.where({ id: webhook.id });
117

118
			// 2. Disable existing webhooks
119
			await trx(TABLE_WEBHOOKS).update({ status: 'inactive' });
120
		}
121
	});
122
}
123

124
/**
125
 * Dont completely drop Webhooks yet.
126
 * Lets preserve the data and drop them in the next release to be extra safe with users data.
127
 */
128
export async function down(knex: Knex): Promise<void> {
129
	// Remove Flows created by the up-migration
130
	const migratedFlowIds = (await knex(TABLE_WEBHOOKS).select(NEW_COLUMN_FLOW).whereNotNull(NEW_COLUMN_FLOW)).map(
131
		(col) => col[NEW_COLUMN_FLOW],
132
	);
133

134
	await knex(TABLE_FLOWS).delete().whereIn('id', migratedFlowIds);
135

136
	// Restore previously activated webhooks
137
	await knex(TABLE_WEBHOOKS)
138
		.update({ status: 'active' })
139
		.where({ [NEW_COLUMN_WAS_ACTIVE]: true });
140

141
	// Cleanup of up-migration
142
	await knex.schema.alterTable(TABLE_WEBHOOKS, (table) => {
143
		table.dropColumns(NEW_COLUMN_WAS_ACTIVE, NEW_COLUMN_FLOW);
144
	});
145
}
146