iopsys-feed/dm-framework/dm-agent/src/dm_agent.c

/*
 * Copyright (c) 2023 Genexis B.V. All rights reserved.
 *
 * This Software and its content are protected by the Dutch Copyright Act
 * ('Auteurswet'). All and any copying and distribution of the software
 * and its content without authorization by Genexis B.V. is
 * prohibited. The prohibition includes every form of reproduction and
 * distribution.
 *
 */

#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include <time.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <libubus.h>
#include <libubox/blobmsg.h>
#include <libubox/blobmsg_json.h>
#include <libubox/list.h>
#include <libubox/uloop.h>
#include <libubox/utils.h>

#include "dm_api.h"
#include "dm_log.h"

#define METHOD_NAME "bbfdm.bridgemngr"
#define BBF_EVENT "bbfdm.event"
#define BBF_ASYNC_COMPLETE_EVENT "bbfdm.dmf.async_complete"
#define DEFAULT_TRANS_TIMEOUT 300*1000

struct agent_context {
	struct ubus_context ubus_ctx;
	struct list_head async_req_list;
	unsigned int last_async_id;
	unsigned int trans_id;
	struct uloop_timeout trans_timer;
};

struct async_operate_request {
	struct ubus_request_data req;
	unsigned int async_id;
	struct list_head list;
};

struct dm_ubus_event_handler {
    struct ubus_event_handler uhandler;
    dm_node_id_t node_id;
};

enum {
	GET_VALUE = 0,
	GET_NAME,
	GET_OBJ_NAME,
	GET_INSTANCE
};

enum instance_mode {
	INSTANCE_MODE_NUMBER,
	INSTANCE_MODE_ALIAS
};

enum dm_proto_type {
	DM_PROTO_CWMP,
	DM_PROTO_USP,
	DM_PROTO_BOTH
};

typedef struct {
	enum dm_proto_type dm_type;
	enum instance_mode instance_mode;
	bool is_raw;
	int trans_id;
	bool command;
	bool event;
	bool parameter;
	int maxdepth;
} bbf_options_t;

static struct agent_context agent_ctx;

// Helper function to check if a parameter is a key parameter
static int is_key_parameter(const dm_node_t *node)
{
	if (!dm_node_is_parameter(node->id)) {
		return 0;
	}

	// Get the parent object node
	dm_node_id_t parent_id = dm_node_i_parent_id(node->id);
	if (parent_id == 0) {
		return 0;
	}

	// Check if parent is an object list (multi-instance object)
	if (!dm_node_is_objectlist(parent_id)) {
		return 0;
	}

	// Get the keys for the parent object
	const char *keys = dm_node_object_keys(parent_id);
	if (!keys) {
		return 0;
	}

	// Get the parameter name from the node path
	dm_path_t node_path;
	dm_node2name(node, node_path, sizeof(dm_path_t));
	
	// Extract just the parameter name (last part after the last dot)
	char *param_name = strrchr(node_path, '.');
	if (!param_name) {
		return 0;
	}
	param_name++; // Skip the dot

	// Check if the parameter name is in the keys list
	char *keys_str = strdup(keys);
	char *key = strtok(keys_str, ",");
	while (key) {
		if (strcmp(key, param_name) == 0) {
			free(keys_str);
			return 1;
		}
		key = strtok(NULL, ",");
	}
	free(keys_str);

	return 0;
}

// return 1 if the proto is allowed for the node, otherwise return 0
static int check_dm_proto(int proto, dm_node_id_t node_id)
{
	if (dm_node_is_internal(node_id)) {
		return 0;
	}

	if (proto == DM_PROTO_BOTH) {
		return 1;
	}

	if (proto == DM_PROTO_CWMP) {
		if (dm_node_is_command(node_id)) {
			return 0;
		}

		if (dm_node_is_usp_only(node_id)) {
			return 0;
		}
	} else {
		// USP
		if (dm_node_is_cwmp_only(node_id)) {
			return 0;
		}
	}

	return 1;
}

static int get_dm_nodes(const bbf_options_t *opts, const dm_node_t *node,
	struct blob_buf *bb, int get_type, int depth)
{
	if (opts->maxdepth > 0 && depth > opts->maxdepth) {
		return 0;
	}

	if (!check_dm_proto(opts->dm_type, node->id)) {
		return 0;
	}

	if (dm_node_is_command(node->id) || dm_node_is_event(node->id)) {
		return 0;
	}

	if (dm_node_is_parameter(node->id)) {
		if (get_type == GET_INSTANCE)
			return 0;

		void *table = blobmsg_open_table(bb, NULL);

		dm_path_t node_path;
		dm_node2name(node, node_path, sizeof(dm_path_t));
		blobmsg_add_string(bb, "path", (char *)node_path);
		if (get_type != GET_OBJ_NAME) {
			char *value = NULL;
			if (dmapi_param_get(node, &value) < 0 || value == NULL) {
				blobmsg_add_string(bb, "data", "");
			} else {
				blobmsg_add_string(bb, "data", value);
				free(value);
			}
		} else {
			blobmsg_add_string(bb, "data", dm_node_is_writable(node->id) ? "1" : "0");
		}

		blobmsg_add_string(bb, "type", dm_node_get_param_xsd_type(node->id));
		
		// Add flags array for key parameters
		if (is_key_parameter(node)) {
			void *flags_array = blobmsg_open_array(bb, "flags");
			blobmsg_add_string(bb, NULL, "Unique");
			blobmsg_close_array(bb, flags_array);
		}
		
		blobmsg_close_table(bb, table);
	} else {
		const struct dm_object *obj = dm_node_get_object(node->id);
		if (obj == NULL)
			return 0;

		if (get_type == GET_NAME || get_type == GET_OBJ_NAME) {
			void *table = blobmsg_open_table(bb, NULL);
			dm_path_t node_path;
			dm_node2name(node, node_path, sizeof(dm_path_t));
			char last = node_path[strlen(node_path) - 1];
			if (last == '.') {
				// remove the {i}
				node_path[strlen(node_path) - 4] = '\0';
			} else {
				strcat(node_path, ".");
			}

			blobmsg_add_string(bb, "path", (char *)node_path);
			if (get_type == GET_OBJ_NAME) {
				blobmsg_add_string(bb, "data", dm_node_is_writable(node->id) ? "1" : "0");
			} else {
				blobmsg_add_string(bb, "data", "0");
			}

			blobmsg_add_string(bb, "type", "xsd:object");
			blobmsg_close_table(bb, table);
		}

		if (dm_node_is_index_complete(node)) {
			if (get_type == GET_INSTANCE && dm_node_is_objectlist(node->id)) {
				void *table = blobmsg_open_table(bb, NULL);
				dm_path_t node_path;
				dm_node2name(node, node_path, sizeof(dm_path_t));
				blobmsg_add_string(bb, "path", (char *)node_path);
				blobmsg_close_table(bb, table);
			}

			if (get_type != GET_INSTANCE) {
				for (int i = 0; i < obj->param_num; i++) {
					dm_node_t child = *node;
					const struct dm_node_info *info = obj->param_list[i];
					child.id = info->node_id;
					get_dm_nodes(opts, &child, bb, get_type, depth+1);
				}
			}

			for (int i = 0; i < obj->object_num; i++) {
				dm_node_t child = *node;
				const struct dm_node_info *info = obj->object_list[i];
				child.id = info->node_id;
				get_dm_nodes(opts, &child, bb, get_type, depth + 1);
			}
		} else {
			if (!dm_node_is_objectlist(node->id)) {
				dmlog_error("unexpected path");
				return -1;
			}
			dm_nodelist_h list = dm_nodelist_get(node);
			const dm_node_t *n = NULL;
			nodelist_for_each_node(n, list)
			{
				get_dm_nodes(opts, n, bb, get_type, depth);
			}
			dm_nodelist_free(list);
		}
	}

	return 0;
}

int get_supported_dm(const bbf_options_t *opts, dm_node_id_t id, struct blob_buf *bb)
{
	if (!check_dm_proto(opts->dm_type, id)) {
		return 0;
	}

	if (!opts->command && dm_node_is_command(id)) {
		return 0;
	}

	if (!opts->parameter && dm_node_is_parameter(id)) {
		return 0;
	}

	if (!opts->event && dm_node_is_event(id)) {
		return 0;
	}

	if (dm_node_is_parameter(id) || dm_node_is_objectlist(id) 
		|| dm_node_is_command(id) || dm_node_is_event(id)) {
		void *table = blobmsg_open_table(bb, NULL);
		dm_path_t node_path;
		dm_node_t node = {id};
		dm_node2name(&node, node_path, sizeof(dm_path_t));
		blobmsg_add_string(bb, "path", (char *)node_path);
		if (!dm_node_is_command(id) && !dm_node_is_event(id)) {
			blobmsg_add_string(bb, "data", dm_node_is_writable(id) ? "1" : "0");
		}
		const char *type;
		if (dm_node_is_objectlist(id)) {
			type = "xsd:object";
		} else if (dm_node_is_command(id)) {
			type = "xsd:command";
		} else if (dm_node_is_event(id)) {
			type = "xsd:event";
		} else {
			type = dm_node_get_param_xsd_type(id);
		}
		blobmsg_add_string(bb, "type", type);
		
		// Add flags array for key parameters
		if (dm_node_is_parameter(id) && is_key_parameter(&node)) {
			void *flags_array = blobmsg_open_array(bb, "flags");
			blobmsg_add_string(bb, NULL, "Unique");
			blobmsg_close_array(bb, flags_array);
		}
		// if (dm_node_is_objectlist(id)) {
		// 	const char *keys = dm_node_object_keys(id);
		// 	if (keys) {
		// 		void *array = blobmsg_open_array(bb, "input");
		// 		char *keys_str = strdup(keys);
		// 		char *key = strtok(keys_str, ",");
		// 		void *tbl = blobmsg_open_table(bb, NULL);
		// 		blobmsg_add_string(bb, "path", key);
		// 		blobmsg_close_table(bb, tbl);
		// 		while ((key = strtok(NULL, ","))) {
		// 			tbl = blobmsg_open_table(bb, NULL);
		// 			blobmsg_add_string(bb, "path", key);
		// 			blobmsg_close_table(bb, tbl);
		// 		}
		// 		blobmsg_close_table(bb, array);
		// 		free(keys_str);
		// 	}
		// }
		if (dm_node_is_command(id)) {
			const struct dm_command *cmd = dm_node_get_command(id);
			blobmsg_add_string(bb, "data", cmd->async ? "async" : "sync");
			if (cmd->inputs_cnt > 0) {
				void *array = blobmsg_open_array(bb, "input");
				for (int i = 0; i < cmd->inputs_cnt; i++) {
					void *tbl = blobmsg_open_table(bb, NULL);
					blobmsg_add_string(bb, "path", cmd->inputs[i].name);
					blobmsg_close_table(bb, tbl);
				}
				blobmsg_close_table(bb, array);
			}
			if (cmd->outputs_cnt > 0) {
				void *array = blobmsg_open_array(bb, "output");
				for (int i = 0; i < cmd->outputs_cnt; i++) {
					void *tbl = blobmsg_open_table(bb, NULL);
					blobmsg_add_string(bb, "path", cmd->outputs[i].name);
					blobmsg_close_table(bb, tbl);
				}
				blobmsg_close_table(bb, array);
			}
		} else if (dm_node_is_event(id)) {
			const struct dm_event *evt = dm_node_get_event(id);
			if (evt->args_cnt > 0) {
				void *array = blobmsg_open_array(bb, "input");
				for (int i = 0; i < evt->args_cnt; i++) {
					void *tbl = blobmsg_open_table(bb, NULL);
					blobmsg_add_string(bb, "path", evt->args[i].name);
					blobmsg_close_table(bb, tbl);
				}
				blobmsg_close_table(bb, array);
			}
		}
		blobmsg_close_table(bb, table);
	}

	if (dm_node_is_object(id) || dm_node_is_objectlist(id)) {
		const struct dm_object *obj = dm_node_get_object(id);
		int i;
		if (obj == NULL)
			return -1;

		for (i = 0; i < obj->param_num; i++) {
			const struct dm_node_info *info = obj->param_list[i];
			get_supported_dm(opts, info->node_id, bb);
		}

		for (i = 0; i < obj->object_num; i++) {
			const struct dm_node_info *info = obj->object_list[i];
			get_supported_dm(opts, info->node_id, bb);
		}

		for (i = 0; i < obj->command_num; i++) {
			const struct dm_node_info *info = obj->command_list[i];
			get_supported_dm(opts, info->node_id, bb);
		}

		for (i = 0; i < obj->event_num; i++) {
			const struct dm_node_info *info = obj->event_list[i];
			get_supported_dm(opts, info->node_id, bb);
		}
	}

	return 0;
}

enum {
	DM_SCHEMA_PATH,
	DM_SCHEMA_OPTIONAL,
	__DM_SCHEMA_MAX
};

static const struct blobmsg_policy dm_schema_policy[] = {
	[DM_SCHEMA_PATH] = { .name = "path", .type = BLOBMSG_TYPE_STRING },
	[DM_SCHEMA_OPTIONAL] = { .name = "optional", .type = BLOBMSG_TYPE_TABLE},
};

static int get_proto_type(struct blob_attr *proto)
{
	int type = DM_PROTO_BOTH;

	if (proto) {
		const char *val = blobmsg_get_string(proto);
		if (strcmp("cwmp", val) == 0)
			type = DM_PROTO_CWMP;
		else if (strcmp("usp", val) == 0)
			type = DM_PROTO_USP;
		else
			type = DM_PROTO_BOTH;
	}

	return type;
}

static int get_instance_mode(struct blob_attr *ins)
{
	int instance_mode = INSTANCE_MODE_NUMBER;

	if (ins)
		instance_mode = blobmsg_get_u32(ins);

	if (instance_mode > INSTANCE_MODE_ALIAS)
		instance_mode = INSTANCE_MODE_NUMBER;

	return instance_mode;
}

static void get_bbf_options(bbf_options_t *options, struct blob_attr *msg)
{
	struct blob_attr *attr;
	size_t rem;

	// set default
	options->dm_type = DM_PROTO_BOTH;
	options->instance_mode = INSTANCE_MODE_NUMBER;
	options->is_raw = true;
	options->command = true;
	options->event = true;
	options->parameter = true;

	blobmsg_for_each_attr(attr, msg, rem) {
		if (!blobmsg_name(attr)[0])
			continue;

		if (strcmp(blobmsg_name(attr), "proto") == 0)
			options->dm_type = get_proto_type(attr);

		if (strcmp(blobmsg_name(attr), "instance_mode") == 0)
			options->instance_mode = get_instance_mode(attr);

		if (strcmp(blobmsg_name(attr), "transaction_id") == 0)
			options->trans_id = blobmsg_get_u32(attr);

		if (strcmp(blobmsg_name(attr), "format") == 0)
			options->is_raw = strcmp(blobmsg_get_string(attr), "raw") == 0 ? true : false;

		if (strcmp(blobmsg_name(attr), "maxdepth") == 0)
			options->maxdepth = blobmsg_get_u32(attr);

		if (strcmp(blobmsg_name(attr), "commands") == 0)
			options->command = blobmsg_get_bool(attr);

		if (strcmp(blobmsg_name(attr), "events") == 0)
			options->event = blobmsg_get_bool(attr);

		if (strcmp(blobmsg_name(attr), "params") == 0)
			options->parameter = blobmsg_get_bool(attr);
	}
}

static int usp_schema_handler(struct ubus_context *ctx, struct ubus_object *obj __attribute__((unused)),
		    struct ubus_request_data *req, const char *method __attribute__((unused)),
		    struct blob_attr *msg)
{
	struct blob_attr *tb[__DM_SCHEMA_MAX];
	bbf_options_t options;
	memset(&options, 0, sizeof(bbf_options_t));
	struct blob_buf bb;
	dm_node_t node;

	memset(&bb, 0, sizeof(struct blob_buf));
	blob_buf_init(&bb, 0);

	if (blobmsg_parse(dm_schema_policy, __DM_SCHEMA_MAX, tb, blob_data(msg), blob_len(msg))) {
		dmlog_error("usp_schema_handler, failed to parse blob");
		return UBUS_STATUS_UNKNOWN_ERROR;
	}

	if (!tb[DM_SCHEMA_PATH]) {
		dmlog_error("usp_schema_handler, missing path parameter");
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	get_bbf_options(&options, tb[DM_SCHEMA_OPTIONAL]);
	unsigned int dm_type = options.dm_type;
	
	bool param_in_opts = false;
	bool cmd_in_opts = false;
	bool evt_in_opts = false;

	if (tb[DM_SCHEMA_OPTIONAL]) {
		struct blob_attr *attr;
		size_t rem;
		blobmsg_for_each_attr(attr, tb[DM_SCHEMA_OPTIONAL], rem)
		{
			if (strcmp(blobmsg_name(attr), "params") == 0)
				param_in_opts = true;
			if (strcmp(blobmsg_name(attr), "commands") == 0)
				cmd_in_opts = true;
			if (strcmp(blobmsg_name(attr), "events") == 0)
				evt_in_opts = true;
		}
	}

	if (!param_in_opts)
		options.parameter = true;

	if (!cmd_in_opts)
		options.command = (dm_type == DM_PROTO_CWMP) ? false : true;

	if (!evt_in_opts)
		options.event = (dm_type == DM_PROTO_CWMP) ? false : true;

	void *array = blobmsg_open_array(&bb, "results");
	const char *path = blobmsg_get_string(tb[DM_SCHEMA_PATH]);
	if (dm_path2node(path, &node) == 0) {
		if (options.dm_type == DM_PROTO_CWMP) {
			get_dm_nodes(&options, &node, &bb, GET_OBJ_NAME, 0);
		} else {
			get_supported_dm(&options, node.id, &bb);
		}
	} else {
		dmlog_error("usp_schema_handler, invalid path: %s", path);
	}

	blobmsg_close_array(&bb, array);
	ubus_send_reply(ctx, req, bb.head);
	blob_buf_free(&bb);
	return 0;
}

enum {
	DM_GET_PATH,
	DM_GET_OPTIONAL,
	__DM_GET_MAX
};

static const struct blobmsg_policy dm_get_policy[] = {
	[DM_GET_PATH] = { .name = "path", .type = BLOBMSG_TYPE_STRING },
	[DM_GET_OPTIONAL] = { .name = "optional", .type = BLOBMSG_TYPE_TABLE},
};

static int usp_get_handler(struct ubus_context *ctx, struct ubus_object *obj __attribute__((unused)),
		    struct ubus_request_data *req, const char *method __attribute__((unused)),
		    struct blob_attr *msg)
{
	struct blob_attr *tb[__DM_GET_MAX];
	bbf_options_t options;
	struct blob_buf bb;
	dm_node_t node;

	memset(&bb, 0, sizeof(struct blob_buf));
	blob_buf_init(&bb, 0);

	memset(&options, 0, sizeof(bbf_options_t));

	if (blobmsg_parse(dm_get_policy, __DM_GET_MAX, tb, blob_data(msg), blob_len(msg))) {
		dmlog_error("usp_get_handler, failed to parse blob");
		return UBUS_STATUS_UNKNOWN_ERROR;
	}

	if (!tb[DM_GET_PATH]) {
		dmlog_error("usp_get_handler, missing the path argument");
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	get_bbf_options(&options, tb[DM_GET_OPTIONAL]);

	void *array = blobmsg_open_array(&bb, "results");
	char *path_str = blobmsg_get_string(tb[DM_GET_PATH]);
	if (dm_path2node(path_str, &node) == 0) {
		get_dm_nodes(&options, &node, &bb, GET_VALUE, 0);
	} else {
		// dmlog_error("usp_get_handler, invalid path: %s", path_str);
	}

	blobmsg_close_array(&bb, array);
	ubus_send_reply(ctx, req, bb.head);
	blob_buf_free(&bb);
	return 0;
}

enum {
	DM_INSTANCES_PATH,
	DM_INSTANCES_OPTIONAL,
	__DM_INSTANCES_MAX
};

static const struct blobmsg_policy dm_instances_policy[] = {
	[DM_INSTANCES_PATH] = { .name = "path", .type = BLOBMSG_TYPE_STRING },
	[DM_INSTANCES_OPTIONAL] = { .name = "optional", .type = BLOBMSG_TYPE_TABLE },
};

static int usp_instances_handler(struct ubus_context *ctx, struct ubus_object *obj __attribute__((unused)),
		    struct ubus_request_data *req, const char *method __attribute__((unused)),
		    struct blob_attr *msg)
{
	struct blob_attr *tb[__DM_INSTANCES_MAX];
	bbf_options_t options;
	struct blob_buf bb;
	dm_node_t node;

	memset(&options, 0, sizeof(bbf_options_t));
	memset(&bb, 0, sizeof(struct blob_buf));
	blob_buf_init(&bb, 0);
	
	if (blobmsg_parse(dm_instances_policy, __DM_INSTANCES_MAX, tb, blob_data(msg), blob_len(msg))) {
		dmlog_debug("usp_instances_handler, failed to parse blob");
		return UBUS_STATUS_UNKNOWN_ERROR;
	}

	if (!tb[DM_INSTANCES_PATH]) {
		dmlog_debug("usp_instances_handler, missing the path");
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	get_bbf_options(&options, tb[DM_INSTANCES_OPTIONAL]);

	void *array = blobmsg_open_array(&bb, "results");
	char *path_str = blobmsg_get_string(tb[DM_INSTANCES_PATH]);
	if (dm_path2node(path_str, &node) == 0) {
		get_dm_nodes(&options, &node, &bb, GET_INSTANCE, 0);
	} else {
		dmlog_error("usp_get_handler, invalid path: %s", path_str);
	}

	blobmsg_close_array(&bb, array);
	ubus_send_reply(ctx, req, bb.head);
	blob_buf_free(&bb);
	return 0;
}

static int get_random_id(void)
{
	int ret;
	srand(time(0));
	ret = rand();
	if (!ret)
		ret = 1;

	return ret;
}

static void trans_timeout_handler(struct uloop_timeout *t __attribute__((unused)))
{
	agent_ctx.trans_id = 0;
	dmapi_session_end(TRANX_ROLLBACK);
}

static int start_trans(int max_timeout)
{
	if (dmapi_in_session()) {
		dmlog_error("failed to start transaction.");
		return -1;
	}

	dmapi_session_start();
	agent_ctx.trans_id = get_random_id();
	memset(&agent_ctx.trans_timer, 0, sizeof(agent_ctx.trans_timer));
	agent_ctx.trans_timer.cb = trans_timeout_handler;
	int timeout = DEFAULT_TRANS_TIMEOUT;
	if (max_timeout > 0) {
		timeout = max_timeout*1000;
	}
	uloop_timeout_set(&agent_ctx.trans_timer, timeout);
	return agent_ctx.trans_id;
}

static int commit_trans(int restart_service)
{
	int ret;
	if (restart_service) {
		ret = dmapi_session_end(TRANX_COMMIT_AND_APPLY);
	} else {
		ret = dmapi_session_end(TRANX_COMMIT);
	}
	uloop_timeout_cancel(&agent_ctx.trans_timer);
	agent_ctx.trans_id = 0;
	return ret;
}

static int abort_trans()
{
	uloop_timeout_cancel(&agent_ctx.trans_timer);
	agent_ctx.trans_id = 0;
	return dmapi_session_end(TRANX_ROLLBACK);
}

enum {
	TRANS_CMD,
	TRANS_OPTIONAL,
	__TRANS_MAX,
};

static const struct blobmsg_policy transaction_policy[] = {
	[TRANS_CMD] = { .name = "cmd", .type = BLOBMSG_TYPE_STRING },
	[TRANS_OPTIONAL] = { .name = "optional", .type = BLOBMSG_TYPE_TABLE },
};

static int usp_transaction_handler(struct ubus_context *ctx, struct ubus_object *obj,
			    struct ubus_request_data *req, const char *method,
			    struct blob_attr *msg)
{
	int ret;
	struct blob_attr *tb[__TRANS_MAX] = {NULL};
	bbf_options_t options;
	bool is_service_restart = true;
	uint32_t max_timeout = 0;
	char *trans_cmd = "status";
	struct blob_buf bb;

	memset(&options, 0, sizeof(bbf_options_t));
	memset(&bb, 0, sizeof(struct blob_buf));
	blob_buf_init(&bb, 0);

	if (blobmsg_parse(transaction_policy, __TRANS_MAX, tb, blob_data(msg), blob_len(msg))) {
		dmlog_debug("usp_transaction_handlerm, failed to parse blob");
		return UBUS_STATUS_UNKNOWN_ERROR;
	}

	if (!tb[TRANS_CMD]) {
		dmlog_error("missing transaction cmd");
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	if (tb[TRANS_CMD])
		trans_cmd = blobmsg_get_string(tb[TRANS_CMD]);

	get_bbf_options(&options, tb[TRANS_OPTIONAL]);

	struct blob_attr *attr;
	size_t rem;
	blobmsg_for_each_attr(attr, tb[TRANS_OPTIONAL], rem) {
		if (strcmp(blobmsg_name(attr), "timeout") == 0)
			max_timeout = blobmsg_get_u32(attr);
		if (strcmp(blobmsg_name(attr), "restart_services") == 0)
			is_service_restart = blobmsg_get_bool(attr);
	}

	if (strcmp(trans_cmd, "start") != 0 && strcmp(trans_cmd, "status") != 0 &&
		(agent_ctx.trans_id == 0 || agent_ctx.trans_id != options.trans_id)) {
		dmlog_error("invalid transaction id for cmd: %s. tid: %d", trans_cmd, options.trans_id);
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	if (strcmp(trans_cmd, "start") == 0) {
		ret = start_trans(max_timeout);
		// ret = transaction_start(max_timeout);
		if (ret > 0) {
			blobmsg_add_u8(&bb, "status", true);
			blobmsg_add_u32(&bb, "transaction_id", ret);
		} else {
			blobmsg_add_u8(&bb, "status", false);
		}
	} else if (strcmp(trans_cmd, "commit") == 0) {
		ret = commit_trans(is_service_restart);
		blobmsg_add_u8(&bb, "status", (ret == 0));
	} else if (strcmp(trans_cmd, "abort") == 0) {
		ret = abort_trans();
		blobmsg_add_u8(&bb, "status", (ret == 0));
	} else if (strcmp(trans_cmd, "status") == 0) {
		int64_t rem = uloop_timeout_remaining64(&agent_ctx.trans_timer);
		blobmsg_add_string(&bb, "status", "on-going");
		blobmsg_add_u32(&bb, "remaining_time", rem / 1000);
	} else {
		dmlog_error("unknown trans method(%s)", method);
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	ubus_send_reply(ctx, req, bb.head);
	blob_buf_free(&bb);
	return 0;
}

enum {
	DM_SET_PATH,
	DM_SET_VALUE,
	DM_SET_OBJ_PATH,
	DM_SET_OPTIONAL,
	__DM_SET_MAX,
};

static int set_blob_value(const dm_node_t *node, struct blob_attr *attr)
{
	int ret;
	if (blob_id(attr) == BLOBMSG_TYPE_STRING) {
		ret = dmapi_param_set(node, blobmsg_get_string(attr));
	} else {
		char * value = NULL;
		switch (blob_id(attr)) {
			case BLOBMSG_TYPE_INT8:
				asprintf(&value, "%d", blobmsg_get_u8(attr));
				break;
			case BLOBMSG_TYPE_INT16:
				asprintf(&value, "%d", blobmsg_get_u16(attr));
				break;
			case BLOBMSG_TYPE_INT32:
				asprintf(&value, "%u", blobmsg_get_u32(attr));
				break;
			case BLOBMSG_TYPE_INT64:
				asprintf(&value, "%"PRIu64"", blobmsg_get_u64(attr));
				break;
			default:
				dmlog_error("Unhandled set request type|%x|", blob_id(attr));
				return -1;
			}

			ret = dmapi_param_set(node, value);
			free(value);
	}

	return ret;
}

static void set_result(struct blob_buf *bb, const char *path, int fault, const char *info)
{
	void *table = blobmsg_open_table(bb, NULL);
	blobmsg_add_string(bb, "path", path);
	if (fault != 0) {
		blobmsg_add_string(bb, "data", "");
		blobmsg_add_u8(bb, "status", false);
		blobmsg_add_u32(bb, "fault", fault);
		blobmsg_add_string(bb, "fault_msg", info);
	} else {
		blobmsg_add_string(bb, "data", info);
		blobmsg_add_u8(bb, "status", true);
	}

	blobmsg_close_table(bb, table);
}

static const struct blobmsg_policy dm_set_policy[] = {
	[DM_SET_PATH] = { .name = "path", .type = BLOBMSG_TYPE_STRING },
	[DM_SET_VALUE] = { .name = "value", .type = BLOBMSG_TYPE_STRING },
	[DM_SET_OBJ_PATH] = { .name = "obj_path", .type = BLOBMSG_TYPE_TABLE },
	[DM_SET_OPTIONAL] = { .name = "optional", .type = BLOBMSG_TYPE_TABLE },
};

int usp_set_handler(struct ubus_context *ctx, struct ubus_object *obj,
	    struct ubus_request_data *req, const char *method,
	    struct blob_attr *msg)
{
	struct blob_attr *tb[__DM_SET_MAX] = {NULL};
	bbf_options_t opts;
	struct blob_buf bb;
	dm_node_t node;

	memset(&bb, 0, sizeof(struct blob_buf));
	blob_buf_init(&bb, 0);

	memset(&opts, 0, sizeof(opts));

	if (blobmsg_parse(dm_set_policy, __DM_SET_MAX, tb, blob_data(msg), blob_len(msg))) {
		dmlog_error("usp_set_handler, failed to parse blob");
		return UBUS_STATUS_UNKNOWN_ERROR;
	}

	if (!tb[DM_SET_PATH]) {
		dmlog_error("usp_set_handler, missing the path");
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	get_bbf_options(&opts, tb[DM_SET_OPTIONAL]);

	char *path_str = blobmsg_get_string(tb[DM_SET_PATH]);
	if (dm_path2node(path_str, &node) != 0 || dm_node_is_command(node.id) || dm_node_is_event(node.id)) {
		dmlog_error("usp_set_handler, invalid path %s", path_str);
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	int is_param = dm_node_is_parameter(node.id);
	if ((is_param && !tb[DM_SET_VALUE]) || (!is_param && !tb[DM_SET_OBJ_PATH])) {
			dmlog_error("usp_set_handler, missing value or obj_path argument");
			return UBUS_STATUS_INVALID_ARGUMENT;
	}

	int internal_trans = 0;
	int res = 0;
	if (agent_ctx.trans_id == 0) {
		internal_trans = 1;
		if (start_trans(0) < 0) {
			return UBUS_STATUS_UNKNOWN_ERROR;
		}
	} else if (opts.trans_id != agent_ctx.trans_id) {
		dmlog_error("in transaction or transaction id is invalid");
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	void *array = blobmsg_open_array(&bb, "results");
	if (dm_node_is_parameter(node.id)) {
		if (set_blob_value(&node, tb[DM_SET_VALUE]) != 0) {
			if (opts.dm_type == DM_PROTO_CWMP)
				set_result(&bb, path_str, 9007, "Invalid parameter value");
			else
				set_result(&bb, path_str, 7012, "Invalid value");
			res = -1;
		} else {
			set_result(&bb, path_str, 0, "1");
		}
	} else {
		struct blob_attr *attr;
		struct blobmsg_hdr *hdr;
		size_t tlen = (size_t)blobmsg_data_len(tb[DM_SET_OBJ_PATH]);

		__blob_for_each_attr(attr, blobmsg_data(tb[DM_SET_OBJ_PATH]), tlen) {
			hdr = blob_data(attr);
			char *path=NULL;
			dm_node_t param_node;
			asprintf(&path, "%s%s", path_str, (char *)hdr->name);
			if (dm_path2node(path, &param_node) != 0 || !dm_node_is_parameter(param_node.id)) {
				dmlog_error("usp_set_handler, invalid path %s", path);
				if (opts.dm_type == DM_PROTO_CWMP)
					set_result(&bb, path_str, 9005, "Invalid parameter name");
				else
					set_result(&bb, path_str, 7026, "Invalid path");
				res = -1;
			} else {
				if (set_blob_value(&param_node, attr) != 0) {
					if (opts.dm_type == DM_PROTO_CWMP)
						set_result(&bb, path_str, 9007, "Invalid parameter value");
					else
						set_result(&bb, path_str, 7012, "Invalid value");
					res = -1;
				} else {
					set_result(&bb, path_str, 0, "1");
				}
			}
			free(path);
			if (res < 0) {
				break;
			}
		}
	}

	if (internal_trans) {
		// Internal transaction: need to commit or revert the changes
		if (res == 0 && commit_trans(1) != 0) {
			dmlog_error("usp_set_handler, failed to commit");
		} else if (res < 0 && abort_trans() < 0) {
			dmlog_error("usp_set_handler, failed to abort");
		}
	}

	blobmsg_close_array(&bb, array);
	ubus_send_reply(ctx, req, bb.head);
	blob_buf_free(&bb);
	return 0;
}

enum {
	DM_ADD_PATH,
	DM_ADD_OBJ_PATH,
	DM_ADD_OPTIONAL,
	__DM_ADD_MAX
};

static const struct blobmsg_policy dm_add_policy[] = {
	[DM_ADD_PATH] = { .name = "path", .type = BLOBMSG_TYPE_STRING },
	[DM_ADD_OBJ_PATH] = { .name = "obj_path", .type = BLOBMSG_TYPE_TABLE },
	[DM_ADD_OPTIONAL] = { .name = "optional", .type = BLOBMSG_TYPE_TABLE },
};

int usp_add_handler(struct ubus_context *ctx, struct ubus_object *obj,
			struct ubus_request_data *req, const char *method,
			struct blob_attr *msg)
{
	struct blob_attr *tb[__DM_ADD_MAX];
	bbf_options_t opts;
	struct blob_buf bb;
	dm_node_t node;
	bool allow_partial = false;

	memset(&bb, 0, sizeof(struct blob_buf));
	blob_buf_init(&bb, 0);

	memset(&opts, 0, sizeof(opts));

	if (blobmsg_parse(dm_add_policy, __DM_ADD_MAX, tb, blob_data(msg), blob_len(msg))) {
		dmlog_error("usp_add_handler, failed to parse blob");
		return UBUS_STATUS_UNKNOWN_ERROR;
	}

	if (!tb[DM_ADD_PATH]) {
		dmlog_error("usp_add_handler, missing path");
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	get_bbf_options(&opts, tb[DM_ADD_OPTIONAL]);
	if (tb[DM_ADD_OPTIONAL]) {
		struct blob_attr *attr;
		size_t rem;
		blobmsg_for_each_attr(attr, tb[DM_ADD_OPTIONAL], rem) {
			if (strcmp(blobmsg_name(attr), "allow_partial") == 0)
				allow_partial = blobmsg_get_bool(attr);
		}
	}

	char *path_str = blobmsg_get_string(tb[DM_ADD_PATH]);
	if (dm_path2node(path_str, &node) != 0 || !dm_node_is_objectlist(node.id)) {
		dmlog_error("usp_add_handler, invalid path %s", path_str);
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	int internal_trans = 0;
	int res = 0;
	if (agent_ctx.trans_id == 0) {
		internal_trans = 1;
		if (start_trans(0) < 0) {
			return UBUS_STATUS_UNKNOWN_ERROR;
		}
	} else if (opts.trans_id != agent_ctx.trans_id) {
		dmlog_error("in transaction or transaction id is invalid");
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	void *array = blobmsg_open_array(&bb, "results");

	if (dmapi_object_add(&node) == 0) {
		dm_index_t idx = dm_node_last_index(&node);
		char *data = NULL;
		asprintf(&data, "%d", idx);
		set_result(&bb, path_str, 0, data);
		free(data);
	} else {
		res = -1;
		if (opts.dm_type == DM_PROTO_CWMP)
			set_result(&bb, path_str, 9005, "Invalid parameter name");
		else
			set_result(&bb, path_str, 7026, "Invalid path");
		blobmsg_close_array(&bb, array);
		goto end;
	}

	if (tb[DM_ADD_OBJ_PATH]) {
		struct blob_attr *attr;
		struct blobmsg_hdr *hdr;
		size_t tlen = (size_t)blobmsg_data_len(tb[DM_ADD_OBJ_PATH]);
		dm_path_t inst_path;
		dm_node2name(&node, inst_path, sizeof(inst_path));
		__blob_for_each_attr(attr, blobmsg_data(tb[DM_ADD_OBJ_PATH]), tlen) {
			hdr = blob_data(attr);
			char *path=NULL;
			dm_node_t param_node;
			asprintf(&path, "%s.%s", inst_path, (char *)hdr->name);
			if (dm_path2node(path, &param_node) != 0 || !dm_node_is_parameter(param_node.id)) {
				dmlog_error("usp_add_handler, invalid path %s", path);
				if (opts.dm_type == DM_PROTO_CWMP)
					set_result(&bb, path, 9005, "Invalid parameter name");
				else
					set_result(&bb, path, 7026, "Invalid path");
				res = -1;
			} else {
				if (set_blob_value(&param_node, attr) != 0) {
					if (opts.dm_type == DM_PROTO_CWMP)
						set_result(&bb, path, 9007, "Invalid parameter value");
					else
						set_result(&bb, path, 7012, "Invalid value");
					res = -1;
				} else {
					set_result(&bb, path, 0, "1");
				}
			}
			free(path);
			if (res < 0) {
				if (!allow_partial) {
					blobmsg_close_array(&bb, array);
					goto end;
				}
				res = 0; // clear error for partial
			}
		}
	}
	blobmsg_close_array(&bb, array);

end:
	if (internal_trans) {
		// Internal transaction: need to commit or revert the changes
		if (res == 0 && commit_trans(1) != 0) {
			dmlog_error("usp_add_handler, failed to commit");
		} else if (res < 0 && abort_trans() < 0) {
			dmlog_error("usp_add_handler, failed to abort");
		}
	}

	ubus_send_reply(ctx, req, bb.head);
	blob_buf_free(&bb);
	return 0;
}

enum {
	DM_DEL_PATH,
	DM_DEL_OPTIONAL,
	__DM_DEL_MAX
};

static const struct blobmsg_policy dm_del_policy[] = {
	[DM_DEL_PATH] = { .name = "path", .type = BLOBMSG_TYPE_STRING },
	[DM_DEL_OPTIONAL] = { .name = "optional", .type = BLOBMSG_TYPE_TABLE },
};

int usp_del_handler(struct ubus_context *ctx, struct ubus_object *obj,
			struct ubus_request_data *req, const char *method,
			struct blob_attr *msg)
{
	struct blob_attr *tb[__DM_DEL_MAX];
	bbf_options_t opts;
	struct blob_buf bb;
	dm_node_t node;

	memset(&bb, 0, sizeof(struct blob_buf));
	blob_buf_init(&bb, 0);
	memset(&opts, 0, sizeof(opts));

	if (blobmsg_parse(dm_del_policy, __DM_DEL_MAX, tb, blob_data(msg), blob_len(msg))) {
		dmlog_error("usp_del_handler, failed to parse blob");
		return UBUS_STATUS_UNKNOWN_ERROR;
	}

	if (!tb[DM_DEL_PATH]) {
		dmlog_error("usp_del_handler, missing path");
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	get_bbf_options(&opts, tb[DM_DEL_OPTIONAL]);

	int internal_trans = 0;
	int res = 0;
	if (agent_ctx.trans_id == 0) {
		internal_trans = 1;
		if (start_trans(0) < 0) {
			return UBUS_STATUS_UNKNOWN_ERROR;
		}
	} else if (opts.trans_id != agent_ctx.trans_id) {
		dmlog_error("in transaction or transaction id is invalid");
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	void *array = blobmsg_open_array(&bb, "results");
	char *path = blobmsg_get_string(tb[DM_DEL_PATH]);
	if (dm_path2node(path, &node) != 0 || dmapi_object_del(&node) < 0) {
		set_result(&bb, path, 9005, "Invalid path");
		res = -1;
		goto end;
	}
	set_result(&bb, path, 0, "1");

end:
	blobmsg_close_array(&bb, array);
	if (internal_trans) {
		// Internal transaction: need to commit or revert the changes
		if (res == 0 && commit_trans(1) != 0) {
			dmlog_error("usp_add_handler, failed to commit");
		} else if (res < 0 && abort_trans() < 0) {
			dmlog_error("usp_add_handler, failed to abort");
		}
	}

	ubus_send_reply(ctx, req, bb.head);
	blob_buf_free(&bb);
	return 0;
}

enum {
	DM_OPERATE_COMMAND,
	DM_OPERATE_OPTIONAL,
	__DM_OPERATE_MAX,
};

static const struct blobmsg_policy dm_operate_policy[__DM_OPERATE_MAX] = {
	[DM_OPERATE_COMMAND] = { .name = "command", .type = BLOBMSG_TYPE_STRING },
	[DM_OPERATE_OPTIONAL] = { .name = "optional", .type = BLOBMSG_TYPE_TABLE },
};

// check if mandatory name is present in the input
static int check_mandatory_input_name(struct blob_attr *input, const char *mandatory_name)
{
	if (input == NULL) {
		return 0;
	}

	struct blob_attr *head = blobmsg_data(input);
	int len = blobmsg_data_len(input);
	struct blob_attr *attr;
	__blob_for_each_attr(attr, head, len)
	{
		if (dm_node_compare_command_arg_name(blobmsg_name(attr), mandatory_name) == 0) {
			return 1;
		}
	}

	return 0;
}

// check if the input names and values are expected as defined in the datamodel
// return 1 if successful, otherwise 0
static int check_operate_inputs(const struct dm_command *cmd, struct blob_attr *input)
{
	// check mandatory input
	for (int i = 0; i < cmd->inputs_cnt; i++) {
		if (cmd->inputs[i].mandatory) {
			if (check_mandatory_input_name(input, cmd->inputs[i].name) != 1) {
				dmlog_error("mandatory input name not found: %s", cmd->inputs[i].name);
				return 0;
			}
		}
	}

	if (input == NULL) {
		return 1;
	}

	struct blob_attr *head = blobmsg_data(input);
	int len = blobmsg_data_len(input);
	struct blob_attr *attr;
	__blob_for_each_attr(attr, head, len)
	{
		const char *name = blobmsg_name(attr);
		if (strcmp(name, "Internal_TimeRef") == 0) {
			continue;
		}
		if (dm_node_verify_command_input(cmd->node.node_id, name, blobmsg_get_string(attr)) != 1) {
			dmlog_error("invalid argument value: %s: %s", name, blobmsg_get_string(attr));
			return 0;
		}
	}

	return 1;
}

void async_operate(struct ubus_context *ctx, const dm_node_t *node, const char *path, const char *cmd_key,
	char *input_json, struct ubus_request_data *req)
{
	agent_ctx.last_async_id++;
	pid_t child = fork();
	if (child == -1) {
		dmlog_error("fork failed");
	} else if (child == 0) {
		struct ubus_context *ctx = ubus_connect(NULL);
		struct json_object *json_output = NULL;
		struct blob_buf bb = {};
		memset(&bb, 0, sizeof(struct blob_buf));
		blob_buf_init(&bb, 0);

		blobmsg_add_u32(&bb, "async_id", agent_ctx.last_async_id);
		void *array = blobmsg_open_array(&bb, "results");
		void *table = blobmsg_open_table(&bb, NULL);
	
		if (dmapi_operate(node, input_json, &json_output) < 0) {
			dmlog_error("async_operate failed");
			set_result(&bb, path, 7022, "Operation failed");
		} else {
			blobmsg_add_string(&bb, "path", path);
			blobmsg_add_string(&bb, "data", cmd_key);
			if (json_output != NULL) {
				blobmsg_add_json_element(&bb, "output", json_output);
				json_object_put(json_output);
			}
			blobmsg_close_table(&bb, table);
		}

		blobmsg_close_array(&bb, array);
		ubus_send_event(ctx, BBF_ASYNC_COMPLETE_EVENT, bb.head);
		blob_buf_free(&bb);
		_exit(0);
	} else {
		struct async_operate_request *async_req = calloc(1, sizeof(struct async_operate_request));
		async_req->async_id = agent_ctx.last_async_id;
		ubus_defer_request(ctx, req, &async_req->req);
		list_add_tail(&async_req->list, &agent_ctx.async_req_list);
	}
}

static int contains_invalid_input_status(const struct dm_command *cmd)
{
	for (int i = 0; i < cmd->outputs_cnt; i++) {
		if (strcmp(cmd->outputs[i].name, "Status") == 0) {
			if (cmd->outputs[i].enum_values) {
				for (int j = 0; cmd->outputs[i].enum_values[j]; j++) {
					if (strcmp(cmd->outputs[i].enum_values[j], "Error_Invalid_Input") == 0) {
						return 1;
					}
				}
			}
		}
	}

	return 0;
}

static void set_invalid_input_status(struct blob_buf *bb, const char *path)
{
	void *table = blobmsg_open_table(bb, NULL);
	blobmsg_add_string(bb, "path", path);
	blobmsg_add_string(bb, "data", "");
	void *output = blobmsg_open_array(bb, "output");
	void *table2 = blobmsg_open_table(bb, NULL);
	blobmsg_add_string(bb, "path", "Status");
	blobmsg_add_string(bb, "data", "Error_Invalid_Input");
	blobmsg_add_string(bb, "type", "xsd:string");
	blobmsg_close_table(bb, table2);
	blobmsg_close_array(bb, output);
	blobmsg_close_table(bb, table);
}

static int usp_operate_handler(struct ubus_context *ctx, struct ubus_object *obj __attribute__((unused)),
		struct ubus_request_data *req, const char *method __attribute__((unused)),
		struct blob_attr *msg)
{
	struct blob_attr *tb[__DM_OPERATE_MAX] = {NULL};
	bbf_options_t opts;
	struct blob_buf bb;
	dm_node_t node;
	struct blob_attr *op_input = NULL;
	const char *cmd_key = "";

	memset(&bb, 0, sizeof(struct blob_buf));
	blob_buf_init(&bb, 0);
	memset(&opts, 0, sizeof(opts));

	if (blobmsg_parse(dm_operate_policy, __DM_OPERATE_MAX, tb, blob_data(msg), blob_len(msg))) {
		dmlog_error("usp_operate_handler, failed to parse blob");
		return UBUS_STATUS_UNKNOWN_ERROR;
	}

	if (!tb[DM_OPERATE_COMMAND]) {
		dmlog_error("usp_operate_handler, missing the operate pathname");
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	get_bbf_options(&opts, tb[DM_OPERATE_OPTIONAL]);

	if (tb[DM_OPERATE_OPTIONAL]) {
		struct blob_attr *attr;
		size_t rem;
		blobmsg_for_each_attr(attr, tb[DM_OPERATE_OPTIONAL], rem) {
			if (strcmp(blobmsg_name(attr), "command_key") == 0)
				cmd_key = blobmsg_get_string(attr);
			if (strcmp(blobmsg_name(attr), "input") == 0)
				op_input = attr;
		}
	}

	void *array = blobmsg_open_array(&bb, "results");
	char *path = blobmsg_get_string(tb[DM_OPERATE_COMMAND]);
	if (dm_path2node(path, &node) != 0 || !dm_node_is_command(node.id)) {
		dmlog_error("usp_operate_handler, invalid operate pathname %s", path);
		if (opts.dm_type == DM_PROTO_CWMP)
			set_result(&bb, path, 9005, "Invalid parameter name");
		else
			set_result(&bb, path, 7026, "Invalid path");
		goto end;
	}

	const struct dm_command *cmd = dm_node_get_command(node.id);
	if (check_operate_inputs(cmd, op_input) != 1) {
		dmlog_error("failed to verify input arguments");
		if (contains_invalid_input_status(cmd)) {
			set_invalid_input_status(&bb, path);
		} else {
			dmlog_error("send 7027");
			set_result(&bb, path, 7027, "Invalid command arguments");
		}
		goto end;
	}

	char *input_json = NULL;
	if (op_input) {
		input_json = blobmsg_format_json(op_input, true);
	}

	if (cmd->async) {
		async_operate(ctx, &node, path, cmd_key, input_json, req);
		blob_buf_free(&bb);
		if (input_json) {
			free(input_json);
		}
		return 0;
	}

	struct json_object *json_output = NULL;
	if (dmapi_operate(&node, input_json, &json_output) < 0) {
		dmlog_error("dmapi_operate failed");
		set_result(&bb, path, 7022, "Command failure");
		goto end;
	} else {
		void *table = blobmsg_open_table(&bb, NULL);
		blobmsg_add_string(&bb, "path", path);
		if (json_output != NULL) {
			blobmsg_add_string(&bb, "data", cmd_key);
			blobmsg_add_json_element(&bb, "output", json_output);
			json_object_put(json_output);
		}
		blobmsg_add_u8(&bb, "status", true);
		blobmsg_close_table(&bb, table);
	}

	if (input_json) {
		free(input_json);
	}

end:
	blobmsg_close_array(&bb, array);
	ubus_send_reply(ctx, req, bb.head);
	blob_buf_free(&bb);
	return 0;
}

enum {
	BBF_NOTIFY_NAME,
	BBF_NOTIFY_PRAMS,
	__BBF_NOTIFY_MAX,
};

static const struct blobmsg_policy dm_notify_event_policy[] = {
	[BBF_NOTIFY_NAME] = { .name = "name", .type = BLOBMSG_TYPE_STRING },
	[BBF_NOTIFY_PRAMS] = { .name = "parameters", .type = BLOBMSG_TYPE_TABLE },
};

static int usp_notify_event(struct ubus_context *ctx, struct ubus_object *obj,
			    struct ubus_request_data *req __attribute__((unused)), const char *method,
			    struct blob_attr *msg)
{
	struct blob_attr *tb[__BBF_NOTIFY_MAX] = {NULL};

	if (blobmsg_parse(dm_notify_event_policy, __BBF_NOTIFY_MAX, tb, blob_data(msg), blob_len(msg))) {
		dmlog_error("usp_notify_event,failed to parse blob");
		return UBUS_STATUS_UNKNOWN_ERROR;
	}

	if (!tb[BBF_NOTIFY_NAME]) {
		dmlog_error("usp_notify_event, missing notify name");
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	ubus_send_event(ctx, BBF_EVENT, msg);

	return 0;
}

static void async_operate_complete_event_handler(struct ubus_context *ctx, struct ubus_event_handler *ev,
                                                 const char *type, struct blob_attr *msg)
{
	(void)ev;
	struct blob_attr *tb[1] = {NULL};
	const struct blobmsg_policy policy[] = {
		[0] = {.name = "async_id", .type = BLOBMSG_TYPE_INT32}
	};

	if (blobmsg_parse(policy, 1, tb, blob_data(msg), blob_len(msg)) || !tb[0]) {
		dmlog_error("Failed to parse msg in async_operate_complete_event_handler");
		return;
	}

	unsigned int async_id = blobmsg_get_u32(tb[0]);
	struct async_operate_request *async_req;
	list_for_each_entry(async_req, &agent_ctx.async_req_list, list)
	{
		if (async_req->async_id == async_id) {
			ubus_send_reply(ctx, &async_req->req, msg);
			ubus_complete_deferred_request(ctx, &async_req->req, 0);
			list_del(&async_req->list);
			free(async_req);
			return;
		}
	}

	dmlog_error("unknown aysnc id from the notify %d", async_id);
}

static int register_async_event()
{
	printf("DEBUG: register_async_event() starting\n");
	struct ubus_event_handler *ev = (struct ubus_event_handler *)malloc(sizeof(struct ubus_event_handler));
	if (!ev) {
		printf("DEBUG: malloc failed in register_async_event()\n");
		return -1;
	}
	memset(ev, 0, sizeof(struct ubus_event_handler));
	ev->cb = async_operate_complete_event_handler;

	if (0 != ubus_register_event_handler(&agent_ctx.ubus_ctx, ev, BBF_ASYNC_COMPLETE_EVENT)) {
		printf("DEBUG: Failed to register ubus event usp_async_complete\n");
		dmlog_error("Failed to register ubus event usp_async_complete");
		return -1;
	}
	printf("DEBUG: register_async_event() completed successfully\n");

	return 0;
}

static void dm_event_handler(struct ubus_context *ctx, struct ubus_event_handler *ev,
                                                 const char *type, struct blob_attr *msg)
{
	struct dm_ubus_event_handler *myhandler = container_of(ev, struct dm_ubus_event_handler, uhandler);
	char *json_str = blobmsg_format_json(msg, true);
	dmlog_debug("dm_event_handler ubus event: %s, for node: %s, msg: %s", type, dm_node_id_str(myhandler->node_id), json_str);

	if (json_str == NULL) {
		return;
	}

	int len = strlen(json_str);
	if (len < 2) {
		return;
	}

	char * json_input = json_str;
	if (json_str[0] == '{' && json_str[1] == '{') {
		// object without name, which is invalid JSON format, remove the outer "{}".
		json_str[0] = '\0';
		json_str[len - 1] = '\0';
		json_input = json_str + 1;
	}
	struct json_object *res = NULL;
	dmapi_handle_ubus_event(myhandler->node_id, json_input, &res);
	if (json_str) {
		free(json_str);
	}
	if (!res) {
		dmlog_error("dm_event_handler, event not handled: %s", type);
		return;
	}
	const struct dm_event *evt_obj = dm_node_get_event(myhandler->node_id);
	if (!evt_obj) {
		dmlog_error("dm_event_handler, event not found: %d", myhandler->node_id);
		return;
	}

	struct blob_buf bb;
	memset(&bb, 0, sizeof(struct blob_buf));
	blob_buf_init(&bb, 0);

	blobmsg_add_string(&bb, "name", dm_node_id_str(myhandler->node_id));
	blobmsg_add_json_element(&bb, "input", res);
	ubus_send_event(ctx, "bbfdm.event", bb.head);
	json_object_put(res);
	blob_buf_free(&bb);
}

static void register_dm_event(const struct dm_object * obj)
{
	printf("DEBUG: register_dm_event() starting, obj=%p\n", obj);
	if (!obj) {
		printf("DEBUG: register_dm_event() obj is NULL\n");
		return;
	}
	
	printf("DEBUG: register_dm_event() event_num=%d\n", obj->event_num);
	for (int i = 0; i < obj->event_num; i++) {
		const struct dm_event * evt = (const struct dm_event*)obj->event_list[i];
		if (evt->ubus_event) {
			printf("DEBUG: registering ubus event: %s\n", evt->ubus_event);
			dmlog_debug("registerred ubus event: %s for node %s", evt->ubus_event, dm_node_id_str(evt->node.node_id));
			struct dm_ubus_event_handler *ev = (struct dm_ubus_event_handler *)malloc(sizeof(struct dm_ubus_event_handler));
			if (!ev) {
				printf("DEBUG: malloc failed for dm_ubus_event_handler\n");
				continue;
			}
			memset(ev, 0, sizeof(struct dm_ubus_event_handler));
			ev->uhandler.cb = dm_event_handler;
			ev->node_id = evt->node.node_id;
			if (0 != ubus_register_event_handler(&agent_ctx.ubus_ctx, &ev->uhandler, evt->ubus_event)) {
				printf("DEBUG: Failed to register ubus event %s\n", evt->ubus_event);
				dmlog_error("Failed to register ubus event %s", evt->ubus_event);
			}
		}
	}

	printf("DEBUG: register_dm_event() object_num=%d\n", obj->object_num);
	for (int i = 0; i < obj->object_num; i++) {
		const struct dm_object * child_obj = (const struct dm_object*)obj->object_list[i];
		register_dm_event(child_obj);
	}
	printf("DEBUG: register_dm_event() completed\n");
}


enum {
	DM_DUMP_BUF_PATH,
	__DM_DUMP_BUF_MAX
};

static const struct blobmsg_policy dm_dump_buf_policy[] = {
	[DM_DUMP_BUF_PATH] = { .name = "path", .type = BLOBMSG_TYPE_STRING },
};
static int usp_dump_buf_handler(struct ubus_context *ctx, struct ubus_object *obj __attribute__((unused)),
		    struct ubus_request_data *req, const char *method __attribute__((unused)),
		    struct blob_attr *msg)
{
	struct blob_attr *tb[__DM_DUMP_BUF_MAX];
	dm_node_t node;

	if (blobmsg_parse(dm_dump_buf_policy, __DM_DUMP_BUF_MAX, tb, blob_data(msg), blob_len(msg))) {
		dmlog_error("usp_dump_buf_handler, failed to parse blob");
		return UBUS_STATUS_UNKNOWN_ERROR;
	}

	if (!tb[DM_DUMP_BUF_PATH]) {
		dmlog_error("usp_dump_buf_handler, missing path parameter");
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	if (dm_path2node(blobmsg_get_string(tb[DM_DUMP_BUF_PATH]), &node) != 0) {
		dmlog_error("usp_dump_buf_handler, missing path parameter");
		return UBUS_STATUS_INVALID_ARGUMENT;
	}

	dmapi_dump_node_buffer(&node);

	return 0;
}

// just to be compatible with bbfdm
int bbfdm_refresh_references_db(struct ubus_context *ctx, struct ubus_object *obj,
			struct ubus_request_data *req, const char *method,
			struct blob_attr *msg)
{
	struct blob_buf bb = {0};

	memset(&bb, 0, sizeof(struct blob_buf));
	blob_buf_init(&bb, 0);

	blobmsg_add_u8(&bb, "status", true);

	ubus_send_reply(ctx, req, bb.head);
	blob_buf_free(&bb);

	return 0;
}

static struct ubus_method bbf_methods[] = {
	UBUS_METHOD("schema", usp_schema_handler, dm_schema_policy),
	UBUS_METHOD("get", usp_get_handler, dm_get_policy),
	UBUS_METHOD("instances", usp_instances_handler, dm_instances_policy),
	UBUS_METHOD("transaction", usp_transaction_handler, transaction_policy),
	UBUS_METHOD("set", usp_set_handler, dm_set_policy),
	UBUS_METHOD("add", usp_add_handler, dm_add_policy),
	UBUS_METHOD("del", usp_del_handler, dm_del_policy),
	UBUS_METHOD("operate", usp_operate_handler, dm_operate_policy),
	UBUS_METHOD("dump-buffer", usp_dump_buf_handler, dm_dump_buf_policy),
	UBUS_METHOD("notify_event", usp_notify_event, dm_notify_event_policy),
	UBUS_METHOD_NOARG("refresh_references_db", bbfdm_refresh_references_db)
};

static struct ubus_object_type bbf_type = UBUS_OBJECT_TYPE(METHOD_NAME, bbf_methods);

static struct ubus_object bbf_object = {
	.name = METHOD_NAME,
	.type = &bbf_type,
	.methods = bbf_methods,
	.n_methods = ARRAY_SIZE(bbf_methods)
};

static int usp_init(struct agent_context *u)
{
	int ret;

	printf("DEBUG: usp_init() starting\n");
	dmlog_debug("Registering bbfdm ubus objects....");
	ret = ubus_add_object(&u->ubus_ctx, &bbf_object);
	if (ret) {
		printf("DEBUG: ubus_add_object failed, ret=%d\n", ret);
		dmlog_error("ubus_add_object failed");
		return ret;
	}
	printf("DEBUG: usp_init() completed successfully\n");

	return ret;
}

int start_dm_agent()
{
	int ret = 0;
	const char *ubus_socket = NULL;

	printf("DEBUG: start_dm_agent() called\n");
	
	memset(&agent_ctx, 0, sizeof(struct agent_context));
	INIT_LIST_HEAD(&agent_ctx.async_req_list);
	printf("DEBUG: agent_ctx initialized\n");

	uloop_init();
	printf("DEBUG: uloop_init() completed\n");

	ret = ubus_connect_ctx(&agent_ctx.ubus_ctx, ubus_socket);
	if (ret != UBUS_STATUS_OK) {
		printf("DEBUG: Failed to connect to ubus, ret=%d\n", ret);
		dmlog_error("Failed to connect to ubus");
		return -1;
	}
	printf("DEBUG: ubus_connect_ctx() successful\n");

	printf("DEBUG: calling register_async_event()\n");
	register_async_event();
	printf("DEBUG: calling register_dm_event()\n");
	register_dm_event(dm_node_get_object(0));
	printf("DEBUG: calling ubus_add_uloop()\n");
	ubus_add_uloop(&agent_ctx.ubus_ctx);
	printf("DEBUG: calling usp_init()\n");
	ret = usp_init(&agent_ctx);
	if (ret != UBUS_STATUS_OK) {
		printf("DEBUG: usp_init failed, ret=%d\n", ret);
		dmlog_error("usp_init failed");
		ret = UBUS_STATUS_UNKNOWN_ERROR;
		goto exit;
	}
	printf("DEBUG: usp_init() successful\n");

	dmlog_info("dm-agent started");
	printf("DEBUG: entering uloop_run()\n");
	uloop_run();
	printf("DEBUG: uloop_run() exited\n");

exit:
	printf("DEBUG: cleaning up and exiting\n");
	ubus_shutdown(&agent_ctx.ubus_ctx);
	uloop_done();

	return ret;
}