/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

#include <qpid/dispatch/ctools.h>
#include "agent_connection.h"
#include <inttypes.h>
#include <stdio.h>

#define QDR_CONNECTION_NAME                   0
#define QDR_CONNECTION_IDENTITY               1
#define QDR_CONNECTION_HOST                   2
#define QDR_CONNECTION_ROLE                   3
#define QDR_CONNECTION_PROTOCOL               4
#define QDR_CONNECTION_DIR                    5
#define QDR_CONNECTION_CONTAINER_ID           6
#define QDR_CONNECTION_SASL_MECHANISMS        7
#define QDR_CONNECTION_IS_AUTHENTICATED       8
#define QDR_CONNECTION_USER                   9
#define QDR_CONNECTION_IS_ENCRYPTED          10
#define QDR_CONNECTION_SSLPROTO              11
#define QDR_CONNECTION_SSLCIPHER             12
#define QDR_CONNECTION_PROPERTIES            13
#define QDR_CONNECTION_SSLSSF                14
#define QDR_CONNECTION_TENANT                15
#define QDR_CONNECTION_TYPE                  16
#define QDR_CONNECTION_SSL                   17
#define QDR_CONNECTION_OPENED                18
#define QDR_CONNECTION_ACTIVE                19
#define QDR_CONNECTION_ADMIN_STATUS          20
#define QDR_CONNECTION_OPER_STATUS           21
#define QDR_CONNECTION_UPTIME_SECONDS        22
#define QDR_CONNECTION_LAST_DLV_SECONDS      23
#define QDR_CONNECTION_ENABLE_PROTOCOL_TRACE 24


const char * const QDR_CONNECTION_DIR_IN  = "in";
const char * const QDR_CONNECTION_DIR_OUT = "out";

const char * QDR_CONNECTION_ADMIN_STATUS_DELETED = "deleted";
const char * QDR_CONNECTION_ADMIN_STATUS_ENABLED = "enabled";

const char * QDR_CONNECTION_OPER_STATUS_UP      = "up";
const char * QDR_CONNECTION_OPER_STATUS_CLOSING = "closing";


const char *qdr_connection_roles[] =
    {"normal",
     "inter-router",
     "route-container",
     "edge",
     0};

const char *qdr_connection_columns[] =
    {"name",
     "identity",
     "host",
     "role",
     "protocol",
     "dir",
     "container",
     "sasl",
     "isAuthenticated",
     "user",
     "isEncrypted",
     "sslProto",
     "sslCipher",
     "properties",
     "sslSsf",
     "tenant",
     "type",
     "ssl",
     "opened",
     "active",
     "adminStatus",
     "operStatus",
     "uptimeSeconds",
     "lastDlvSeconds",
     "enableProtocolTrace",
     0};

const char *CONNECTION_TYPE = "org.apache.qpid.dispatch.connection";

static void qd_get_next_pn_data(pn_data_t **data, const char **d, int *d1)
{
    if (pn_data_next(*data)) {
        switch (pn_data_type(*data)) {
            case PN_STRING:
                *d = pn_data_get_string(*data).start;
                break;
            case PN_SYMBOL:
                *d = pn_data_get_symbol(*data).start;
                break;
            case PN_INT:
                *d1 = pn_data_get_int(*data);
                break;
            case PN_LONG:
                *d1 = pn_data_get_long(*data);
                break;
            default:
                break;
        }
    }
    }


static void qdr_connection_insert_column_CT(qdr_core_t *core, qdr_connection_t *conn, int col, qd_composed_field_t *body, bool as_map)
{
    char id_str[100];
    const char *text = 0;

    if (!conn)
        return;

    if (as_map)
        qd_compose_insert_string(body, qdr_connection_columns[col]);

    switch(col) {
    case QDR_CONNECTION_NAME:
        qd_compose_insert_string2(body, "connection/", conn->connection_info->host);
        break;

    case QDR_CONNECTION_IDENTITY: {
        snprintf(id_str, 100, "%"PRId64, conn->identity);
        qd_compose_insert_string(body, id_str);
        break;
    }

    case QDR_CONNECTION_HOST:
        qd_compose_insert_string(body, conn->connection_info->host);
        break;

    case QDR_CONNECTION_ROLE:
        qd_compose_insert_string(body, qdr_connection_roles[conn->connection_info->role]);
        break;

    case QDR_CONNECTION_PROTOCOL:
        qd_compose_insert_string(body, conn->protocol_adaptor->name);
        break;

    case QDR_CONNECTION_DIR:
        if (conn->connection_info->dir == QD_INCOMING)
            qd_compose_insert_string(body, QDR_CONNECTION_DIR_IN);
        else
            qd_compose_insert_string(body, QDR_CONNECTION_DIR_OUT);
        break;

    case QDR_CONNECTION_CONTAINER_ID:
        if (conn->connection_info->container)
            qd_compose_insert_string(body, conn->connection_info->container);
        else
            qd_compose_insert_null(body);
        break;

    case QDR_CONNECTION_SASL_MECHANISMS:
        if (conn->connection_info->sasl_mechanisms) 
            qd_compose_insert_string(body, conn->connection_info->sasl_mechanisms);
	else
	    qd_compose_insert_null(body);
        break;

    case QDR_CONNECTION_IS_AUTHENTICATED:
        qd_compose_insert_bool(body, conn->connection_info->is_authenticated);
        break;

    case QDR_CONNECTION_USER:
        if (conn->connection_info->user)
            qd_compose_insert_string(body, conn->connection_info->user);
        else
            qd_compose_insert_null(body);
        break;

    case QDR_CONNECTION_IS_ENCRYPTED:
        qd_compose_insert_bool(body, conn->connection_info->is_encrypted);
        break;

    case QDR_CONNECTION_SSLPROTO:
        if (conn->connection_info->ssl_proto[0] != '\0')
            qd_compose_insert_string(body, conn->connection_info->ssl_proto);
        else
            qd_compose_insert_null(body);
        break;

    case QDR_CONNECTION_SSLCIPHER:
        if (conn->connection_info->ssl_cipher[0] != '\0')
            qd_compose_insert_string(body, conn->connection_info->ssl_cipher);
        else
            qd_compose_insert_null(body);
        break;

    case QDR_CONNECTION_SSLSSF:
        qd_compose_insert_long(body, conn->connection_info->ssl_ssf);
        break;

    case QDR_CONNECTION_TENANT:
        if (conn->tenant_space)
            qd_compose_insert_string(body, conn->tenant_space);
        else
            qd_compose_insert_null(body);
        break;

    case QDR_CONNECTION_TYPE:
        qd_compose_insert_string(body, CONNECTION_TYPE);
        break;

    case QDR_CONNECTION_SSL:
        qd_compose_insert_bool(body, conn->connection_info->ssl);
        break;

    case QDR_CONNECTION_OPENED:
        qd_compose_insert_bool(body, conn->connection_info->opened);
        break;

    case QDR_CONNECTION_ENABLE_PROTOCOL_TRACE:
        qd_compose_insert_bool(body, conn->enable_protocol_trace);
        break;

    case QDR_CONNECTION_ACTIVE:
        if (conn->role == QDR_ROLE_EDGE_CONNECTION) {
            if (core->router_mode == QD_ROUTER_MODE_INTERIOR) {
                qd_compose_insert_bool(body, true);
            }
            else if (core->router_mode  == QD_ROUTER_MODE_EDGE){
                if (core->active_edge_connection == conn)
                    qd_compose_insert_bool(body, true);
                else
                    qd_compose_insert_bool(body, false);
            }
        }
        else {
            qd_compose_insert_bool(body, true);
        }
        break;

    case QDR_CONNECTION_ADMIN_STATUS:
        text = conn->closed ? QDR_CONNECTION_ADMIN_STATUS_DELETED : QDR_CONNECTION_ADMIN_STATUS_ENABLED;
        qd_compose_insert_string(body, text);
        break;

    case QDR_CONNECTION_OPER_STATUS:
        text = conn->closed ? QDR_CONNECTION_OPER_STATUS_CLOSING : QDR_CONNECTION_OPER_STATUS_UP;
        qd_compose_insert_string(body, text);
        break;

    case QDR_CONNECTION_UPTIME_SECONDS:
        qd_compose_insert_uint(body, core->uptime_ticks - conn->conn_uptime);
        break;

    case QDR_CONNECTION_LAST_DLV_SECONDS:
        if (conn->last_delivery_time==0)
            qd_compose_insert_null(body);
        else
            qd_compose_insert_uint(body, core->uptime_ticks - conn->last_delivery_time);
        break;

    case QDR_CONNECTION_PROPERTIES: {
        pn_data_t *data = conn->connection_info->connection_properties;
        qd_compose_start_map(body);
        if (data) {
            pn_data_next(data);
            size_t count = pn_data_get_map(data);
            pn_data_enter(data);

            if (count > 0) {

                for (size_t i = 0; i < count/2; i++) {
                    const char *key   = 0;
                    // We are assuming for now that all keys are strings
                    qd_get_next_pn_data(&data, &key, 0);

                    const char *value_string = 0;
                    int value_int = 0;
                    // We are assuming for now that all values are either strings or integers
                    qd_get_next_pn_data(&data, &value_string, &value_int);

                    // We now have the key and the value. Do not insert the key or the value if key is empty
                    if (key) {
                        qd_compose_insert_string(body, key);
                        if (value_string)
                            qd_compose_insert_string(body, value_string);
                        else if (value_int)
                            qd_compose_insert_int(body, value_int);
                    }

                }
            }

            pn_data_exit(data);

        }
        qd_compose_end_map(body);
    }
    break;
    }
}


static void qdr_agent_write_connection_CT(qdr_core_t *core, qdr_query_t *query,  qdr_connection_t *conn)
{
    qd_composed_field_t *body = query->body;

    qd_compose_start_list(body);

    if (conn) {
        int i = 0;
        while (query->columns[i] >= 0) {
            qdr_connection_insert_column_CT(core, conn, query->columns[i], body, false);
            i++;
        }
    }
    qd_compose_end_list(body);
}


static void qdr_manage_advance_connection_CT(qdr_query_t *query, qdr_connection_t *conn)
{
    if (conn) {
        query->next_offset++;
        conn = DEQ_NEXT(conn);
        query->more = !!conn;
    }
    else {
        query->more = false;
    }
}


void qdra_connection_get_first_CT(qdr_core_t *core, qdr_query_t *query, int offset)
{
    //
    // Queries that get this far will always succeed.
    //
    query->status = QD_AMQP_OK;

    //
    // If the offset goes beyond the set of objects, end the query now.
    //
    if (offset >= DEQ_SIZE(core->open_connections)) {
        query->more = false;
        qdr_agent_enqueue_response_CT(core, query);
        return;
    }

    //
    // Run to the object at the offset.
    //
    qdr_connection_t *conn = DEQ_HEAD(core->open_connections);
    for (int i = 0; i < offset && conn; i++)
        conn = DEQ_NEXT(conn);
    assert(conn);

    if (conn) {
        //
        // Write the columns of the object into the response body.
        //
        qdr_agent_write_connection_CT(core, query, conn);

        //
        // Advance to the next connection
        //
        query->next_offset = offset;
        qdr_manage_advance_connection_CT(query, conn);
    }
    else {
        query->more = false;
    }

    //
    // Enqueue the response.
    //
    qdr_agent_enqueue_response_CT(core, query);
}

void qdra_connection_get_next_CT(qdr_core_t *core, qdr_query_t *query)
{
    qdr_connection_t *conn = 0;

    if (query->next_offset < DEQ_SIZE(core->open_connections)) {
        conn = DEQ_HEAD(core->open_connections);
        for (int i = 0; i < query->next_offset && conn; i++)
            conn = DEQ_NEXT(conn);
    }

    if (conn) {
        //
        // Write the columns of the connection entity into the response body.
        //
        qdr_agent_write_connection_CT(core, query, conn);

        //
        // Advance to the next object
        //
        qdr_manage_advance_connection_CT(query, conn);
    } else
        query->more = false;

    //
    // Enqueue the response.
    //
    qdr_agent_enqueue_response_CT(core, query);
}

static void qdr_manage_write_connection_map_CT(qdr_core_t          *core,
                                               qdr_connection_t    *conn,
                                               qd_composed_field_t *body,
                                               const char          *qdr_connection_columns[])
{
    qd_compose_start_map(body);

    for(int i = 0; i < QDR_CONNECTION_COLUMN_COUNT; i++) {
        qd_compose_insert_string(body, qdr_connection_columns[i]);
        qdr_connection_insert_column_CT(core, conn, i, body, false);
    }

    qd_compose_end_map(body);
}

static qdr_connection_t *_find_conn_CT(qdr_core_t *core, uint64_t conn_id)
{
    qdr_connection_t *conn = DEQ_HEAD(core->open_connections);
    while (conn) {
        if (conn->identity == conn_id)
            break;
        conn = DEQ_NEXT(conn);
    }
    return conn;
}


static qdr_connection_t *qdr_connection_find_by_identity_CT(qdr_core_t *core, qd_iterator_t *identity)
{
    if (!identity)
        return 0;

    qdr_connection_t *conn = DEQ_HEAD(core->open_connections);
    while (conn) {
        // Convert the passed in identity to a char*
        char id[100];
        snprintf(id, 100, "%"PRId64, conn->identity);
        if (qd_iterator_equal(identity, (const unsigned char*) id))
            break;
        conn = DEQ_NEXT(conn);
    }

    return conn;

}

void qdra_connection_get_CT(qdr_core_t    *core,
                            qd_iterator_t *name,
                            qd_iterator_t *identity,
                            qdr_query_t   *query,
                            const char    *qdr_connection_columns[])
{
    qdr_connection_t *conn = 0;

    if (!identity) {
        query->status = QD_AMQP_BAD_REQUEST;
        query->status.description = "Name not supported. Identity required";
        qd_log(core->agent_log, QD_LOG_ERROR, "Error performing READ of %s: %s", CONNECTION_TYPE, query->status.description);
    }
    else {
        conn = qdr_connection_find_by_identity_CT(core, identity);

        if (conn == 0) {
            // Send back a 404
            query->status = QD_AMQP_NOT_FOUND;
        }
        else {
            //
            // Write the columns of the connection entity into the response body.
            //
            qdr_manage_write_connection_map_CT(core, conn, query->body, qdr_connection_columns);
            query->status = QD_AMQP_OK;
        }
    }

    //
    // Enqueue the response.
    //
    qdr_agent_enqueue_response_CT(core, query);
}


static void qdra_connection_set_bad_request(qdr_query_t *query)
{
    query->status = QD_AMQP_BAD_REQUEST;
    qd_compose_start_map(query->body);
    qd_compose_end_map(query->body);
}


static void qdra_connection_update_set_status(qdr_core_t *core, qdr_query_t *query, qdr_connection_t *conn, qd_parsed_field_t *admin_state)
{
    if (conn) {
        qd_iterator_t *admin_status_iter = qd_parse_raw(admin_state);

        if (qd_iterator_equal(admin_status_iter, (unsigned char*) QDR_CONNECTION_ADMIN_STATUS_DELETED)) {
            // This connection has been force-closed.
            // Inter-router and edge connections may not be force-closed
            if (conn->role != QDR_ROLE_INTER_ROUTER && conn->role != QDR_ROLE_EDGE_CONNECTION) {
                qdr_close_connection_CT(core, conn);
                qd_log(core->log, QD_LOG_INFO, "[C%"PRIu64"] Connection force-closed by request from connection [C%"PRIu64"]", conn->identity, query->in_conn);
                query->status = QD_AMQP_OK;
                qdr_manage_write_connection_map_CT(core, conn, query->body, qdr_connection_columns);
            }
            else {
                //
                // You are trying to delete an inter-router connection and that is always forbidden, no matter what
                // policy rights you have.
                //
                query->status = QD_AMQP_FORBIDDEN;
                query->status.description = "You are not allowed to perform this operation.";
                qd_compose_start_map(query->body);
                qd_compose_end_map(query->body);
            }

        }
        else if (qd_iterator_equal(admin_status_iter, (unsigned char*) QDR_CONNECTION_ADMIN_STATUS_ENABLED)) {
            query->status = QD_AMQP_OK;
            qdr_manage_write_connection_map_CT(core, conn, query->body, qdr_connection_columns);
        }
        else {
            qdra_connection_set_bad_request(query);
        }
    }
    else {
        query->status = QD_AMQP_NOT_FOUND;
        qd_compose_start_map(query->body);
        qd_compose_end_map(query->body);
    }
}


void qdra_connection_update_CT(qdr_core_t      *core,
                             qd_iterator_t     *name,
                             qd_iterator_t     *identity,
                             qdr_query_t       *query,
                             qd_parsed_field_t *in_body)
{
    // If the request was successful then the statusCode MUST contain 200 (OK) and the body of the message
    // MUST contain a map containing the actual attributes of the entity updated. These MAY differ from those
    // requested.
    // A map containing attributes that are not applicable for the entity being created, or invalid values for a
    // given attribute, MUST result in a failure response with a statusCode of 400 (Bad Request).
    if (qd_parse_is_map(in_body) && identity) {
        // The absence of an attribute name implies that the entity should retain its already existing value.
        // If the map contains a key-value pair where the value is null then the updated entity should have no value
        // for that attribute, removing any previous value.
        qd_parsed_field_t *admin_state = qd_parse_value_by_key(in_body, qdr_connection_columns[QDR_CONNECTION_ADMIN_STATUS]);

        // Find the connection that the user connected on. This connection must have the correct policy rights which
        // will allow the user on this connection to terminate some other connection.
        qdr_connection_t *user_conn = _find_conn_CT(core, query->in_conn);
        qd_parsed_field_t *trace_field   = qd_parse_value_by_key(in_body, qdr_connection_columns[QDR_CONNECTION_ENABLE_PROTOCOL_TRACE]);


        //
        // The only two fields that can be updated on a connection is the enableProtocolTrace flag and the admin state.
        // If both these fields are not there, this is a bad request
        // For example, this qdmanage is a bad request - qdmanage update --type=connection identity=1
        //
        if (!trace_field && !admin_state) {
            qdra_connection_set_bad_request(query);
            qdr_agent_enqueue_response_CT(core, query);
            return;
        }

        bool enable_protocol_trace = !!trace_field ? qd_parse_as_bool(trace_field) : false;

        qdr_connection_t *conn = qdr_connection_find_by_identity_CT(core, identity);

        if (!conn) {
            //
            // The identity supplied was used to obtain the connection. If the connection was not found,
            // it is possible that the connection went away or the wrong identity was provided.
            // Either way we will have to let the caller know that this is a bad request.
            //
            qdra_connection_set_bad_request(query);
            qdr_agent_enqueue_response_CT(core, query);
            return;
        }

        bool admin_status_bad_or_forbidden = false;

        if (admin_state) {
            if (!user_conn) {
                // This is bad. The user connection (that was requesting that some
                // other connection be dropped) is gone
                query->status.description = "Parent connection no longer exists";
                qdra_connection_set_bad_request(query);
                admin_status_bad_or_forbidden = true;
            }
            else {
                bool allow = user_conn->policy_spec ? user_conn->policy_spec->allowAdminStatusUpdate : true;
                if (!allow) {
                    //
                    // Policy on the connection that is requesting that some other connection be deleted does not allow
                    // for the other connection to be deleted.Set the status to QD_AMQP_FORBIDDEN and just quit.
                    //
                    query->status = QD_AMQP_FORBIDDEN;
                    query->status.description = "You are not allowed to perform this operation.";
                    qd_compose_start_map(query->body);
                    qd_compose_end_map(query->body);
                    admin_status_bad_or_forbidden = true;
                 }
                else {
                    qdra_connection_update_set_status(core, query, conn, admin_state);
                }
            }

            if (admin_status_bad_or_forbidden) {
                //
                // Enqueue the response and return
                //
                qdr_agent_enqueue_response_CT(core, query);
                return;
            }
        }

        if (trace_field) {
            //
            // Trace logging needs to be turned on if enableProtocolTrace is true.
            // Trace logging needs to be turned off if enableProtocolTrace is false.
            //
            if (conn->enable_protocol_trace != enable_protocol_trace) {
                qdr_connection_work_type_t work_type = QDR_CONNECTION_WORK_TRACING_ON;
                conn->enable_protocol_trace = enable_protocol_trace;
                if (!enable_protocol_trace) {
                    work_type = QDR_CONNECTION_WORK_TRACING_OFF;
                }
                qdr_connection_work_t *work = new_qdr_connection_work_t();
                ZERO(work);
                work->work_type = work_type;
                qdr_connection_enqueue_work_CT(core, conn, work);

            }
            query->status = QD_AMQP_OK;
            qdr_manage_write_connection_map_CT(core, conn, query->body, qdr_connection_columns);
        }

    }    // if (qd_parse_is_map(in_body) && identity)
    else {
        qdra_connection_set_bad_request(query);
    }

    //
    // Enqueue the response.
    //
    qdr_agent_enqueue_response_CT(core, query);
}