datahub-restful-core/routes/sql-mappings-router.js

'use strict';

const debug = require('debug')('krh:route');
const route = require('koa-router');

const util = require('./util');
const SqlParser = require('../core/sql/sqlParser');
const knexHelper = require('../core/knex-helper');
const ObjStream = require('objstream')
/**
 * Coded at Chiangmai >_<
 * Drafted upon cloud...
 *
 * @param this.knexHelper
 * @returns {Router|exports|module.exports}
 */

let router = new route();
let sqlParser = new SqlParser();

let mappingExecutor = function* (next) {

    let possibleConnectionId = yield this.knexHelper.findConnectionByDbkey(this.params.sqlKey);

    if (possibleConnectionId) {
        this.params.sqlKey = possibleConnectionId;
    }

    // 检查权限
    if (!this.privilegeHelper.canExecuteSQLMapping(this.state.database, this.state.schema, this.params.sqlKey)) {
        this.throw(403, '无权限')
        return;
    }

    let sqlMappings = yield this.knexHelper.getMetaConnection().select().from('sql_mapping')
        .where({ 'sql_key': this.params.sqlKey });

    debug('SQL IS', sqlMappings);

    let mappingObject = sqlMappings[0];

    //still not found.
    if (!mappingObject) {
        this.status = 404;
    } else {

        let accessTime = new Date();
        let source = this.headers['Datahub-Source'] || this.origin || 'unknown';

        //log start
        let params = {};

        if (JSON.stringify(this.query) !== '{}') {
            params = this.query;
        } else if (JSON.stringify(this.request.body) !== '{}') {
            params = this.request.body;
        }

        let logId = yield this.knexHelper.logSqlAccessing(
            this.ip,
            accessTime,
            mappingObject.id,
            mappingObject.sql,
            JSON.stringify(params),
            source);

        this.state.mappingSql = sqlParser.cleanAndReplaceParams(
            sqlParser.formatSql(mappingObject.sql),
            params,
            this.state.schema
        );

        //query sql executing
        yield next;

        let costTime = new Date().getTime() - accessTime.getTime();

        yield this.knexHelper.updateCostTime(logId, costTime);
        //log end
        //this.res.setHeader('Sql-Mapping', mappingObject.sql);
    }
};

router
    .param('schema', function* (schema, next) {
        if (this.state.schema) {
            return yield next;
        }

        debug('schema param', schema);
        //resolve path to state.schema for table usage
        if (!schema) {
            this.body = '缺少数据库schema';
            return this.status = 400;
        }

        // 检查权限
        if (!this.privilegeHelper.canAccessSchema(this.state.database, schema)) {
            this.throw(403, '无权限')
            return;
        }

        this.state.schema = schema;
        yield next;

    })

router.post('/sqlMapping', function* () {
    let dbConfig = yield this.knexHelper.getMetaConnection().insert(this.request.body).into('sql_mapping');
    this.body = dbConfig;
});

/**
 * Find sql by key
 * And execute query
 */
router.get('/:sqlKey',
    mappingExecutor,
    knexHelper.readOnlyTx.bind(this)(),
    function* () {
        yield this.state.connection.raw('use :schema:', { schema: this.state.schema });

        let sql = sqlParser.formatSql(this.state.mappingSql).replace(/\\\\:/g, ":");

        for (let i in this.query) {
            try {
                this.query[i] = JSON.parse(this.query[i]);
            } catch (e) {
                this.query[i] = this.query[i];
            }
        }

        try {
            let result = this.state.connection
                .raw(sql, this.query).stream().pipe(new ObjStream());

            this.body = result;
        } catch (err) {
            this.status = 500;
            this.body = err
        }
    });

router.get('/:sqlKey/pagination(.*)',
    mappingExecutor,
    knexHelper.readOnlyTx.bind(this)(),
    function* () {
        yield this.state.connection.raw('use :schema:', { schema: this.state.schema });

        let sql = sqlParser.formatSql(this.state.mappingSql).replace(/\\\\:/g, ":");
        try {
            let data = yield this.knexHelper.paginate(sql,
                this.query,
                this.state.connection);

            this.body = util.wrapPagingResult(data.result, data.options);
        } catch (err) {
            this.status = 500;
            this.body = err
        }

    });

//TODO support tx
router.post('/:sqlKey', mappingExecutor,
    function* () {
        yield this.state.connection.raw('use :schema:', { schema: this.state.schema });

        let sql = sqlParser.formatSql(this.state.mappingSql).replace(/\\\\:/g, ":");

        for (let i in this.request.body) {
            try {
                this.request.body[i] = JSON.parse(this.request.body[i]);
            } catch (e) {
                this.request.body[i] = this.request.body[i];
            }
        }

        try {
            let result = this.state.connection
                .raw(sql, this.request.body).stream().pipe(new ObjStream());

            this.body = result;
        } catch (err) {
            this.status = 500;
            this.body = err
        }
    });

module.exports = router;