/** * @node postgresql-query * @name PostgreSQL Query * @category database * @version 1.0.0 * @description Execute CRUD operations on PostgreSQL databases with parameterized queries * @icon database */ const configSchema = { type: 'object', properties: { credentialId: { type: 'string', title: 'PostgreSQL Credential', description: 'Select the PostgreSQL credential to use for connection', dynamicOptions: { source: 'credentials', filter: { type: 'postgresql' } } }, database: { type: 'string', title: 'Database', description: 'Database name (overrides credential default if set)' }, table: { type: 'string', title: 'Table', description: 'Table name for query operations' }, operation: { type: 'string', title: 'Operation', description: 'Type of operation to perform', enum: ['select', 'insert', 'update', 'delete', 'upsert', 'bulkInsert', 'customSql'], enumLabels: ['Select', 'Insert', 'Update', 'Delete', 'Upsert', 'Bulk Insert', 'Custom SQL'], default: 'select' }, columns: { type: 'array', title: 'Columns', description: 'Columns to select (empty for all)', items: { type: 'string' }, showWhen: { field: 'operation', value: 'select' } }, whereConditions: { type: 'array', title: 'WHERE Conditions', description: 'Filter conditions for the query', items: { type: 'object', properties: { column: { type: 'string', title: 'Column' }, operator: { type: 'string', title: 'Operator', enum: ['=', '!=', '<', '>', '<=', '>=', 'LIKE', 'ILIKE', 'NOT LIKE', 'IN', 'NOT IN', 'IS NULL', 'IS NOT NULL'], default: '=' }, value: { type: 'string', title: 'Value' } } }, showWhen: { field: 'operation', valueIn: ['select', 'update', 'delete'] } }, orderBy: { type: 'array', title: 'ORDER BY', description: 'Sorting options', items: { type: 'object', properties: { column: { type: 'string', title: 'Column' }, direction: { type: 'string', title: 'Direction', enum: ['ASC', 'DESC'], default: 'ASC' } } }, showWhen: { field: 'operation', value: 'select' } }, limit: { type: 'number', title: 'Limit', description: 'Maximum number of rows to return (0 for no limit)', default: 100, showWhen: { field: 'operation', value: 'select' } }, offset: { type: 'number', title: 'Offset', description: 'Number of rows to skip', default: 0, showWhen: { field: 'operation', value: 'select' } }, insertData: { type: 'string', title: 'Data to Insert (JSON)', description: 'JSON object with column:value pairs. Supports {{variable}} interpolation.', showWhen: { field: 'operation', value: 'insert' } }, updateData: { type: 'string', title: 'Data to Update (JSON)', description: 'JSON object with column:value pairs to set. Supports {{variable}} interpolation.', showWhen: { field: 'operation', value: 'update' } }, upsertData: { type: 'string', title: 'Data to Upsert (JSON)', description: 'JSON object with column:value pairs. Supports {{variable}} interpolation.', showWhen: { field: 'operation', value: 'upsert' } }, conflictKey: { type: 'array', title: 'Conflict Key Columns', description: 'Columns that form the unique/primary key for conflict detection', items: { type: 'string' }, showWhen: { field: 'operation', value: 'upsert' } }, updateOnConflict: { type: 'array', title: 'Update Columns on Conflict', description: 'Columns to update when a conflict is detected (empty for all non-key columns)', items: { type: 'string' }, showWhen: { field: 'operation', value: 'upsert' } }, bulkData: { type: 'string', title: 'Bulk Data (JSON Array)', description: 'JSON array of objects to insert. Supports {{variable}} interpolation.', showWhen: { field: 'operation', value: 'bulkInsert' } }, batchSize: { type: 'number', title: 'Batch Size', description: 'Number of rows per INSERT statement (0 for all at once)', default: 100, showWhen: { field: 'operation', value: 'bulkInsert' } }, customQuery: { type: 'string', title: 'Custom SQL', description: 'Custom SQL query with $1, $2, etc. placeholders for parameters. Supports {{variable}} interpolation.', showWhen: { field: 'operation', value: 'customSql' } }, parameters: { type: 'array', title: 'Query Parameters', description: 'Parameters for $1, $2, etc. placeholders in custom SQL', items: { type: 'string' }, showWhen: { field: 'operation', value: 'customSql' } } }, required: ['credentialId', 'operation'] }; const inputSchema = { type: 'object', properties: { data: { type: 'any', description: 'Input data available for interpolation in queries' }, parameters: { type: 'array', description: 'Override query parameters from input' }, rows: { type: 'array', description: 'Array of objects for bulk insert (alternative to bulkData config)' } } }; const outputSchema = { type: 'object', properties: { rows: { type: 'array', description: 'Query result rows as array of objects' }, metadata: { type: 'object', description: 'Operation metadata', properties: { affectedRows: { type: 'number', description: 'Number of affected rows' }, insertId: { type: 'number', description: 'Last auto-generated insert ID (if using RETURNING)' }, rowCount: { type: 'number', description: 'Number of rows returned' } } }, columns: { type: 'array', description: 'Column names from the result' }, query: { type: 'string', description: 'The executed SQL query (for debugging)' } } }; const outputs = [ { name: 'main', displayName: 'Query Result', type: 'object', color: '#336791' } ]; function parseJsonData(jsonString, data, fieldName) { if (!jsonString) { throw new Error(fieldName + ' is required'); } let parsed = jsonString; if (typeof parsed === 'string') { parsed = smartbotic.utils.interpolate(parsed, data); try { parsed = JSON.parse(parsed); } catch (e) { throw new Error(fieldName + ' must be valid JSON: ' + e.message); } } return parsed; } function escapeIdentifier(name) { // PostgreSQL uses double quotes for identifiers return '"' + name.replace(/"/g, '""') + '"'; } function buildWhereClause(conditions, data, paramOffset) { if (!conditions || conditions.length === 0) { return { clause: '', params: [], nextParam: paramOffset }; } const clauses = []; const params = []; let paramIdx = paramOffset; for (const condition of conditions) { if (!condition.column) continue; const col = escapeIdentifier(condition.column); const op = condition.operator || '='; let value = condition.value; if (typeof value === 'string') { value = smartbotic.utils.interpolate(value, data); } if (op === 'IS NULL') { clauses.push(col + ' IS NULL'); } else if (op === 'IS NOT NULL') { clauses.push(col + ' IS NOT NULL'); } else if (op === 'IN' || op === 'NOT IN') { let values = value; if (typeof values === 'string') { try { values = JSON.parse(values); } catch (e) { values = values.split(',').map(v => v.trim()); } } if (!Array.isArray(values)) { values = [values]; } const placeholders = values.map(() => '$' + (paramIdx++)).join(', '); clauses.push(col + ' ' + op + ' (' + placeholders + ')'); params.push(...values); } else { clauses.push(col + ' ' + op + ' $' + (paramIdx++)); params.push(value); } } return { clause: clauses.length > 0 ? ' WHERE ' + clauses.join(' AND ') : '', params, nextParam: paramIdx }; } function buildSelectQuery(config, data) { const table = config.table; if (!table) { throw new Error('Table is required for SELECT operations'); } const columns = config.columns && config.columns.length > 0 ? config.columns.map(escapeIdentifier).join(', ') : '*'; let query = 'SELECT ' + columns + ' FROM ' + escapeIdentifier(table); const params = []; let paramIdx = 1; const where = buildWhereClause(config.whereConditions, data, paramIdx); query += where.clause; params.push(...where.params); paramIdx = where.nextParam; if (config.orderBy && config.orderBy.length > 0) { const orderClauses = config.orderBy .filter(o => o.column) .map(o => escapeIdentifier(o.column) + ' ' + (o.direction || 'ASC')); if (orderClauses.length > 0) { query += ' ORDER BY ' + orderClauses.join(', '); } } if (config.limit && config.limit > 0) { query += ' LIMIT $' + (paramIdx++); params.push(config.limit); if (config.offset && config.offset > 0) { query += ' OFFSET $' + (paramIdx++); params.push(config.offset); } } return { query, params }; } function buildInsertQuery(config, data) { const table = config.table; if (!table) { throw new Error('Table is required for INSERT operations'); } const insertData = parseJsonData(config.insertData, data, 'Insert data'); if (!insertData || typeof insertData !== 'object' || Array.isArray(insertData)) { throw new Error('Insert data must be a JSON object'); } const columns = Object.keys(insertData); if (columns.length === 0) { throw new Error('Insert data cannot be empty'); } const values = Object.values(insertData); const placeholders = columns.map((_, i) => '$' + (i + 1)).join(', '); // Use RETURNING to get the inserted ID if available const query = 'INSERT INTO ' + escapeIdentifier(table) + ' (' + columns.map(escapeIdentifier).join(', ') + ')' + ' VALUES (' + placeholders + ')' + ' RETURNING *'; return { query, params: values }; } function buildUpdateQuery(config, data) { const table = config.table; if (!table) { throw new Error('Table is required for UPDATE operations'); } const updateData = parseJsonData(config.updateData, data, 'Update data'); if (!updateData || typeof updateData !== 'object' || Array.isArray(updateData)) { throw new Error('Update data must be a JSON object'); } const columns = Object.keys(updateData); if (columns.length === 0) { throw new Error('Update data cannot be empty'); } const values = Object.values(updateData); let paramIdx = 1; const setClause = columns.map(col => escapeIdentifier(col) + ' = $' + (paramIdx++)).join(', '); let query = 'UPDATE ' + escapeIdentifier(table) + ' SET ' + setClause; const params = [...values]; const where = buildWhereClause(config.whereConditions, data, paramIdx); if (!where.clause) { throw new Error('WHERE conditions are required for UPDATE operations to prevent accidental data modification'); } query += where.clause; params.push(...where.params); return { query, params }; } function buildDeleteQuery(config, data) { const table = config.table; if (!table) { throw new Error('Table is required for DELETE operations'); } let query = 'DELETE FROM ' + escapeIdentifier(table); const params = []; const where = buildWhereClause(config.whereConditions, data, 1); if (!where.clause) { throw new Error('WHERE conditions are required for DELETE operations to prevent accidental data loss'); } query += where.clause; params.push(...where.params); return { query, params }; } function buildUpsertQuery(config, data) { const table = config.table; if (!table) { throw new Error('Table is required for UPSERT operations'); } const upsertData = parseJsonData(config.upsertData, data, 'Upsert data'); if (!upsertData || typeof upsertData !== 'object' || Array.isArray(upsertData)) { throw new Error('Upsert data must be a JSON object'); } const columns = Object.keys(upsertData); if (columns.length === 0) { throw new Error('Upsert data cannot be empty'); } const values = Object.values(upsertData); const placeholders = columns.map((_, i) => '$' + (i + 1)).join(', '); const conflictKeys = config.conflictKey || []; if (conflictKeys.length === 0) { throw new Error('Conflict key columns are required for UPSERT operations in PostgreSQL'); } let updateColumns = config.updateOnConflict || []; if (updateColumns.length === 0) { updateColumns = columns.filter(col => !conflictKeys.includes(col)); } if (updateColumns.length === 0) { updateColumns = columns; } const conflictClause = conflictKeys.map(escapeIdentifier).join(', '); const updateClause = updateColumns .map(col => escapeIdentifier(col) + ' = EXCLUDED.' + escapeIdentifier(col)) .join(', '); // PostgreSQL ON CONFLICT syntax const query = 'INSERT INTO ' + escapeIdentifier(table) + ' (' + columns.map(escapeIdentifier).join(', ') + ')' + ' VALUES (' + placeholders + ')' + ' ON CONFLICT (' + conflictClause + ')' + ' DO UPDATE SET ' + updateClause + ' RETURNING *'; return { query, params: values }; } function buildBulkInsertQueries(config, data, input) { const table = config.table; if (!table) { throw new Error('Table is required for BULK INSERT operations'); } let bulkData = input.rows; if (!bulkData) { bulkData = parseJsonData(config.bulkData, data, 'Bulk data'); } if (!Array.isArray(bulkData)) { throw new Error('Bulk data must be a JSON array of objects'); } if (bulkData.length === 0) { throw new Error('Bulk data array cannot be empty'); } const columns = Object.keys(bulkData[0]); if (columns.length === 0) { throw new Error('Bulk data objects cannot be empty'); } const batchSize = config.batchSize > 0 ? config.batchSize : bulkData.length; const queries = []; for (let i = 0; i < bulkData.length; i += batchSize) { const batch = bulkData.slice(i, i + batchSize); const allValues = []; const rowPlaceholders = []; let paramIdx = 1; for (const row of batch) { const rowValues = columns.map(col => { const val = row[col]; return val !== undefined ? val : null; }); allValues.push(...rowValues); const rowParams = columns.map(() => '$' + (paramIdx++)); rowPlaceholders.push('(' + rowParams.join(', ') + ')'); } const query = 'INSERT INTO ' + escapeIdentifier(table) + ' (' + columns.map(escapeIdentifier).join(', ') + ')' + ' VALUES ' + rowPlaceholders.join(', '); queries.push({ query, params: allValues, rowCount: batch.length }); } return queries; } function buildCustomQuery(config, data, input) { if (!config.customQuery) { throw new Error('Custom SQL query is required'); } const query = smartbotic.utils.interpolate(config.customQuery, data); let params = input.parameters || config.parameters || []; if (params.length > 0) { params = params.map(p => { if (typeof p === 'string') { return smartbotic.utils.interpolate(p, data); } return p; }); } return { query, params }; } module.exports = { configSchema, inputSchema, outputSchema, outputs, async execute(config, input) { const credentialId = config.credentialId; const data = input.data || input; if (!credentialId) { throw new Error('PostgreSQL credential is required'); } const operation = config.operation || 'select'; let totalAffectedRows = 0; let lastInsertId = null; let executedQuery = ''; if (operation === 'bulkInsert') { const queries = buildBulkInsertQueries(config, data, input); smartbotic.log.info('PostgreSQL BULK INSERT: ' + queries.length + ' batch(es)'); for (const queryInfo of queries) { const options = { credentialId, database: config.database, query: queryInfo.query, params: queryInfo.params }; const result = smartbotic.postgresql.query(options); if (!result.success) { smartbotic.log.error('PostgreSQL bulk insert failed: ' + result.error); throw new Error('PostgreSQL bulk insert failed: ' + result.error); } totalAffectedRows += result.affectedRows || 0; if (result.insertId) { lastInsertId = result.insertId; } executedQuery = queryInfo.query; } smartbotic.log.info('PostgreSQL BULK INSERT completed: ' + totalAffectedRows + ' rows inserted'); return { rows: [], metadata: { affectedRows: totalAffectedRows, insertId: lastInsertId, rowCount: 0 }, columns: [], query: executedQuery }; } let queryInfo; switch (operation) { case 'select': queryInfo = buildSelectQuery(config, data); break; case 'insert': queryInfo = buildInsertQuery(config, data); break; case 'update': queryInfo = buildUpdateQuery(config, data); break; case 'delete': queryInfo = buildDeleteQuery(config, data); break; case 'upsert': queryInfo = buildUpsertQuery(config, data); break; case 'customSql': queryInfo = buildCustomQuery(config, data, input); break; default: throw new Error('Invalid operation: ' + operation); } const { query, params } = queryInfo; smartbotic.log.info('PostgreSQL ' + operation.toUpperCase() + ': ' + query.substring(0, 100) + (query.length > 100 ? '...' : '')); const options = { credentialId, database: config.database, query, params }; const result = smartbotic.postgresql.query(options); if (!result.success) { smartbotic.log.error('PostgreSQL query failed: ' + result.error); throw new Error('PostgreSQL query failed: ' + result.error); } smartbotic.log.info('PostgreSQL ' + operation.toUpperCase() + ' completed: ' + (result.affectedRows || 0) + ' affected, ' + (result.rows || []).length + ' rows returned'); return { rows: result.rows || [], metadata: { affectedRows: result.affectedRows || 0, insertId: result.insertId || null, rowCount: (result.rows || []).length }, columns: result.columns || [], query: query }; } };