From f597fd7e213c55c1f1a7377d13cda35616101c5c Mon Sep 17 00:00:00 2001
From: eeff <33566201+eeff@users.noreply.github.com>
Date: Tue, 12 Jul 2022 14:23:30 +0800
Subject: [PATCH 1/8] feat(persistence): migrate adapter persistence to sqlite

---
 CMakeLists.txt              |   2 +-
 src/core/manager_internal.c |   4 +-
 src/core/manager_internal.h |   4 +-
 src/core/storage.c          |  30 ++--
 src/persist/persist.c       | 286 +++++++++++++++---------------------
 src/persist/persist.h       |  69 ++++-----
 6 files changed, 166 insertions(+), 229 deletions(-)

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 915a4581..af823b52 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -121,7 +121,7 @@ set(CMAKE_BUILD_RPATH ./)
 add_executable(neuron)
 target_sources(neuron PRIVATE ${NEURON_SOURCES}) 
 target_include_directories(neuron PRIVATE include/neuron src plugins)
-target_link_libraries(neuron dl neuron-base)
+target_link_libraries(neuron dl neuron-base sqlite3)
 
 #copy file for run
 file(COPY ${CMAKE_SOURCE_DIR}/zlog.conf DESTINATION ${CMAKE_BINARY_DIR}/config)
diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c
index 4ac66047..891b7832 100644
--- a/src/core/manager_internal.c
+++ b/src/core/manager_internal.c
@@ -137,8 +137,8 @@ UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app)
     return neu_subscribe_manager_get(manager->subscribe_manager, app);
 }
 
-int neu_manager_get_adapter_info(neu_manager_t *manager, const char *name,
-                                 neu_persist_adapter_info_t *info)
+int neu_manager_get_node_info(neu_manager_t *manager, const char *name,
+                              neu_persist_node_info_t *info)
 {
     neu_adapter_t *adapter = neu_node_manager_find(manager->node_manager, name);
 
diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h
index ba5ba173..90b27c66 100644
--- a/src/core/manager_internal.h
+++ b/src/core/manager_internal.h
@@ -57,8 +57,8 @@ int       neu_manager_unsubscribe(neu_manager_t *manager, const char *app,
                                   const char *driver, const char *group);
 UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app);
 
-int neu_manager_get_adapter_info(neu_manager_t *manager, const char *name,
-                                 neu_persist_adapter_info_t *info);
+int neu_manager_get_node_info(neu_manager_t *manager, const char *name,
+                              neu_persist_node_info_t *info);
 
 void neu_manager_notify_app_sub(neu_manager_t *manager, const char *app,
                                 const char *driver, const char *group);
diff --git a/src/core/storage.c b/src/core/storage.c
index c298107b..53f66422 100644
--- a/src/core/storage.c
+++ b/src/core/storage.c
@@ -37,26 +37,26 @@ void manager_strorage_plugin(neu_manager_t *manager)
 
 void manager_storage_add_node(neu_manager_t *manager, const char *node)
 {
-    int                        rv           = 0;
-    neu_persist_adapter_info_t adapter_info = {};
+    int                     rv        = 0;
+    neu_persist_node_info_t node_info = {};
 
-    rv = neu_manager_get_adapter_info(manager, node, &adapter_info);
+    rv = neu_manager_get_node_info(manager, node, &node_info);
     if (0 != rv) {
         nlog_error("unable to get adapter:%s info", node);
         return;
     }
 
-    rv = neu_persister_store_adapter(manager->persister, &adapter_info);
+    rv = neu_persister_store_node(manager->persister, &node_info);
     if (0 != rv) {
         nlog_error("failed to store adapter info");
     }
 
-    neu_persist_adapter_info_fini(&adapter_info);
+    neu_persist_node_info_fini(&node_info);
 }
 
 void manager_storage_del_node(neu_manager_t *manager, const char *node)
 {
-    neu_persister_delete_adapter(manager->persister, node);
+    neu_persister_delete_node(manager->persister, node);
 }
 
 void manager_storage_subscribe(neu_manager_t *manager, const char *app)
@@ -97,26 +97,26 @@ int manager_load_plugin(neu_manager_t *manager)
 
 int manager_load_node(neu_manager_t *manager)
 {
-    UT_array *adapter_infos = NULL;
-    int       rv            = 0;
+    UT_array *node_infos = NULL;
+    int       rv         = 0;
 
-    rv = neu_persister_load_adapters(manager->persister, &adapter_infos);
+    rv = neu_persister_load_nodes(manager->persister, &node_infos);
     if (0 != rv) {
         nlog_error("failed to load adapter infos");
         return -1;
     }
 
-    utarray_foreach(adapter_infos, neu_persist_adapter_info_t *, adapter_info)
+    utarray_foreach(node_infos, neu_persist_node_info_t *, node_info)
     {
-        rv = neu_manager_add_node(manager, adapter_info->name,
-                                  adapter_info->plugin_name);
+        rv                    = neu_manager_add_node(manager, node_info->name,
+                                  node_info->plugin_name);
         const char *ok_or_err = (0 == rv) ? "success" : "fail";
         nlog_info("load adapter %s type:%" PRId64 ", name:%s plugin:%s",
-                  ok_or_err, adapter_info->type, adapter_info->name,
-                  adapter_info->plugin_name);
+                  ok_or_err, node_info->type, node_info->name,
+                  node_info->plugin_name);
     }
 
-    neu_persist_adapter_infos_free(adapter_infos);
+    utarray_free(node_infos);
     return rv;
 }
 
diff --git a/src/persist/persist.c b/src/persist/persist.c
index 52bd0605..b0a21a98 100644
--- a/src/persist/persist.c
+++ b/src/persist/persist.c
@@ -17,6 +17,7 @@
  * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  **/
 
+#include <stdarg.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -28,6 +29,8 @@
 #include <sys/types.h>
 #include <unistd.h>
 
+#include <sqlite3.h>
+
 #include "errcodes.h"
 #include "utils/log.h"
 
@@ -57,8 +60,8 @@ static inline bool ends_with(const char *str, const char *suffix)
 
 typedef struct neu_persister {
     const char *persist_dir;
-    const char *adapters_fname;
     const char *plugins_fname;
+    sqlite3 *   db;
 } neu_persister_t;
 
 const char *neu_persister_get_persist_dir(neu_persister_t *persister)
@@ -70,15 +73,6 @@ const char *neu_persister_get_persist_dir(neu_persister_t *persister)
     return persister->persist_dir;
 }
 
-const char *neu_persister_get_adapters_fname(neu_persister_t *persister)
-{
-    if (persister == NULL) {
-        return NULL;
-    }
-
-    return persister->adapters_fname;
-}
-
 const char *neu_persister_get_plugins_fname(neu_persister_t *persister)
 {
     if (persister == NULL) {
@@ -450,39 +444,6 @@ error_open:
     return rv;
 }
 
-// file tree walking callback for collecting adapter infos
-static int read_adapter_cb(const char *fpath, bool is_dir, void *arg)
-{
-    UT_array *adapter_infos = arg;
-    int       rv            = 0;
-
-    if (is_dir) {
-        return 0;
-    }
-
-    if (!ends_with(fpath, "adapter.json")) {
-        return 0;
-    }
-
-    char *json_str = NULL;
-    rv             = read_file_string(fpath, &json_str);
-    if (0 != rv) {
-        return rv;
-    }
-
-    neu_json_node_req_node_t *node = NULL;
-    rv = neu_json_decode_node_req_node(json_str, &node);
-    free(json_str);
-    if (0 != rv) {
-        return 0; // ignore bad adapter data
-    }
-
-    nlog_info("read %s", fpath);
-    utarray_push_back(adapter_infos, node);
-    free(node);
-    return rv;
-}
-
 // file tree walking callback for collecting adapter group config infos
 static int read_group_config_cb(const char *fpath, bool is_dir, void *arg)
 {
@@ -521,7 +482,6 @@ neu_persister_t *neu_persister_create(const char *dir_name)
 {
     int   rv                  = 0;
     char *persist_dir         = NULL;
-    char *adapters_fname      = NULL;
     char *plugins_fname       = NULL;
     char  path[PATH_MAX_SIZE] = { 0 };
 
@@ -553,17 +513,6 @@ neu_persister_t *neu_persister_create(const char *dir_name)
         goto error;
     }
 
-    n = path_cat(path, dir_len, sizeof(path), "adapters.json");
-    if (sizeof(path) == n) {
-        nlog_error("path too long: %s", path);
-        goto error;
-    }
-    adapters_fname = strdup(path);
-    if (NULL == adapters_fname) {
-        nlog_error("fail to strdup: %s", path);
-        goto error;
-    }
-
     n = path_cat(path, dir_len, sizeof(path), "plugins.json");
     if (sizeof(path) == n) {
         nlog_error("path too long: %s", path);
@@ -586,15 +535,33 @@ neu_persister_t *neu_persister_create(const char *dir_name)
         goto error;
     }
 
-    persister->plugins_fname  = plugins_fname;
-    persister->adapters_fname = adapters_fname;
-    persister->persist_dir    = persist_dir;
+    persister->plugins_fname = plugins_fname;
+    persister->persist_dir   = persist_dir;
+
+    n = path_cat(path, dir_len, sizeof(path), "sqlite.db");
+    if (sizeof(path) == n) {
+        nlog_error("path too long: %s", path);
+        goto error;
+    }
+    rv = sqlite3_open(path, &persister->db);
+    if (SQLITE_OK != rv) {
+        nlog_error("db `%s` fail: %s", path, sqlite3_errstr(rv));
+        goto error;
+    }
+    // we rely on sqlite foreign key support
+    if (SQLITE_OK !=
+        sqlite3_exec(persister->db, "PRAGMA foreign_keys=ON", NULL, NULL,
+                     NULL)) {
+        nlog_error("db foreign key support fail: %s",
+                   sqlite3_errmsg(persister->db));
+        sqlite3_close(persister->db);
+        goto error;
+    }
 
     return persister;
 
 error:
     free(plugins_fname);
-    free(adapters_fname);
     free(persist_dir);
     return NULL;
 }
@@ -653,139 +620,118 @@ static inline int persister_group_config_dir(char *buf, size_t size,
 
 void neu_persister_destroy(neu_persister_t *persister)
 {
-    free((char *) persister->adapters_fname);
+    sqlite3_close(persister->db);
     free((char *) persister->plugins_fname);
     free((char *) persister->persist_dir);
     free(persister);
 }
 
-static int migrate_adapters_file(neu_persister_t *persister)
+int neu_persister_sql_exec(neu_persister_t *persister, const char *sql, ...)
 {
-    char *json_str = NULL;
-    int   rv       = read_file_string(persister->adapters_fname, &json_str);
-    if (rv != 0) {
-        return 0;
+    int rv = 0;
+
+    va_list args;
+    va_start(args, sql);
+    char *query = sqlite3_vmprintf(sql, args);
+    va_end(args);
+
+    if (NULL == query) {
+        nlog_error("allocate SQL `%s` fail", query);
+        return NEU_ERR_EINTERNAL;
     }
 
-    neu_json_node_req_t *node_req = NULL;
-    rv = neu_json_decode_node_req(json_str, &node_req);
-    free(json_str);
-    json_str = NULL;
-    if (rv != 0) {
-        nlog_error("decode adapters json file fail");
-        return rv;
+    char *err_msg = NULL;
+    if (SQLITE_OK != sqlite3_exec(persister->db, query, NULL, NULL, &err_msg)) {
+        nlog_error("query `%s` fail: %s", query, err_msg);
+        rv = NEU_ERR_EINTERNAL;
     }
 
-    bool all_migrated = true;
-    for (int i = 0; i < node_req->n_node; ++i) {
-        neu_json_node_req_node_t *node = &node_req->nodes[i];
-        rv = neu_persister_store_adapter(persister, node);
-        if (0 != rv) {
-            nlog_error("migrate adapter %s fail", node->name);
-            all_migrated = false;
+    sqlite3_free(err_msg);
+    sqlite3_free(query);
+
+    return rv;
+}
+
+int neu_persister_store_node(neu_persister_t *        persister,
+                             neu_persist_node_info_t *info)
+{
+    return neu_persister_sql_exec(persister,
+                                  "INSERT INTO nodes (name, type, state, "
+                                  "plugin_name) VALUES (%Q, %i, %i, %Q)",
+                                  info->name, info->type, info->state,
+                                  info->plugin_name);
+}
+
+static UT_icd node_info_icd = {
+    sizeof(neu_persist_node_info_t),
+    NULL,
+    NULL,
+    (dtor_f *) neu_persist_node_info_fini,
+};
+
+int neu_persister_load_nodes(neu_persister_t *persister, UT_array **node_infos)
+{
+    int           rv    = 0;
+    sqlite3_stmt *stmt  = NULL;
+    const char *  query = "SELECT name, type, state, plugin_name FROM nodes;";
+
+    utarray_new(*node_infos, &node_info_icd);
+
+    if (SQLITE_OK !=
+        sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) {
+        nlog_error("prepare `%s` fail: %s", query,
+                   sqlite3_errmsg(persister->db));
+        utarray_free(*node_infos);
+        *node_infos = NULL;
+        return NEU_ERR_EINTERNAL;
+    }
+
+    int step = sqlite3_step(stmt);
+    while (SQLITE_ROW == step) {
+        neu_persist_node_info_t info = {};
+        char *name = strdup((char *) sqlite3_column_text(stmt, 0));
+        if (NULL == name) {
+            break;
         }
+
+        char *plugin_name = strdup((char *) sqlite3_column_text(stmt, 3));
+        if (NULL == plugin_name) {
+            free(name);
+            break;
+        }
+
+        info.name        = name;
+        info.type        = sqlite3_column_int(stmt, 1);
+        info.state       = sqlite3_column_int(stmt, 2);
+        info.plugin_name = plugin_name;
+        utarray_push_back(*node_infos, &info);
+
+        step = sqlite3_step(stmt);
     }
 
-    if (all_migrated) {
-        nlog_info("remove %s", persister->adapters_fname);
-        remove(persister->adapters_fname);
+    if (SQLITE_DONE != step) {
+        nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(persister->db));
+        // do not set return code, return partial or empty result
     }
 
-    neu_json_decode_node_req_free(node_req);
+    sqlite3_finalize(stmt);
     return rv;
 }
 
-int neu_persister_store_adapter(neu_persister_t *           persister,
-                                neu_persist_adapter_info_t *info)
+int neu_persister_delete_node(neu_persister_t *persister, const char *node_name)
 {
-    char path[PATH_MAX_SIZE] = { 0 };
-
-    int n = persister_adapter_dir(path, sizeof(path), persister, info->name);
-    if (sizeof(path) == n) {
-        nlog_error("persister path too long: %s", path);
-        return -1;
-    }
-    if (0 != create_dir_recursive(path)) {
-        nlog_error("persister failed to create dir: %s", path);
-        return -1;
-    }
-
-    n = path_cat(path, n, sizeof(path), "adapter.json");
-    if (sizeof(path) == n) {
-        nlog_error("persister path too long: %s", path);
-        return -1;
-    }
-
-    char *result = NULL;
-    int   rv =
-        neu_json_encode_by_fn(info, neu_json_encode_node_resp_node, &result);
-    if (rv != 0) {
-        return rv;
-    }
-
-    rv = write_file_string(path, result);
-
-    free(result);
-    return rv;
+    // rely on foreign key constraints to remove settings, groups, tags and
+    // subscriptions
+    return neu_persister_sql_exec(persister, "DELETE FROM nodes WHERE name=%Q;",
+                                  node_name);
 }
 
-int neu_persister_load_adapters(neu_persister_t *persister,
-                                UT_array **      adapter_infos)
+int neu_persister_update_node(neu_persister_t *persister, const char *node_name,
+                              const char *new_name)
 {
-    char   path[PATH_MAX_SIZE] = { 0 };
-    UT_icd icd = { sizeof(neu_persist_adapter_info_t), NULL, NULL, NULL };
-
-    migrate_adapters_file(persister);
-
-    if (sizeof(path) == persister_adapters_dir(path, sizeof(path), persister)) {
-        return -1;
-    }
-
-    utarray_new(*adapter_infos, &icd);
-
-    int rv = file_tree_walk(path, read_adapter_cb, *adapter_infos);
-    if (0 != rv) {
-        neu_persist_adapter_infos_free(*adapter_infos);
-    }
-
-    return rv;
-}
-
-int neu_persister_delete_adapter(neu_persister_t *persister,
-                                 const char *     adapter_name)
-{
-    char path[PATH_MAX_SIZE] = { 0 };
-    int  n = persister_adapter_dir(path, sizeof(path), persister, adapter_name);
-    if (sizeof(path) == n) {
-        nlog_error("persister path too long: %s", path);
-        return -1;
-    }
-    rmdir_recursive(path);
-    return 0;
-}
-
-int neu_persister_update_adapter(neu_persister_t *persister,
-                                 const char *adapter_name, const char *new_name)
-{
-    char path[PATH_MAX_SIZE] = { 0 };
-    int  n = persister_adapter_dir(path, sizeof(path), persister, adapter_name);
-    if (sizeof(path) == n) {
-        nlog_error("persister path too long: %s", path);
-        return -1;
-    }
-
-    char new_path[PATH_MAX_SIZE] = { 0 };
-    n = persister_adapter_dir(new_path, sizeof(new_path), persister, new_name);
-    if (sizeof(new_path) == n) {
-        nlog_error("persister new path too long: %s", new_path);
-        return -1;
-    }
-
-    if (0 != rename(path, new_path)) {
-        return -1;
-    }
-
-    return 0;
+    return neu_persister_sql_exec(persister,
+                                  "UPDATE nodes SET name=%Q WHERE name=%Q;",
+                                  new_name, node_name);
 }
 
 int neu_persister_store_plugins(neu_persister_t *persister,
diff --git a/src/persist/persist.h b/src/persist/persist.h
index 2e64c208..7c9c9ec4 100644
--- a/src/persist/persist.h
+++ b/src/persist/persist.h
@@ -25,42 +25,36 @@ extern "C" {
 #endif
 
 #include "adapter/adapter_info.h"
-#include "persist/json/persist_json_adapter.h"
 #include "persist/json/persist_json_datatags.h"
 #include "persist/json/persist_json_group_configs.h"
 #include "persist/json/persist_json_plugin.h"
 #include "persist/json/persist_json_subs.h"
 
-typedef neu_json_node_req_node_t     neu_persist_adapter_info_t;
 typedef neu_json_plugin_req_plugin_t neu_persist_plugin_info_t;
+
+typedef struct {
+    char *  name;
+    int64_t type;
+    char *  plugin_name;
+    int64_t state;
+} neu_persist_node_info_t;
+
 typedef neu_json_datatag_req_tag_t   neu_persist_datatag_info_t;
 typedef neu_json_group_configs_req_t neu_persist_group_config_info_t;
 typedef neu_json_subscriptions_req_subscription_t
     neu_persist_subscription_info_t;
 
-static inline void
-neu_persist_adapter_info_fini(neu_persist_adapter_info_t *adapter_info)
-{
-    free(adapter_info->name);
-    free(adapter_info->plugin_name);
-}
-
-static inline void neu_persist_adapter_infos_free(UT_array *adapter_infos)
-{
-    utarray_foreach(adapter_infos, neu_persist_adapter_info_t *, p)
-    {
-        free(p->name);
-        free(p->plugin_name);
-    }
-
-    utarray_free(adapter_infos);
-}
-
 static inline void neu_persist_plugin_infos_free(UT_array *plugin_infos)
 {
     utarray_free(plugin_infos);
 }
 
+static inline void neu_persist_node_info_fini(neu_persist_node_info_t *info)
+{
+    free(info->name);
+    free(info->plugin_name);
+}
+
 static inline void
 neu_persist_group_config_infos_free(UT_array *group_config_infos)
 {
@@ -95,7 +89,6 @@ neu_persist_subscription_infos_free(UT_array *subscription_infos)
 typedef struct neu_persister neu_persister_t;
 
 const char *neu_persister_get_persist_dir(neu_persister_t *persister);
-const char *neu_persister_get_adapters_fname(neu_persister_t *persister);
 const char *neu_persister_get_plugins_fname(neu_persister_t *persister);
 
 /**
@@ -110,41 +103,39 @@ neu_persister_t *neu_persister_create(const char *dir_name);
 void neu_persister_destroy(neu_persister_t *persiter);
 
 /**
- * Persist adapters.
+ * Persist nodes.
  * @param persister                 persiter object.
- * @param adapter_infos             neu_persist_adapter_info_t.
+ * @param node_info                 neu_persist_node_info_t.
  * @return 0 on success, non-zero on failure
  */
-int neu_persister_store_adapter(neu_persister_t *           persister,
-                                neu_persist_adapter_info_t *info);
+int neu_persister_store_node(neu_persister_t *        persister,
+                             neu_persist_node_info_t *info);
 /**
- * Load adapter infos.
+ * Load node infos.
  * @param persister                 persiter object.
- * @param[out] adapter_infos        used to return pointer to heap allocated
- * vector of neu_persist_adapter_info_t.
+ * @param[out] node_infos           used to return pointer to heap allocated
+ *                                  vector of neu_persist_node_info_t.
  * @return 0 on success, none-zero on failure
  */
-int neu_persister_load_adapters(neu_persister_t *persister,
-                                UT_array **      adapter_infos);
+int neu_persister_load_nodes(neu_persister_t *persister, UT_array **node_infos);
 /**
- * Delete adapter.
+ * Delete node.
  * @param persister                 persiter object.
- * @param adapter_name              name of the adapter to delete.
+ * @param node_name                 name of the node to delete.
  * @return 0 on success, none-zero on failure
  */
-int neu_persister_delete_adapter(neu_persister_t *persister,
-                                 const char *     adapter_name);
+int neu_persister_delete_node(neu_persister_t *persister,
+                              const char *     node_name);
 
 /**
- * Update adapter.
+ * Update node.
  * @param persister                 persiter object.
- * @param adapter_name              name of the adapter to update.
+ * @param node_name                 name of the node to update.
  * @param new_name                  new name of the adapter.
  * @return 0 on success, none-zero on failure
  */
-int neu_persister_update_adapter(neu_persister_t *persister,
-                                 const char *     adapter_name,
-                                 const char *     new_name);
+int neu_persister_update_node(neu_persister_t *persister, const char *node_name,
+                              const char *new_name);
 
 /**
  * Persist plugins.

From 4e85fab19a4524019c5450c229b130eb4616a090 Mon Sep 17 00:00:00 2001
From: eeff <33566201+eeff@users.noreply.github.com>
Date: Wed, 13 Jul 2022 11:00:01 +0800
Subject: [PATCH 2/8] feat(persistence): migrate setting persistence to sqlite

---
 src/adapter/storage.c |  6 +--
 src/persist/persist.c | 91 ++++++++++++++++++++++---------------------
 src/persist/persist.h | 30 +++++++-------
 3 files changed, 64 insertions(+), 63 deletions(-)

diff --git a/src/adapter/storage.c b/src/adapter/storage.c
index 2fc8edab..79b11a38 100644
--- a/src/adapter/storage.c
+++ b/src/adapter/storage.c
@@ -26,7 +26,7 @@
 void adapter_storage_setting(neu_persister_t *persister, const char *node,
                              const char *setting)
 {
-    neu_persister_store_adapter_setting(persister, node, setting);
+    neu_persister_store_node_setting(persister, node, setting);
 }
 
 void adapter_storage_add_group(neu_persister_t *persister, const char *node,
@@ -70,8 +70,8 @@ void adapter_storage_tag(neu_persister_t *persister, neu_adapter_t *adapter,
 int adapter_load_setting(neu_persister_t *persister, const char *node,
                          char **setting)
 {
-    int rv = neu_persister_load_adapter_setting(persister, node,
-                                                (const char **) setting);
+    int rv = neu_persister_load_node_setting(persister, node,
+                                             (const char **) setting);
     if (0 != rv) {
         nlog_info("load %s setting fail", node);
         return -1;
diff --git a/src/persist/persist.c b/src/persist/persist.c
index b0a21a98..0de12587 100644
--- a/src/persist/persist.c
+++ b/src/persist/persist.c
@@ -1077,60 +1077,61 @@ int neu_persister_delete_group_config(neu_persister_t *persister,
     return rv;
 }
 
-int neu_persister_store_adapter_setting(neu_persister_t *persister,
-                                        const char *     adapter_name,
-                                        const char *     setting)
+int neu_persister_store_node_setting(neu_persister_t *persister,
+                                     const char *node_name, const char *setting)
 {
-    char path[PATH_MAX_SIZE] = { 0 };
+    return neu_persister_sql_exec(
+        persister,
+        "INSERT OR REPLACE INTO settings (node_name, setting) VALUES (%Q, %Q)",
+        node_name, setting);
+}
 
-    int n = persister_adapter_dir(path, sizeof(path), persister, adapter_name);
-    if (sizeof(path) == n) {
-        nlog_error("persister path too long: %s", path);
-        return -1;
-    }
-    if (0 != create_dir(path)) {
-        nlog_error("persister failed to create dir: %s", path);
-        return -1;
+int neu_persister_load_node_setting(neu_persister_t *  persister,
+                                    const char *       node_name,
+                                    const char **const setting)
+{
+    int           rv    = 0;
+    sqlite3_stmt *stmt  = NULL;
+    const char *  query = "SELECT setting FROM settings WHERE node_name=?";
+
+    if (SQLITE_OK !=
+        sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) {
+        nlog_error("prepare `%s` with `%s` fail: %s", query, node_name,
+                   sqlite3_errmsg(persister->db));
+        return NEU_ERR_EINTERNAL;
     }
 
-    n = path_cat(path, n, sizeof(path), "settings.json");
-    if (sizeof(path) == n) {
-        nlog_error("persister path too long: %s", path);
-        return -1;
+    if (SQLITE_OK != sqlite3_bind_text(stmt, 1, node_name, -1, NULL)) {
+        nlog_error("bind `%s` with `%s` fail: %s", query, node_name,
+                   sqlite3_errmsg(persister->db));
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
     }
 
-    int rv = write_file_string(path, setting);
+    if (SQLITE_ROW != sqlite3_step(stmt)) {
+        nlog_warn("SQL `%s` with `%s` fail: %s", query, node_name,
+                  sqlite3_errmsg(persister->db));
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
+    }
 
+    char *s = strdup((char *) sqlite3_column_text(stmt, 0));
+    if (NULL == s) {
+        nlog_error("strdup fail");
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
+    }
+
+    *setting = s;
+
+end:
+    sqlite3_finalize(stmt);
     return rv;
 }
 
-int neu_persister_load_adapter_setting(neu_persister_t *  persister,
-                                       const char *       adapter_name,
-                                       const char **const setting)
+int neu_persister_delete_node_setting(neu_persister_t *persister,
+                                      const char *     node_name)
 {
-    char path[PATH_MAX_SIZE] = { 0 };
-
-    int n = persister_adapter_dir(path, sizeof(path), persister, adapter_name);
-    if (sizeof(path) == n) {
-        nlog_error("persister path too long: %s", path);
-        return -1;
-    }
-
-    n = path_cat(path, n, sizeof(path), "settings.json");
-    if (sizeof(path) == n) {
-        nlog_error("persister path too long: %s", path);
-        return -1;
-    }
-
-    int rv = read_file_string(path, (char **) setting);
-
-    return rv;
-}
-
-int neu_persister_delete_adapter_setting(neu_persister_t *persister,
-                                         const char *     adapter_name)
-{
-    (void) persister;
-    (void) adapter_name;
-    return 0;
+    return neu_persister_sql_exec(
+        persister, "DELETE FROM settings WHERE node_name=%Q", node_name);
 }
diff --git a/src/persist/persist.h b/src/persist/persist.h
index 7c9c9ec4..f6ed6260 100644
--- a/src/persist/persist.h
+++ b/src/persist/persist.h
@@ -241,33 +241,33 @@ int neu_persister_delete_group_config(neu_persister_t *persister,
                                       const char *     group_config_name);
 
 /**
- * Persist adapter setting.
+ * Persist node setting.
  * @param persister                 persiter object.
  * @param adapter_name              name of the adapter who owns the setting.
- * @param setting                   adapter setting string.
+ * @param setting                   node setting string.
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_store_adapter_setting(neu_persister_t *persister,
-                                        const char *     adapter_name,
-                                        const char *     setting);
+int neu_persister_store_node_setting(neu_persister_t *persister,
+                                     const char *     node_name,
+                                     const char *     setting);
 /**
- * Load adapter setting.
+ * Load node setting.
  * @param persister                 persiter object.
- * @param adapter_name              name of the adapter.
- * @param[out] setting              used to return adapter setting string.
+ * @param node_name                 name of the node.
+ * @param[out] setting              used to return node setting string.
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_load_adapter_setting(neu_persister_t *  persister,
-                                       const char *       adapter_name,
-                                       const char **const setting);
+int neu_persister_load_node_setting(neu_persister_t *  persister,
+                                    const char *       node_name,
+                                    const char **const setting);
 /**
- * Delete adapter setting.
+ * Delete node setting.
  * @param persister                 persiter object.
- * @param adapter_name              name of the adapter.
+ * @param node_name                 name of the node.
  * @return 0 on success, none-zero on failure
  */
-int neu_persister_delete_adapter_setting(neu_persister_t *persister,
-                                         const char *     adapter_name);
+int neu_persister_delete_node_setting(neu_persister_t *persister,
+                                      const char *     node_name);
 
 #ifdef __cplusplus
 }

From dc8e69fea0311ebcb6738f965a4ef403c3f7aa1c Mon Sep 17 00:00:00 2001
From: eeff <33566201+eeff@users.noreply.github.com>
Date: Wed, 13 Jul 2022 14:43:27 +0800
Subject: [PATCH 3/8] feat(persistence): migrate group persistence to sqlite

---
 src/adapter/storage.c |  30 +++--
 src/persist/persist.c | 262 ++++++++++--------------------------------
 src/persist/persist.h |  52 ++++-----
 3 files changed, 97 insertions(+), 247 deletions(-)

diff --git a/src/adapter/storage.c b/src/adapter/storage.c
index 79b11a38..b11e90a2 100644
--- a/src/adapter/storage.c
+++ b/src/adapter/storage.c
@@ -32,19 +32,18 @@ void adapter_storage_setting(neu_persister_t *persister, const char *node,
 void adapter_storage_add_group(neu_persister_t *persister, const char *node,
                                const char *group, uint32_t interval)
 {
-    neu_persist_group_config_info_t info = {
-        .group_config_name = (char *) group,
-        .read_interval     = interval,
-        .adapter_name      = (char *) node,
+    neu_persist_group_info_t info = {
+        .name     = (char *) group,
+        .interval = interval,
     };
 
-    neu_persister_store_group_config(persister, node, &info);
+    neu_persister_store_group(persister, node, &info);
 }
 
 void adapter_storage_del_group(neu_persister_t *persister, const char *node,
                                const char *group)
 {
-    neu_persister_delete_group_config(persister, node, group);
+    neu_persister_delete_group(persister, node, group);
 }
 
 void adapter_storage_tag(neu_persister_t *persister, neu_adapter_t *adapter,
@@ -86,35 +85,32 @@ int adapter_load_group_and_tag(neu_persister_t *     persister,
     UT_array *     group_infos = NULL;
     neu_adapter_t *adapter     = (neu_adapter_t *) driver;
 
-    int rv = neu_persister_load_group_configs(persister, adapter->name,
-                                              &group_infos);
+    int rv = neu_persister_load_groups(persister, adapter->name, &group_infos);
     if (0 != rv) {
         nlog_warn("load %s group fail", adapter->name);
         return rv;
     }
 
-    utarray_foreach(group_infos, neu_persist_group_config_info_t *, p)
+    utarray_foreach(group_infos, neu_persist_group_info_t *, p)
     {
         UT_array *tags = NULL;
-        neu_adapter_driver_add_group(driver, p->group_config_name,
-                                     p->read_interval);
+        neu_adapter_driver_add_group(driver, p->name, p->interval);
 
-        rv = neu_persister_load_datatags(persister, adapter->name,
-                                         p->group_config_name, &tags);
+        rv = neu_persister_load_datatags(persister, adapter->name, p->name,
+                                         &tags);
         if (0 != rv) {
-            nlog_warn("load %s:%s tags fail", adapter->name,
-                      p->group_config_name);
+            nlog_warn("load %s:%s tags fail", adapter->name, p->name);
             continue;
         }
 
         utarray_foreach(tags, neu_datatag_t *, tag)
         {
-            neu_adapter_driver_add_tag(driver, p->group_config_name, tag);
+            neu_adapter_driver_add_tag(driver, p->name, tag);
         }
 
         neu_persist_datatag_infos_free(tags);
     }
 
-    neu_persist_group_config_infos_free(group_infos);
+    utarray_free(group_infos);
     return rv;
 }
\ No newline at end of file
diff --git a/src/persist/persist.c b/src/persist/persist.c
index 0de12587..30366af9 100644
--- a/src/persist/persist.c
+++ b/src/persist/persist.c
@@ -230,108 +230,6 @@ static int create_dir_recursive(const char *path)
     return rv;
 }
 
-/**
- * Type of callback invoked in file tree walking.
- */
-typedef int (*file_tree_walk_cb_t)(const char *fpath, bool is_dir, void *arg);
-
-// depth first file tree walking implementation.
-// `path_buf` is buffer for accumulating file path,
-// `len` is the length of the directory path of the current invocation,
-// `size` is the path buffer size.
-static int file_tree_walk_(char *path_buf, size_t len, size_t size,
-                           file_tree_walk_cb_t cb, void *ctx)
-{
-    int            rv = 0;
-    DIR *          dirp;
-    struct dirent *dent;
-
-    // if not possible to read the directory for this user
-    if ((dirp = opendir(path_buf)) == NULL) {
-        rv = -1;
-        return rv;
-    }
-
-    while (NULL != (dent = readdir(dirp))) {
-        if (DT_DIR == dent->d_type) {
-            // skip entries "." and ".."
-            if (!strcmp(dent->d_name, ".") || !strcmp(dent->d_name, "..")) {
-                continue;
-            }
-        } else if (DT_REG != dent->d_type) {
-            continue;
-        }
-
-        // determinate a full path of an entry
-        size_t n = path_cat(path_buf, len, size, dent->d_name);
-        if (size == n) {
-            rv = -1;
-            break;
-        }
-
-        if (DT_DIR == dent->d_type) {
-            rv = file_tree_walk_(path_buf, n, size, cb, ctx);
-        } else {
-            rv = cb(path_buf, false, ctx);
-        }
-
-        path_buf[len] = '\0'; // restore current directory path
-
-        if (0 != rv) {
-            break;
-        }
-    }
-
-    if (0 == rv) {
-        rv = cb(path_buf, true, ctx);
-    }
-
-    closedir(dirp);
-    return rv;
-}
-
-/**
- * Depth first traversal of file tree.
- *
- * @param dir       path of directory
- * @param cb        callback to invoke for each directory entry
- * @param ctx       argument to callback
- *
- * @return  If `cb` returns nonzero, then the tree walk is terminated and the
- *          value returned by `cb` is returned as the result
- */
-static int file_tree_walk(const char *dir, file_tree_walk_cb_t cb, void *ctx)
-{
-    char buf[PATH_MAX_SIZE] = { 0 };
-    int  n                  = path_cat(buf, 0, sizeof(buf), dir);
-    if (sizeof(buf) == n) {
-        return -1;
-    }
-    return file_tree_walk_(buf, n, sizeof(buf), cb, ctx);
-}
-
-// file tree walking callback for `rmdir_recursive`
-static int remove_cb(const char *fpath, bool is_dir, void *arg)
-{
-    (void) arg;
-
-    if (!is_dir && !ends_with(fpath, ".json")) {
-        // we are being defensive here, ignoring non-json files
-        return 0;
-    }
-    nlog_info("remove %s", fpath);
-    return remove(fpath);
-}
-
-/**
- * Remove directory recursively.
- * @param dir         path of directory to remove
- */
-static int rmdir_recursive(const char *dir)
-{
-    return file_tree_walk(dir, remove_cb, NULL);
-}
-
 /**
  * Ensure that a file with the given name exists.
  *
@@ -444,40 +342,6 @@ error_open:
     return rv;
 }
 
-// file tree walking callback for collecting adapter group config infos
-static int read_group_config_cb(const char *fpath, bool is_dir, void *arg)
-{
-    UT_array *group_config_infos = arg;
-    int       rv                 = 0;
-
-    if (is_dir) {
-        return 0;
-    }
-
-    if (!ends_with(fpath, "config.json")) {
-        return 0;
-    }
-
-    char *json_str = NULL;
-    rv             = read_file_string(fpath, &json_str);
-    if (0 != rv) {
-        return rv;
-    }
-
-    neu_json_group_configs_req_t *group_config_req = NULL;
-    rv = neu_json_decode_group_configs_req(json_str, &group_config_req);
-    free(json_str);
-    if (0 != rv) {
-        return 0; // ignore bad group config data
-    }
-
-    nlog_info("read %s", fpath);
-    utarray_push_back(group_config_infos, group_config_req);
-    free(group_config_req);
-
-    return rv;
-}
-
 neu_persister_t *neu_persister_create(const char *dir_name)
 {
     int   rv                  = 0;
@@ -999,82 +863,80 @@ int neu_persister_load_subscriptions(neu_persister_t *persister,
     return 0;
 }
 
-int neu_persister_store_group_config(
-    neu_persister_t *persister, const char *adapter_name,
-    neu_persist_group_config_info_t *group_config_info)
+int neu_persister_store_group(neu_persister_t *         persister,
+                              const char *              driver_name,
+                              neu_persist_group_info_t *group_info)
 {
-    char path[PATH_MAX_SIZE] = { 0 };
-
-    int n =
-        persister_group_config_dir(path, sizeof(path), persister, adapter_name,
-                                   group_config_info->group_config_name);
-    if (sizeof(path) == n) {
-        nlog_error("persister path too long: %s", path);
-        return -1;
-    }
-
-    int rv = create_dir_recursive(path);
-    if (0 != rv) {
-        nlog_error("persister failed to create dir: %s", path);
-        return rv;
-    }
-
-    n = path_cat(path, n, sizeof(path), "config.json");
-    if (sizeof(path) == n) {
-        nlog_error("persister path too long: %s", path);
-        return -1;
-    }
-
-    char *result = NULL;
-    rv           = neu_json_encode_by_fn(group_config_info,
-                               neu_json_encode_group_configs_resp, &result);
-    if (rv == 0) {
-        rv = write_file_string(path, result);
-    }
-
-    free(result);
-    return rv;
+    return neu_persister_sql_exec(
+        persister,
+        "INSERT INTO groups (driver_name, name, interval) VALUES (%Q, %Q, %i)",
+        driver_name, group_info->name, group_info->interval);
 }
 
-int neu_persister_load_group_configs(neu_persister_t *persister,
-                                     const char *     adapter_name,
-                                     UT_array **      group_config_infos)
+static UT_icd group_info_icd = {
+    sizeof(neu_persist_group_info_t),
+    NULL,
+    NULL,
+    (dtor_f *) neu_persist_group_info_fini,
+};
+
+int neu_persister_load_groups(neu_persister_t *persister,
+                              const char *driver_name, UT_array **group_infos)
 {
-    char   path[PATH_MAX_SIZE] = { 0 };
-    UT_icd icd = { sizeof(neu_persist_group_config_info_t), NULL, NULL, NULL };
+    sqlite3_stmt *stmt = NULL;
+    const char *query = "SELECT name, interval FROM groups WHERE driver_name=?";
 
-    int n = persister_group_configs_dir(path, sizeof(path), persister,
-                                        adapter_name);
-    if (sizeof(path) == n) {
-        nlog_error("persister path too long: %s", path);
-        return -1;
+    utarray_new(*group_infos, &group_info_icd);
+
+    if (SQLITE_OK !=
+        sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) {
+        nlog_error("prepare `%s` fail: %s", query,
+                   sqlite3_errmsg(persister->db));
+        goto error;
     }
 
-    utarray_new(*group_config_infos, &icd);
-
-    int rv = file_tree_walk(path, read_group_config_cb, *group_config_infos);
-    if (0 != rv) {
-        neu_persist_group_config_infos_free(*group_config_infos);
+    if (SQLITE_OK != sqlite3_bind_text(stmt, 1, driver_name, -1, NULL)) {
+        nlog_error("bind `%s` with `%s` fail: %s", query, driver_name,
+                   sqlite3_errmsg(persister->db));
+        goto error;
     }
 
-    return rv;
+    int step = sqlite3_step(stmt);
+    while (SQLITE_ROW == step) {
+        neu_persist_group_info_t info = {};
+        char *name = strdup((char *) sqlite3_column_text(stmt, 0));
+        if (NULL == name) {
+            break;
+        }
+
+        info.name     = name;
+        info.interval = sqlite3_column_int(stmt, 1);
+        utarray_push_back(*group_infos, &info);
+
+        step = sqlite3_step(stmt);
+    }
+
+    if (SQLITE_DONE != step) {
+        nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(persister->db));
+        // do not set return code, return partial or empty result
+    }
+
+    sqlite3_finalize(stmt);
+    return 0;
+
+error:
+    utarray_free(*group_infos);
+    *group_infos = NULL;
+    return NEU_ERR_EINTERNAL;
 }
 
-int neu_persister_delete_group_config(neu_persister_t *persister,
-                                      const char *     adapter_name,
-                                      const char *     group_config_name)
+int neu_persister_delete_group(neu_persister_t *persister,
+                               const char *driver_name, const char *group_name)
 {
-    char path[PATH_MAX_SIZE] = { 0 };
-
-    int n = persister_group_config_dir(path, sizeof(path), persister,
-                                       adapter_name, group_config_name);
-    if (sizeof(path) == n) {
-        nlog_error("persister path too long: %s", path);
-        return -1;
-    }
-
-    int rv = rmdir_recursive(path);
-    return rv;
+    // rely on foreign key constraints to delete tags and subscriptions
+    return neu_persister_sql_exec(
+        persister, "DELETE FROM groups WHERE driver_name=%Q AND name=%Q",
+        driver_name, group_name);
 }
 
 int neu_persister_store_node_setting(neu_persister_t *persister,
diff --git a/src/persist/persist.h b/src/persist/persist.h
index f6ed6260..764710a5 100644
--- a/src/persist/persist.h
+++ b/src/persist/persist.h
@@ -26,7 +26,6 @@ extern "C" {
 
 #include "adapter/adapter_info.h"
 #include "persist/json/persist_json_datatags.h"
-#include "persist/json/persist_json_group_configs.h"
 #include "persist/json/persist_json_plugin.h"
 #include "persist/json/persist_json_subs.h"
 
@@ -40,7 +39,12 @@ typedef struct {
 } neu_persist_node_info_t;
 
 typedef neu_json_datatag_req_tag_t   neu_persist_datatag_info_t;
-typedef neu_json_group_configs_req_t neu_persist_group_config_info_t;
+
+typedef struct {
+    uint32_t interval;
+    char *   name;
+} neu_persist_group_info_t;
+
 typedef neu_json_subscriptions_req_subscription_t
     neu_persist_subscription_info_t;
 
@@ -55,16 +59,9 @@ static inline void neu_persist_node_info_fini(neu_persist_node_info_t *info)
     free(info->plugin_name);
 }
 
-static inline void
-neu_persist_group_config_infos_free(UT_array *group_config_infos)
+static inline void neu_persist_group_info_fini(neu_persist_group_info_t *info)
 {
-    utarray_foreach(group_config_infos, neu_persist_group_config_info_t *, p)
-    {
-        free(p->group_config_name);
-        free(p->adapter_name);
-    }
-
-    utarray_free(group_config_infos);
+    free(info->name);
 }
 
 static inline void neu_persist_datatag_infos_free(UT_array *datatag_infos)
@@ -208,37 +205,32 @@ int neu_persister_load_subscriptions(neu_persister_t *persister,
 /**
  * Persist group config.
  * @param persister                 persiter object.
- * @param adapter_name              name of the adapter who owns the group
- * config.
- * @param group_config_info         group config info to persist.
+ * @param driver_name               name of the driver who owns the group
+ * @param group_info                group info to persist.
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_store_group_config(
-    neu_persister_t *persister, const char *adapter_name,
-    neu_persist_group_config_info_t *group_config_info);
+int neu_persister_store_group(neu_persister_t *         persister,
+                              const char *              driver_name,
+                              neu_persist_group_info_t *group_info);
 /**
  * Load all group config infos under an adapter.
  * @param persister                 persiter object.
- * @param adapter_name              name of the adapter who owns the group
- * configs.
- * @param[out] group_config_infos   used to return pointer to heap allocated
- * vector of neu_persist_group_config_info_t.
+ * @param driver_name               name of the driver who owns the group
+ * @param[out] group_infos          used to return pointer to heap allocated
+ *                                  vector of neu_persist_group_info_t.
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_load_group_configs(neu_persister_t *persister,
-                                     const char *     adapter_name,
-                                     UT_array **      group_config_infos);
+int neu_persister_load_groups(neu_persister_t *persister,
+                              const char *driver_name, UT_array **group_infos);
 /**
  * Delete group config.
  * @param persister                 persiter object.
- * @param adapter_name              name of the adapter who owns the group
- * config.
- * @param group_config_name         name of the group config.
+ * @param driver_name               name of the driver who owns the group
+ * @param group_name                name of the group.
  * @return 0 on success, none-zero on failure
  */
-int neu_persister_delete_group_config(neu_persister_t *persister,
-                                      const char *     adapter_name,
-                                      const char *     group_config_name);
+int neu_persister_delete_group(neu_persister_t *persister,
+                               const char *driver_name, const char *group_name);
 
 /**
  * Persist node setting.

From 384f231da5d840ed8c6399180edd3fccdf56e542 Mon Sep 17 00:00:00 2001
From: eeff <33566201+eeff@users.noreply.github.com>
Date: Thu, 14 Jul 2022 11:31:23 +0800
Subject: [PATCH 4/8] feat(persistence): migrate tags persistence to sqlite

---
 src/adapter/adapter.c |  29 +++---
 src/adapter/storage.c |  42 +++++----
 src/adapter/storage.h |   8 +-
 src/persist/persist.c | 206 ++++++++++++++++--------------------------
 src/persist/persist.h |  63 ++++++++-----
 5 files changed, 158 insertions(+), 190 deletions(-)

diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c
index de0078b6..ef9dc621 100644
--- a/src/adapter/adapter.c
+++ b/src/adapter/adapter.c
@@ -466,8 +466,12 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
             error.error = NEU_ERR_GROUP_NOT_ALLOW;
         } else {
             for (int i = 0; i < cmd->n_tag; i++) {
-                neu_adapter_driver_del_tag((neu_adapter_driver_t *) adapter,
-                                           cmd->group, cmd->tags[i]);
+                int ret = neu_adapter_driver_del_tag(
+                    (neu_adapter_driver_t *) adapter, cmd->group, cmd->tags[i]);
+                if (0 == ret) {
+                    adapter_storage_del_tag(adapter->persister, cmd->driver,
+                                            cmd->group, cmd->tags[i]);
+                }
             }
         }
 
@@ -476,11 +480,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
         }
         free(cmd->tags);
 
-        if (error.error == NEU_ERR_SUCCESS) {
-            adapter_storage_tag(adapter->persister, adapter, cmd->driver,
-                                cmd->group);
-        }
-
         neu_msg_exchange(header);
         header->type = NEU_RESP_ERROR;
         reply(adapter, header, &error);
@@ -498,7 +497,10 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
                     neu_adapter_driver_add_tag((neu_adapter_driver_t *) adapter,
                                                cmd->group, &cmd->tags[i]);
                 if (ret == 0) {
+                    adapter_storage_add_tag(adapter->persister, cmd->driver,
+                                            cmd->group, &cmd->tags[i]);
                     resp.index += 1;
+
                 } else {
                     resp.error = ret;
                     break;
@@ -512,11 +514,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
         }
         free(cmd->tags);
 
-        if (resp.index > 0) {
-            adapter_storage_tag(adapter->persister, adapter, cmd->driver,
-                                cmd->group);
-        }
-
         neu_msg_exchange(header);
         header->type = NEU_RESP_ADD_TAG;
         reply(adapter, header, &resp);
@@ -534,6 +531,9 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
                     (neu_adapter_driver_t *) adapter, cmd->group,
                     &cmd->tags[i]);
                 if (ret == 0) {
+                    adapter_storage_update_tag(adapter->persister, cmd->driver,
+                                               cmd->group, &cmd->tags[i]);
+
                     resp.index += 1;
                 } else {
                     resp.error = ret;
@@ -548,11 +548,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
         }
         free(cmd->tags);
 
-        if (resp.index > 0) {
-            adapter_storage_tag(adapter->persister, adapter, cmd->driver,
-                                cmd->group);
-        }
-
         neu_msg_exchange(header);
         header->type = NEU_RESP_UPDATE_TAG;
         reply(adapter, header, &resp);
diff --git a/src/adapter/storage.c b/src/adapter/storage.c
index b11e90a2..07c54674 100644
--- a/src/adapter/storage.c
+++ b/src/adapter/storage.c
@@ -46,24 +46,33 @@ void adapter_storage_del_group(neu_persister_t *persister, const char *node,
     neu_persister_delete_group(persister, node, group);
 }
 
-void adapter_storage_tag(neu_persister_t *persister, neu_adapter_t *adapter,
-                         const char *node, const char *group)
+void adapter_storage_add_tag(neu_persister_t *persister, const char *node,
+                             const char *group, const neu_datatag_t *tag)
 {
-    UT_array *tags = NULL;
-
-    int rv = neu_adapter_driver_get_tag((neu_adapter_driver_t *) adapter, group,
-                                        &tags);
-    if (rv != 0) {
-        nlog_warn("fail get tag infos");
-        return;
-    }
-
-    rv = neu_persister_store_datatags(persister, node, group, tags);
+    int rv = neu_persister_store_tag(persister, node, group, tag);
     if (0 != rv) {
-        nlog_error("fail store adapter:%s grp:%s datatag infos", node, group);
+        nlog_error("fail store tag:%s adapter:%s grp:%s", tag->name, node,
+                   group);
     }
+}
 
-    neu_persist_datatag_infos_free(tags);
+void adapter_storage_update_tag(neu_persister_t *persister, const char *node,
+                                const char *group, const neu_datatag_t *tag)
+{
+    int rv = neu_persister_update_tag(persister, node, group, tag);
+    if (0 != rv) {
+        nlog_error("fail update tag:%s adapter:%s grp:%s", tag->name, node,
+                   group);
+    }
+}
+
+void adapter_storage_del_tag(neu_persister_t *persister, const char *node,
+                             const char *group, const char *name)
+{
+    int rv = neu_persister_delete_tag(persister, node, group, name);
+    if (0 != rv) {
+        nlog_error("fail del tag:%s adapter:%s grp:%s", name, node, group);
+    }
 }
 
 int adapter_load_setting(neu_persister_t *persister, const char *node,
@@ -96,8 +105,7 @@ int adapter_load_group_and_tag(neu_persister_t *     persister,
         UT_array *tags = NULL;
         neu_adapter_driver_add_group(driver, p->name, p->interval);
 
-        rv = neu_persister_load_datatags(persister, adapter->name, p->name,
-                                         &tags);
+        rv = neu_persister_load_tags(persister, adapter->name, p->name, &tags);
         if (0 != rv) {
             nlog_warn("load %s:%s tags fail", adapter->name, p->name);
             continue;
@@ -108,7 +116,7 @@ int adapter_load_group_and_tag(neu_persister_t *     persister,
             neu_adapter_driver_add_tag(driver, p->name, tag);
         }
 
-        neu_persist_datatag_infos_free(tags);
+        utarray_free(tags);
     }
 
     utarray_free(group_infos);
diff --git a/src/adapter/storage.h b/src/adapter/storage.h
index e09b0ea1..24549e58 100644
--- a/src/adapter/storage.h
+++ b/src/adapter/storage.h
@@ -30,8 +30,12 @@ void adapter_storage_add_group(neu_persister_t *persister, const char *node,
                                const char *group, uint32_t interval);
 void adapter_storage_del_group(neu_persister_t *persister, const char *node,
                                const char *group);
-void adapter_storage_tag(neu_persister_t *persister, neu_adapter_t *adapter,
-                         const char *node, const char *group);
+void adapter_storage_add_tag(neu_persister_t *persister, const char *node,
+                             const char *group, const neu_datatag_t *tag);
+void adapter_storage_update_tag(neu_persister_t *persister, const char *node,
+                                const char *group, const neu_datatag_t *tag);
+void adapter_storage_del_tag(neu_persister_t *persister, const char *node,
+                             const char *group, const char *name);
 
 int adapter_load_setting(neu_persister_t *persister, const char *node,
                          char **setting);
diff --git a/src/persist/persist.c b/src/persist/persist.c
index 30366af9..46aa8474 100644
--- a/src/persist/persist.c
+++ b/src/persist/persist.c
@@ -194,42 +194,6 @@ static inline int create_dir(char *dir_name)
     return rv;
 }
 
-/**
- * Create a directory and any intermediate directory if necessary.
- */
-static int create_dir_recursive(const char *path)
-{
-    int  rv               = 0;
-    char p[PATH_MAX_SIZE] = { 0 };
-
-    rv = snprintf(p, sizeof(p), "%s", path);
-    if (sizeof(p) == rv) {
-        return -1;
-    }
-
-    int i = 0;
-    while (1) {
-        while (p[i] && PATH_SEP_CHAR == p[i]) {
-            ++i;
-        }
-        if ('\0' == p[i]) {
-            break;
-        }
-        while (p[i] && PATH_SEP_CHAR != p[i]) {
-            ++i;
-        }
-
-        char c = p[i];
-        p[i]   = '\0';
-        rv     = create_dir(p);
-        if (0 != rv) {
-            break;
-        }
-        p[i] = c;
-    }
-    return rv;
-}
-
 /**
  * Ensure that a file with the given name exists.
  *
@@ -657,109 +621,93 @@ int neu_persister_load_plugins(neu_persister_t *persister,
     return 0;
 }
 
-int neu_persister_store_datatags(neu_persister_t *persister,
-                                 const char *     adapter_name,
-                                 const char *     group_config_name,
-                                 UT_array *       datatag_infos)
+int neu_persister_store_tag(neu_persister_t *persister, const char *driver_name,
+                            const char *group_name, const neu_datatag_t *tag)
 {
-    char path[PATH_MAX_SIZE] = { 0 };
-
-    int n = persister_group_config_dir(path, sizeof(path), persister,
-                                       adapter_name, group_config_name);
-    if (sizeof(path) == n) {
-        nlog_error("path too long: %s", path);
-        return -1;
-    }
-
-    int rv = create_dir_recursive(path);
-    if (0 != rv) {
-        nlog_error("fail to create dir: %s", path);
-        return -1;
-    }
-
-    n = path_cat(path, n, sizeof(path), "datatags.json");
-    if (sizeof(path) == n) {
-        nlog_error("path too long: %s", path);
-        return -1;
-    }
-
-    neu_json_datatag_req_t datatags_resp = { 0 };
-    datatags_resp.n_tag                  = utarray_len(datatag_infos);
-    datatags_resp.tags =
-        calloc(datatags_resp.n_tag, sizeof(neu_json_datatag_req_tag_t));
-    int index = 0;
-    utarray_foreach(datatag_infos, neu_datatag_t *, tag)
-    {
-        datatags_resp.tags[index].name        = tag->name;
-        datatags_resp.tags[index].description = tag->description;
-        datatags_resp.tags[index].address     = tag->address;
-        datatags_resp.tags[index].attribute   = tag->attribute;
-        datatags_resp.tags[index].type        = tag->type;
-        index += 1;
-    }
-
-    char *result = NULL;
-    rv = neu_json_encode_by_fn(&datatags_resp, neu_json_encode_datatag_resp,
-                               &result);
-
-    if (rv == 0) {
-        write_file_string(path, result);
-    }
-
-    free(result);
-    free(datatags_resp.tags);
-
-    return rv;
+    return neu_persister_sql_exec(
+        persister,
+        "INSERT INTO tags (driver_name, group_name, name, address, attribute, "
+        "type, description) VALUES(%Q, %Q, %Q, %Q, %i, %i, %Q)",
+        driver_name, group_name, tag->name, tag->address, tag->attribute,
+        tag->type, tag->description);
 }
 
-int neu_persister_load_datatags(neu_persister_t *persister,
-                                const char *     adapter_name,
-                                const char *group_config_name, UT_array **tags)
+int neu_persister_load_tags(neu_persister_t *persister, const char *driver_name,
+                            const char *group_name, UT_array **tags)
 {
-    char path[PATH_MAX_SIZE] = { 0 };
-
-    int n = persister_group_config_dir(path, sizeof(path), persister,
-                                       adapter_name, group_config_name);
-    if (sizeof(path) == n) {
-        nlog_error("path too long: %s", path);
-        return -1;
-    }
-
-    n = path_cat(path, n, sizeof(path), "datatags.json");
-    if (sizeof(path) == n) {
-        nlog_error("path too long: %s", path);
-        return -1;
-    }
-
-    char *json_str = NULL;
-    int   rv       = read_file_string(path, &json_str);
-    if (rv != 0) {
-        return rv;
-    }
-
-    neu_json_datatag_req_t *datatag_req = NULL;
-    rv = neu_json_decode_datatag_req(json_str, &datatag_req);
-    free(json_str);
-    if (rv != 0) {
-        return rv;
-    }
+    sqlite3_stmt *stmt  = NULL;
+    const char *  query = "SELECT name, address, attribute, type, description "
+                        "FROM tags WHERE driver_name=? AND group_name=?";
 
     utarray_new(*tags, neu_tag_get_icd());
-    for (int i = 0; i < datatag_req->n_tag; i++) {
-        neu_datatag_t tag = {
-            .name        = datatag_req->tags[i].name,
-            .address     = datatag_req->tags[i].address,
-            .type        = datatag_req->tags[i].type,
-            .attribute   = datatag_req->tags[i].attribute,
-            .description = datatag_req->tags[i].description,
-        };
 
-        utarray_push_back(*tags, &tag);
+    if (SQLITE_OK !=
+        sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) {
+        nlog_error("prepare `%s` fail: %s", query,
+                   sqlite3_errmsg(persister->db));
+        goto error;
     }
 
-    neu_json_decode_datatag_req_free(datatag_req);
+    if (SQLITE_OK != sqlite3_bind_text(stmt, 1, driver_name, -1, NULL)) {
+        nlog_error("bind `%s` with `%s` fail: %s", query, driver_name,
+                   sqlite3_errmsg(persister->db));
+        goto error;
+    }
 
-    return rv;
+    if (SQLITE_OK != sqlite3_bind_text(stmt, 2, group_name, -1, NULL)) {
+        nlog_error("bind `%s` with `%s` fail: %s", query, group_name,
+                   sqlite3_errmsg(persister->db));
+        goto error;
+    }
+
+    int step = sqlite3_step(stmt);
+    while (SQLITE_ROW == step) {
+        neu_datatag_t tag = {
+            .name        = (char *) sqlite3_column_text(stmt, 0),
+            .address     = (char *) sqlite3_column_text(stmt, 1),
+            .attribute   = sqlite3_column_int(stmt, 2),
+            .type        = sqlite3_column_int(stmt, 3),
+            .description = (char *) sqlite3_column_text(stmt, 4),
+        };
+        utarray_push_back(*tags, &tag);
+
+        step = sqlite3_step(stmt);
+    }
+
+    if (SQLITE_DONE != step) {
+        nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(persister->db));
+        // do not set return code, return partial or empty result
+    }
+
+    sqlite3_finalize(stmt);
+    return 0;
+
+error:
+    utarray_free(*tags);
+    *tags = NULL;
+    return NEU_ERR_EINTERNAL;
+}
+
+int neu_persister_update_tag(neu_persister_t *persister,
+                             const char *driver_name, const char *group_name,
+                             const neu_datatag_t *tag)
+{
+    return neu_persister_sql_exec(
+        persister,
+        "UPDATE tags SET address=%Q, attribute=%i, type=%i, description=%Q "
+        "WHERE driver_name=%Q AND group_name=%Q AND name=%Q",
+        tag->address, tag->attribute, tag->type, tag->description, driver_name,
+        group_name, tag->name);
+}
+
+int neu_persister_delete_tag(neu_persister_t *persister,
+                             const char *driver_name, const char *group_name,
+                             const char *tag_name)
+{
+    return neu_persister_sql_exec(
+        persister,
+        "DELETE FROM tags WHERE driver_name=%Q AND group_name=%Q AND name=%Q",
+        driver_name, group_name, tag_name);
 }
 
 int neu_persister_store_subscriptions(neu_persister_t *persister,
diff --git a/src/persist/persist.h b/src/persist/persist.h
index 764710a5..a033033b 100644
--- a/src/persist/persist.h
+++ b/src/persist/persist.h
@@ -25,7 +25,6 @@ extern "C" {
 #endif
 
 #include "adapter/adapter_info.h"
-#include "persist/json/persist_json_datatags.h"
 #include "persist/json/persist_json_plugin.h"
 #include "persist/json/persist_json_subs.h"
 
@@ -38,8 +37,6 @@ typedef struct {
     int64_t state;
 } neu_persist_node_info_t;
 
-typedef neu_json_datatag_req_tag_t   neu_persist_datatag_info_t;
-
 typedef struct {
     uint32_t interval;
     char *   name;
@@ -64,11 +61,6 @@ static inline void neu_persist_group_info_fini(neu_persist_group_info_t *info)
     free(info->name);
 }
 
-static inline void neu_persist_datatag_infos_free(UT_array *datatag_infos)
-{
-    utarray_free(datatag_infos);
-}
-
 static inline void
 neu_persist_subscription_infos_free(UT_array *subscription_infos)
 {
@@ -153,30 +145,51 @@ int neu_persister_load_plugins(neu_persister_t *persister,
                                UT_array **      plugin_infos);
 
 /**
- * Persist adapter datatags.
+ * Persist node tags.
  * @param persister                 persiter object.
- * @param adapter_name              name of the adapter who owns the datatags.
- * @param group_config_name         name of the group config.
- * @param datatag_infos             vector of neu_persist_datatag_info_t.
+ * @param driver_name               name of the driver who owns the tags
+ * @param group_name                name of the group
+ * @param tag                       the tag to store
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_store_datatags(neu_persister_t *persister,
-                                 const char *     adapter_name,
-                                 const char *     group_config_name,
-                                 UT_array *       datatag_infos);
+int neu_persister_store_tag(neu_persister_t *persister, const char *driver_name,
+                            const char *group_name, const neu_datatag_t *tag);
+
 /**
- * Load adapter datatag infos.
+ * Load node tag infos.
  * @param persister                 persiter object.
- * @param adapter_name              name of the adapter who owns the datatags.
- * @param group_config_name         name of the group config.
- * @param[out] datatag_infos        used to return pointer to heap allocated
- * vector of neu_persist_datatag_info_t.
+ * @param node_name                 name of the node who owns the tags
+ * @param group_name                name of the group
+ * @param[out] tag_infos            used to return pointer to heap allocated
+ *                                  vector of neu_datatag_t
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_load_datatags(neu_persister_t *persister,
-                                const char *     adapter_name,
-                                const char *     group_config_name,
-                                UT_array **      datatag_infos);
+int neu_persister_load_tags(neu_persister_t *persister, const char *driver_name,
+                            const char *group_name, UT_array **tag_infos);
+
+/**
+ * Update node tags.
+ * @param persister                 persiter object.
+ * @param driver_name               name of the driver who owns the tags
+ * @param group_name                name of the group
+ * @param tag                       the tag to update
+ * @return 0 on success, non-zero otherwise
+ */
+int neu_persister_update_tag(neu_persister_t *persister,
+                             const char *driver_name, const char *group_name,
+                             const neu_datatag_t *tag);
+
+/**
+ * Delete node tags.
+ * @param persister                 persiter object.
+ * @param driver_name               name of the driver who owns the tags
+ * @param group_name                name of the group
+ * @param tag_name                  name of the tag
+ * @return 0 on success, non-zero otherwise
+ */
+int neu_persister_delete_tag(neu_persister_t *persister,
+                             const char *driver_name, const char *group_name,
+                             const char *tag_name);
 
 /**
  * Persist adapter subscriptions.

From 6b6ed493de2b8205603dbd7781c043bec3fe78f6 Mon Sep 17 00:00:00 2001
From: eeff <33566201+eeff@users.noreply.github.com>
Date: Thu, 14 Jul 2022 16:16:27 +0800
Subject: [PATCH 5/8] feat(persistence): migrate subscription persistence to
 sqlite

---
 src/core/manager.c    |   6 +-
 src/core/storage.c    |  40 ++++++-----
 src/core/storage.h    |   5 +-
 src/persist/persist.c | 152 +++++++++++++++++++-----------------------
 src/persist/persist.h |  56 +++++++++-------
 5 files changed, 134 insertions(+), 125 deletions(-)

diff --git a/src/core/manager.c b/src/core/manager.c
index 62835ea7..f59bc449 100644
--- a/src/core/manager.c
+++ b/src/core/manager.c
@@ -297,7 +297,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
             neu_manager_subscribe(manager, cmd->app, cmd->driver, cmd->group);
 
         if (error.error == NEU_ERR_SUCCESS) {
-            manager_storage_subscribe(manager, cmd->app);
+            manager_storage_subscribe(manager, cmd->app, cmd->driver,
+                                      cmd->group);
             neu_manager_notify_app_sub(manager, cmd->app, cmd->driver,
                                        cmd->group);
         }
@@ -315,7 +316,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
             neu_manager_unsubscribe(manager, cmd->app, cmd->driver, cmd->group);
 
         if (error.error == NEU_ERR_SUCCESS) {
-            manager_storage_subscribe(manager, cmd->app);
+            manager_storage_unsubscribe(manager, cmd->app, cmd->driver,
+                                        cmd->group);
             neu_manager_notify_app_unsub(manager, cmd->app, cmd->driver,
                                          cmd->group);
         }
diff --git a/src/core/storage.c b/src/core/storage.c
index 53f66422..54b22673 100644
--- a/src/core/storage.c
+++ b/src/core/storage.c
@@ -59,18 +59,26 @@ void manager_storage_del_node(neu_manager_t *manager, const char *node)
     neu_persister_delete_node(manager->persister, node);
 }
 
-void manager_storage_subscribe(neu_manager_t *manager, const char *app)
+void manager_storage_subscribe(neu_manager_t *manager, const char *app,
+                               const char *driver, const char *group)
 {
-
-    UT_array *sub_infos = neu_manager_get_sub_group(manager, app);
-
-    int rv =
-        neu_persister_store_subscriptions(manager->persister, app, sub_infos);
+    int rv = neu_persister_store_subscription(manager->persister, app, driver,
+                                              group);
     if (0 != rv) {
-        nlog_error("fail store adapter:%s subscription infos", app);
+        nlog_error("fail store subscription app:%s driver:%s group:%s", app,
+                   driver, group);
     }
+}
 
-    utarray_free(sub_infos);
+void manager_storage_unsubscribe(neu_manager_t *manager, const char *app,
+                                 const char *driver, const char *group)
+{
+    int rv = neu_persister_delete_subscription(manager->persister, app, driver,
+                                               group);
+    if (0 != rv) {
+        nlog_error("fail delete subscription app:%s driver:%s group:%s", app,
+                   driver, group);
+    }
 }
 
 int manager_load_plugin(neu_manager_t *manager)
@@ -137,21 +145,19 @@ int manager_load_subscribe(neu_manager_t *manager)
         } else {
             utarray_foreach(sub_infos, neu_persist_subscription_info_t *, info)
             {
-                rv = neu_manager_subscribe(manager, info->sub_adapter_name,
-                                           info->src_adapter_name,
-                                           info->group_config_name);
+                rv = neu_manager_subscribe(manager, node->node,
+                                           info->driver_name, info->group_name);
                 const char *ok_or_err = (0 == rv) ? "success" : "fail";
                 nlog_info("%s load subscription app:%s driver:%s grp:%s",
-                          ok_or_err, info->sub_adapter_name,
-                          info->src_adapter_name, info->group_config_name);
+                          ok_or_err, node->node, info->driver_name,
+                          info->group_name);
                 if (rv == 0) {
-                    neu_manager_notify_app_sub_update(manager,
-                                                      info->src_adapter_name,
-                                                      info->group_config_name);
+                    neu_manager_notify_app_sub_update(
+                        manager, info->driver_name, info->group_name);
                 }
             }
 
-            neu_persist_subscription_infos_free(sub_infos);
+            utarray_free(sub_infos);
         }
     }
 
diff --git a/src/core/storage.h b/src/core/storage.h
index db8ecd65..5151898f 100644
--- a/src/core/storage.h
+++ b/src/core/storage.h
@@ -27,7 +27,10 @@
 void manager_strorage_plugin(neu_manager_t *manager);
 void manager_storage_del_node(neu_manager_t *manager, const char *node);
 void manager_storage_add_node(neu_manager_t *manager, const char *node);
-void manager_storage_subscribe(neu_manager_t *manager, const char *node);
+void manager_storage_subscribe(neu_manager_t *manager, const char *app,
+                               const char *driver, const char *group);
+void manager_storage_unsubscribe(neu_manager_t *manager, const char *app,
+                                 const char *driver, const char *group);
 
 int manager_load_plugin(neu_manager_t *manager);
 int manager_load_node(neu_manager_t *manager);
diff --git a/src/persist/persist.c b/src/persist/persist.c
index 46aa8474..c1e96a9c 100644
--- a/src/persist/persist.c
+++ b/src/persist/persist.c
@@ -710,105 +710,93 @@ int neu_persister_delete_tag(neu_persister_t *persister,
         driver_name, group_name, tag_name);
 }
 
-int neu_persister_store_subscriptions(neu_persister_t *persister,
-                                      const char *     adapter_name,
-                                      UT_array *       subscription_infos)
+int neu_persister_store_subscription(neu_persister_t *persister,
+                                     const char *     app_name,
+                                     const char *     driver_name,
+                                     const char *     group_name)
 {
-    char path[PATH_MAX_SIZE] = { 0 };
-
-    int n = persister_adapter_dir(path, sizeof(path), persister, adapter_name);
-    if (sizeof(path) == n) {
-        nlog_error("path too long: %s", path);
-        return -1;
-    }
-
-    int rv = create_dir(path);
-    if (0 != rv) {
-        nlog_error("fail to create dir: %s", path);
-        return rv;
-    }
-
-    n = path_cat(path, n, sizeof(path), "subscriptions.json");
-    if (sizeof(path) == n) {
-        nlog_error("path too long: %s", path);
-        return -1;
-    }
-
-    neu_json_subscriptions_req_t subs_resp = { 0 };
-    int                          index     = 0;
-    subs_resp.n_subscription               = utarray_len(subscription_infos);
-    subs_resp.subscriptions =
-        calloc(subs_resp.n_subscription,
-               sizeof(neu_json_subscriptions_req_subscription_t));
-    utarray_foreach(subscription_infos, neu_resp_subscribe_info_t *, info)
-    {
-        subs_resp.subscriptions[index].group_config_name = info->group;
-        subs_resp.subscriptions[index].src_adapter_name  = info->driver;
-        subs_resp.subscriptions[index].sub_adapter_name  = info->app;
-        index += 1;
-    }
-
-    char *result = NULL;
-    rv = neu_json_encode_by_fn(&subs_resp, neu_json_encode_subscriptions_resp,
-                               &result);
-    if (rv == 0) {
-        rv = write_file_string(path, result);
-    }
-
-    free(result);
-    free(subs_resp.subscriptions);
-
-    return rv;
+    return neu_persister_sql_exec(
+        persister,
+        "INSERT INTO subscriptions (app_name, driver_name, group_name) VALUES "
+        "(%Q, %Q, %Q)",
+        app_name, driver_name, group_name);
 }
 
+static UT_icd subscription_info_icd = {
+    sizeof(neu_persist_subscription_info_t),
+    NULL,
+    NULL,
+    (dtor_f *) neu_persist_subscription_info_fini,
+};
+
 int neu_persister_load_subscriptions(neu_persister_t *persister,
-                                     const char *     adapter_name,
+                                     const char *     app_name,
                                      UT_array **      subscription_infos)
 {
-    char path[PATH_MAX_SIZE] = { 0 };
+    sqlite3_stmt *stmt = NULL;
+    const char *  query =
+        "SELECT driver_name, group_name FROM subscriptions WHERE app_name=?";
 
-    int n = persister_adapter_dir(path, sizeof(path), persister, adapter_name);
-    if (sizeof(path) == n) {
-        nlog_error("path too long: %s", path);
-        return -1;
+    utarray_new(*subscription_infos, &subscription_info_icd);
+
+    if (SQLITE_OK !=
+        sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) {
+        nlog_error("prepare `%s` fail: %s", query,
+                   sqlite3_errmsg(persister->db));
+        goto error;
     }
 
-    n = path_cat(path, n, sizeof(path), "subscriptions.json");
-    if (sizeof(path) == n) {
-        nlog_error("path too long: %s", path);
-        return -1;
+    if (SQLITE_OK != sqlite3_bind_text(stmt, 1, app_name, -1, NULL)) {
+        nlog_error("bind `%s` with `%s` fail: %s", query, app_name,
+                   sqlite3_errmsg(persister->db));
+        goto error;
     }
 
-    char *json_str = NULL;
-    int   rv       = read_file_string(path, &json_str);
-    if (rv != 0) {
-        return rv;
-    }
+    int step = sqlite3_step(stmt);
+    while (SQLITE_ROW == step) {
+        char *driver_name = strdup((char *) sqlite3_column_text(stmt, 0));
+        if (NULL == driver_name) {
+            break;
+        }
 
-    neu_json_subscriptions_req_t *subs_req = NULL;
-    rv = neu_json_decode_subscriptions_req(json_str, &subs_req);
-    free(json_str);
-    if (0 != rv) {
-        return rv;
-    }
-
-    UT_icd icd = { sizeof(neu_persist_subscription_info_t), NULL, NULL, NULL };
-    utarray_new(*subscription_infos, &icd);
-
-    for (int i = 0; i < subs_req->n_subscription; i++) {
-        neu_persist_subscription_info_t info = { 0 };
-        info.group_config_name =
-            strdup(subs_req->subscriptions[i].group_config_name);
-        info.src_adapter_name =
-            strdup(subs_req->subscriptions[i].src_adapter_name);
-        info.sub_adapter_name =
-            strdup(subs_req->subscriptions[i].sub_adapter_name);
+        char *group_name = strdup((char *) sqlite3_column_text(stmt, 1));
+        if (NULL == group_name) {
+            free(driver_name);
+            break;
+        }
 
+        neu_persist_subscription_info_t info = {
+            .driver_name = driver_name,
+            .group_name  = group_name,
+        };
         utarray_push_back(*subscription_infos, &info);
+
+        step = sqlite3_step(stmt);
     }
 
-    neu_json_decode_subscriptions_req_free(subs_req);
+    if (SQLITE_DONE != step) {
+        nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(persister->db));
+        // do not set return code, return partial or empty result
+    }
+
+    sqlite3_finalize(stmt);
     return 0;
+
+error:
+    utarray_free(*subscription_infos);
+    *subscription_infos = NULL;
+    return NEU_ERR_EINTERNAL;
+}
+
+int neu_persister_delete_subscription(neu_persister_t *persister,
+                                      const char *     app_name,
+                                      const char *     driver_name,
+                                      const char *     group_name)
+{
+    return neu_persister_sql_exec(persister,
+                                  "DELETE FROM subscriptions WHERE app_name=%Q "
+                                  "AND driver_name=%Q AND group_name=%Q",
+                                  app_name, driver_name, group_name);
 }
 
 int neu_persister_store_group(neu_persister_t *         persister,
diff --git a/src/persist/persist.h b/src/persist/persist.h
index a033033b..35de0551 100644
--- a/src/persist/persist.h
+++ b/src/persist/persist.h
@@ -26,7 +26,6 @@ extern "C" {
 
 #include "adapter/adapter_info.h"
 #include "persist/json/persist_json_plugin.h"
-#include "persist/json/persist_json_subs.h"
 
 typedef neu_json_plugin_req_plugin_t neu_persist_plugin_info_t;
 
@@ -42,8 +41,10 @@ typedef struct {
     char *   name;
 } neu_persist_group_info_t;
 
-typedef neu_json_subscriptions_req_subscription_t
-    neu_persist_subscription_info_t;
+typedef struct {
+    char *driver_name;
+    char *group_name;
+} neu_persist_subscription_info_t;
 
 static inline void neu_persist_plugin_infos_free(UT_array *plugin_infos)
 {
@@ -62,15 +63,10 @@ static inline void neu_persist_group_info_fini(neu_persist_group_info_t *info)
 }
 
 static inline void
-neu_persist_subscription_infos_free(UT_array *subscription_infos)
+neu_persist_subscription_info_fini(neu_persist_subscription_info_t *info)
 {
-    utarray_foreach(subscription_infos, neu_persist_subscription_info_t *, info)
-    {
-        free(info->group_config_name);
-        free(info->src_adapter_name);
-        free(info->sub_adapter_name);
-    }
-    utarray_free(subscription_infos);
+    free(info->driver_name);
+    free(info->group_name);
 }
 
 /**
@@ -192,29 +188,43 @@ int neu_persister_delete_tag(neu_persister_t *persister,
                              const char *tag_name);
 
 /**
- * Persist adapter subscriptions.
+ * Persist subscriptions.
  * @param persister                 persiter object.
- * @param adapter_name              name of the adapter who owns the
- * subscriptions.
- * @param subscription_infos        vector of neu_persist_subscription_info_t.
+ * @param app_name                  name of the app node
+ * @param driver_name               name of the driver node
+ * @param group_name                name of the group
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_store_subscriptions(neu_persister_t *persister,
-                                      const char *     adapter_name,
-                                      UT_array *       subscription_infos);
+int neu_persister_store_subscription(neu_persister_t *persister,
+                                     const char *     app_name,
+                                     const char *     driver_name,
+                                     const char *     group_name);
+
 /**
- * Load adapter subscription infos.
+ * Load adapter subscriptions.
  * @param persister                 persiter object.
- * @param adapter_name              name of the adapter who owns the
- * subscriptions.
+ * @param app_name                  name of the app node
  * @param[out] subscription_infos   used to return pointer to heap allocated
- * vector of neu_persist_subscription_info_t.
+ *                                  vector of neu_persist_subscription_info_t.
  * @return 0 on success, non-zero otherwise
  */
 int neu_persister_load_subscriptions(neu_persister_t *persister,
-                                     const char *     adapter_name,
+                                     const char *     app_name,
                                      UT_array **      subscription_infos);
 
+/**
+ * Persist subscriptions.
+ * @param persister                 persiter object.
+ * @param app_name                  name of the app node
+ * @param driver_name               name of the driver node
+ * @param group_name                name of the group
+ * @return 0 on success, non-zero otherwise
+ */
+int neu_persister_delete_subscription(neu_persister_t *persister,
+                                      const char *     app_name,
+                                      const char *     driver_name,
+                                      const char *     group_name);
+
 /**
  * Persist group config.
  * @param persister                 persiter object.

From b91d62828dd8533a8f47aeea3246d9e1ab69b614 Mon Sep 17 00:00:00 2001
From: eeff <33566201+eeff@users.noreply.github.com>
Date: Fri, 15 Jul 2022 18:18:54 +0800
Subject: [PATCH 6/8] feat(persistence): apply schema files when persister init

---
 src/persist/persist.c | 323 ++++++++++++++++++++++++++++++++++--------
 1 file changed, 264 insertions(+), 59 deletions(-)

diff --git a/src/persist/persist.c b/src/persist/persist.c
index c1e96a9c..6110a71a 100644
--- a/src/persist/persist.c
+++ b/src/persist/persist.c
@@ -306,6 +306,236 @@ error_open:
     return rv;
 }
 
+static inline int execute_sql(sqlite3 *db, const char *sql, ...)
+{
+    int rv = 0;
+
+    va_list args;
+    va_start(args, sql);
+    char *query = sqlite3_vmprintf(sql, args);
+    va_end(args);
+
+    if (NULL == query) {
+        nlog_error("allocate SQL `%s` fail", sql);
+        return NEU_ERR_EINTERNAL;
+    }
+
+    char *err_msg = NULL;
+    if (SQLITE_OK != sqlite3_exec(db, query, NULL, NULL, &err_msg)) {
+        nlog_error("query `%s` fail: %s", query, err_msg);
+        rv = NEU_ERR_EINTERNAL;
+    }
+
+    sqlite3_free(err_msg);
+    sqlite3_free(query);
+
+    return rv;
+}
+
+static int get_schema_version(sqlite3 *db, char **version_p, bool *dirty_p)
+{
+    sqlite3_stmt *stmt  = NULL;
+    const char *  query = "SELECT version, dirty FROM migrations ORDER BY "
+                        "migration_id DESC LIMIT 1";
+
+    if (SQLITE_OK != sqlite3_prepare_v2(db, query, -1, &stmt, NULL)) {
+        nlog_error("prepare `%s` fail: %s", query, sqlite3_errmsg(db));
+        return NEU_ERR_EINTERNAL;
+    }
+
+    int step = sqlite3_step(stmt);
+    if (SQLITE_ROW == step) {
+        char *version = strdup((char *) sqlite3_column_text(stmt, 0));
+        if (NULL == version) {
+            sqlite3_finalize(stmt);
+            return NEU_ERR_EINTERNAL;
+        }
+
+        *version_p = version;
+        *dirty_p   = sqlite3_column_int(stmt, 1);
+    } else if (SQLITE_DONE != step) {
+        nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(db));
+    }
+
+    sqlite3_finalize(stmt);
+    return 0;
+}
+
+static int extract_schema_info(const char *file, char **version_p,
+                               char **description_p)
+{
+    if (!ends_with(file, ".sql")) {
+        return NEU_ERR_EINTERNAL;
+    }
+
+    char *sep = strchr(file, '_');
+    if (NULL == sep) {
+        return NEU_ERR_EINTERNAL;
+    }
+
+    size_t n = sep - file;
+    if (4 != n) {
+        // invalid version
+        return NEU_ERR_EINTERNAL;
+    }
+
+    if (strcmp(++sep, ".sql") == 0) {
+        // no description
+        return NEU_ERR_EINTERNAL;
+    }
+
+    char *version = calloc(n + 1, sizeof(char));
+    if (NULL == version) {
+        return NEU_ERR_EINTERNAL;
+    }
+    strncat(version, file, n);
+
+    n                 = strlen(sep) - 4;
+    char *description = calloc(n + 1, sizeof(char));
+    if (NULL == description) {
+        free(version);
+        return NEU_ERR_EINTERNAL;
+    }
+    strncat(description, sep, n);
+
+    *version_p     = version;
+    *description_p = description;
+
+    return 0;
+}
+
+static int apply_schema_file(sqlite3 *db, const char *dir, const char *file)
+{
+    int   rv          = 0;
+    char *version     = NULL;
+    char *description = NULL;
+    char *sql         = NULL;
+    char  path[128]   = {};
+    int   n           = 0;
+
+    if (sizeof(path) == (n = path_cat(path, 0, sizeof(path), dir)) ||
+        sizeof(path) == path_cat(path, n, sizeof(path), file)) {
+        nlog_error("path too long: %s", path);
+        return -1;
+    }
+
+    if (0 != extract_schema_info(file, &version, &description)) {
+        nlog_warn("extract `%s` schema info fail, ignore", path);
+        return 0;
+    }
+
+    rv = read_file_string(path, &sql);
+    if (0 != rv) {
+        goto end;
+    }
+
+    rv = execute_sql(db,
+                     "INSERT INTO migrations (version, description, dirty) "
+                     "VALUES (%Q, %Q, 1)",
+                     version, description);
+    if (0 != rv) {
+        goto end;
+    }
+
+    char *err_msg = NULL;
+    rv            = sqlite3_exec(db, sql, NULL, NULL, &err_msg);
+    if (SQLITE_OK != rv) {
+        nlog_error("execute %s fail: %s", path, err_msg);
+        sqlite3_free(err_msg);
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
+    }
+
+    rv = execute_sql(db, "UPDATE migrations SET dirty = 0 WHERE version=%Q",
+                     version);
+
+end:
+    if (0 == rv) {
+        nlog_info("success apply schema `%s`, version=`%s` description=`%s`",
+                  path, version, description);
+    } else {
+        nlog_error("fail apply schema `%s`, version=`%s` description=`%s`",
+                   path, version, description);
+    }
+
+    free(sql);
+    free(version);
+    free(description);
+    return rv;
+}
+
+static int apply_schemas(sqlite3 *db, const char *dir)
+{
+    int rv = 0;
+
+    const char *sql = "CREATE TABLE IF NOT EXISTS migrations ( migration_id "
+                      "INTEGER PRIMARY KEY, version TEXT NOT NULL UNIQUE, "
+                      "description TEXT NOT NULL, dirty INTEGER NOT NULL, "
+                      "created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP) ";
+    if (0 != execute_sql(db, sql)) {
+        nlog_error("create migration table fail");
+        return NEU_ERR_EINTERNAL;
+    }
+
+    bool  dirty   = false;
+    char *version = NULL;
+    if (0 != get_schema_version(db, &version, &dirty)) {
+        nlog_error("find schema version fail");
+        return NEU_ERR_EINTERNAL;
+    }
+
+    nlog_info("schema head version=%s", version ? version : "none");
+
+    if (dirty) {
+        nlog_error("database is dirty, need manual intervention");
+        return NEU_ERR_EINTERNAL;
+    }
+
+    DIR *          dirp = NULL;
+    struct dirent *dent = NULL;
+    if ((dirp = opendir(dir)) == NULL) {
+        nlog_error("fail open dir: %s", dir);
+        free(version);
+        return NEU_ERR_EINTERNAL;
+    }
+
+    UT_array *files = NULL;
+    utarray_new(files, &ut_str_icd);
+
+    while (NULL != (dent = readdir(dirp))) {
+        if (ends_with(dent->d_name, ".sql")) {
+            char *file = dent->d_name;
+            utarray_push_back(files, &file);
+        }
+    }
+
+    closedir(dirp);
+    dirp = NULL;
+
+    if (0 == utarray_len(files)) {
+        nlog_warn("directory `%s` contains no schema files", dir);
+    }
+
+    utarray_sort(files, (int (*)(const void *, const void *)) strcmp);
+
+    utarray_foreach(files, char **, file)
+    {
+        if (version && strncmp(*file, version, strlen(version)) <= 0) {
+            continue;
+        }
+
+        if (0 != apply_schema_file(db, dir, *file)) {
+            rv = NEU_ERR_EINTERNAL;
+            break;
+        }
+    }
+
+    free(version);
+    utarray_free(files);
+
+    return rv;
+}
+
 neu_persister_t *neu_persister_create(const char *dir_name)
 {
     int   rv                  = 0;
@@ -385,6 +615,9 @@ neu_persister_t *neu_persister_create(const char *dir_name)
         sqlite3_close(persister->db);
         goto error;
     }
+    if (0 != apply_schemas(persister->db, persister->persist_dir)) {
+        goto error;
+    }
 
     return persister;
 
@@ -454,40 +687,13 @@ void neu_persister_destroy(neu_persister_t *persister)
     free(persister);
 }
 
-int neu_persister_sql_exec(neu_persister_t *persister, const char *sql, ...)
-{
-    int rv = 0;
-
-    va_list args;
-    va_start(args, sql);
-    char *query = sqlite3_vmprintf(sql, args);
-    va_end(args);
-
-    if (NULL == query) {
-        nlog_error("allocate SQL `%s` fail", query);
-        return NEU_ERR_EINTERNAL;
-    }
-
-    char *err_msg = NULL;
-    if (SQLITE_OK != sqlite3_exec(persister->db, query, NULL, NULL, &err_msg)) {
-        nlog_error("query `%s` fail: %s", query, err_msg);
-        rv = NEU_ERR_EINTERNAL;
-    }
-
-    sqlite3_free(err_msg);
-    sqlite3_free(query);
-
-    return rv;
-}
-
 int neu_persister_store_node(neu_persister_t *        persister,
                              neu_persist_node_info_t *info)
 {
-    return neu_persister_sql_exec(persister,
-                                  "INSERT INTO nodes (name, type, state, "
-                                  "plugin_name) VALUES (%Q, %i, %i, %Q)",
-                                  info->name, info->type, info->state,
-                                  info->plugin_name);
+    return execute_sql(persister->db,
+                       "INSERT INTO nodes (name, type, state, plugin_name) "
+                       "VALUES (%Q, %i, %i, %Q)",
+                       info->name, info->type, info->state, info->plugin_name);
 }
 
 static UT_icd node_info_icd = {
@@ -550,16 +756,15 @@ int neu_persister_delete_node(neu_persister_t *persister, const char *node_name)
 {
     // rely on foreign key constraints to remove settings, groups, tags and
     // subscriptions
-    return neu_persister_sql_exec(persister, "DELETE FROM nodes WHERE name=%Q;",
-                                  node_name);
+    return execute_sql(persister->db, "DELETE FROM nodes WHERE name=%Q;",
+                       node_name);
 }
 
 int neu_persister_update_node(neu_persister_t *persister, const char *node_name,
                               const char *new_name)
 {
-    return neu_persister_sql_exec(persister,
-                                  "UPDATE nodes SET name=%Q WHERE name=%Q;",
-                                  new_name, node_name);
+    return execute_sql(persister->db, "UPDATE nodes SET name=%Q WHERE name=%Q;",
+                       new_name, node_name);
 }
 
 int neu_persister_store_plugins(neu_persister_t *persister,
@@ -624,8 +829,8 @@ int neu_persister_load_plugins(neu_persister_t *persister,
 int neu_persister_store_tag(neu_persister_t *persister, const char *driver_name,
                             const char *group_name, const neu_datatag_t *tag)
 {
-    return neu_persister_sql_exec(
-        persister,
+    return execute_sql(
+        persister->db,
         "INSERT INTO tags (driver_name, group_name, name, address, attribute, "
         "type, description) VALUES(%Q, %Q, %Q, %Q, %i, %i, %Q)",
         driver_name, group_name, tag->name, tag->address, tag->attribute,
@@ -692,8 +897,8 @@ int neu_persister_update_tag(neu_persister_t *persister,
                              const char *driver_name, const char *group_name,
                              const neu_datatag_t *tag)
 {
-    return neu_persister_sql_exec(
-        persister,
+    return execute_sql(
+        persister->db,
         "UPDATE tags SET address=%Q, attribute=%i, type=%i, description=%Q "
         "WHERE driver_name=%Q AND group_name=%Q AND name=%Q",
         tag->address, tag->attribute, tag->type, tag->description, driver_name,
@@ -704,8 +909,8 @@ int neu_persister_delete_tag(neu_persister_t *persister,
                              const char *driver_name, const char *group_name,
                              const char *tag_name)
 {
-    return neu_persister_sql_exec(
-        persister,
+    return execute_sql(
+        persister->db,
         "DELETE FROM tags WHERE driver_name=%Q AND group_name=%Q AND name=%Q",
         driver_name, group_name, tag_name);
 }
@@ -715,10 +920,10 @@ int neu_persister_store_subscription(neu_persister_t *persister,
                                      const char *     driver_name,
                                      const char *     group_name)
 {
-    return neu_persister_sql_exec(
-        persister,
-        "INSERT INTO subscriptions (app_name, driver_name, group_name) VALUES "
-        "(%Q, %Q, %Q)",
+    return execute_sql(
+        persister->db,
+        "INSERT INTO subscriptions (app_name, driver_name, group_name) "
+        "VALUES (%Q, %Q, %Q)",
         app_name, driver_name, group_name);
 }
 
@@ -793,18 +998,18 @@ int neu_persister_delete_subscription(neu_persister_t *persister,
                                       const char *     driver_name,
                                       const char *     group_name)
 {
-    return neu_persister_sql_exec(persister,
-                                  "DELETE FROM subscriptions WHERE app_name=%Q "
-                                  "AND driver_name=%Q AND group_name=%Q",
-                                  app_name, driver_name, group_name);
+    return execute_sql(persister->db,
+                       "DELETE FROM subscriptions WHERE app_name=%Q AND "
+                       "driver_name=%Q AND group_name=%Q",
+                       app_name, driver_name, group_name);
 }
 
 int neu_persister_store_group(neu_persister_t *         persister,
                               const char *              driver_name,
                               neu_persist_group_info_t *group_info)
 {
-    return neu_persister_sql_exec(
-        persister,
+    return execute_sql(
+        persister->db,
         "INSERT INTO groups (driver_name, name, interval) VALUES (%Q, %Q, %i)",
         driver_name, group_info->name, group_info->interval);
 }
@@ -870,16 +1075,16 @@ int neu_persister_delete_group(neu_persister_t *persister,
                                const char *driver_name, const char *group_name)
 {
     // rely on foreign key constraints to delete tags and subscriptions
-    return neu_persister_sql_exec(
-        persister, "DELETE FROM groups WHERE driver_name=%Q AND name=%Q",
-        driver_name, group_name);
+    return execute_sql(persister->db,
+                       "DELETE FROM groups WHERE driver_name=%Q AND name=%Q",
+                       driver_name, group_name);
 }
 
 int neu_persister_store_node_setting(neu_persister_t *persister,
                                      const char *node_name, const char *setting)
 {
-    return neu_persister_sql_exec(
-        persister,
+    return execute_sql(
+        persister->db,
         "INSERT OR REPLACE INTO settings (node_name, setting) VALUES (%Q, %Q)",
         node_name, setting);
 }
@@ -930,6 +1135,6 @@ end:
 int neu_persister_delete_node_setting(neu_persister_t *persister,
                                       const char *     node_name)
 {
-    return neu_persister_sql_exec(
-        persister, "DELETE FROM settings WHERE node_name=%Q", node_name);
+    return execute_sql(persister->db, "DELETE FROM settings WHERE node_name=%Q",
+                       node_name);
 }

From f4ebb6c00d7a06dffad2441c06e91de91321643b Mon Sep 17 00:00:00 2001
From: eeff <33566201+eeff@users.noreply.github.com>
Date: Fri, 15 Jul 2022 18:35:00 +0800
Subject: [PATCH 7/8] fix: persistence functionality tests

---
 ft/Keyword/Neuron.py     | 23 +++++++++++++----------
 ft/persist_test.robot    | 18 ++++++++++--------
 ft/resource/api.resource | 10 ----------
 3 files changed, 23 insertions(+), 28 deletions(-)

diff --git a/ft/Keyword/Neuron.py b/ft/Keyword/Neuron.py
index 756fe671..a79e14fb 100644
--- a/ft/Keyword/Neuron.py
+++ b/ft/Keyword/Neuron.py
@@ -6,6 +6,12 @@ import shutil
 import time
 
 
+def prepare_persistence_dir():
+    os.system("mkdir -p build/persistence")
+    os.system("cp -r persistence build/")
+    os.system("cp default_plugins.json build/persistence/plugins.json")
+
+
 class Neuron(object):
     process = 0
     profiler_process = None
@@ -14,8 +20,7 @@ class Neuron(object):
         pass
 
     def Start_Neuron(self):
-        os.system("mkdir -p build/persistence")
-        os.system("cp default_plugins.json build/persistence/plugins.json")
+        prepare_persistence_dir()
         self.process = subprocess.Popen(['./neuron', '--log'], cwd='build/')
         return self.process
 
@@ -25,17 +30,15 @@ class Neuron(object):
             time.sleep(5)
         self.process.kill()
         if remove_persistence_data:
-            shutil.rmtree("build/persistence", ignore_errors=True)
-            os.system("mkdir -p build/persistence")
-            os.system("cp default_plugins.json build/persistence/plugins.json")
+            os.system("rm build/persistence/sqlite.db")
+            prepare_persistence_dir()
 
     def End_Neuron(self, process):
         process.kill()
 
     def Remove_Persistence(self):
-        shutil.rmtree("build/persistence", ignore_errors=True)
-        os.system("mkdir -p build/persistence")
-        os.system("cp default_plugins.json build/persistence/plugins.json")
+        os.system("rm build/persistence/sqlite.db")
+        prepare_persistence_dir()
 
     def Profile_Neuron(self, interval, result_dir):
         cmd = [
@@ -151,9 +154,9 @@ class Tag(object):
     def Tag_Find_By_Name(self, tags, name):
         for tag in tags:
             if tag['name'] == name:
-                return tag['id']
+                return tag
 
-        return -1
+        return None
 
 
 class Subscribe(object):
diff --git a/ft/persist_test.robot b/ft/persist_test.robot
index 96b242d0..07594773 100644
--- a/ft/persist_test.robot
+++ b/ft/persist_test.robot
@@ -237,15 +237,17 @@ Get the update tags after stopping ande restarting Neuron, it should return the
 	${len} =	Get Length	${res}[tags]
  	Should Be Equal As Integers	${len}		2
 
-	Should Be Equal As Strings	${res}[tags][1][name]		tag1
-	Should Be Equal As Strings	${res}[tags][1][address]	1!400002
-	Should Be Equal As Strings	${res}[tags][1][attribute]	${TAG_ATTRIBUTE_READ}
-	Should Be Equal As Strings	${res}[tags][1][type]		${TAG_TYPE_INT16}
+  ${tag1} =   Tag Find By Name  ${res}[tags]  tag1
+	Should Be Equal As Strings	${tag1}[name]		tag1
+	Should Be Equal As Strings	${tag1}[address]	1!400002
+	Should Be Equal As Strings	${tag1}[attribute]	${TAG_ATTRIBUTE_READ}
+	Should Be Equal As Strings	${tag1}[type]		${TAG_TYPE_INT16}
 
-	Should Be Equal As Strings	${res}[tags][0][name]		tag2
-	Should Be Equal As Strings	${res}[tags][0][address]	1!00001
-	Should Be Equal As Strings	${res}[tags][0][attribute]	${TAG_ATTRIBUTE_RW}
-	Should Be Equal As Strings	${res}[tags][0][type]		${TAG_TYPE_BIT}
+  ${tag2} =   Tag Find By Name  ${res}[tags]  tag2
+	Should Be Equal As Strings	${tag2}[name]		tag2
+	Should Be Equal As Strings	${tag2}[address]	1!00001
+	Should Be Equal As Strings	${tag2}[attribute]	${TAG_ATTRIBUTE_RW}
+	Should Be Equal As Strings	${tag2}[type]		${TAG_TYPE_BIT}
 
    	[Teardown]    Neuron Stop Running    ${process}
 
diff --git a/ft/resource/api.resource b/ft/resource/api.resource
index b762a3e7..23c67e8a 100644
--- a/ft/resource/api.resource
+++ b/ft/resource/api.resource
@@ -168,16 +168,6 @@ Del Tags
 
   [RETURN]            ${res}[0]
 
-Get Tag ID
-  [Arguments]         ${node_id}              ${group}                      ${tag_name}
-  ${res} =            Get Tags                ${node_id}                    ${group}
-  Check Response Status                       ${res}                        200
-
-  ${id} =             Tag Find By Name        ${res}[tags]                  ${tag_name}
-  Should Not Be Equal As Integers             ${id}                         0
-
-  [Return]            ${id}
-
 Read Tags
   [Arguments]         ${node}              ${group}
   POST                /api/v2/read            {"node": "${node}", "group": "${group}"}

From f4e20e7059a0ac9d617d5fd7db5af0618726920b Mon Sep 17 00:00:00 2001
From: eeff <33566201+eeff@users.noreply.github.com>
Date: Sun, 17 Jul 2022 22:56:03 +0800
Subject: [PATCH 8/8] fix: cross compiling and static linking for sqlite

---
 CMakeLists.txt | 2 +-
 install.sh     | 4 +++-
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/CMakeLists.txt b/CMakeLists.txt
index af823b52..ea991b1b 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -121,7 +121,7 @@ set(CMAKE_BUILD_RPATH ./)
 add_executable(neuron)
 target_sources(neuron PRIVATE ${NEURON_SOURCES}) 
 target_include_directories(neuron PRIVATE include/neuron src plugins)
-target_link_libraries(neuron dl neuron-base sqlite3)
+target_link_libraries(neuron dl neuron-base sqlite3 -lm)
 
 #copy file for run
 file(COPY ${CMAKE_SOURCE_DIR}/zlog.conf DESTINATION ${CMAKE_BINARY_DIR}/config)
diff --git a/install.sh b/install.sh
index 31596d11..d8154324 100755
--- a/install.sh
+++ b/install.sh
@@ -83,7 +83,9 @@ function build_sqlite3() {
     tar xzf sqlite3.tar.gz --strip-components=1 -C sqlite3
     cd sqlite3
     ./configure --prefix=${install_dir} \
-      && make CC=${compiler_prefix}-gcc -j4 \
+                --disable-shared --disable-readline \
+                --host ${arch} CC=${compiler_prefix}-gcc \
+      && make -j4 \
       && sudo make install
     cd ../
 }