@ datafn/ 服务器

DataFn的HTTP服务器运行时间. 提供DFQL查询执行,以一能和乐观的相通性进行突变,原子交易,完全双向同步(clone/pull/push/reconcile/seed),每个用户/租户隔离,REST端点生成,WebSocket实时更新,插件钩,可配置限制.

安装

npm install @datafn/server @datafn/core @superfunctions/db @superfunctions/http @superfunctions/http-hono hono @hono/node-server

特征

多区域网关模式是可选入的,并保持所有权与 权限指令投影。 见 multi-region-gateway.mdx 用于配置、错误、 WebSocket、一致性、迁移和操作 合同.

初步地区安置

DataFn 在不嵌入产品数据的情况下提供可重复使用的区域选择 居住或能力政策:

import {
  readDatafnCloudflarePlacementLocation,
  selectDatafnPlacementRegion,
} from "@datafn/server/placement";

const decision = selectDatafnPlacementRegion({
  candidates: regions,
  location: readDatafnCloudflarePlacementLocation(request),
  constraints: [
    (region) => allowedRegionIds.includes(region.regionId),
    (region) => healthyRegionIds.has(region.regionId),
  ],
});

符合条件的明确偏好先取胜, 信任的坐标或大陆 元数据按距离排列其余区域,可选 stableKey 能够确定后退。 返回的决定记录是 已使用。 申请仍负责界定合格区域,收集 明确组织选择,并解剖性地声称由此产生的命名空间 通过放置目录.

快速启动

import { createDatafnServer } from "@datafn/server";
import { drizzleAdapter } from "@superfunctions/db/adapters";
import { toHono } from "@superfunctions/http-hono";
import { Hono } from "hono";
import { serve } from "@hono/node-server";

const server = await createDatafnServer({
  schema: {
    resources: [
      {
        name: "tasks",
        version: 1,
        fields: [
          { name: "id", type: "string", required: true, unique: true },
          { name: "title", type: "string", required: true },
          { name: "completed", type: "boolean", required: true, default: false },
        ],
      },
    ],
  },
  database: drizzleAdapter({ db: myDrizzleInstance, dialect: "postgres" }),
  authorize: async (ctx, action, payload) => {
    return true; // Allow all (replace with real logic)
  },
});

const app = new Hono();
app.route("/", toHono(server.router));
serve({ fetch: app.fetch, port: 3000 });

配置

数据fnServer 配置

interface DatafnServerConfig<TContext = any> {
  /** DataFn schema — validated and normalized at startup */
  schema: DatafnSchema;

  /** Database adapter (e.g. drizzleAdapter, memoryAdapter) */
  database?: Adapter;

  /** Server-side plugins */
  plugins?: DatafnPlugin[];

  /**
   * Authorization callback. Called on every request after JSON parsing.
   * Return true to allow, false to reject with 403 FORBIDDEN.
   */
  authorize?: (
    ctx: TContext,
    action:
      | "status" | "query" | "mutation" | "transact"
      | "seed" | "clone" | "pull" | "push" | "reconcile",
    payload: unknown,
  ) => Promise<boolean> | boolean;

  /** Static request context or a request-scoped context factory. */
  context?: TContext | ((request: Request) => Promise<TContext> | TContext);

  /** Configurable limits */
  limits?: {
    /** Maximum number of records per query (default: 100) */
    maxLimit?: number;
    /** Maximum transaction steps */
    maxTransactSteps?: number;
    /** Maximum request body size in bytes */
    maxPayloadBytes?: number;
  };

  /** Custom server time provider (for testing) */
  getServerTime?: () => number;

  /**
   * Enable REST endpoint generation.
   * When true, generates REST wrappers alongside DFQL endpoints.
   */
  rest?: boolean;

  /**
   * Auth context provider for per-user/tenant isolation.
   * Extracts user/tenant context from request to create isolated namespaces.
   */
  authContextProvider?: {
    getContext: (ctx: TContext) => AuthContext | Promise<AuthContext>;
  };

  /** Trusted namespace and optional actor derivation from request context. */
  namespaceProvider?: {
    getNamespace: (ctx: TContext) => string | Promise<string>;
    getActorId?: (ctx: TContext) => string | undefined | Promise<string | undefined>;
  };

  /**
   * Redis adapter for atomic operations (serverSeq, rate limiting, caching).
   * Used when dbMapping.serverseq = "redis".
   */
  redis?: RedisAdapter;

  /**
   * KV store adapter for simple key-value operations.
   * Must support incr() for serverSeq. Used when dbMapping.serverseq = "kv".
   */
  kvStore?: KVStoreAdapter;

  /**
   * Database mapping — choose which backend for each concern.
   * Default: main database for everything.
   */
  dbMapping?: DbMapping;
}

绘图

将不同的服务器关切运行到不同的后端 :

type DbMapping = {
  /** Backend for serverSeq counter: "db" | "redis" | "kv" */
  serverseq?: "db" | "redis" | "kv";
  /** Backend for rate limiting */
  ratelimiting?: "db" | "redis" | "kv";
  /** Backend for caching */
  cache?: "db" | "redis" | "kv";
};

终点

获取/datafn/状态

服务器健康检查,计划散列,能力和限制.

{
  "ok": true,
  "result": {
    "schemaHash": "a1b2c3...",
    "capabilities": [
      "dfql.query", "dfql.mutation", "dfql.transact",
      "sync.seed", "sync.clone", "sync.pull", "sync.push", "sync.reconcile"
    ],
    "limits": { "maxLimit": 100 },
    "serverTimeMs": 1707000000000
  }
}

POST / 数据fn/克瑞

执行 DFQL 查询,包括全过滤、排序、页码和关系扩展.

请:

{
  "resource": "tasks",
  "version": 1,
  "select": ["id", "title", "completed"],
  "filters": { "completed": false },
  "sort": ["title:asc"],
  "limit": 20
}

回应:

{
  "ok": true,
  "result": {
    "data": [
      { "id": "task:1", "title": "Buy milk", "completed": false }
    ],
    "nextCursor": null
  }
}

过滤操作符 :

逻辑组 :

{
  "$or": [
    { "status": "active" },
    { "priority": { "gte": 4 } }
  ]
}

POST /数据fn/搜索

通过配置的服务器执行跨资源搜索 searchProvider。 。 。 .

请:

{
  "query": "test",
  "resources": ["tasks", "projects"],
  "limit": 20,
  "prefix": true,
  "fuzzy": 0.2,
  "fieldBoosts": { "title": 2, "name": 1 }
}

回应:

{
  "ok": true,
  "result": {
    "results": [
      { "resource": "tasks", "id": "task:1", "score": 12.34, "data": {} }
    ]
  }
}

与客户端路径的搜索选项均等 :

• prefix 启用前缀匹配 .
• fuzzy 允许模糊匹配 .
• fieldBoosts 应用字段级加权.

这些选项在路由/平流边界被验证后被传送到提供者执行.

汇总查询 :

{
  "resource": "tasks",
  "version": 1,
  "groupBy": ["status"],
  "aggregations": {
    "count": { "fn": "count" },
    "avgPriority": { "fn": "avg", "field": "priority" }
  },
  "having": { "count": { "gt": 5 } }
}

POST/数据fn/计算

以一能和乐观的货币来执行突变.

插入:

{
  "resource": "tasks",
  "version": 1,
  "operation": "insert",
  "clientId": "client:device-1",
  "mutationId": "m-001",
  "record": { "title": "New task", "completed": false }
}

合并( 部分更新) :

{
  "resource": "tasks",
  "version": 1,
  "operation": "merge",
  "clientId": "client:device-1",
  "mutationId": "m-002",
  "id": "task:abc",
  "record": { "completed": true }
}

删除:

{
  "resource": "tasks",
  "version": 1,
  "operation": "delete",
  "clientId": "client:device-1",
  "mutationId": "m-003",
  "id": "task:abc"
}

乐观货币( 如果是警卫):

{
  "resource": "tasks",
  "version": 1,
  "operation": "merge",
  "id": "task:abc",
  "record": { "status": "completed" },
  "if": { "status": "in_progress" }
}

变异被拒绝 CONFLICT 如果看守条件没有得到满足.

关系业务:

{
  "resource": "tasks",
  "version": 1,
  "operation": "relate",
  "id": "task:1",
  "relation": "tags",
  "targetId": "tag:urgent"
}

回应:

{
  "ok": true,
  "result": {
    "ok": true,
    "mutationId": "m-001",
    "affectedIds": ["task:new-uuid"],
    "errors": [],
    "deduped": false
  }
}

• deduped: true 表示已应用了突变(偶发性).

POST/数据fn/交易

原子多步骤交易混合了询问和突变.

原子执行需要数据库交易. 适配器报告时 capabilities.transactions.supported === false (例如SQLite或Cloudflare D1上的Drizzle适配器),一个带有任何变异步骤的原子变换器返回 HTTP 400 与 DFQL_UNSUPPORTED (单位:千美元)details.path: "atomic")在任何东西被写入之前. 发送 "atomic": false 一次实施步骤一.

请:

{
  "transactionId": "tx-batch",
  "atomic": true,
  "steps": [
    {
      "query": {
        "resource": "tasks",
        "version": 1,
        "select": ["id"],
        "filters": { "completed": false }
      }
    },
    {
      "mutation": {
        "resource": "tasks",
        "version": 1,
        "operation": "merge",
        "id": "task:1",
        "record": { "completed": true }
      }
    }
  ]
}

回应:

{
  "ok": true,
  "result": {
    "ok": true,
    "results": [
      { "kind": "query", "ok": true, "result": { "data": [...] } },
      { "kind": "mutation", "ok": true, "result": { ... } }
    ]
  }
}

POST / 数据fn/ clone

初始水分完全数据下载. 支持帕吉纳克.

请:

{
  "version": 1,
  "tables": ["tasks", "categories"]
}

回应:

{
  "ok": true,
  "result": {
    "ok": true,
    "data": {
      "tasks": [{ "id": "task:1", "title": "..." }],
      "categories": [{ "id": "cat:1", "name": "..." }]
    },
    "cursors": {
      "tasks": "42",
      "categories": "15"
    }
  }
}

POST/数据fn/推

递增同步——仅返回自上个光标后的变化 .

请:

{
  "version": 1,
  "cursors": {
    "tasks": "42",
    "categories": "15"
  }
}

回应:

{
  "ok": true,
  "result": {
    "ok": true,
    "records": {
      "tasks": [{ "id": "task:3", "title": "New since last pull", "__deleted": false }]
    },
    "cursors": {
      "tasks": "44",
      "categories": "15"
    }
  }
}

POST/数据fn/推

从客户端更改日志上传下线突变.

请:

{
  "version": 1,
  "mutations": [
    {
      "resource": "tasks",
      "version": 1,
      "operation": "insert",
      "clientId": "client:device-1",
      "mutationId": "m-offline-1",
      "record": { "title": "Created offline" }
    }
  ]
}

POST /数据fn/调节

探测并解决本地和偏远地区之间的漂移.

请:

{
  "version": 1,
  "counts": {
    "tasks": 42,
    "categories": 8
  }
}

POST /数据fn/种子

种子数据输入数据库.

请:

{
  "data": {
    "tasks": [
      { "id": "task:seed-1", "title": "Seeded Task", "completed": false }
    ]
  }
}

多用途/多用途隔离

按用户或租户单独跟踪变化情况。 每个命名空间都有自己的 serverSeq 计数器,确保用户只看到自己的修改.

const server = await createDatafnServer({
  schema,
  database: adapter,
  authContextProvider: {
    getContext: (ctx) => ({
      userId: ctx.session?.userId,
      tenantId: ctx.session?.tenantId,
    }),
  },
});

命名空间格式 :

• 无认证 : datafn (默认)
• 单租户: user:${userId}
• 多租户: tenant:${tenantId}:user:${userId}

行为:

1. 每一个要求, authContextProvider.getContext(ctx) 名为
2. 命名空间来源于 userId 和可选 tenantId
3. 所有更改跟踪( 服务器Seq, 更改记录) 都使用此命名空间
4. Pull/ clone 只返回用户命名空间的更改
5. 若为 getContext() 失败, 回到默认状态 datafn 命名空间

序列存储

服务器使用 SequenceStore 用于生成单调增序数。 选择适合您的基础设施的后端 :

// Default: use main database
const server = await createDatafnServer({
  schema,
  database: adapter,
});

// Redis for high-throughput atomic INCR
const server = await createDatafnServer({
  schema,
  database: adapter,
  redis: myRedisAdapter,
  dbMapping: { serverseq: "redis" },
});

// KV store (e.g. Cloudflare KV, DynamoDB)
const server = await createDatafnServer({
  schema,
  database: adapter,
  kvStore: myKVAdapter,
  dbMapping: { serverseq: "kv" },
});

可用的执行:

创建序列Store( 配置)

创建适当商店的工厂 dbMapping数字 :

import { createSequenceStore } from "@datafn/server";

const store = createSequenceStore({
  db: myAdapter,
  redis: myRedisAdapter,
  dbMapping: { serverseq: "redis" },
});

REST 终点( 可选)

启用 DFQL 端点旁的 REST 包装器 :

const server = await createDatafnServer({
  schema,
  database: adapter,
  rest: true, // Enable REST endpoints
});

生成路线 :

网络安全套

服务器显示一个用于实时光标广播的 WebSocket 处理器:

const server = await createDatafnServer({ schema, database: adapter });

// Wire WebSocket connections to server handler
wss.on("connection", (ws) => {
  const client = { id: ws.id, send: (msg) => ws.send(msg) };
  server.websocketHandler.addClient(client);
  ws.on("message", (data) => server.websocketHandler.handleMessage(client, data));
  ws.on("close", () => server.websocketHandler.removeClient(client));
});

插件

服务器端插件拦截查询、突变和同步 :

const auditPlugin: DatafnPlugin = {
  name: "server-audit",
  runsOn: ["server"],

  afterMutation(ctx, mutation, result) {
    console.log("Server mutation:", mutation);
  },

  beforeSync(ctx, phase, payload) {
    console.log(`Sync ${phase} started`);
    return payload; // Return modified payload or original
  },
};

const server = await createDatafnServer({
  schema,
  database: adapter,
  plugins: [auditPlugin],
});

服务器实例

createDatafnServer 返回 a DatafnServer数字 :

interface DatafnServer<TContext = any> {
  /** HTTP router — wire to your framework (Hono, Express, etc.) */
  router: Router<TContext>;

  /** WebSocket handler for real-time updates */
  websocketHandler: {
    addClient(client: WebSocketClient): void;
    removeClient(client: WebSocketClient): void;
    handleMessage(client: WebSocketClient, data: string): void;
  };
}

使用 @superfunctions/http-hono 或者说 @superfunctions/http-express 以挂载路由器 :

// Hono
import { toHono } from "@superfunctions/http-hono";
app.route("/", toHono(server.router));

// Express
import { toExpress } from "@superfunctions/http-express";
app.use("/", toExpress(server.router));

出口

// Server factory and types
export { createDatafnServer }
export type { DatafnServerConfig, DatafnServer }
export type { StatusResult }

// Sequence store
export type { SequenceStore, DbMapping }
export {
  createSequenceStore,
  RedisSequenceStore,
  KVSequenceStore,
  DatabaseSequenceStore,
  ChainedSequenceStore,
}

许可证

麻省理工学院

修复继承的不活动

recomputeAncestorInactive 在一个命名空间中扫描已限定的批次; 以 nextCursor。 。 。 . recomputeAncestorInactiveAll 重复扫描直到验证 扫描找不到更改或跳过比较和设置的写入 。 检查 converged· ; updated: 0 单靠自己不能证明完成. skipped 计数同时写入 冲突。 干运行数不匹配而不写入 .

修复从祖先存档/崩溃状态和关系链接中获取值, 从不存储 isAncestorInactive 旗帜。 周期、缺失/畸形父母, 越轨限制的违反会造成明显的错误. batchSize, (中文(简体) ), maxSweeps 财务报告和审定财务报表 maxGraphNodes 必须是正安全整数; maxGraphNodes 默认为 每行修补一万个节点。 失败的批次可以安全重放; 之前完成的写作是一能的.

以祖先/链接突变进行修复 比较和设定保护 每个字段来自同时传播的文字;它不提供 所有祖先记录的原子快照 同时恢复和核实 活动在依赖修复的价值进行消费者迁移之前就已解决.