@dbx-工具/邮递

使用连接正确的 PostgreSQL 咨询锁和结构化 LISTEN编号 :NOTIFY 主题总线来自Node.js. 软件包还提供了可选 消息持续性、发送者身份和连接帮助者.

都与平地为敌 pg.Pool 或任何结构上与一个, 包括 AppKit 的 Lakebase 插件导出, 因此一个 Databricks App 获得 它们没有第二个数据库客户端或连接池.

快开始

在持有锁的 PostgreSQL 连接上运行受保护的工作 :

import { postgresConnectionOptions, withAdvisoryLock } from "@dbx-tools/postgres";
import { Pool } from "pg";

const pool = new Pool(postgresConnectionOptions({}));
const invoiceId = 42;

await withAdvisoryLock(pool, ["invoice", invoiceId], async (client) => {
  await client.query("UPDATE invoices SET status = 'sent' WHERE id = $1", [invoiceId]);
});

能力

• 转换任意结构键(字符串、阵列、对象、, Date,或一个 明确 bigint)成为稳定签名的64位咨询锁标识符.
• 从任何识别频道中获取合法的 Postgres 频道名称,所以不 调用网站必须将应用程序名称或租户身份标识进行消毒.
• 对整个召回的专用集合连接有一个会话锁定, 并且把连接到回调器上 所以保护的活在它上.
• 持有解析释放的交易锁定 COMMIT 或者说 ROLLBACK,这是 一次性安装需要什么.
• 广播a type编号 :metadata编号 :body 每个监听过程的封装 pg_notify,按正在处理的专题过滤.
• 填充工程、机器、进程、部署和可选的 AppKit 发送器 上下文,调用元数据总是获胜.
• 拒绝不会在JSON往返中幸存的有效载荷 电话站而不是电线上.
• 重新连接一个已丢弃的收听器, 用户仍保留时, 设定了后退 .
• 可选地坚持每条已发布的消息到一个软件包的表格,所以 未收听的订阅者可以通过光标重放历史, 默认 TTL和清理在插入.
• 两种有效载荷样式:通知中的整个信封,或指向 听者读回. 指针抬起八千字节 NOTIFY 限制和保留 表格赠款背后的内容.

会议作用

设定 DBX_TOOLS_POSTGRES_ROLE 当每个仓库拥有的 PostgreSQL 连接时 认证后应承担一个角色. postgresConnectionOptions() 添加 作用 pg 和 AppKit Lakebase 池的启动选项,同时 postgresServerSettings() 返回等同点 asyncpg 服务器设置.

import { postgresConnectionOptions } from "@dbx-tools/postgres";
import { Pool } from "pg";

const pool = new Pool(postgresConnectionOptions({}));

明确的第二个参数压倒环境值。 帮助者保留 现有 options,包括语句超时,并附加角色设置. 使用 postgresRoleStatement() 仅用于设置工作,该工作必须在 连接已经打开 .

当这些选项配置 AppKit Lakebase 路由池时,它们适用于其 服务-主要和每个用户OBO集合。 每一个可以到达的登录身份 因此,人才库必须是所配置角色的成员,否则 权限 SET ROLE,在游泳池成功连接之前.

与 AppKit Lakebase 一同使用

使用 AppKit lakebase() 以获取和刷新 PostgreSQL 池。 这个包 不替换该插件。 当工作需要连接正确咨询时添加 锁或交叉连接 LISTEN编号 :NOTIFY 交付、能力 Lakebase 插件不提供 .

咨询锁

import { postgresConnectionOptions, withAdvisoryLock } from "@dbx-tools/postgres";
import { Pool } from "pg";

const pool = new Pool(postgresConnectionOptions({}));

await withAdvisoryLock(pool, ["invoice", invoiceId], async (client) => {
  await client.query("UPDATE invoices SET status = 'sent' WHERE id = $1", [invoiceId]);
});

运行保护工程 client 调用接收。 A 类 pool.query() 内召回可能降落 在不同的连接上, 这不是一个 拿着锁.

使用交易范围帮助一次性数据库工作 :

import { withAdvisoryTransactionLock } from "@dbx-tools/postgres";

await withAdvisoryTransactionLock(pool, { schema: "my_feature", version: 1 }, async (client) => {
  await client.query("CREATE TABLE IF NOT EXISTS my_feature.events (id bigint primary key)");
});

identity.advisoryLockId(key) 暴露与 JSON 安全小数相同的减少 绳子。 A 类 bigint 键为 直接使用( 缩到64比特) 而不是散列, 因此锁可以交互操作 和另一个发布其数字锁标识的应用程序,例如 pgmq是因为 例如安装器锁. 任何其它东西都用 object.toStableKey 从 @dbx-tools/shared-core {\fn黑体\fs22\bord1\shad0\3ahbe\4ah00\fscx67\fscy66\2chffffff\3ch808080}那么关键顺序 对象在 a 时无关紧要 1 (单位:千美元) "1" 保持不同的锁。 一个阵列是 读作若干部分,所以 ["invoice", 7] 财务报告和审定财务报表 "invoice_7" 是不同的锁。 A 类 非无限数字或循环键抛出 TypeError 而不是产生一个 两个打电话的人可能不同意.

主题总线

import { PostgresTopicBus } from "@dbx-tools/postgres";

const bus = new PostgresTopicBus(pool, {
  metadata: async () => ({ publicIp: await resolvePublicIp() }),
  onError: (cause) => logger.error("bus", { cause }),
});

const unsubscribe = await bus.listen("orders", ({ type, metadata, body }) => {
  console.log(type, metadata.project, body);
});

await bus.broadcast("orders", {
  type: "order.updated",
  metadata: { traceId: "abc-123" },
  body: { orderId: "123", status: "ready" },
});

await unsubscribe();
await bus.close();

在频道上监听的每个过程都会收到每个消息. 没有 竞争的消费者。 巴士适合每个浏览器的SSE流等现场播放 选项卡,缓存无效或存在 ping。 它不分配工作. 默认只提供现场直播。 打开 persist 当 a 订阅者需要重放,需要时使用真正的队列.

信件信封

publishedAt 财务报告和审定财务报表 id 用于显示和调试。 也没有命令信件 不同的情况可靠.

元数据优先

最弱到最强:自动机器上下文,然后是AppKit发送器身份,然后是 公交车 metadata 选项,然后是每个消息 metadata键入 更坚固的地层从不被覆盖 所以通过 project 明确替换 推断出一个 明确 null 也保存并阻止自动值 从用于那个键.

自动键, 当解析为无时, 每个键都会省略 : project (从) DATABRICKS_APP_NAME, (中文(简体) ), DATABRICKS_BUNDLE_NAME, (中文(简体) ), PROJECT_NAME,或 npm_package_name, (中文(简体) ), hostname, (中文(简体) ), platform, (中文(简体) ), environment, (中文(简体) ), appName, (中文(简体) ), deploymentId,以及 databricksHost. 公共/客户IP、进程ID、CPU 架构、运行时名称和运行时版本被故意忽略: 皮条很吵(而且往往很敏感),其余的都是每次部署的常数,以及 每条信息都会按有效载荷限制支付 通过 a metadata 函数,当一个进程需要其中一个这些值时.

上面的例子将一个发现的公用IP附以这种方式. 函数运行于 每一个广播, 所以记住任何昂贵的.

与可选 @databricks/appkit 已安装端点和用户执行上下文 活动, senderId, (中文(简体) ), senderName,以及 senderEmail 还加上了。 AppKit 为 进口懒惰且不需要。 在请求之外,或者没有包裹 公交车跳过那些键.

支持的信息值

身体和元数据必须活下来 JSON.parse(JSON.stringify(x)) 不变,哪些 更为严格 JSON.stringify 不扔东西。 A 类 Date 成为弦, NaN 财务报告和审定财务报表 Infinity 变成 null页:1 Map 变成 {},以及 undefined 失踪了. 每一个都到达听众的身边, 与所发送的东西不同, 所以所有他们 掷出 TypeError 在电话中代替。 周期、功能、符号、比值和类 由于同样的原因,拒绝提交案件.

规矩在 @dbx-tools/shared-core 作为 object.isSerializableValue 财务报告和审定财务报表 object.SerializableValue,因此,一个路由可以对不可信 请求正文并回答400而不是500,浏览器侧调用器可以共享 类型.

编码后的信封被封为 7900 字节, 在 PostgreSQL 的 8000 字节下 NOTIFY 有限度的,有大势的 RangeError自动元数据计数 反之,所以广播一个参考 并让接收器获取任何重大的东西.

连接、生命周期和失败

出版每通电话借用集合连接. 听力无法 : LISTEN 实值 会话状态,所以总线会检查一个专用客户端,并保持它多久 因为它有订阅者,但许多专题和听众都登记在册。 大小 带着那长生不老的检查表去游泳.

第一次之前没有联系 listen,或明示 start() 你何时会 最好找出一个连接问题 在启动. close() 释放客户端 并且是必需的。 用主机的停机钩登记 封闭的公交车停放 关闭并投掷,而不是静默地重新连接.

当通知连接中断时,总线会自行重新连接:立即 首先,将上限从250米翻一番,无限期地翻一番 订户仍然存在。 Postgres 重启或旋转 Lakebase 证书回收 没有干预。 空隙期间发布的消息会丢失 。 每一个失败 尝试,每个中断的连接, 和每个播放者 丢弃 结束 onError,它默认吞下它们。 配置登录器 任何长期运行。 一个失败的听众从来不影响出版商或其他人 听众.

一个频道包含许多话题,所以增加一个话题并不花费什么. 给一个 不相关的高音量流 自己的频道,因为每次听 会话解码频道上的每个消息.

重放为 persist

NOTIFY 没有重放: 一个不听的过程永远不会知道它是什么 错过了。 设定 persist 并且每个已发布的消息都是相同的 交易作为其通知,因此光标可以读取历史.

const bus = new PostgresTopicBus(pool, { persist: true });

await bus.broadcast("orders", { type: "order.updated", body: { id: 7 } });

let cursor: string | undefined;
do {
  const page = await bus.history("orders", { after: cursor, limit: 100 });
  for (const { message, expiresAt } of page.messages) handle(message, expiresAt);
  cursor = page.nextCursor;
} while (cursor);

该行和通知共同承诺,这就是整个财产 设计的基础是: NOTIFY 是在承诺时而不是在发言时交付的,所以 永远不能告诉听众它还不能读的一行.

消息 EXPIRE 默认。 24小时TTL是默认的,正因为 未绑定的消息表是此特性否则会引入的失败 . 已过期的行被删除,在每份持续出版的限定批次 cleanupExpired() 按要求进行同样的扫荡. ttl: false 选择一个总线(或 { ttl: false } 单曲 broadcast)出为已出. 虽然有,但.. 这不是默认.

两层, open 财务报告和审定财务报表 restricted,所以一个信道可以携带信息 不同的读者群:它们是单独的表格,所以一个角色被赋予了开放性 表格不能读取限制性历史. messageBusGrantStatements(roles, scope, options) 发放的补助额为: history 光标已绑定 至其所出之阶. 在限定的表格中恢复开放页面 舍去掉别人的行 发布到分级 每个消息 broadcast(topic, { scope: "restricted", ... })。 。 。 .

设定 provision: false 当移民拥有这个计划时 声明是: 在咨询交易锁下,否则适用一次,因此同时适用 在启动时安装的例不比赛 .

信封或指针

payload: "envelope" (缺省)将整个消息放入通知. 一次往返,不读放大,以PostgreSQL的有效载荷限制为界.

payload: "pointer" 仅以路线和身份通知 (单位:千美元){ v, scope, topic, id, sequence }),收听者读回信封. 这样就大了 完全限制(一个大于7900字节的身体只能以这种方式公布)和 意思是没有级表补助的订户得知出了事 没有学会什么。 费用为每份通知读取一次。 听众收到 同样的 TopicMessage 无论是哪种方式。 上面的区别是看不见的 listen。 。 。 .

已经走光的指针( TTL 清理赢得比赛) 或无法读取 提高不会产生任何效果, 使用 decodePointer 如果非公共汽车消费者需要阅读原始通知.

命名频道

channel 使用任何识别通道, 而不是一个预先保密的标识 : 姓名、房客身份证、 [env, feature] 组合,或配置对象。 一个或多个值: 一个阵列被读取为多个部分,其他任何部分被读取为一体.

new PostgresTopicBus(pool, { channel: "billing-events" }).channelName;
// "billing_events_3x55ck"
new PostgresTopicBus(pool, { channel: ["billing", "production"] }).channelName;
// "billing_production_008jgf"

部分被标记成可读的半个和短的散列 窗体( e)object.toStableKey页:1 大麻是制图的原因 信得过: 孤独的象征是迷失的,所以 my-app, (中文(简体) ), my_app,以及 myApp 将 崩溃到一个信道上,并且名称长度足以击中 Postgres 的 63 个字符 限制会与任何分享主要标志的东西相撞。 用后缀 可读部分保持可读性,不同投入保持差异,包括 ["billing", "prod"] 对 "billing_prod",结构不同.

衍生是决定 跨过程和运行, 所以每个参与者 同一航道上降落时没有协调 对象密钥 秩序无关紧要。 不同的拼写方式有 读取 bus.channelName 来看 一组部分是否同意或确认两个进程。 默认为 a 共享 dbx_tools_topic_bus 频道.

identity.channelName(parts) 不搭公交车就暴露出原因, 用于配置检查和交叉运行的合同测试.

连接寿命周期

咨询锁是连接范围,不正确的池使用可能隐蔽地失败: 带着锁 pool.query() 与他人一起从事受保护的工作 pool.query() 可以用两个不同的连接, 所以锁保护什么 代码看起来是正确的。 这同样适用于 LISTEN,即会话状态 a 集合查询无法维持 。 这些帮手将两个生命周期都保持专注 连接和仅依赖 pg。 。 。 .

AppKit供应了"湖基"游泳池. 这些帮助者添加了锁和交叉连接 在它上面发送消息.

模块映射

两个模块也被平整到软件包根上,所以 import { withAdvisoryLock, PostgresTopicBus } from "@dbx-tools/postgres" 工作内容 没有命名空间.