InstantDB 集成

概述

InstantDB 的 Node.js 集成，使 OpenClaw 能够执行管理操作，并通过 WebSocket 订阅实时监控数据变更。

配置

安装依赖：

npm install

设置环境变量：

export INSTANTDB_APP_ID="your-app-id"
export INSTANTDB_ADMIN_TOKEN="your-admin-token"

核心能力

1. 查询数据

使用 InstantDB 的查询语法获取数据：

const { InstantDBClient } = require('./scripts/instantdb.js');

const client = new InstantDBClient(appId, adminToken);
const result = await client.query({
  tasks: {
    $: {
      where: { status: 'active' }
    }
  }
});

CLI 命令：

./scripts/instantdb.js query '{"tasks": {}}'

2. 创建实体

向命名空间添加新实体：

const { entityId, result } = await client.createEntity('tasks', {
  title: 'Process data',
  status: 'pending',
  priority: 'high'
});

CLI 命令：

./scripts/instantdb.js create tasks '{"title": "Process data", "status": "pending"}'

指定自定义实体 ID（可选）：

./scripts/instantdb.js create tasks '{"title": "Task"}' custom-entity-id

3. 更新实体

修改现有实体的属性：

await client.updateEntity(entityId, 'tasks', {
  status: 'completed'
});

CLI 命令：

./scripts/instantdb.js update  tasks '{"status": "completed"}'

4. 删除实体

移除实体：

await client.deleteEntity(entityId, 'tasks');

CLI 命令：

./scripts/instantdb.js delete  tasks

5. 关联实体

在实体之间建立关系：

await client.linkEntities(taskId, assigneeId, 'assignees');

CLI 命令：

./scripts/instantdb.js link   assignees

6. 解除实体关联

移除实体间的关系：

await client.unlinkEntities(taskId, assigneeId, 'assignees');

CLI 命令：

./scripts/instantdb.js unlink   assignees

7. 实时订阅

通过 WebSocket 监控数据变更：

const subscriptionId = client.subscribe(
  { tasks: { $: { where: { status: 'active' } } } },
  (data) => {
    console.log('Data updated:', data);
  },
  (error) => {
    console.error('Subscription error:', error);
  }
);

// 后续：client.unsubscribe(subscriptionId);

CLI 命令（按指定时长监听）：

./scripts/instantdb.js subscribe '{"tasks": {}}' 60  # 监听 60 秒

8. 事务

使用 tx 构建器原子性地执行多个操作：

const { tx, id } = require('@instantdb/admin');

await client.transact([
  tx.tasks[id()].update({ title: 'Task 1' }),
  tx.tasks[id()].update({ title: 'Task 2' })
]);

CLI 命令：

./scripts/instantdb.js transact '[{"op": "update", "id": "...", "data": {...}}]'

OpenClaw 使用模式

操作状态更新

向人工观察者实时推送进度：

const { id } = require('@instantdb/admin');

// 创建状态实体
const actionId = id();
await client.createEntity('actions', {
  type: 'file_processing',
  status: 'started',
  progress: 0,
  timestamp: Date.now()
}, actionId);

// 更新进度
await client.updateEntity(actionId, 'actions', {
  progress: 50,
  status: 'processing'
});

// 标记为完成
await client.updateEntity(actionId, 'actions', {
  progress: 100,
  status: 'completed'
});

多步骤工作流追踪

追踪复杂操作流程：

const { tx, id } = require('@instantdb/admin');

const workflowId = id();
const steps = ['Extract', 'Transform', 'Validate', 'Load', 'Verify'];

// 初始化工作流并关联各步骤
const txs = [
  tx.workflows[workflowId].update({
    name: 'Data Pipeline',
    status: 'running',
    currentStep: 1,
    totalSteps: steps.length
  })
];

const stepIds = steps.map((name, i) => {
  const stepId = id();
  txs.push(
    tx.steps[stepId].update({
      name,
      order: i + 1,
      status: 'pending'
    }),
    tx.workflows[workflowId].link({ steps: stepId })
  );
  return stepId;
});

await client.transact(txs);

// 步骤完成后逐个更新
for (let i = 0; i < stepIds.length; i++) {
  await client.updateEntity(stepIds[i], 'steps', { 
    status: 'completed' 
  });
  await client.updateEntity(workflowId, 'workflows', { 
    currentStep: i + 2 
  });
}

人工监控模式

人工用户通过前端订阅以观察 OpenClaw 的操作行为：

// 人工用户的前端代码
import { init } from '@instantdb/react';

const db = init({ appId });

function ActionMonitor() {
  const { data } = db.useQuery({
    actions: {
      $: {
        where: { status: { in: ['started', 'processing'] } }
      }
    }
  });
  
  return data?.actions?.map(action => (
    

      {action.type}: {action.progress}%
    
  ));
}

流式进度更新

对于长时间运行的操作，持续流式推送更新：

const { id } = require('@instantdb/admin');

async function processLargeDataset(items) {
  const progressId = id();
  
  await client.createEntity('progress', {
    total: items.length,
    completed: 0,
    status: 'running'
  }, progressId);

  for (let i = 0; i < items.length; i++) {
    // 处理单个项目...
    await processItem(items[i]);
    
    // 每处理 10 个项目更新一次进度
    if (i % 10 === 0) {
      await client.updateEntity(progressId, 'progress', {
        completed: i + 1,
        percentage: Math.round(((i + 1) / items.length) * 100)
      });
    }
  }

  await client.updateEntity(progressId, 'progress', {
    completed: items.length,
    percentage: 100,
    status: 'completed'
  });
}

事务模式

详细事务模式（包括批量操作、关系管理、条件更新、状态机、级联操作等）请参阅：references/transactions.md

• 批量操作
• 关系管理
• 条件更新
• 状态机
• 级联操作

错误处理

所有操作均返回 Promise，在失败时拒绝该 Promise：

try {
  const result = await client.createEntity('tasks', data);
} catch (error) {
  console.error('Operation failed:', error.message);
}

查询语法

完整查询示例（包括 where 子句与运算符、关系遍历、排序与分页、多层嵌套等）请参阅：references/query_syntax.md

• Where 子句与运算符
• 关系遍历
• 排序和分页
• 多级嵌套

参考资料

• InstantDB 文档：https://www.instantdb.com/docs
• Admin SDK：https://www.instantdb.com/docs/admin
• 查询参考：参见 references/query_syntax.md
• 事务模式：参见 references/transactions.md