smartbotic/nodes/postgresql/postgresql-query.js

/**
 * @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
        };
    }
};