mudhorn/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2024-11-10 06:34:05 -08:00
Code Issues Actions Packages Projects Releases Wiki Activity

⚡ Run migration in chunks (#2393)

Browse source
This commit is contained in:
Ben Hesseldieck 2021-11-03 16:12:48 +01:00 committed by GitHub
parent 4f9aee14b5
commit 0877f485d9
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 560 additions and 247 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

39
packages/cli/src/databases/MigrationHelpers.ts Normal file
View file

@ -0,0 +1,39 @@
import { QueryRunner } from 'typeorm';
export class MigrationHelpers {
queryRunner: QueryRunner;
constructor(queryRunner: QueryRunner) {
this.queryRunner = queryRunner;
}
// runs an operation sequential on chunks of a query that returns a potentially large Array.
/* eslint-disable no-await-in-loop */
async runChunked(
query: string,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
operation: (results: any[]) => Promise<void>,
limit = 100,
): Promise<void> {
let offset = 0;
let chunkedQuery: string;
let chunkedQueryResults: unknown[];
do {
chunkedQuery = this.chunkQuery(query, limit, offset);
chunkedQueryResults = (await this.queryRunner.query(chunkedQuery)) as unknown[];
// pass a copy to prevent errors from mutation
await operation([...chunkedQueryResults]);
offset += limit;
} while (chunkedQueryResults.length === limit);
}
/* eslint-enable no-await-in-loop */
private chunkQuery(query: string, limit: number, offset = 0): string {
return `
${query}
LIMIT ${limit}
OFFSET ${offset}
`;
}
}

159
packages/cli/src/databases/mysqldb/migrations/1630451444017-UpdateWorkflowCredentials.ts
View file

@ -1,5 +1,6 @@
import { MigrationInterface, QueryRunner } from 'typeorm';
import config = require('../../../../config');
import { MigrationHelpers } from '../../MigrationHelpers';
// replacing the credentials in workflows and execution
// `nodeType: name` changes to `nodeType: { id, name }`
@ -8,18 +9,22 @@ export class UpdateWorkflowCredentials1630451444017 implements MigrationInterfac
name = 'UpdateWorkflowCredentials1630451444017';
public async up(queryRunner: QueryRunner): Promise<void> {
console.log('Start migration', this.name);
console.time(this.name);
const tablePrefix = config.get('database.tablePrefix');
const helpers = new MigrationHelpers(queryRunner);
const credentialsEntities = await queryRunner.query(`
SELECT id, name, type
FROM ${tablePrefix}credentials_entity
`);
const workflows = await queryRunner.query(`
const workflowsQuery = `
SELECT id, nodes
FROM ${tablePrefix}workflow_entity
`);
`;
// @ts-ignore
await helpers.runChunked(workflowsQuery, (workflows) => {
workflows.forEach(async (workflow) => {
const nodes = workflow.nodes;
let credentialsUpdated = false;
@ -29,7 +34,6 @@ export class UpdateWorkflowCredentials1630451444017 implements MigrationInterfac
const allNodeCredentials = Object.entries(node.credentials);
for (const [type, name] of allNodeCredentials) {
if (typeof name === 'string') {
// @ts-ignore
const matchingCredentials = credentialsEntities.find(
// @ts-ignore
(credentials) => credentials.name === name && credentials.type === type,
@ -41,7 +45,8 @@ export class UpdateWorkflowCredentials1630451444017 implements MigrationInterfac
}
});
if (credentialsUpdated) {
const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters(
const [updateQuery, updateParams] =
queryRunner.connection.driver.escapeQueryWithParameters(
`
UPDATE ${tablePrefix}workflow_entity
SET nodes = :nodes
@ -51,25 +56,19 @@ export class UpdateWorkflowCredentials1630451444017 implements MigrationInterfac
{},
);
await queryRunner.query(updateQuery, updateParams);
queryRunner.query(updateQuery, updateParams);
}
});
});
const waitingExecutions = await queryRunner.query(`
const waitingExecutionsQuery = `
SELECT id, workflowData
FROM ${tablePrefix}execution_entity
WHERE waitTill IS NOT NULL AND finished = 0
`);
const retryableExecutions = await queryRunner.query(`
SELECT id, workflowData
FROM ${tablePrefix}execution_entity
WHERE waitTill IS NULL AND finished = 0 AND mode != 'retry'
ORDER BY startedAt DESC
LIMIT 200
`);
[...waitingExecutions, ...retryableExecutions].forEach(async (execution) => {
`;
// @ts-ignore
await helpers.runChunked(waitingExecutionsQuery, (waitingExecutions) => {
waitingExecutions.forEach(async (execution) => {
const data = execution.workflowData;
let credentialsUpdated = false;
// @ts-ignore
@ -78,7 +77,50 @@ export class UpdateWorkflowCredentials1630451444017 implements MigrationInterfac
const allNodeCredentials = Object.entries(node.credentials);
for (const [type, name] of allNodeCredentials) {
if (typeof name === 'string') {
const matchingCredentials = credentialsEntities.find(
// @ts-ignore
(credentials) => credentials.name === name && credentials.type === type,
);
node.credentials[type] = { id: matchingCredentials?.id.toString() || null, name };
credentialsUpdated = true;
}
}
}
});
if (credentialsUpdated) {
const [updateQuery, updateParams] =
queryRunner.connection.driver.escapeQueryWithParameters(
`
UPDATE ${tablePrefix}execution_entity
SET workflowData = :data
WHERE id = '${execution.id}'
`,
{ data: JSON.stringify(data) },
{},
);
queryRunner.query(updateQuery, updateParams);
}
});
});
const retryableExecutions = await queryRunner.query(`
SELECT id, workflowData
FROM ${tablePrefix}execution_entity
WHERE waitTill IS NULL AND finished = 0 AND mode != 'retry'
ORDER BY startedAt DESC
LIMIT 200
`);
// @ts-ignore
retryableExecutions.forEach(async (execution) => {
const data = execution.workflowData;
let credentialsUpdated = false;
// @ts-ignore
data.nodes.forEach((node) => {
if (node.credentials) {
const allNodeCredentials = Object.entries(node.credentials);
for (const [type, name] of allNodeCredentials) {
if (typeof name === 'string') {
const matchingCredentials = credentialsEntities.find(
// @ts-ignore
(credentials) => credentials.name === name && credentials.type === type,
@ -100,29 +142,80 @@ export class UpdateWorkflowCredentials1630451444017 implements MigrationInterfac
{},
);
await queryRunner.query(updateQuery, updateParams);
queryRunner.query(updateQuery, updateParams);
}
});
console.timeEnd(this.name);
}
public async down(queryRunner: QueryRunner): Promise<void> {
const tablePrefix = config.get('database.tablePrefix');
const helpers = new MigrationHelpers(queryRunner);
const credentialsEntities = await queryRunner.query(`
SELECT id, name, type
FROM ${tablePrefix}credentials_entity
`);
const workflows = await queryRunner.query(`
const workflowsQuery = `
SELECT id, nodes
FROM ${tablePrefix}workflow_entity
`);
`;
// @ts-ignore
await helpers.runChunked(workflowsQuery, (workflows) => {
workflows.forEach(async (workflow) => {
const nodes = workflow.nodes;
let credentialsUpdated = false;
// @ts-ignore
nodes.forEach((node) => {
if (node.credentials) {
const allNodeCredentials = Object.entries(node.credentials);
for (const [type, creds] of allNodeCredentials) {
if (typeof creds === 'object') {
const matchingCredentials = credentialsEntities.find(
// @ts-ignore
(credentials) => credentials.id === creds.id && credentials.type === type,
);
if (matchingCredentials) {
node.credentials[type] = matchingCredentials.name;
} else {
// @ts-ignore
node.credentials[type] = creds.name;
}
credentialsUpdated = true;
}
}
}
});
if (credentialsUpdated) {
const [updateQuery, updateParams] =
queryRunner.connection.driver.escapeQueryWithParameters(
`
UPDATE ${tablePrefix}workflow_entity
SET nodes = :nodes
WHERE id = '${workflow.id}'
`,
{ nodes: JSON.stringify(nodes) },
{},
);
queryRunner.query(updateQuery, updateParams);
}
});
});
const waitingExecutionsQuery = `
SELECT id, workflowData
FROM ${tablePrefix}execution_entity
WHERE waitTill IS NOT NULL AND finished = 0
`;
// @ts-ignore
await helpers.runChunked(waitingExecutionsQuery, (waitingExecutions) => {
waitingExecutions.forEach(async (execution) => {
const data = execution.workflowData;
let credentialsUpdated = false;
// @ts-ignore
data.nodes.forEach((node) => {
if (node.credentials) {
const allNodeCredentials = Object.entries(node.credentials);
for (const [type, creds] of allNodeCredentials) {
@ -144,25 +237,21 @@ export class UpdateWorkflowCredentials1630451444017 implements MigrationInterfac
}
});
if (credentialsUpdated) {
const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters(
const [updateQuery, updateParams] =
queryRunner.connection.driver.escapeQueryWithParameters(
`
UPDATE ${tablePrefix}workflow_entity
SET nodes = :nodes
WHERE id = '${workflow.id}'
UPDATE ${tablePrefix}execution_entity
SET workflowData = :data
WHERE id = '${execution.id}'
`,
{ nodes: JSON.stringify(nodes) },
{ data: JSON.stringify(data) },
{},
);
await queryRunner.query(updateQuery, updateParams);
queryRunner.query(updateQuery, updateParams);
}
});
const waitingExecutions = await queryRunner.query(`
SELECT id, workflowData
FROM ${tablePrefix}execution_entity
WHERE waitTill IS NOT NULL AND finished = 0
`);
});
const retryableExecutions = await queryRunner.query(`
SELECT id, workflowData
@ -171,8 +260,8 @@ export class UpdateWorkflowCredentials1630451444017 implements MigrationInterfac
ORDER BY startedAt DESC
LIMIT 200
`);
[...waitingExecutions, ...retryableExecutions].forEach(async (execution) => {
// @ts-ignore
retryableExecutions.forEach(async (execution) => {
const data = execution.workflowData;
let credentialsUpdated = false;
// @ts-ignore
@ -208,7 +297,7 @@ export class UpdateWorkflowCredentials1630451444017 implements MigrationInterfac
{},
);
await queryRunner.query(updateQuery, updateParams);
queryRunner.query(updateQuery, updateParams);
}
});
}

128
packages/cli/src/databases/postgresdb/migrations/1630419189837-UpdateWorkflowCredentials.ts
View file

@ -1,5 +1,6 @@
import { MigrationInterface, QueryRunner } from 'typeorm';
import config = require('../../../../config');
import { MigrationHelpers } from '../../MigrationHelpers';
// replacing the credentials in workflows and execution
// `nodeType: name` changes to `nodeType: { id, name }`
@ -8,22 +9,26 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac
name = 'UpdateWorkflowCredentials1630419189837';
public async up(queryRunner: QueryRunner): Promise<void> {
console.log('Start migration', this.name);
console.time(this.name);
let tablePrefix = config.get('database.tablePrefix');
const schema = config.get('database.postgresdb.schema');
if (schema) {
tablePrefix = schema + '.' + tablePrefix;
}
const helpers = new MigrationHelpers(queryRunner);
const credentialsEntities = await queryRunner.query(`
SELECT id, name, type
FROM ${tablePrefix}credentials_entity
`);
const workflows = await queryRunner.query(`
const workflowsQuery = `
SELECT id, nodes
FROM ${tablePrefix}workflow_entity
`);
`;
// @ts-ignore
await helpers.runChunked(workflowsQuery, (workflows) => {
workflows.forEach(async (workflow) => {
const nodes = workflow.nodes;
let credentialsUpdated = false;
@ -33,7 +38,6 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac
const allNodeCredentials = Object.entries(node.credentials);
for (const [type, name] of allNodeCredentials) {
if (typeof name === 'string') {
// @ts-ignore
const matchingCredentials = credentialsEntities.find(
// @ts-ignore
(credentials) => credentials.name === name && credentials.type === type,
@ -45,7 +49,8 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac
}
});
if (credentialsUpdated) {
const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters(
const [updateQuery, updateParams] =
queryRunner.connection.driver.escapeQueryWithParameters(
`
UPDATE ${tablePrefix}workflow_entity
SET nodes = :nodes
@ -55,15 +60,53 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac
{},
);
await queryRunner.query(updateQuery, updateParams);
queryRunner.query(updateQuery, updateParams);
}
});
});
const waitingExecutions = await queryRunner.query(`
const waitingExecutionsQuery = `
SELECT id, "workflowData"
FROM ${tablePrefix}execution_entity
WHERE "waitTill" IS NOT NULL AND finished = FALSE
`);
`;
// @ts-ignore
await helpers.runChunked(waitingExecutionsQuery, (waitingExecutions) => {
waitingExecutions.forEach(async (execution) => {
const data = execution.workflowData;
let credentialsUpdated = false;
// @ts-ignore
data.nodes.forEach((node) => {
if (node.credentials) {
const allNodeCredentials = Object.entries(node.credentials);
for (const [type, name] of allNodeCredentials) {
if (typeof name === 'string') {
const matchingCredentials = credentialsEntities.find(
// @ts-ignore
(credentials) => credentials.name === name && credentials.type === type,
);
node.credentials[type] = { id: matchingCredentials?.id.toString() || null, name };
credentialsUpdated = true;
}
}
}
});
if (credentialsUpdated) {
const [updateQuery, updateParams] =
queryRunner.connection.driver.escapeQueryWithParameters(
`
UPDATE ${tablePrefix}execution_entity
SET "workflowData" = :data
WHERE id = '${execution.id}'
`,
{ data: JSON.stringify(data) },
{},
);
queryRunner.query(updateQuery, updateParams);
}
});
});
const retryableExecutions = await queryRunner.query(`
SELECT id, "workflowData"
@ -73,7 +116,8 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac
LIMIT 200
`);
[...waitingExecutions, ...retryableExecutions].forEach(async (execution) => {
// @ts-ignore
retryableExecutions.forEach(async (execution) => {
const data = execution.workflowData;
let credentialsUpdated = false;
// @ts-ignore
@ -104,9 +148,10 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac
{},
);
await queryRunner.query(updateQuery, updateParams);
queryRunner.query(updateQuery, updateParams);
}
});
console.timeEnd(this.name);
}
public async down(queryRunner: QueryRunner): Promise<void> {
@ -115,17 +160,19 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac
if (schema) {
tablePrefix = schema + '.' + tablePrefix;
}
const helpers = new MigrationHelpers(queryRunner);
const credentialsEntities = await queryRunner.query(`
SELECT id, name, type
FROM ${tablePrefix}credentials_entity
`);
const workflows = await queryRunner.query(`
const workflowsQuery = `
SELECT id, nodes
FROM ${tablePrefix}workflow_entity
`);
`;
// @ts-ignore
await helpers.runChunked(workflowsQuery, (workflows) => {
workflows.forEach(async (workflow) => {
const nodes = workflow.nodes;
let credentialsUpdated = false;
@ -152,7 +199,8 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac
}
});
if (credentialsUpdated) {
const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters(
const [updateQuery, updateParams] =
queryRunner.connection.driver.escapeQueryWithParameters(
`
UPDATE ${tablePrefix}workflow_entity
SET nodes = :nodes
@ -162,15 +210,59 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac
{},
);
await queryRunner.query(updateQuery, updateParams);
queryRunner.query(updateQuery, updateParams);
}
});
});
const waitingExecutions = await queryRunner.query(`
const waitingExecutionsQuery = `
SELECT id, "workflowData"
FROM ${tablePrefix}execution_entity
WHERE "waitTill" IS NOT NULL AND finished = FALSE
`);
`;
// @ts-ignore
await helpers.runChunked(waitingExecutionsQuery, (waitingExecutions) => {
waitingExecutions.forEach(async (execution) => {
const data = execution.workflowData;
let credentialsUpdated = false;
// @ts-ignore
data.nodes.forEach((node) => {
if (node.credentials) {
const allNodeCredentials = Object.entries(node.credentials);
for (const [type, creds] of allNodeCredentials) {
if (typeof creds === 'object') {
// @ts-ignore
const matchingCredentials = credentialsEntities.find(
// @ts-ignore
(credentials) => credentials.id === creds.id && credentials.type === type,
);
if (matchingCredentials) {
node.credentials[type] = matchingCredentials.name;
} else {
// @ts-ignore
node.credentials[type] = creds.name;
}
credentialsUpdated = true;
}
}
}
});
if (credentialsUpdated) {
const [updateQuery, updateParams] =
queryRunner.connection.driver.escapeQueryWithParameters(
`
UPDATE ${tablePrefix}execution_entity
SET "workflowData" = :data
WHERE id = '${execution.id}'
`,
{ data: JSON.stringify(data) },
{},
);
queryRunner.query(updateQuery, updateParams);
}
});
});
const retryableExecutions = await queryRunner.query(`
SELECT id, "workflowData"
@ -179,8 +271,8 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac
ORDER BY "startedAt" DESC
LIMIT 200
`);
[...waitingExecutions, ...retryableExecutions].forEach(async (execution) => {
// @ts-ignore
retryableExecutions.forEach(async (execution) => {
const data = execution.workflowData;
let credentialsUpdated = false;
// @ts-ignore
@ -216,7 +308,7 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac
{},
);
await queryRunner.query(updateQuery, updateParams);
queryRunner.query(updateQuery, updateParams);
}
});
}

149
packages/cli/src/databases/sqlite/migrations/1630330987096-UpdateWorkflowCredentials.ts
View file

@ -1,5 +1,6 @@
import { MigrationInterface, QueryRunner } from 'typeorm';
import config = require('../../../../config');
import { MigrationHelpers } from '../../MigrationHelpers';
// replacing the credentials in workflows and execution
// `nodeType: name` changes to `nodeType: { id, name }`
@ -8,18 +9,23 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac
name = 'UpdateWorkflowCredentials1630330987096';
public async up(queryRunner: QueryRunner): Promise<void> {
console.log('Start migration', this.name);
console.time(this.name);
const tablePrefix = config.get('database.tablePrefix');
const helpers = new MigrationHelpers(queryRunner);
const credentialsEntities = await queryRunner.query(`
SELECT id, name, type
FROM "${tablePrefix}credentials_entity"
`);
const workflows = await queryRunner.query(`
const workflowsQuery = `
SELECT id, nodes
FROM "${tablePrefix}workflow_entity"
`);
`;
// @ts-ignore
await helpers.runChunked(workflowsQuery, (workflows) => {
workflows.forEach(async (workflow) => {
const nodes = JSON.parse(workflow.nodes);
let credentialsUpdated = false;
@ -29,7 +35,6 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac
const allNodeCredentials = Object.entries(node.credentials);
for (const [type, name] of allNodeCredentials) {
if (typeof name === 'string') {
// @ts-ignore
const matchingCredentials = credentialsEntities.find(
// @ts-ignore
(credentials) => credentials.name === name && credentials.type === type,
@ -41,7 +46,8 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac
}
});
if (credentialsUpdated) {
const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters(
const [updateQuery, updateParams] =
queryRunner.connection.driver.escapeQueryWithParameters(
`
UPDATE "${tablePrefix}workflow_entity"
SET nodes = :nodes
@ -51,25 +57,19 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac
{},
);
await queryRunner.query(updateQuery, updateParams);
queryRunner.query(updateQuery, updateParams);
}
});
});
const waitingExecutions = await queryRunner.query(`
const waitingExecutionsQuery = `
SELECT id, "workflowData"
FROM "${tablePrefix}execution_entity"
WHERE "waitTill" IS NOT NULL AND finished = 0
`);
const retryableExecutions = await queryRunner.query(`
SELECT id, "workflowData"
FROM "${tablePrefix}execution_entity"
WHERE "waitTill" IS NULL AND finished = 0 AND mode != 'retry'
ORDER BY "startedAt" DESC
LIMIT 200
`);
[...waitingExecutions, ...retryableExecutions].forEach(async (execution) => {
`;
// @ts-ignore
await helpers.runChunked(waitingExecutionsQuery, (waitingExecutions) => {
waitingExecutions.forEach(async (execution) => {
const data = JSON.parse(execution.workflowData);
let credentialsUpdated = false;
// @ts-ignore
@ -78,7 +78,50 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac
const allNodeCredentials = Object.entries(node.credentials);
for (const [type, name] of allNodeCredentials) {
if (typeof name === 'string') {
const matchingCredentials = credentialsEntities.find(
// @ts-ignore
(credentials) => credentials.name === name && credentials.type === type,
);
node.credentials[type] = { id: matchingCredentials?.id || null, name };
credentialsUpdated = true;
}
}
}
});
if (credentialsUpdated) {
const [updateQuery, updateParams] =
queryRunner.connection.driver.escapeQueryWithParameters(
`
UPDATE "${tablePrefix}execution_entity"
SET "workflowData" = :data
WHERE id = '${execution.id}'
`,
{ data: JSON.stringify(data) },
{},
);
queryRunner.query(updateQuery, updateParams);
}
});
});
const retryableExecutions = await queryRunner.query(`
SELECT id, "workflowData"
FROM "${tablePrefix}execution_entity"
WHERE "waitTill" IS NULL AND finished = 0 AND mode != 'retry'
ORDER BY "startedAt" DESC
LIMIT 200
`);
// @ts-ignore
retryableExecutions.forEach(async (execution) => {
const data = JSON.parse(execution.workflowData);
let credentialsUpdated = false;
// @ts-ignore
data.nodes.forEach((node) => {
if (node.credentials) {
const allNodeCredentials = Object.entries(node.credentials);
for (const [type, name] of allNodeCredentials) {
if (typeof name === 'string') {
const matchingCredentials = credentialsEntities.find(
// @ts-ignore
(credentials) => credentials.name === name && credentials.type === type,
@ -100,23 +143,28 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac
{},
);
await queryRunner.query(updateQuery, updateParams);
queryRunner.query(updateQuery, updateParams);
}
});
console.timeEnd(this.name);
}
public async down(queryRunner: QueryRunner): Promise<void> {
const tablePrefix = config.get('database.tablePrefix');
const helpers = new MigrationHelpers(queryRunner);
const credentialsEntities = await queryRunner.query(`
SELECT id, name, type
FROM "${tablePrefix}credentials_entity"
`);
const workflows = await queryRunner.query(`
const workflowsQuery = `
SELECT id, nodes
FROM "${tablePrefix}workflow_entity"
`);
`;
// @ts-ignore
await helpers.runChunked(workflowsQuery, (workflows) => {
// @ts-ignore
workflows.forEach(async (workflow) => {
const nodes = JSON.parse(workflow.nodes);
@ -127,7 +175,6 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac
const allNodeCredentials = Object.entries(node.credentials);
for (const [type, creds] of allNodeCredentials) {
if (typeof creds === 'object') {
// @ts-ignore
const matchingCredentials = credentialsEntities.find(
// @ts-ignore
(credentials) => credentials.id === creds.id && credentials.type === type,
@ -144,7 +191,8 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac
}
});
if (credentialsUpdated) {
const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters(
const [updateQuery, updateParams] =
queryRunner.connection.driver.escapeQueryWithParameters(
`
UPDATE "${tablePrefix}workflow_entity"
SET nodes = :nodes
@ -154,15 +202,60 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac
{},
);
await queryRunner.query(updateQuery, updateParams);
queryRunner.query(updateQuery, updateParams);
}
});
});
const waitingExecutions = await queryRunner.query(`
const waitingExecutionsQuery = `
SELECT id, "workflowData"
FROM "${tablePrefix}execution_entity"
WHERE "waitTill" IS NOT NULL AND finished = 0
`);
`;
// @ts-ignore
await helpers.runChunked(waitingExecutionsQuery, (waitingExecutions) => {
// @ts-ignore
waitingExecutions.forEach(async (execution) => {
const data = JSON.parse(execution.workflowData);
let credentialsUpdated = false;
// @ts-ignore
data.nodes.forEach((node) => {
if (node.credentials) {
const allNodeCredentials = Object.entries(node.credentials);
for (const [type, creds] of allNodeCredentials) {
if (typeof creds === 'object') {
const matchingCredentials = credentialsEntities.find(
// @ts-ignore
(credentials) => credentials.id === creds.id && credentials.type === type,
);
if (matchingCredentials) {
node.credentials[type] = matchingCredentials.name;
} else {
// @ts-ignore
node.credentials[type] = creds.name;
}
credentialsUpdated = true;
}
}
}
});
if (credentialsUpdated) {
const [updateQuery, updateParams] =
queryRunner.connection.driver.escapeQueryWithParameters(
`
UPDATE "${tablePrefix}execution_entity"
SET "workflowData" = :data
WHERE id = '${execution.id}'
`,
{ data: JSON.stringify(data) },
{},
);
await queryRunner.query(updateQuery, updateParams);
}
});
});
const retryableExecutions = await queryRunner.query(`
SELECT id, "workflowData"
@ -172,7 +265,8 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac
LIMIT 200
`);
[...waitingExecutions, ...retryableExecutions].forEach(async (execution) => {
// @ts-ignore
retryableExecutions.forEach(async (execution) => {
const data = JSON.parse(execution.workflowData);
let credentialsUpdated = false;
// @ts-ignore
@ -181,7 +275,6 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac
const allNodeCredentials = Object.entries(node.credentials);
for (const [type, creds] of allNodeCredentials) {
if (typeof creds === 'object') {
// @ts-ignore
const matchingCredentials = credentialsEntities.find(
// @ts-ignore
(credentials) => credentials.id === creds.id && credentials.type === type,
@ -208,7 +301,7 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac
{},
);
await queryRunner.query(updateQuery, updateParams);
queryRunner.query(updateQuery, updateParams);
}
});
}