qing
80969adc9b
- db2.cpp: 移除 RpcClient/parse_ini/IniMap，load_config 改为 默认值 → ENV → CLI 三层；新增 --db-url/--retention-days/ --max-records/--cleanup-interval CLI 参数 - core/services.yaml: db2 条目展开为完整 CLI 声明（10个配置项） - db2/README.md: 配置章节重写为 CLI 参数表 + 环境变量表 + services.yaml 示例 - README.md: 配置链路移除 db2，confs 配置表删除 db.ini 行 与 ca/redirect80 保持一致的配置模式，db2 不再依赖 confs 服务。
969 lines
39 KiB
C++
969 lines
39 KiB
C++
/** db2 — C++20 数据库 RPC 服务（对齐 Python db 模块的 execute 路由）
|
||
*
|
||
* 协议约定（与 Python/Go 客户端字节级兼容）：
|
||
* 请求: {"_route": "execute", "action": "select|insert|update|delete",
|
||
* "table": "T", "columns": [...], "values": {...},
|
||
* "where": {"col": val | {"op": val}}, "order_by": [...],
|
||
* "limit": N, "offset": N}
|
||
* 响应: {"success": true, "data": [...]}
|
||
* {"success": false, "error": "..."}
|
||
*
|
||
* 依赖: libmsg (libus_msg.so), boost::json (-lboost_json), sqlite3 (-lsqlite3)
|
||
*/
|
||
|
||
#include <us/RpcServer.hpp>
|
||
|
||
#include <boost/json.hpp>
|
||
|
||
#include <sqlite3.h>
|
||
|
||
#include <algorithm>
|
||
#include <atomic>
|
||
#include <chrono>
|
||
#include <condition_variable>
|
||
#include <cstdint>
|
||
#include <cstring>
|
||
#include <fstream>
|
||
#include <iostream>
|
||
#include <map>
|
||
#include <mutex>
|
||
#include <optional>
|
||
#include <sstream>
|
||
#include <string>
|
||
#include <thread>
|
||
#include <unordered_map>
|
||
#include <vector>
|
||
|
||
#include <cstdio>
|
||
#include <ctime>
|
||
|
||
namespace json = boost::json;
|
||
using namespace pmc::net;
|
||
|
||
// ============================================================================
|
||
// 全局常量
|
||
// ============================================================================
|
||
|
||
static constexpr size_t kMaxMessage = 8192;
|
||
static constexpr size_t kMaxExtra = 4096;
|
||
|
||
// 输入规模限制（防资源耗尽）
|
||
static constexpr size_t kMaxWhereConditions = 64; // 单请求 WHERE 条件数上限
|
||
static constexpr size_t kMaxInElements = 1000; // 单个 IN 运算符元素上限
|
||
|
||
// ============================================================================
|
||
// 工具：当前 UTC 时间 ISO 8601
|
||
// ============================================================================
|
||
|
||
static std::string utc_now() {
|
||
auto now = std::chrono::system_clock::now();
|
||
std::time_t t = std::chrono::system_clock::to_time_t(now);
|
||
struct tm tmbuf;
|
||
gmtime_r(&t, &tmbuf);
|
||
char buf[32];
|
||
std::strftime(buf, sizeof(buf), "%Y-%m-%dT%H:%M:%S", &tmbuf);
|
||
return buf;
|
||
}
|
||
|
||
// ============================================================================
|
||
// 工具：安全字符串转 int（失败时抛出带字段名的友好错误）
|
||
// ============================================================================
|
||
|
||
static int safe_stoi(const std::string& s, const char* field_name) {
|
||
try {
|
||
size_t pos = 0;
|
||
int val = std::stoi(s, &pos);
|
||
if (pos != s.size()) {
|
||
throw std::runtime_error(std::string("invalid ") + field_name
|
||
+ ": '" + s + "' (trailing characters)");
|
||
}
|
||
return val;
|
||
} catch (const std::runtime_error&) {
|
||
throw; // 已有友好信息，直接传播
|
||
} catch (const std::exception& e) {
|
||
throw std::runtime_error(std::string("invalid ") + field_name
|
||
+ ": '" + s + "' (" + e.what() + ")");
|
||
}
|
||
}
|
||
|
||
// ============================================================================
|
||
// 配置
|
||
// ============================================================================
|
||
|
||
struct Config {
|
||
std::string listen_host = "127.0.0.1";
|
||
int listen_port = 8005;
|
||
int workers = 4;
|
||
std::string db_url = "sqlite:///database.db";
|
||
|
||
// 日志清理
|
||
int retention_days = 30;
|
||
int max_records = 1000000;
|
||
int cleanup_interval = 3600;
|
||
|
||
// 标准化 CLI 参数
|
||
std::string log_level = "info"; // debug|info|warn|error
|
||
std::string instance = "db2"; // 实例名（注册中心用）
|
||
};
|
||
|
||
// 从 sqlite:///path 或 sqlite:////abs/path 提取文件路径
|
||
static std::string db_url_to_path(const std::string& url) {
|
||
const std::string prefix = "sqlite:///";
|
||
if (url.rfind(prefix, 0) == 0) {
|
||
return url.substr(prefix.size());
|
||
}
|
||
return url; // 非 sqlite URL，原样返回（暂不支持其他后端）
|
||
}
|
||
|
||
static Config load_config(int argc, char* argv[]) {
|
||
Config cfg;
|
||
|
||
// 0. 预扫描 --help/-h
|
||
for (int i = 1; i < argc; ++i) {
|
||
std::string arg = argv[i];
|
||
if (arg == "--help" || arg == "-h") {
|
||
std::printf(
|
||
"db2 — C++20 数据库 RPC 服务\n\n"
|
||
"用法: db2 [选项]\n\n"
|
||
"选项:\n"
|
||
" --host <addr> 绑定地址 (别名: --addr/-a)\n"
|
||
" --port <num> 监听端口 (别名: -p)\n"
|
||
" --workers <num> 工作线程数 (别名: -w)\n"
|
||
" --db-url <url> 数据库路径 (如 sqlite:///database.db)\n"
|
||
" --retention-days <num> 日志保留天数 (默认: 30)\n"
|
||
" --max-records <num> 日志最大条数 (默认: 1000000)\n"
|
||
" --cleanup-interval <num> 日志清理间隔秒数 (默认: 3600)\n"
|
||
" --log-level <level> debug|info|warn|error (默认: info)\n"
|
||
" --instance <name> 实例名 (默认: db2)\n"
|
||
" --help, -h 显示帮助并退出\n\n"
|
||
"配置优先级: CLI > 环境变量 > 默认值\n");
|
||
std::exit(0);
|
||
}
|
||
}
|
||
|
||
// 1. 环境变量覆盖（Docker 兼容）
|
||
if (const char* v = std::getenv("LISTEN_ADDR")) cfg.listen_host = v;
|
||
if (const char* v = std::getenv("LISTEN_PORT")) cfg.listen_port = safe_stoi(v, "LISTEN_PORT");
|
||
if (const char* v = std::getenv("WORKERS")) cfg.workers = safe_stoi(v, "WORKERS");
|
||
if (const char* v = std::getenv("DATABASE_URL")) cfg.db_url = v;
|
||
if (const char* v = std::getenv("RETENTION_DAYS")) cfg.retention_days = safe_stoi(v, "RETENTION_DAYS");
|
||
if (const char* v = std::getenv("MAX_RECORDS")) cfg.max_records = safe_stoi(v, "MAX_RECORDS");
|
||
if (const char* v = std::getenv("CLEANUP_INTERVAL")) cfg.cleanup_interval = safe_stoi(v, "CLEANUP_INTERVAL");
|
||
|
||
// 2. 命令行参数（最高优先级）
|
||
for (int i = 1; i < argc; ++i) {
|
||
std::string arg = argv[i];
|
||
if ((arg == "--addr" || arg == "-a" || arg == "--host") && i + 1 < argc) cfg.listen_host = argv[++i];
|
||
else if ((arg == "--port" || arg == "-p") && i + 1 < argc) cfg.listen_port = safe_stoi(argv[++i], "--port");
|
||
else if ((arg == "--workers" || arg == "-w") && i + 1 < argc) cfg.workers = safe_stoi(argv[++i], "--workers");
|
||
else if (arg == "--db-url" && i + 1 < argc) cfg.db_url = argv[++i];
|
||
else if (arg == "--retention-days" && i + 1 < argc) cfg.retention_days = safe_stoi(argv[++i], "--retention-days");
|
||
else if (arg == "--max-records" && i + 1 < argc) cfg.max_records = safe_stoi(argv[++i], "--max-records");
|
||
else if (arg == "--cleanup-interval" && i + 1 < argc) cfg.cleanup_interval = safe_stoi(argv[++i], "--cleanup-interval");
|
||
else if (arg == "--log-level" && i + 1 < argc) cfg.log_level = argv[++i];
|
||
else if (arg == "--instance" && i + 1 < argc) cfg.instance = argv[++i];
|
||
}
|
||
|
||
// 3. 数值配置范围校验（防误配置导致异常行为）
|
||
if (cfg.listen_port < 1 || cfg.listen_port > 65535) {
|
||
std::fprintf(stderr, "[db2] invalid listen_port: %d (must be 1-65535)\n", cfg.listen_port);
|
||
std::exit(1);
|
||
}
|
||
if (cfg.workers < 1) {
|
||
std::fprintf(stderr, "[db2] invalid workers: %d (must be >= 1)\n", cfg.workers);
|
||
std::exit(1);
|
||
}
|
||
if (cfg.cleanup_interval < 1) {
|
||
std::fprintf(stderr, "[db2] invalid cleanup_interval: %d (must be >= 1 second)\n", cfg.cleanup_interval);
|
||
std::exit(1);
|
||
}
|
||
if (cfg.retention_days < 0) {
|
||
std::fprintf(stderr, "[db2] invalid retention_days: %d (must be >= 0)\n", cfg.retention_days);
|
||
std::exit(1);
|
||
}
|
||
if (cfg.max_records < 1) {
|
||
std::fprintf(stderr, "[db2] invalid max_records: %d (must be >= 1)\n", cfg.max_records);
|
||
std::exit(1);
|
||
}
|
||
|
||
// log_level 合法性校验
|
||
{
|
||
std::string lv = cfg.log_level;
|
||
std::transform(lv.begin(), lv.end(), lv.begin(), ::tolower);
|
||
if (lv != "debug" && lv != "info" && lv != "warn" && lv != "error") {
|
||
std::fprintf(stderr, "[db2] invalid log-level: %s (must be debug|info|warn|error)\n",
|
||
cfg.log_level.c_str());
|
||
std::exit(1);
|
||
}
|
||
cfg.log_level = lv;
|
||
}
|
||
|
||
return cfg;
|
||
}
|
||
|
||
// ============================================================================
|
||
// SQLite 连接（per-thread）
|
||
// ============================================================================
|
||
|
||
class DbConn {
|
||
public:
|
||
explicit DbConn(const std::string& path) {
|
||
if (sqlite3_open(path.c_str(), &_db) != SQLITE_OK) {
|
||
std::string err = sqlite3_errmsg(_db);
|
||
sqlite3_close(_db);
|
||
_db = nullptr;
|
||
throw std::runtime_error("sqlite3_open failed: " + err);
|
||
}
|
||
_exec_pragma("PRAGMA journal_mode=WAL");
|
||
_exec_pragma("PRAGMA synchronous=NORMAL");
|
||
_exec_pragma("PRAGMA foreign_keys=ON");
|
||
_exec_pragma("PRAGMA busy_timeout=5000");
|
||
_exec_pragma("PRAGMA mmap_size=268435456");
|
||
}
|
||
|
||
~DbConn() {
|
||
if (_db) { sqlite3_close_v2(_db); _db = nullptr; }
|
||
}
|
||
|
||
sqlite3* get() const { return _db; }
|
||
|
||
// 执行无结果集 SQL（建表、PRAGMA 等）
|
||
void exec(const std::string& sql) {
|
||
char* errmsg = nullptr;
|
||
int rc = sqlite3_exec(_db, sql.c_str(), nullptr, nullptr, &errmsg);
|
||
if (rc != SQLITE_OK) {
|
||
std::string err = errmsg ? errmsg : "unknown";
|
||
sqlite3_free(errmsg);
|
||
throw std::runtime_error("sqlite3_exec failed: " + err);
|
||
}
|
||
}
|
||
|
||
DbConn(const DbConn&) = delete;
|
||
DbConn& operator=(const DbConn&) = delete;
|
||
|
||
private:
|
||
sqlite3* _db = nullptr;
|
||
|
||
void _exec_pragma(const std::string& sql) {
|
||
char* errmsg = nullptr;
|
||
sqlite3_exec(_db, sql.c_str(), nullptr, nullptr, &errmsg);
|
||
if (errmsg) sqlite3_free(errmsg);
|
||
}
|
||
};
|
||
|
||
// ============================================================================
|
||
// 列元信息（来自 PRAGMA table_info）
|
||
// ============================================================================
|
||
|
||
struct ColInfo {
|
||
std::string name;
|
||
std::string type; // 原始类型字符串，大写，如 "DATETIME" "TEXT" "INTEGER"
|
||
};
|
||
|
||
// 表 schema 缓存项，携带 PRAGMA schema_version 以感知 DDL 变更
|
||
struct SchemaCacheEntry {
|
||
int schema_version;
|
||
std::vector<ColInfo> cols;
|
||
};
|
||
|
||
// ============================================================================
|
||
// DB2Server
|
||
// ============================================================================
|
||
|
||
class DB2Server {
|
||
public:
|
||
explicit DB2Server(const Config& cfg)
|
||
: _cfg(cfg), _db_path(db_url_to_path(cfg.db_url)) {}
|
||
|
||
void start() {
|
||
// 用主线程连接初始化 schema
|
||
{
|
||
DbConn conn(_db_path);
|
||
_ensure_logs_table(conn);
|
||
}
|
||
|
||
_log_internal("INFO", "db2", "db2 starting on "
|
||
+ _cfg.listen_host + ":" + std::to_string(_cfg.listen_port)
|
||
+ ", db=" + _db_path, "", "");
|
||
|
||
// 启动后台清理线程
|
||
_cleanup_running = true;
|
||
_cleanup_thread = std::thread(&DB2Server::_cleanup_loop, this);
|
||
|
||
// RAII 作用域守卫：无论 rpc.start() 正常返回还是抛出（socket/bind/listen 失败、
|
||
// 事件循环异常），都停止并 join 清理线程。否则异常路径下 _cleanup_thread 保持
|
||
// joinable，栈展开时 std::thread 析构会触发 std::terminate() 硬崩溃，掩盖真实错误。
|
||
struct CleanupGuard {
|
||
DB2Server& s;
|
||
~CleanupGuard() {
|
||
s._cleanup_running = false;
|
||
s._cleanup_cv.notify_all();
|
||
if (s._cleanup_thread.joinable()) s._cleanup_thread.join();
|
||
}
|
||
} guard{*this};
|
||
|
||
// 启动 RPC 服务
|
||
MsgServerConfig srv_cfg;
|
||
srv_cfg.workers = _cfg.workers;
|
||
|
||
RpcServer rpc(_cfg.listen_host, _cfg.listen_port, srv_cfg,
|
||
[](const std::string& route, const std::exception& e) {
|
||
std::fprintf(stderr, "[db2] handler error route=%s: %s\n",
|
||
route.c_str(), e.what());
|
||
});
|
||
|
||
rpc.add_route("execute", [this](const json::object& req) -> json::value {
|
||
return _handle_execute(req);
|
||
});
|
||
|
||
rpc.start();
|
||
}
|
||
|
||
private:
|
||
const Config& _cfg;
|
||
std::string _db_path;
|
||
|
||
// 表列信息缓存
|
||
std::mutex _schema_mtx;
|
||
std::unordered_map<std::string, SchemaCacheEntry> _schema_cache;
|
||
|
||
// 后台清理线程
|
||
std::atomic<bool> _cleanup_running{false};
|
||
std::thread _cleanup_thread;
|
||
std::mutex _cleanup_mtx;
|
||
std::condition_variable _cleanup_cv;
|
||
|
||
// ---- per-thread SQLite 连接 -----------------------------------------------
|
||
|
||
DbConn& _get_conn() {
|
||
thread_local std::unique_ptr<DbConn> conn;
|
||
if (!conn) conn = std::make_unique<DbConn>(_db_path);
|
||
return *conn;
|
||
}
|
||
|
||
// ---- 列 schema 反射 --------------------------------------------------------
|
||
|
||
static int _get_schema_version(sqlite3* db) {
|
||
sqlite3_stmt* stmt = nullptr;
|
||
if (sqlite3_prepare_v2(db, "PRAGMA schema_version", -1, &stmt, nullptr) != SQLITE_OK) {
|
||
throw std::runtime_error("PRAGMA schema_version failed");
|
||
}
|
||
int version = -1;
|
||
if (sqlite3_step(stmt) == SQLITE_ROW) {
|
||
version = sqlite3_column_int(stmt, 0);
|
||
}
|
||
sqlite3_finalize(stmt);
|
||
return version;
|
||
}
|
||
|
||
std::vector<ColInfo> _get_schema(const std::string& table) {
|
||
DbConn& conn = _get_conn();
|
||
const int current_version = _get_schema_version(conn.get());
|
||
|
||
{
|
||
std::lock_guard<std::mutex> lk(_schema_mtx);
|
||
auto it = _schema_cache.find(table);
|
||
if (it != _schema_cache.end() && it->second.schema_version == current_version) {
|
||
return it->second.cols;
|
||
}
|
||
}
|
||
|
||
// PRAGMA table_info 返回: cid, name, type, notnull, dflt_value, pk
|
||
std::string sql = "PRAGMA table_info(" + table + ")";
|
||
sqlite3_stmt* stmt = nullptr;
|
||
if (sqlite3_prepare_v2(conn.get(), sql.c_str(), -1, &stmt, nullptr) != SQLITE_OK) {
|
||
throw std::runtime_error("PRAGMA table_info failed for: " + table);
|
||
}
|
||
|
||
std::vector<ColInfo> cols;
|
||
while (sqlite3_step(stmt) == SQLITE_ROW) {
|
||
ColInfo ci;
|
||
const char* name = reinterpret_cast<const char*>(sqlite3_column_text(stmt, 1));
|
||
const char* type = reinterpret_cast<const char*>(sqlite3_column_text(stmt, 2));
|
||
ci.name = name ? name : "";
|
||
ci.type = type ? type : "";
|
||
// 统一大写便于比较
|
||
std::transform(ci.type.begin(), ci.type.end(), ci.type.begin(), ::toupper);
|
||
cols.push_back(ci);
|
||
}
|
||
sqlite3_finalize(stmt);
|
||
|
||
if (cols.empty()) {
|
||
throw std::runtime_error("table not found: " + table);
|
||
}
|
||
|
||
std::lock_guard<std::mutex> lk(_schema_mtx);
|
||
_schema_cache[table] = SchemaCacheEntry{current_version, std::move(cols)};
|
||
return _schema_cache[table].cols;
|
||
}
|
||
|
||
// 判断列是否为日期时间类型
|
||
static bool _is_datetime_col(const ColInfo& ci) {
|
||
// 按类型字符串
|
||
if (ci.type.find("DATETIME") != std::string::npos) return true;
|
||
if (ci.type.find("DATE") != std::string::npos) return true;
|
||
if (ci.type.find("TIME") != std::string::npos) return true;
|
||
// 按列名约定
|
||
std::string n = ci.name;
|
||
std::transform(n.begin(), n.end(), n.begin(), ::tolower);
|
||
return n.find("_at") != std::string::npos
|
||
|| n.find("time") != std::string::npos
|
||
|| n.find("date") != std::string::npos;
|
||
}
|
||
|
||
// 将 JSON 值做必要转换（$now 替换为 UTC 时间），但保留原始类型用于绑定
|
||
static json::value _coerce_value(const ColInfo& ci, const json::value& jv) {
|
||
if (jv.is_string()) {
|
||
std::string s(jv.as_string());
|
||
if (_is_datetime_col(ci) && s == "$now") return json::value(utc_now());
|
||
}
|
||
return jv;
|
||
}
|
||
|
||
static void _bind_value(sqlite3_stmt* stmt, int idx, const json::value& jv) {
|
||
if (jv.is_null()) {
|
||
sqlite3_bind_null(stmt, idx);
|
||
} else if (jv.is_int64()) {
|
||
sqlite3_bind_int64(stmt, idx, jv.as_int64());
|
||
} else if (jv.is_uint64()) {
|
||
std::uint64_t u = jv.as_uint64();
|
||
if (u > static_cast<std::uint64_t>(INT64_MAX))
|
||
throw std::runtime_error("uint64 value exceeds signed 64-bit integer range");
|
||
sqlite3_bind_int64(stmt, idx, static_cast<sqlite3_int64>(u));
|
||
} else if (jv.is_double()) {
|
||
sqlite3_bind_double(stmt, idx, jv.as_double());
|
||
} else if (jv.is_bool()) {
|
||
sqlite3_bind_int(stmt, idx, jv.as_bool() ? 1 : 0);
|
||
} else if (jv.is_string()) {
|
||
std::string s(jv.as_string());
|
||
sqlite3_bind_text(stmt, idx, s.c_str(), -1, SQLITE_TRANSIENT);
|
||
} else {
|
||
std::string s = json::serialize(jv);
|
||
sqlite3_bind_text(stmt, idx, s.c_str(), -1, SQLITE_TRANSIENT);
|
||
}
|
||
}
|
||
|
||
// 校验列名：必须存在于表 schema 中，拒绝未知列名（防 SQL 注入）
|
||
static ColInfo _validate_col(const std::string& col,
|
||
const std::unordered_map<std::string, ColInfo>& col_map) {
|
||
auto it = col_map.find(col);
|
||
if (it == col_map.end())
|
||
throw std::runtime_error("unknown column: " + col);
|
||
return it->second;
|
||
}
|
||
|
||
// ---- WHERE 子句构建 --------------------------------------------------------
|
||
|
||
struct WhereClause {
|
||
std::string sql; // "col1 = ? AND col2 > ?"
|
||
std::vector<json::value> params; // 绑定参数值（顺序对应 ?），保留 JSON 类型
|
||
};
|
||
|
||
WhereClause _build_where(const json::object& where,
|
||
const std::vector<ColInfo>& cols) {
|
||
WhereClause wc;
|
||
std::vector<std::string> parts;
|
||
|
||
if (where.size() > kMaxWhereConditions) {
|
||
throw std::runtime_error("too many where conditions: "
|
||
+ std::to_string(where.size())
|
||
+ " (max " + std::to_string(kMaxWhereConditions) + ")");
|
||
}
|
||
|
||
// 建立列名 -> ColInfo 映射
|
||
std::unordered_map<std::string, ColInfo> col_map;
|
||
for (auto& c : cols) col_map[c.name] = c;
|
||
|
||
for (auto& [key, val] : where) {
|
||
std::string col(key);
|
||
ColInfo ci = _validate_col(col, col_map);
|
||
|
||
if (val.is_object()) {
|
||
// {"op": value} 形式
|
||
for (auto& [op_key, op_val] : val.as_object()) {
|
||
std::string op(op_key);
|
||
if (op == "eq") {
|
||
parts.push_back(col + " = ?");
|
||
wc.params.push_back(_coerce_value(ci, op_val));
|
||
} else if (op == "ne") {
|
||
parts.push_back(col + " != ?");
|
||
wc.params.push_back(_coerce_value(ci, op_val));
|
||
} else if (op == "gt") {
|
||
parts.push_back(col + " > ?");
|
||
wc.params.push_back(_coerce_value(ci, op_val));
|
||
} else if (op == "lt") {
|
||
parts.push_back(col + " < ?");
|
||
wc.params.push_back(_coerce_value(ci, op_val));
|
||
} else if (op == "ge") {
|
||
parts.push_back(col + " >= ?");
|
||
wc.params.push_back(_coerce_value(ci, op_val));
|
||
} else if (op == "le") {
|
||
parts.push_back(col + " <= ?");
|
||
wc.params.push_back(_coerce_value(ci, op_val));
|
||
} else if (op == "like") {
|
||
parts.push_back(col + " LIKE ?");
|
||
wc.params.push_back(_coerce_value(ci, op_val));
|
||
} else if (op == "in") {
|
||
if (!op_val.is_array())
|
||
throw std::runtime_error("'in' operator requires an array value");
|
||
if (op_val.as_array().empty()) {
|
||
parts.push_back("0");
|
||
continue;
|
||
}
|
||
if (op_val.as_array().size() > kMaxInElements) {
|
||
throw std::runtime_error("too many elements in 'in' operator: "
|
||
+ std::to_string(op_val.as_array().size())
|
||
+ " (max " + std::to_string(kMaxInElements) + ")");
|
||
}
|
||
std::string placeholders;
|
||
for (auto& item : op_val.as_array()) {
|
||
if (!placeholders.empty()) placeholders += ", ";
|
||
placeholders += "?";
|
||
wc.params.push_back(_coerce_value(ci, item));
|
||
}
|
||
parts.push_back(col + " IN (" + placeholders + ")");
|
||
} else {
|
||
throw std::runtime_error("unknown operator: " + op);
|
||
}
|
||
}
|
||
} else {
|
||
// 直接等于
|
||
parts.push_back(col + " = ?");
|
||
wc.params.push_back(_coerce_value(ci, val));
|
||
}
|
||
}
|
||
|
||
for (size_t i = 0; i < parts.size(); ++i) {
|
||
if (i > 0) wc.sql += " AND ";
|
||
wc.sql += parts[i];
|
||
}
|
||
return wc;
|
||
}
|
||
|
||
// ---- 参数绑定辅助 ----------------------------------------------------------
|
||
|
||
static void _bind_params(sqlite3_stmt* stmt, int start_idx,
|
||
const std::vector<json::value>& params) {
|
||
for (size_t i = 0; i < params.size(); ++i) {
|
||
_bind_value(stmt, start_idx + static_cast<int>(i), params[i]);
|
||
}
|
||
}
|
||
|
||
// ---- execute 路由 ----------------------------------------------------------
|
||
|
||
json::value _handle_execute(const json::object& req) {
|
||
// 提取公共字段
|
||
std::string action = "select";
|
||
if (auto it = req.find("action"); it != req.end()) {
|
||
if (!it->value().is_string())
|
||
throw std::runtime_error("'action' must be a string");
|
||
action = std::string(it->value().as_string());
|
||
}
|
||
|
||
if (!req.contains("table") || !req.at("table").is_string())
|
||
throw std::runtime_error("missing required field: table");
|
||
std::string table(req.at("table").as_string());
|
||
|
||
// 验证表名（仅允许字母数字下划线，防止 SQL 注入）
|
||
for (char c : table) {
|
||
if (!std::isalnum(static_cast<unsigned char>(c)) && c != '_')
|
||
throw std::runtime_error("invalid table name: " + table);
|
||
}
|
||
|
||
std::vector<ColInfo> cols = _get_schema(table);
|
||
|
||
// 'where' 存在但非对象时立即报错，不静默忽略（与 action/columns 校验一致）
|
||
if (auto wit = req.find("where"); wit != req.end() && !wit->value().is_object()) {
|
||
throw std::runtime_error("'where' must be an object");
|
||
}
|
||
|
||
if (action == "select") return _do_select(req, table, cols);
|
||
if (action == "insert") return _do_insert(req, table, cols);
|
||
if (action == "update") return _do_update(req, table, cols);
|
||
if (action == "delete") return _do_delete(req, table, cols);
|
||
throw std::runtime_error("unknown action: " + action);
|
||
}
|
||
|
||
// ---- SELECT ----------------------------------------------------------------
|
||
|
||
json::value _do_select(const json::object& req,
|
||
const std::string& table,
|
||
const std::vector<ColInfo>& cols) {
|
||
DbConn& conn = _get_conn();
|
||
|
||
std::unordered_map<std::string, ColInfo> col_map;
|
||
for (auto& c : cols) col_map[c.name] = c;
|
||
|
||
// columns
|
||
std::string col_expr = "*";
|
||
if (auto it = req.find("columns");
|
||
it != req.end() && it->value().is_array() && !it->value().as_array().empty()) {
|
||
col_expr.clear();
|
||
for (auto& c : it->value().as_array()) {
|
||
if (!c.is_string())
|
||
throw std::runtime_error("columns array must contain only strings");
|
||
std::string col_name = std::string(c.as_string());
|
||
_validate_col(col_name, col_map);
|
||
if (!col_expr.empty()) col_expr += ", ";
|
||
col_expr += col_name;
|
||
}
|
||
}
|
||
|
||
std::string sql = "SELECT " + col_expr + " FROM " + table;
|
||
WhereClause wc;
|
||
|
||
if (auto it = req.find("where"); it != req.end() && it->value().is_object())
|
||
wc = _build_where(it->value().as_object(), cols);
|
||
if (!wc.sql.empty()) sql += " WHERE " + wc.sql;
|
||
|
||
// order_by
|
||
if (auto it = req.find("order_by");
|
||
it != req.end() && it->value().is_array()) {
|
||
std::string order;
|
||
for (auto& ob : it->value().as_array()) {
|
||
if (!ob.is_object())
|
||
throw std::runtime_error("order_by array must contain only objects");
|
||
auto& o = ob.as_object();
|
||
if (!o.contains("column") || !o.at("column").is_string())
|
||
throw std::runtime_error("order_by entry requires 'column' string field");
|
||
std::string col_name = std::string(o.at("column").as_string());
|
||
_validate_col(col_name, col_map);
|
||
if (!order.empty()) order += ", ";
|
||
order += col_name;
|
||
if (o.contains("desc")) {
|
||
if (!o.at("desc").is_bool())
|
||
throw std::runtime_error("order_by 'desc' must be boolean");
|
||
if (o.at("desc").as_bool()) order += " DESC";
|
||
}
|
||
}
|
||
if (!order.empty()) sql += " ORDER BY " + order;
|
||
}
|
||
|
||
// limit / offset（类型不符立即报错，不静默忽略，与 action 校验一致）
|
||
if (auto it = req.find("limit"); it != req.end()) {
|
||
if (!it->value().is_int64())
|
||
throw std::runtime_error("limit must be an integer");
|
||
std::int64_t limit = it->value().as_int64();
|
||
if (limit < 0)
|
||
throw std::runtime_error("limit must be non-negative");
|
||
sql += " LIMIT " + std::to_string(limit);
|
||
}
|
||
if (auto it = req.find("offset"); it != req.end()) {
|
||
if (!it->value().is_int64())
|
||
throw std::runtime_error("offset must be an integer");
|
||
std::int64_t offset = it->value().as_int64();
|
||
if (offset < 0)
|
||
throw std::runtime_error("offset must be non-negative");
|
||
sql += " OFFSET " + std::to_string(offset);
|
||
}
|
||
|
||
sqlite3_stmt* stmt = nullptr;
|
||
if (sqlite3_prepare_v2(conn.get(), sql.c_str(), -1, &stmt, nullptr) != SQLITE_OK) {
|
||
throw std::runtime_error(std::string("prepare failed: ") + sqlite3_errmsg(conn.get()));
|
||
}
|
||
_bind_params(stmt, 1, wc.params);
|
||
|
||
// 获取结果列名（可能是 * 展开后的实际列名）
|
||
int ncols = sqlite3_column_count(stmt);
|
||
std::vector<std::string> result_cols;
|
||
result_cols.reserve(ncols);
|
||
for (int i = 0; i < ncols; ++i) {
|
||
const char* n = sqlite3_column_name(stmt, i);
|
||
result_cols.push_back(n ? n : "");
|
||
}
|
||
|
||
json::array rows;
|
||
while (sqlite3_step(stmt) == SQLITE_ROW) {
|
||
json::object row;
|
||
for (int i = 0; i < ncols; ++i) {
|
||
int ctype = sqlite3_column_type(stmt, i);
|
||
if (ctype == SQLITE_NULL) {
|
||
row[result_cols[i]] = nullptr;
|
||
} else if (ctype == SQLITE_INTEGER) {
|
||
row[result_cols[i]] = sqlite3_column_int64(stmt, i);
|
||
} else if (ctype == SQLITE_FLOAT) {
|
||
row[result_cols[i]] = sqlite3_column_double(stmt, i);
|
||
} else {
|
||
const unsigned char* txt = sqlite3_column_text(stmt, i);
|
||
row[result_cols[i]] = txt ? reinterpret_cast<const char*>(txt) : "";
|
||
}
|
||
}
|
||
rows.push_back(row);
|
||
}
|
||
sqlite3_finalize(stmt);
|
||
return rows;
|
||
}
|
||
|
||
// ---- INSERT ----------------------------------------------------------------
|
||
|
||
json::value _do_insert(const json::object& req,
|
||
const std::string& table,
|
||
const std::vector<ColInfo>& cols) {
|
||
if (!req.contains("values") || !req.at("values").is_object())
|
||
throw std::runtime_error("insert requires 'values' object");
|
||
|
||
DbConn& conn = _get_conn();
|
||
const json::object& vals = req.at("values").as_object();
|
||
if (vals.empty())
|
||
throw std::runtime_error("insert requires non-empty 'values'");
|
||
|
||
// 建列名 -> ColInfo 映射
|
||
std::unordered_map<std::string, ColInfo> col_map;
|
||
for (auto& c : cols) col_map[c.name] = c;
|
||
|
||
std::string col_list, ph_list;
|
||
std::vector<json::value> params;
|
||
|
||
for (auto& [k, v] : vals) {
|
||
std::string col_name(k);
|
||
ColInfo ci = _validate_col(col_name, col_map);
|
||
if (!col_list.empty()) { col_list += ", "; ph_list += ", "; }
|
||
col_list += col_name;
|
||
ph_list += "?";
|
||
params.push_back(_coerce_value(ci, v));
|
||
}
|
||
|
||
std::string sql = "INSERT INTO " + table + " (" + col_list + ") VALUES (" + ph_list + ")";
|
||
sqlite3_stmt* stmt = nullptr;
|
||
if (sqlite3_prepare_v2(conn.get(), sql.c_str(), -1, &stmt, nullptr) != SQLITE_OK)
|
||
throw std::runtime_error(std::string("prepare failed: ") + sqlite3_errmsg(conn.get()));
|
||
|
||
_bind_params(stmt, 1, params);
|
||
int rc = sqlite3_step(stmt);
|
||
sqlite3_finalize(stmt);
|
||
if (rc != SQLITE_DONE)
|
||
throw std::runtime_error(std::string("insert failed: ") + sqlite3_errmsg(conn.get()));
|
||
|
||
sqlite3_int64 rowid = sqlite3_last_insert_rowid(conn.get());
|
||
return json::array{json::object{{"lastrowid", rowid}}};
|
||
}
|
||
|
||
// ---- UPDATE ----------------------------------------------------------------
|
||
|
||
json::value _do_update(const json::object& req,
|
||
const std::string& table,
|
||
const std::vector<ColInfo>& cols) {
|
||
if (!req.contains("values") || !req.at("values").is_object())
|
||
throw std::runtime_error("update requires 'values' object");
|
||
|
||
DbConn& conn = _get_conn();
|
||
const json::object& vals = req.at("values").as_object();
|
||
if (vals.empty())
|
||
throw std::runtime_error("update requires non-empty 'values'");
|
||
|
||
std::unordered_map<std::string, ColInfo> col_map;
|
||
for (auto& c : cols) col_map[c.name] = c;
|
||
|
||
std::string set_clause;
|
||
std::vector<json::value> params;
|
||
|
||
for (auto& [k, v] : vals) {
|
||
std::string col_name(k);
|
||
ColInfo ci = _validate_col(col_name, col_map);
|
||
if (!set_clause.empty()) set_clause += ", ";
|
||
set_clause += col_name + " = ?";
|
||
params.push_back(_coerce_value(ci, v));
|
||
}
|
||
|
||
std::string sql = "UPDATE " + table + " SET " + set_clause;
|
||
WhereClause wc;
|
||
if (auto it = req.find("where"); it != req.end() && it->value().is_object())
|
||
wc = _build_where(it->value().as_object(), cols);
|
||
if (!wc.sql.empty()) {
|
||
sql += " WHERE " + wc.sql;
|
||
params.insert(params.end(), wc.params.begin(), wc.params.end());
|
||
}
|
||
|
||
sqlite3_stmt* stmt = nullptr;
|
||
if (sqlite3_prepare_v2(conn.get(), sql.c_str(), -1, &stmt, nullptr) != SQLITE_OK)
|
||
throw std::runtime_error(std::string("prepare failed: ") + sqlite3_errmsg(conn.get()));
|
||
|
||
_bind_params(stmt, 1, params);
|
||
int rc = sqlite3_step(stmt);
|
||
sqlite3_finalize(stmt);
|
||
if (rc != SQLITE_DONE)
|
||
throw std::runtime_error(std::string("update failed: ") + sqlite3_errmsg(conn.get()));
|
||
|
||
int rowcount = sqlite3_changes(conn.get());
|
||
return json::array{json::object{{"rowcount", rowcount}}};
|
||
}
|
||
|
||
// ---- DELETE ----------------------------------------------------------------
|
||
|
||
json::value _do_delete(const json::object& req,
|
||
const std::string& table,
|
||
const std::vector<ColInfo>& cols) {
|
||
DbConn& conn = _get_conn();
|
||
|
||
std::string sql = "DELETE FROM " + table;
|
||
WhereClause wc;
|
||
if (auto it = req.find("where"); it != req.end() && it->value().is_object())
|
||
wc = _build_where(it->value().as_object(), cols);
|
||
if (!wc.sql.empty()) sql += " WHERE " + wc.sql;
|
||
|
||
sqlite3_stmt* stmt = nullptr;
|
||
if (sqlite3_prepare_v2(conn.get(), sql.c_str(), -1, &stmt, nullptr) != SQLITE_OK)
|
||
throw std::runtime_error(std::string("prepare failed: ") + sqlite3_errmsg(conn.get()));
|
||
|
||
_bind_params(stmt, 1, wc.params);
|
||
int rc = sqlite3_step(stmt);
|
||
sqlite3_finalize(stmt);
|
||
if (rc != SQLITE_DONE)
|
||
throw std::runtime_error(std::string("delete failed: ") + sqlite3_errmsg(conn.get()));
|
||
|
||
int rowcount = sqlite3_changes(conn.get());
|
||
return json::array{json::object{{"rowcount", rowcount}}};
|
||
}
|
||
|
||
// ---- Logs 表初始化 ---------------------------------------------------------
|
||
|
||
void _ensure_logs_table(DbConn& conn) {
|
||
conn.exec(R"(
|
||
CREATE TABLE IF NOT EXISTS Logs (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
ts TEXT NOT NULL,
|
||
service TEXT NOT NULL,
|
||
level TEXT NOT NULL,
|
||
message TEXT NOT NULL,
|
||
trace_id TEXT,
|
||
extra TEXT
|
||
)
|
||
)");
|
||
conn.exec("CREATE INDEX IF NOT EXISTS idx_logs_service_ts ON Logs(service, ts)");
|
||
conn.exec("CREATE INDEX IF NOT EXISTS idx_logs_level_ts ON Logs(level, ts)");
|
||
conn.exec("CREATE INDEX IF NOT EXISTS idx_logs_ts ON Logs(ts)");
|
||
conn.exec("CREATE INDEX IF NOT EXISTS idx_logs_trace ON Logs(trace_id)");
|
||
}
|
||
|
||
// ---- 内部日志直写（绕过 RPC，防递归） --------------------------------------
|
||
|
||
void _log_internal(const std::string& level, const std::string& service,
|
||
std::string message, const std::string& trace_id,
|
||
const std::string& extra) {
|
||
// 日志级别过滤
|
||
static const std::unordered_map<std::string, int> kLogLevels = {
|
||
{"debug", 0}, {"info", 1}, {"warn", 2}, {"warning", 2}, {"error", 3},
|
||
{"DEBUG", 0}, {"INFO", 1}, {"WARN", 2}, {"WARNING", 2}, {"ERROR", 3}
|
||
};
|
||
auto it_cfg = kLogLevels.find(_cfg.log_level);
|
||
auto it_msg = kLogLevels.find(level);
|
||
if (it_cfg != kLogLevels.end() && it_msg != kLogLevels.end()
|
||
&& it_msg->second < it_cfg->second) {
|
||
return; // 低于设定级别，丢弃
|
||
}
|
||
|
||
if (message.size() > kMaxMessage) message.resize(kMaxMessage);
|
||
std::string ex = extra;
|
||
if (ex.size() > kMaxExtra) ex.resize(kMaxExtra);
|
||
|
||
try {
|
||
DbConn& conn = _get_conn();
|
||
std::string sql =
|
||
"INSERT INTO Logs (ts, service, level, message, trace_id, extra) "
|
||
"VALUES (?, ?, ?, ?, ?, ?)";
|
||
sqlite3_stmt* stmt = nullptr;
|
||
if (sqlite3_prepare_v2(conn.get(), sql.c_str(), -1, &stmt, nullptr) != SQLITE_OK)
|
||
return;
|
||
|
||
std::string ts = utc_now();
|
||
sqlite3_bind_text(stmt, 1, ts.c_str(), -1, SQLITE_TRANSIENT);
|
||
sqlite3_bind_text(stmt, 2, service.c_str(), -1, SQLITE_TRANSIENT);
|
||
sqlite3_bind_text(stmt, 3, level.c_str(), -1, SQLITE_TRANSIENT);
|
||
sqlite3_bind_text(stmt, 4, message.c_str(), -1, SQLITE_TRANSIENT);
|
||
if (trace_id.empty()) sqlite3_bind_null(stmt, 5);
|
||
else sqlite3_bind_text(stmt, 5, trace_id.c_str(), -1, SQLITE_TRANSIENT);
|
||
if (ex.empty()) sqlite3_bind_null(stmt, 6);
|
||
else sqlite3_bind_text(stmt, 6, ex.c_str(), -1, SQLITE_TRANSIENT);
|
||
|
||
sqlite3_step(stmt);
|
||
sqlite3_finalize(stmt);
|
||
} catch (const std::exception& e) {
|
||
std::fprintf(stderr, "[db2] _log_internal failed: %s\n", e.what());
|
||
}
|
||
}
|
||
|
||
// ---- 后台日志清理线程 -------------------------------------------------------
|
||
|
||
void _cleanup_loop() {
|
||
while (_cleanup_running) {
|
||
std::unique_lock<std::mutex> lk(_cleanup_mtx);
|
||
_cleanup_cv.wait_for(lk,
|
||
std::chrono::seconds(_cfg.cleanup_interval),
|
||
[this] { return !_cleanup_running.load(); });
|
||
if (!_cleanup_running) break;
|
||
_do_cleanup();
|
||
}
|
||
}
|
||
|
||
void _do_cleanup() {
|
||
try {
|
||
DbConn& conn = _get_conn();
|
||
|
||
// 1. 删除超过 retention_days 的记录
|
||
std::string cutoff_sql =
|
||
"DELETE FROM Logs WHERE ts < strftime('%Y-%m-%dT%H:%M:%S','now', '-"
|
||
+ std::to_string(_cfg.retention_days) + " days')";
|
||
conn.exec(cutoff_sql);
|
||
int expired = sqlite3_changes(conn.get());
|
||
|
||
// 2. 若总条数超过 max_records，删除最旧的超出部分
|
||
int trimmed = 0;
|
||
{
|
||
sqlite3_stmt* stmt = nullptr;
|
||
std::string cnt_sql = "SELECT COUNT(*) FROM Logs";
|
||
if (sqlite3_prepare_v2(conn.get(), cnt_sql.c_str(), -1, &stmt, nullptr) == SQLITE_OK) {
|
||
if (sqlite3_step(stmt) == SQLITE_ROW) {
|
||
int total = sqlite3_column_int(stmt, 0);
|
||
if (total > _cfg.max_records) {
|
||
int to_delete = total - _cfg.max_records;
|
||
sqlite3_finalize(stmt);
|
||
stmt = nullptr;
|
||
std::string trim_sql =
|
||
"DELETE FROM Logs WHERE id IN "
|
||
"(SELECT id FROM Logs ORDER BY id ASC LIMIT "
|
||
+ std::to_string(to_delete) + ")";
|
||
conn.exec(trim_sql);
|
||
trimmed = sqlite3_changes(conn.get());
|
||
}
|
||
}
|
||
} else {
|
||
std::fprintf(stderr, "[db2] cleanup COUNT prepare failed: %s\n",
|
||
sqlite3_errmsg(conn.get()));
|
||
}
|
||
if (stmt) sqlite3_finalize(stmt);
|
||
}
|
||
|
||
if (expired > 0 || trimmed > 0) {
|
||
_log_internal("INFO", "db2",
|
||
"cleanup: expired=" + std::to_string(expired)
|
||
+ " trimmed=" + std::to_string(trimmed), "", "");
|
||
}
|
||
} catch (const std::exception& e) {
|
||
std::fprintf(stderr, "[db2] cleanup error: %s\n", e.what());
|
||
}
|
||
}
|
||
};
|
||
|
||
// ============================================================================
|
||
// main
|
||
// ============================================================================
|
||
|
||
int main(int argc, char* argv[]) {
|
||
try {
|
||
Config cfg = load_config(argc, argv);
|
||
|
||
std::printf("[db2] listen=%s:%d workers=%d db=%s instance=%s log-level=%s\n",
|
||
cfg.listen_host.c_str(), cfg.listen_port,
|
||
cfg.workers, cfg.db_url.c_str(),
|
||
cfg.instance.c_str(), cfg.log_level.c_str());
|
||
std::fflush(stdout);
|
||
|
||
DB2Server server(cfg);
|
||
server.start();
|
||
} catch (const std::exception& e) {
|
||
std::fprintf(stderr, "[db2] fatal: %s\n", e.what());
|
||
return 1;
|
||
}
|
||
return 0;
|
||
}
|