JSON 数据库同步

一个使用 Rust 构建的客户端-服务器同步系统，具备实时 WebSocket 通信、基于补丁的双向版本控制以及自动冲突解决功能。

功能

实时同步

基于 WebSocket 的双向同步
基础冲突解决，采用服务器优先的回退策略
离线优先 设计，支持本地存储
并发更新处理，具备冲突检测

版本控制

JSON 补丁：用于文档变更的前向补丁
文档版本化：每个文档都有版本和修订跟踪
变更事件：所有修改均记录序列号
审计跟踪：记录变更并关联用户

数据库架构

客户端：SQLite，支持离线队列和文档缓存
服务器：PostgreSQL，支持 JSONB 和事件日志
变更事件：基于序列的同步，存储前向/反向补丁

架构

sync-core：共享库，包含数据模型、JSON 补丁操作和同步协议
sync-client：客户端库，包含 SQLite 存储、离线队列和 C FFI 导出
sync-server：服务器二进制文件，包含 PostgreSQL 存储、WebSocket 支持和事件日志

快速开始

先决条件

Rust 1.75+
PostgreSQL 16+
Docker（可选）

选项 1：Docker（推荐）

git clone https://github.com/yourusername/json-db-sync.git
cd json-db-sync/sync-workspace
docker-compose up -d

这将启动 PostgreSQL 和同步服务器，端口为 8080。

选项 2：手动设置

安装依赖项：

cargo build --workspace

设置 PostgreSQL：

createdb sync_db
export DATABASE_URL="postgres://user:password@localhost/sync_db"

运行迁移：

cd sync-server
sqlx migrate run

生成 API 凭据：

cargo run --bin sync-server generate-credentials --name "My Application"

安全地保存生成的 API 密钥和密钥。密钥将不会再次显示。

启动服务器：

cargo run --bin sync-server serve

试用交互式客户端

# Using your generated API credentials
cargo run --package sync-client --example interactive_client -- \
    --api-key "rpa_your_api_key_here"

# Or specify a different database and user
cargo run --package sync-client --example interactive_client -- \
    --database bob \
    --api-key "rpa_your_api_key_here" \
    --user-id "user@example.com"

交互式客户端提供任务管理界面，包括：

创建、编辑和完成任务
优先级级别和标签
带有状态指示符的丰富任务列表
跨多个客户端的实时同步
离线支持，重新连接时自动同步

API 使用

Rust 客户端库

use sync_client::SyncEngine;
use serde_json::json;

// Connect to server with HMAC authentication
let engine = SyncEngine::new(
    "sqlite:client.db?mode=rwc",
    "ws://localhost:8080/ws",
    "rpa_your_api_key_here",
    "user@example.com"
).await?;

// Create a document
let doc = engine.create_document(
    json!({
        "title": "My Document",
        "description": "Task description",
        "status": "pending",
        "priority": "medium",
        "tags": ["work", "important"]
    })
).await?;

// Update document
engine.update_document(doc.id, json!({
    "status": "completed"
})).await?;

Rust 事件回调

use sync_client::events::{EventDispatcher, EventType};

// Get event dispatcher from engine
let events = engine.event_dispatcher();

// Register callback for document events
events.register_callback(
    |event_type, document_id, title, content, error, numeric_data, boolean_data, context| {
        match event_type {
            EventType::DocumentCreated => {
                println!("New document: {}", title.unwrap_or("untitled"));
            },
            EventType::DocumentUpdated => {
                println!("Updated: {}", title.unwrap_or("untitled"));
            },
            EventType::SyncCompleted => {
                println!("Synced {} documents", numeric_data);
            },
            _ => {}
        }
    },
    std::ptr::null_mut(), // context
    None // filter (None = all events)
)?;

// In your main loop
loop {
    let processed = events.process_events()?;
    if processed > 0 {
        println!("Processed {} events", processed);
    }
    
    tokio::time::sleep(tokio::time::Duration::from_millis(16)).await;
}

WebSocket API

连接到 ws://localhost:8080/ws 并使用 HMAC 签名进行身份验证：

{
  "type": "authenticate",
  "email": "user@example.com",
  "client_id": "550e8400-e29b-41d4-a716-446655440000",
  "api_key": "rpa_your_api_key_here",
  "signature": "calculated_hmac_signature",
  "timestamp": 1736525432
}

HMAC 签名计算如下：HMAC-SHA256(secret, "timestamp.email.api_key.body")

创建文档：

{
  "type": "create_document",
  "document": {
    "id": "550e8400-e29b-41d4-a716-446655440001",
    "title": "My Document",
    "content": {"text": "Hello, World!"}
  }
}

使用 JSON 补丁更新：

{
  "type": "update_document",
  "patch": {
    "document_id": "550e8400-e29b-41d4-a716-446655440001",
    "patch": [
      {"op": "replace", "path": "/text", "value": "Updated text"}
    ]
  }
}

C/C++ 集成

同步客户端提供了一个 C API，可从 C、C++ 及其他语言中使用。构建分发 SDK：

# Build the complete SDK with headers and libraries
./scripts/build_dist.sh

# This creates:
# - dist/include/sync_client.h    (C header)
# - dist/include/sync_client.hpp  (C++ wrapper)
# - dist/lib/                     (compiled libraries)
# - dist/examples/                (working examples)

事件回调系统

同步客户端包含一个全面的事件回调系统，用于实时通知文档更改、同步操作和连接状态。

主要特性：

线程安全设计，用户代码中无需加锁
支持 Rust、C 和 C++ 应用程序
事件过滤和上下文传递
离线和在线事件通知

快速 C 示例：

#include "sync_client_events.h"

void my_callback(const SyncEventData* event, void* context) {
    switch (event->event_type) {
        case SYNC_EVENT_DOCUMENT_CREATED:
            printf("Document created: %s\n", event->title);
            break;
        case SYNC_EVENT_SYNC_COMPLETED:
            printf("Sync completed: %llu documents\n", event->numeric_data);
            break;
    }
}

int main() {
    struct CSyncEngine* engine = sync_engine_create(
        "sqlite:client.db",
        "ws://localhost:8080/ws",
        "user@example.com",
        "rpa_your_api_key_here",
        "rps_your_secret_here");
    
    // Register callback for all events
    sync_engine_register_event_callback(engine, my_callback, NULL, -1);
    
    // Main loop - process events regularly
    while (running) {
        uint32_t processed_count;
        sync_engine_process_events(engine, &processed_count);
        // ... do other work
        usleep(16000); // ~60 FPS
    }
    
    sync_engine_destroy(engine);
    return 0;
}

带 Lambda 的 C++ 示例：

#include "sync_client_events.h"

int main() {
    // Simple RAII wrapper
    SyncEngine engine("sqlite:client.db", "ws://localhost:8080/ws",
                     "user@example.com", "rpa_your_api_key_here",
                     "rps_your_secret_here");
    
    // Lambda callback with capture
    int event_count = 0;
    auto callback = [&event_count](const SyncEventData* event, void* context) {
        event_count++;
        std::cout << "Event #" << event_count << ": " 
                 << get_event_name(event->event_type) << std::endl;
    };
    
    // Register callback
    engine.register_callback(callback);
    
    // Main loop
    while (running) {
        engine.process_events();
        std::this_thread::sleep_for(std::chrono::milliseconds(16));
    }
    
    return 0;
}

有关完整文档和高级示例，请参阅 EVENT_CALLBACKS.md。

框架集成

事件回调系统通过定时器与桌面框架集成：

JUCE：使用 juce::Timer 每 16ms 调用一次 process_events()
Qt：使用 QTimer 配合 Qt 的信号/槽系统
GTK：使用 g_timeout_add() 进行周期性处理
所有回调均在框架的主线程上执行

请参阅 FRAMEWORK_INTEGRATION.md 获取完整示例。

C API 示例：

#include "sync_client.h"
#include <stdio.h>
#include <stdlib.h>

int main() {
    // Create sync engine with HMAC authentication
    struct CSyncEngine* engine = sync_engine_create(
        "sqlite:client.db?mode=rwc",
        "ws://localhost:8080/ws",
        "user@example.com",
        "rpa_your_api_key_here",
        "rps_your_secret_here"
    );
    
    if (!engine) {
        printf("Failed to create sync engine\n");
        return 1;
    }
    
    // Get version
    char* version = sync_get_version();
    if (version) {
        printf("Sync client version: %s\n", version);
        sync_string_free(version);
    }
    
    // Create a document
    char doc_id[37] = {0}; // UUID string + null terminator
    enum CSyncResult result = sync_engine_create_document(
        engine,
        "{\"title\":\"My Document\",\"content\":\"Hello World\",\"type\":\"note\",\"priority\":\"medium\"}",
        doc_id
    );
    
    if (result == Success) {
        printf("Created document: %s\n", doc_id);
        
        // Update the document
        result = sync_engine_update_document(
            engine,
            doc_id,
            "{\"content\":\"Hello Updated World\",\"type\":\"note\",\"priority\":\"high\"}"
        );
        
        if (result == Success) {
            printf("Updated document successfully\n");
        }
    }
    
    // Clean up
    sync_engine_destroy(engine);
    return 0;
}

C++ 集成：

#include "replicant.hpp"
#include <iostream>

int main()
{
    try
    {
        // Create client with HMAC authentication
        replicant::Client client(
            "sqlite:client.db?mode=rwc",
            "ws://localhost:8080/ws",
            "user@example.com",
            "rpa_your_api_key_here",
            "rps_your_secret_here"
        );

        std::cout << "Replicant version: " << replicant::Client::get_version() << std::endl;

        // Create a document
        auto doc_id = client.create_document(
            R"({"title":"My Document","content":"Hello World","type":"note","priority":"medium"})"
        );

        std::cout << "Created document: " << doc_id << std::endl;

        // Update the document
        client.update_document(
            doc_id,
            R"({"content":"Hello Updated World","type":"note","priority":"high"})"
        );

        std::cout << "Updated document successfully" << std::endl;
    }
    catch (const replicant::SyncException& e)
    {
        std::cerr << "Replicant error: " << e.what() << std::endl;
        return 1;
    }

    return 0;
}

构建与链接：

使用分发 SDK（推荐）：

# CMakeLists.txt
cmake_minimum_required(VERSION 3.15)
project(MyProject)

set(CMAKE_CXX_STANDARD 11)

# Find the sync client SDK
find_package(sync_client REQUIRED PATHS /path/to/sync-workspace/dist)

# Your C++ application
add_executable(my_app main.cpp)
target_link_libraries(my_app sync_client)

或直接编译：

# Build the distribution first
./scripts/build_dist.sh

# Compile your C/C++ code (example for macOS/Linux)
gcc -I./dist/include your_code.c -L./dist/lib -lsync_client -framework Security -o your_program

# For C++
g++ -std=c++11 -I./dist/include your_code.cpp -L./dist/lib -lsync_client -framework Security -o your_program

双向补丁系统

系统为每次文档变更同时存储正向和反向补丁：

工作原理

CREATE 事件：

forward_patch: 包含作为初始状态的完整文档
reverse_patch: null (创建操作不可撤销)

UPDATE 事件：

forward_patch: 用于应用变更的 JSON 补丁
reverse_patch: 用于撤销变更的 JSON 补丁

DELETE 事件：

forward_patch: null (删除是隐式的)
reverse_patch: 包含完整文档，以便在取消删除时恢复

优势

完整的审计追踪：所有变更均被记录，并具备恢复能力
高效的存储：JSON 补丁仅存储版本之间的差异
版本历史：支持正向和反向补丁，为未来的撤销功能提供支持

冲突解决

系统通过基本的冲突检测机制处理并发更新：

多个客户端 可以同时更新同一文档
服务器按顺序处理更新，并使用向量时钟检测冲突
服务器优先回退 用于冲突解决（客户端接受服务器状态）
冲突检测 通过向量时钟比较实现

测试

单元测试

cargo test --lib --bins

集成测试

# Local setup with PostgreSQL (fast)
./test/run_integration_tests_local.sh

# Docker-based setup (consistent environment)
./test/run_integration_tests_docker.sh

# Manual setup
docker-compose -f docker-compose.test.yml up -d
export RUN_INTEGRATION_TESTS=1
export TEST_DATABASE_URL="postgres://postgres:postgres@localhost:5433/sync_test_db"
cargo test integration -- --test-threads=1

测试覆盖率

sync-core：7 个测试（向量时钟、文档修订、JSON 补丁）
sync-client：1 个测试（基本功能）
sync-server：7 个测试（身份验证、WebSocket 协议、并发场景）

身份验证

系统使用基于 HMAC 请求签名的两级身份验证模型：

认证级别

应用级别：API 凭据用于认证应用程序
- API Key 格式：rpa_* (Replicant API)
- Secret 格式：rps_* (Replicant Secret)
- 凭据使用 CLI 工具生成
- 每个应用程序/环境一套凭据
用户级别：电子邮件用于识别要访问哪个用户的数据
- 用户通过电子邮件地址识别
- 无需密码（安全性在应用级别处理）
- 用户记录在首次认证时自动创建

生成凭据

使用服务器 CLI 生成 API 凭据：

cargo run --bin sync-server generate-credentials --name "Production"

这输出：

API Key:    rpa_a8d73487645ef2b9c3d4e5f6a7b8c9d0
Secret:     rps_1f2e3d4c5b6a798897a6b5c4d3e2f1a0

重要：请安全地保存该密钥 - 它不会再次显示。

HMAC 签名

所有经过身份验证的请求都需要一个 HMAC-SHA256 签名：

Message format: timestamp.email.api_key.body
Timestamp validation: Requests expire after 5 minutes
Signature verification: Prevents tampering and replay attacks

安全注意事项

传输安全：在生产环境中使用 WSS/HTTPS
凭据存储：安全地存储 API 密钥（环境变量、密钥管理器）
速率限制：实施速率限制以防止暴力破解尝试
审计日志：跟踪身份验证尝试及失败

性能与安全

优化

针对 PostgreSQL 和 SQLite 的连接池
用于快速文档搜索的 JSONB 索引
用于高效变更跟踪的基于补丁的存储

安全特性

基于 HMAC 的认证，包含时间戳验证
API 凭据存储在数据库中（MVP 阶段为明文）
针对 JSON 补丁的输入验证
通过时间戳检查防止重放攻击

许可证

MIT

贡献

Fork the repository
Create a feature branch
Add tests for new functionality
Run cargo test and ./test/run_integration_tests_docker.sh
Submit a pull request

参见 TESTING.md 获取详细的测试指南，以及 EXAMPLES.md 获取使用示例。