fastify-statistics

描述

基于 Fastify 的数据采集与多周期聚合统计插件，支持缓冲写入、时区查询和自动 Cron 聚合

安装

npm i --save @kne/fastify-statistics

概述

项目概述

@kne/fastify-statistics 是一个基于 Fastify 的数据采集与多周期聚合统计插件。它提供从原始数据上报到多级周期聚合、灵活查询与实时推送的完整数据管道，适用于 IoT 传感器数据、业务指标监控、多通道多属性聚合等场景。

核心架构与数据流

插件的核心理念是逐级聚合——原始数据采集后，按 h→d→w/m→q→y 的依赖链自动滚动聚合，每一级只从其直接上游读取数据，形成清晰的数据流管道：

数据采集(collect) → data-record 表
                          ↓ h 周期聚合 (Cron: 每小时第1分钟)
                    period-stat (period=h)
                          ↓ d 周期聚合 (Cron: 每日0:01)
                    period-stat (period=d)
                       ↙        ↘
              w 聚合(周一0:01)   m 聚合(每月1日0:01)
                    ↓                ↓
              period-stat(w)   period-stat(m)
                                   ↓ q 聚合(每季首月1日0:01)
                              period-stat(q)
                                   ↓ y 聚合(每年1月1日0:01)
                              period-stat(y)

聚合依赖关系（PERIOD_DEPENDENCY）：

周期	数据来源	上游周期
h	data-record	-
d	period-stat	h
w	period-stat	d
m	period-stat	d
q	period-stat	m
y	period-stat	q

六种统计周期

周期	key	Cron 表达式	时间截断规则	数据来源
时	h	1 * * * *	startOf('hour')	data-record
日	d	1 0 * * *	startOf('day')	period-stat(h)
周	w	1 0 * * 1	startOf('week')+1天 (周一)	period-stat(d)
月	m	1 0 1 * *	startOf('month')	period-stat(d)
季	q	1 0 1 1,4,7,10 *	Math.floor(month/3)*3 月首日	period-stat(m)
年	y	1 0 1 1 *	startOf('year')	period-stat(q)

聚合方法与级联计算

五种聚合方法在聚合过程中协同计算，确保高级别周期可以正确推导：

方法	key	从 data-record 聚合	从 period-stat 聚合
合计	sum	SUM(data)	各子窗口 sum 求和
计数	count	COUNT(data)	各子窗口 count 求和
平均	avg	AVG(data)	sum总 / count总（非 AVG(AVG)）
最小	min	MIN(data)	各子窗口 min 取最小值
最大	max	MAX(data)	各子窗口 max 取最大值

关键设计：avg 不直接对上游 avg 取平均，而是用 sum/count 重新计算，避免二次平均偏差。

聚合区间语义

所有聚合使用左闭右开区间 [startTime, endTime)：

• startTime：当前时间窗口的截断起始
• endTime：下一个时间窗口的起始（getNextStart(startTime)）
• 查询条件使用 Op.gte(startTime) + Op.lt(endTime)，确保 endTime 所在时刻的数据不被包含

这一设计修复了此前使用 Op.between（闭区间）导致边界数据被重复聚合到两个窗口的 bug。

Channel 通道层级设计

Channel（数据通道） 采用冒号分隔的多级结构（a:b:c），核心思想是从宏观到微观的层级划分：

• 一级 channel（如 sales）是根通道，对应唯一的 channel-meta 记录
• 多级 channel（如 sales:beijing、sales:beijing:team-a）是更细粒度的子通道
• 采集时自动展开：company:sales:beijing 展开为 company、company:sales、company:sales:beijing 三条记录
• 查询时：默认精确匹配；includeChildren=true 时匹配通道及所有子通道，返回树形结构

AttributeName（属性名） 是同一 channel 下的第二级分类维度：

• 默认值 default，适用于单指标场景
• data 传入对象时自动展开（如 {revenue: 10000, orders: 50} → 两条记录）
• unit 支持字符串（所有属性共用）或对象（按 attributeName 映射不同单位）

通道统计设计案例

当一个业务希望统计“某类事件在不同来源、不同小时桶中的完成量”时，容易把 统计周期、通道层级 和 查询返回结构 混在一起。正确做法是先确定需要在查询结果中保留的维度，再把这些维度建模到 channel 或 attributeName 中。

场景抽象

需求	推荐建模	说明
查询总量	event	根通道自动汇总所有子通道
按来源查询	event:web、event:api	第二级通道表示来源
按小时桶查询	event:web:00、event:web:01	第三级通道表示固定小时桶
按指标类型查询	attributeName	同一通道下多个数值指标更适合放在 attributeName

关键设计：period=h/d/m 是系统内部按时间窗口聚合后的存储周期，不应承担业务维度拆分职责。业务需要稳定输出的维度，应在采集时进入 channel 或 attributeName。

数据流

采集 event:web:13
        ↓ 自动展开
event:web:13 → event:web → event
        ↓ 聚合
period-stat(h/d/m/...)
        ↓ 查询
精确通道列表 或 includeChildren 树形结果

阶段	行为	注意事项
采集	多级 channel 自动展开为自身及所有父级	父级已经包含子级汇总
聚合	每个展开后的 channel 独立生成周期统计	父级和子级不是互斥数据
查询	默认只匹配传入的精确 channel	需要子树时才使用 includeChildren=true
消费	根据返回结构选择汇总方式	不要把父级与子级再次相加

常见错误

错误做法	问题	正确做法
长区间 Dashboard 直查 period=h	h 仅保留当月，历史会被清理	使用 query() / queryFlat() / queryTotals()
查询根通道后再期望自动得到每个小时桶	默认查询只返回精确匹配通道	显式查询所有叶子通道，或使用 channelScope: 'descendantsFlat'
对 includeChildren=true 的树形结果做扁平求和	父级已经包含子级汇总，容易重复计数	使用 channelScope: 'descendantsFlat' 只取叶子，或只查询叶子通道
依赖查询结果中的 period=h 构建长范围小时分布	查询会按对齐窗口选择较粗周期，长范围可能返回 d/m/y	将小时桶作为 channel 维度，如 event:web:13
用一个宽泛 channel 表达多个正交维度	后续筛选和拆桶需要猜测字符串含义	固定维度顺序，例如 event:{source}:{hour}

推荐查询

channelScope（比 includeChildren 更语义化）：

channelScope	行为	典型场景
exact（默认）	仅精确匹配 channels	已知叶子 channel
descendantsFlat	匹配子通道，返回扁平叶子 channel	按父 channel 查所有子项且避免重复计数
descendantsTree	匹配子通道，返回树形结构	等同 includeChildren=true，层级浏览

采集时 expandChannel('a:b:c') 会同时写入 a、a:b、a:b:c。descendantsFlat 默认只返回叶子 channel，避免父级与子级同时汇总导致重复计数。可选 maxDepth 限制冒号层级。

如果目标是得到每个来源在 24 个小时桶中的分布，可枚举叶子通道（channelScope: 'exact'），或使用 descendantsFlat 传入父 channel：

// 方式一：显式枚举叶子通道
const sources = ['web', 'api'];
const hours = Array.from({ length: 24 }, (_, i) => String(i).padStart(2, '0'));
const channels = sources.flatMap(source => hours.map(hour => `event:${source}:${hour}`));

const result = await fastify.statistics.services.query({
  channels,
  startTime: '2026-05-01T00:00:00.000Z',
  endTime: '2026-06-01T00:00:00.000Z',
  aggregates: ['sum']
});

// 方式二：父 channel + descendantsFlat（无需手动枚举）
const flatResult = await fastify.statistics.services.periodStat.queryFlat({
  channels: ['event'],
  channelScope: 'descendantsFlat',
  maxDepth: 3,
  startTime: '2026-05-01T00:00:00.000Z',
  endTime: '2026-06-01T00:00:00.000Z',
  aggregates: ['sum']
});

返回结果是扁平列表，调用方可以直接按 channel 解析来源和小时桶：

{
  "channelMetas": {
    "event": { "channel": "event", "title": "事件统计", "description": "事件完成量" }
  },
  "list": [
    {
      "channel": "event:web:13",
      "period": "m",
      "time": "2026-05-01T00:00:00.000Z",
      "data": { "default": 128 }
    }
  ]
}

使用 includeChildren 的边界

includeChildren=true 适合展示通道树，或让使用方按层级浏览数据；它不适合直接作为“按叶子维度分桶”的扁平数据源。

查询方式	返回形态	适合场景
channels=['event:web:13']	扁平列表	精确叶子桶统计
channels=['event:web']	扁平列表	来源汇总
channels=['event:web'], includeChildren=true	树形结构	展示来源及其小时子桶
channels=[所有叶子通道]	扁平列表	构建固定维度趋势图

经验教训：通道层级查询的第一原则是先决定“结果要哪个层级”。如果结果要叶子桶，就枚举叶子桶；如果结果要父级汇总，就查父级；如果结果要层级浏览，再使用 includeChildren=true。

水位线机制与补偿聚合

水位线（aggregation-watermark） 记录每个周期下一次应聚合的起始时间，是补偿聚合的核心：

• 每个周期一条记录：(period, nextTime)
• 聚合完成后，nextTime 推进到下一窗口起始
• 补偿时从 nextTime 开始逐步向前，追上当前时间

启动初始化流程（period-stat.init()）：

1. 按 h→d→w→m→q→y 顺序依次处理
2. 对每个周期：
   • 若水位线存在且过期 → 从水位线开始补偿
   • 若水位线不存在 → 调用 determineStartFromSource 从源数据推断起始点
   • 若无任何源数据 → 水位线设为当前截断时间（全新系统）
3. 逐窗口执行聚合，每完成一个窗口就推进水位线
4. 上游未完成时自动先补偿上游（如聚合 d 前确保 h 已完成）

determineStartFromSource 推断逻辑：

• h 周期：从 data-record 的 MIN(time) 截断到小时
• 其他周期：从上游 period-stat 的 MIN(time) 截断到当前周期起始

早期版本使用 MAX(time) + nextStart 推断起始点，导致遗漏上游最早数据。现已改为 MIN(time) 截断，确保所有历史数据都被覆盖。

运行时补偿（Cron 触发）：

• 每个 Cron 周期调用 compensate(period)
• 每次最多处理 compensationBatchSize 个窗口（默认 24）
• 连续失败 maxCompensationFailCount 次（默认 3）后停止，下次 Cron 继续
• 每个周期有独立锁（compensatingLocks），防止并发补偿

数据保留策略

通过 Cron 定时清理过期数据，避免数据无限增长。可通过 periodStat.getRetentionPolicy() 读取当前策略。

数据类型	保留策略	清理时机	安全检查
data-record	dataRetentionDays 天（默认 7 天）	每天 02:00	不超过 h 周期水位线
period-stat(h)	当月	每天 03:00	不超过 d 周期水位线
period-stat(d)	当年	每天 03:00	不超过 w、m 周期水位线
period-stat(w)	当年	每天 03:00	无下游依赖
period-stat(m/q/y)	永久保留	不清理	-

安全检查：删除前检查下游水位线，确保尚未聚合的数据不会被提前删除。

为什么不能只查 period=h

period=h 会在每月初之后被清理，仅保留当月数据。若应用层直接 findAll({ where: { period: 'h' } }) 做长区间 KPI，统计会在运行一段时间后「突然变少」——新增一条采集后可能只剩当月/当天少量数据。

正确做法：使用 query() / queryFlat() / queryTotals()，由 buildQueryWindows 按区间自动组合 y → q → m → w → d → h，并合并当前小时未聚合的 data_record。

缓冲写入模式

当配置 cache 实例时，采集数据先写入内存缓冲区，再定时批量持久化：

• 缓冲区达到 collectMaxBufferSize（默认 1000）时触发 flush
• 定时 collectFlushInterval（默认 5000ms）自动 flush
• 缓冲区溢出上限 collectMaxBufferOverflow（默认 maxBufferSize * 2），超出时丢弃最旧数据
• 进程关闭时（onClose hook）持久化缓冲区到 cache 并执行最终 flush
• 启动时从 cache 恢复缓冲区数据（restoreBuffer）

无 cache 时，每次采集直接写入数据库（collectImmediate）。

查询缓存

查询结果自动缓存，减少重复查询的数据库压力：

特性	说明
外部缓存	配置 cache 时使用，支持 TTL
内存 LRU 缓存	无外部缓存时使用，最大 queryCacheMaxEntries 条
版本校验	缓存条目记录写入时的通道版本号，读取时校验版本是否变化
TTL 策略	实时查询（endTime 在当前小时内）用 queryCacheTTL（30s），历史查询用 queryCacheHistoryTTL（3600s）
补偿期间	正在补偿聚合时查询不走缓存，确保数据实时性

缓存失效：采集数据时自动调用 invalidateQueryCache(affectedChannels)，递增对应通道及其所有前缀的版本号。

查询辅助 API

在保留策略下做长区间统计时，优先使用以下 API，避免手写 period 组合与 flatten 逻辑：

场景	API
需要原始扁平行	periodStat.queryFlat(params)
需要全局/按 channel 汇总	periodStat.queryTotals(params)
需要树形子通道	query({ channelScope: 'descendantsTree' })
需要扁平叶子子通道	query({ channelScope: 'descendantsFlat', maxDepth? })

queryTotals 返回 totals（全局 sum）、totalsByChannel、maxByChannel（对 max 取 Math.max，禁止对小时 max 求和）、attrStats 等。HTTP 查询支持 format=flat|totals。

const result = await fastify.statistics.services.periodStat.queryTotals({
  channels: [`interview:${clientId}`],
  channelScope: 'descendantsFlat',
  maxDepth: 3,
  startTime,
  endTime,
  aggregates: ['sum', 'max']
});
// result.totals / result.totalsByChannel / result.maxByChannel

SSE 实时推送

基于 Server-Sent Events 的实时统计推送：

• 按 interval（默认 5 秒）定时调用 fetchData 获取最新数据推送
• 心跳保活（默认 30 秒），防止连接被代理/负载均衡器断开
• 最大连接时长（默认 30 分钟），超时自动断开并推送 timeout 事件
• 防止推送重叠：当前推送未完成时跳过下一次
• 缓存复用：相同 name+params+interval 在同一时间窗口内命中缓存

长任务进度 SSE（sseStream.runTask）：用于聚合重建等耗时操作，推送 progress / done / error 事件，而非轮询 query。HTTP：GET {prefix}/rebuild/sse。

重置、重建与修复

提供 resetPeriodStats 方法用于修复错误的聚合数据：

• 删除指定周期和时间范围的 period-stat 记录
• 重置水位线到指定起始点
• 支持 cascade=true 级联重置下游周期（如重置 h 时同时重置依赖 h 的 d、w、m、q、y）

聚合重建（periodStat.rebuild）用于历史修复或初始化回填后的重聚合：

mode	行为
aggregate-only（默认）	flush buffer → 从 MIN(data_record.time) 或 startTime 重聚合 h→y
reset-and-aggregate	先 clearAll → 宿主 beforeAggregate hook 写入 data_record → 聚合
repair	指定 [startTime, endTime) 区间局部 resetPeriodStats + 重聚合

await fastify.statistics.services.periodStat.rebuild({
  mode: 'reset-and-aggregate',
  onProgress: payload => { /* stage: start | flush | clear | aggregate | done */ },
  beforeAggregate: async () => { /* 宿主业务采集 */ },
  afterAggregate: async () => { /* 宿主一致性校验 */ }
});

宿主集成建议：

1. 禁止业务 Dashboard 直查 period=h 做全量 KPI
2. 按父 channel 查所有子项时，用 channelScope: 'descendantsFlat' 代替手动枚举
3. 回填脚本只保留领域 collect/verify 逻辑，聚合循环交给 rebuild

技术栈

• Fastify + fastify-plugin + fastify-namespace
• Sequelize（数据持久化，支持多种数据库）
• fastify-cron（定时聚合与清理任务）
• dayjs（时间处理，支持 UTC 和时区扩展）
• lodash（工具函数）

主要特性

• 数据采集：支持单条和批量数据上报，自动展开多属性对象和多级通道
• 缓冲写入：支持缓存缓冲模式，定时批量写入数据库，减少写入压力
• 多周期聚合：h/d/w/m/q/y 六种统计周期，自动通过 Cron 定时聚合
• 聚合方法：sum、avg、count、min、max 五种聚合计算
• 灵活查询：按通道、时间范围、属性名、聚合方法查询统计结果，自动合并当前小时未聚合的原始数据；queryFlat / queryTotals 开箱即用
• 子通道查询：channelScope 支持扁平叶子子通道（descendantsFlat）与树形（descendantsTree）
• 聚合重建：rebuild + SSE 进度，宿主通过 hook 注入采集与校验
• SSE 实时推送：基于 Server-Sent Events 的实时数据推送
• 时区支持：查询时支持传入客户端时区（IANA 格式）
• 事务安全：所有数据库写操作使用事务保证原子性，聚合操作支持幂等（upsert）
• 查询缓存：支持内存 LRU 或外部缓存，带版本校验和 TTL 策略

使用方法

快速开始

const fastify = require('fastify')();

// 注册依赖插件
fastify.register(require('@kne/fastify-sequelize'), { /* sequelize 配置 */ });
fastify.register(require('@kne/fastify-cron'), { /* cron 配置 */ });

// 注册统计插件
fastify.register(require('@kne/fastify-statistics'), {
  prefix: '/api/statistics',
  cache: redisCacheInstance,   // 传入缓存实例启用缓冲模式和查询缓存
  compensationEnabled: true,   // 启动时自动补偿聚合
  compensationBatchSize: 24,   // 每次补偿最多24个时间窗口
  dataRetentionDays: 7,        // 原始数据保留7天
  queryCacheEnabled: true,     // 启用查询缓存
  queryCacheTTL: 30,           // 实时查询缓存30秒
  queryCacheHistoryTTL: 3600,  // 历史查询缓存1小时
  queryCacheMaxEntries: 100,    // 内存缓存最大100条（无外部缓存时生效）
  getAuthenticate: type => {
    // type 为 'read' 或 'write'，返回认证信息
  }
});

fastify.listen({ port: 3000 });

Channel 与 AttributeName 的设计理念

Channel（数据通道） 是数据的第一级分类维度，采用冒号分隔的多级结构（a:b:c）。它的核心思想是：从宏观到微观的层级划分。

• 一级 channel（如 sales）是根通道，对应唯一的 channel-meta 记录（标题、描述）
• 多级 channel（如 sales:beijing、sales:beijing:team-a）是更细粒度的子通道
• 查询时传入一级 channel 即可匹配所有子通道的数据
• 同一根通道下的所有子通道共享同一个 channel-meta

AttributeName（属性名） 是数据的第二级分类维度，用于在同一 channel 下区分不同的数据指标。

• 默认值为 default，适用于单一指标的场景
• 当 data 传入对象时自动展开为多属性（如 {revenue: 10000, orders: 50} 拆分为两条记录）

实际场景：企业部门数据统计

假设一家公司要统计各部门的经营数据，我们可以这样设计 channel：

company                     ← 根通道：公司整体
company:sales               ← 子通道：销售部
company:sales:beijing       ← 子通道：销售部北京分部
company:sales:shanghai      ← 子通道：销售部上海分部
company:rd                  ← 子通道：研发部
company:rd:frontend          ← 子通道：研发部前端组
company:rd:backend           ← 子通道：研发部后端组
company:hr                  ← 子通道：人力资源部

对应的 channel-meta 只需为根通道 company 创建一条记录：

channel	title	description
company	公司经营数据	各部门经营数据统计

采集数据：

// 1. 销售部北京分部上报单指标（默认 attributeName=default）
await fastify.statistics.services.collect({
  channel: 'company:sales:beijing',
  data: 58000,
  unit: '元',
  title: '公司',
  description: '各部门经营数据统计'
});

// 2. 销售部上海分部上报多指标，unit 为字符串时所有属性共用同一单位
await fastify.statistics.services.collect({
  channel: 'company:sales:shanghai',
  data: { revenue: 72000, orders: 150 },
  unit: '元',
  title: '公司',
  description: '各部门经营数据统计'
});

// 3. 研发部前端组上报，unit 为对象时按 attributeName 映射不同单位
await fastify.statistics.services.collect({
  channel: 'company:rd:frontend',
  data: { tasks: 12, bugs: 3 },
  unit: { tasks: '个', bugs: '个' },
  title: '公司',
  description: '各部门经营数据统计'
});

采集后数据会自动展开并入库：

channel	attributeName	data	unit
company	default	58000	元
company:sales	default	58000	元
company:sales:beijing	default	58000	元
company	revenue	72000	元
company	orders	150	元
company:sales	revenue	72000	元
company:sales	orders	150	元
company:sales:shanghai	revenue	72000	元
company:sales:shanghai	orders	150	元
...	...	...	...

通道展开规则：company:sales:beijing 自动展开为 company、company:sales、company:sales:beijing 三条记录，确保每一级都能查到汇总数据。

查询数据：

// 1. 查询销售部本月合计（仅自身）
const salesResult = await fastify.statistics.services.query({
  channels: ['company:sales'],
  startTime: '2026-05-01T00:00:00.000Z',
  endTime: '2026-06-01T00:00:00.000Z',
  aggregates: ['sum']
});

// 2. 查询公司所有部门的本月合计（传入一级 channel + includeChildren）
const companyResult = await fastify.statistics.services.query({
  channels: ['company'],
  startTime: '2026-05-01T00:00:00.000Z',
  endTime: '2026-06-01T00:00:00.000Z',
  aggregates: ['sum'],
  includeChildren: true
});

// 3. 同时查询多个通道（仅自身）
const multiResult = await fastify.statistics.services.query({
  channels: ['company:sales', 'company:rd'],
  startTime: '2026-05-01T00:00:00.000Z',
  endTime: '2026-06-01T00:00:00.000Z',
  aggregates: ['sum']
});

// 4. 查询 revenue 和 orders 两个属性的合计与平均
const revenueResult = await fastify.statistics.services.query({
  channels: ['company'],
  startTime: '2026-05-01T00:00:00.000Z',
  endTime: '2026-06-01T00:00:00.000Z',
  attributeNames: ['revenue', 'orders'],
  aggregates: ['sum', 'avg']
});

查询返回格式：

注意：默认（includeChildren=false）只返回精确匹配通道的扁平列表。includeChildren=true 时按通道构建树形结构，子通道数据嵌套在 children 数组中。data 字段始终为对象（按属性名映射），例如单聚合时 data 为 {"default": 58000}，多聚合时 data 为 {"sum": {"default": 58000}, "avg": {"default": 29000}}。

查询销售部（channels=['company:sales']，默认不包含子通道）返回：

{
  "channelMetas": {
    "company": { "channel": "company", "title": "公司", "description": "各部门经营数据统计" }
  },
  "list": [
    {
      "channel": "company:sales",
      "period": "m",
      "time": "2026-05-01T00:00:00.000Z",
      "data": { "default": 130000, "revenue": 72000, "orders": 150 },
      "unit": { "default": "元", "revenue": "元", "orders": "元" }
    }
  ]
}

查询整个公司（channels=['company'], includeChildren=true）返回：

{
  "channelMetas": {
    "company": { "channel": "company", "title": "公司", "description": "各部门经营数据统计" }
  },
  "list": [
    {
      "channel": "company",
      "items": [
        {
          "period": "m",
          "time": "2026-05-01T00:00:00.000Z",
          "data": { "default": 130000, "revenue": 72000, "orders": 150, "tasks": 12, "bugs": 3 },
          "unit": { "default": "元", "revenue": "元", "orders": "元", "tasks": "个", "bugs": "个" }
        }
      ],
      "children": [
        {
          "channel": "company:sales",
          "items": [
            {
              "period": "m",
              "time": "2026-05-01T00:00:00.000Z",
              "data": { "default": 130000, "revenue": 72000, "orders": 150 },
              "unit": { "default": "元", "revenue": "元", "orders": "元" }
            }
          ],
          "children": [
            {
              "channel": "company:sales:beijing",
              "items": [
                {
                  "period": "m",
                  "time": "2026-05-01T00:00:00.000Z",
                  "data": { "default": 58000 },
                  "unit": { "default": "元" }
                }
              ]
            },
            {
              "channel": "company:sales:shanghai",
              "items": [
                {
                  "period": "m",
                  "time": "2026-05-01T00:00:00.000Z",
                  "data": { "revenue": 72000, "orders": 150 },
                  "unit": { "revenue": "元", "orders": "元" }
                }
              ]
            }
          ]
        },
        {
          "channel": "company:rd",
          "items": [
            {
              "period": "m",
              "time": "2026-05-01T00:00:00.000Z",
              "data": { "tasks": 12, "bugs": 3 },
              "unit": { "tasks": "个", "bugs": "个" }
            }
          ],
          "children": [
            {
              "channel": "company:rd:frontend",
              "items": [
                {
                  "period": "m",
                  "time": "2026-05-01T00:00:00.000Z",
                  "data": { "tasks": 12, "bugs": 3 },
                  "unit": { "tasks": "个", "bugs": "个" }
                }
              ]
            }
          ]
        }
      ]
    }
  ]
}

查询多个通道（channels=['company:sales','company:rd']，默认不包含子通道）返回：

{
  "channelMetas": {
    "company": { "channel": "company", "title": "公司", "description": "各部门经营数据统计" }
  },
  "list": [
    {
      "channel": "company:sales",
      "period": "m",
      "time": "2026-05-01T00:00:00.000Z",
      "data": { "default": 130000, "revenue": 72000, "orders": 150 },
      "unit": { "default": "元", "revenue": "元", "orders": "元" }
    },
    {
      "channel": "company:rd",
      "period": "m",
      "time": "2026-05-01T00:00:00.000Z",
      "data": { "tasks": 12, "bugs": 3 },
      "unit": { "tasks": "个", "bugs": "个" }
    }
  ]
}

查询 revenue 和 orders 两个属性的合计与平均（channels=['company'], attributeNames=['revenue','orders'], aggregates=['sum','avg']）返回：

{
  "channelMetas": {
    "company": { "channel": "company", "title": "公司", "description": "各部门经营数据统计" }
  },
  "list": [
    {
      "channel": "company",
      "period": "m",
      "time": "2026-05-01T00:00:00.000Z",
      "data": { "sum": { "revenue": 72000, "orders": 150 }, "avg": { "revenue": 72000, "orders": 150 } },
      "unit": { "revenue": "元", "orders": "元" }
    }
  ]
}

channelMetas 按 root channel 去重，所有子通道共享同一份元数据，避免数据冗余。

Channel Meta 管理

通道元数据在首次采集时自动创建，也可通过服务接口管理：

// 查询通道元数据
const meta = await fastify.statistics.services.channelMeta.detail({
  channel: 'company'
});

// 列出所有元数据
const list = await fastify.statistics.services.channelMeta.list();

// 按通道筛选
const list = await fastify.statistics.services.channelMeta.list({
  filter: { channel: 'company' }
});

// 修改元数据
await fastify.statistics.services.channelMeta.save({
  channel: 'company',
  title: '企业经营数据总览',
  description: '全公司各部门经营指标汇总'
});

SSE 实时推送

通过 HTTP 接口或程序化 API 获取实时统计数据推送：

// HTTP 接口：GET /api/statistics/sse?channel=company&aggregates=sum&interval=5
// 浏览器端使用 EventSource 接收
const eventSource = new EventSource('/api/statistics/sse?channel=company&aggregates=sum&interval=5');
eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log(data); // { channelMetas, list: [{ channel, items, children }] }
};

// 程序化调用（在 Fastify 路由中）
fastify.get('/my-sse', async (request, reply) => {
  const sseContext = await fastify.statistics.services.sseStream.send(reply, {
    name: 'my-sse-channel',
    params: {
      channel: ['company'],
      startTime: new Date(Date.now() - 3600000).toISOString(),
      endTime: new Date().toISOString(),
      aggregates: ['sum']
    },
    fetchData: async (params) => {
      return fastify.statistics.services.query(params);
    },
    interval: 5,
    heartbeatInterval: 30000,
    maxDuration: 1800000
  });

  // 可手动关闭
  // sseContext.close();

  // 监听关闭事件
  sseContext.onClose(() => {
    console.log('SSE 连接已关闭');
  });
});

SSE 事件类型：

事件	说明
data（默认）	正常数据推送，内容为查询结果 JSON
timeout	连接超过 maxDuration 后自动断开通知
error	fetchData 出错时的错误事件
心跳（: heartbeat）	保活注释行

SSE 上下文方法：

方法	说明
isConnected()	返回当前连接状态
close()	手动关闭 SSE 连接
onClose(callback)	注册连接关闭回调，若已断开则立即执行

手动触发聚合与重置

// 手动触发指定周期的聚合
await fastify.statistics.services.periodStat.aggregate('h');
await fastify.statistics.services.periodStat.aggregate('d', {
  startTime: new Date('2026-05-01'),
  endTime: new Date('2026-05-02')
});

// 重置 h 周期数据并级联重置所有下游
const result = await fastify.statistics.services.periodStat.resetPeriodStats('h', {
  startTime: new Date('2026-05-01'),
  endTime: new Date('2026-05-02'),
  cascade: true
});
// result: { period: 'h', deletedCount: 48, nextTime: '2026-05-01T00:00:00.000Z', cascade_d: {...}, cascade_w: {...}, ... }

// 刷新缓冲区
await fastify.statistics.services.dataRecord.flush();

// 清理过期数据
await fastify.statistics.services.dataRecord.cleanup();
await fastify.statistics.services.periodStat.cleanupOldPeriodStats();

示例

API

插件配置

属性名	类型	默认值	说明
prefix	string	/api/statistics	路由前缀
dbTableNamePrefix	string	t_	数据库表名前缀
name	string	statistics	命名空间名称
collectFlushInterval	number	5000	缓冲刷新间隔(ms)
collectMaxBufferSize	number	1000	缓冲区最大条数
collectMaxBufferOverflow	number	maxBufferSize * 2	缓冲区溢出上限
cache	object	null	缓存实例（提供时启用缓冲模式和查询缓存）
compensationEnabled	boolean	true	是否启用启动时自动补偿聚合
compensationBatchSize	number	24	每次补偿聚合的最大窗口数
dataRetentionDays	number	7	原始数据保留天数
queryCacheEnabled	boolean	true	是否启用查询缓存
queryCacheTTL	number	30	实时查询缓存TTL(秒)
queryCacheHistoryTTL	number	3600	历史查询缓存TTL(秒)
queryCacheMaxEntries	number	100	内存查询缓存最大条数（仅无外部缓存时生效）
onRebuild	object	null	重建 hook：{ beforeAggregate, afterAggregate }，参数为 (fastify, ctx)

数据采集

POST {prefix}/collect

采集数据，支持单条或批量上报。

请求体（单条）：

属性名	类型	必填	默认值	说明
channel	string	是	-	数据通道，支持多级格式如 device:sensor1
data	number / object	是	-	数据值。数字为单值，对象为多属性如 {temp: 25, humidity: 60}
title	string	否	channel	标题
description	string	否	-	描述
attributeName	string	否	default	属性名
unit	string / object	否	-	数据单位。字符串时所有属性共用同一单位；对象时按 attributeName 映射单位
time	string	否	当前时间	采集时间(ISO格式)

请求体（批量）：以上对象的数组。

通道展开规则：device:sensor1 展开为 device 和 device:sensor1 两个通道记录。

数据展开规则：data 为对象时按属性名拆分。如 data: {temp: 25, humidity: 60} 拆分为 {attributeName: 'temp', data: 25} 和 {attributeName: 'humidity', data: 60}。

单位展开规则：unit 为字符串时所有属性共用同一单位；为对象时按 attributeName 映射，未匹配到的不设置单位。

缓冲模式：配置 cache 时，采集数据先写入内存缓冲区，定时或缓冲区满时批量写入数据库；否则直接写入。

统计查询

GET {prefix}/query

获取统计结果，自动合并当前小时未聚合的原始数据。

查询参数：

属性名	类型	必填	默认值	说明
channels	string	是	-	数据通道(逗号分隔多个通道)
startTime	string	是	-	开始时间(ISO格式)
endTime	string	是	-	结束时间(ISO格式)
attributeNames	string	否	全部	属性名列表(逗号分隔)
aggregates	string	否	全部	聚合方法列表(逗号分隔): sum,avg,count,min,max
timezone	string	否	服务器时区	客户端时区(IANA格式，如 Asia/Shanghai)
includeChildren	boolean	否	false	是否包含子通道数据（等同 channelScope=descendantsTree）
channelScope	string	否	exact	exact / descendantsFlat / descendantsTree
maxDepth	number	否	-	descendantsFlat 时限制 channel 层级
format	string	否	default	default / flat / totals

通道匹配规则：

• exact（默认）：仅精确匹配 channels
• descendantsTree / includeChildren=true：匹配子通道，返回树形结构
• descendantsFlat：匹配子通道，返回扁平叶子 channel 列表（避免父级重复计数）

返回格式：

默认（includeChildren=false）返回扁平列表：

{
  "channelMetas": {
    "sensor": { "channel": "sensor", "title": "传感器", "description": "" }
  },
  "list": [
    {
      "channel": "sensor",
      "period": "h",
      "time": "2026-05-22T08:00:00.000Z",
      "data": { "default": 100 },
      "unit": { "default": "℃" }
    }
  ]
}

includeChildren=true 时返回树形结构：

{
  "channelMetas": {
    "sensor": { "channel": "sensor", "title": "传感器", "description": "" }
  },
  "list": [
    {
      "channel": "sensor",
      "items": [{ "period": "h", "time": "...", "data": {"default": 100}, "unit": {"default": "℃"} }],
      "children": [
        {
          "channel": "sensor:temp",
          "items": [{ "period": "h", "time": "...", "data": {"default": 25}, "unit": {"default": "℃"} }]
        }
      ]
    }
  ]
}

includeChildren=true 时 list 中每个节点包含：

字段	类型	说明
channel	string	通道名称
items	array	该通道的统计结果数组（按时间排序），每项包含 period、time、data、unit
children	array	子通道数组（递归结构，仅存在子通道时返回）

data 字段格式：

条件	data 格式	示例
单聚合	object	{"default": 100} 或 {"temperature": 25, "humidity": 60}
多聚合	嵌套object	{"sum": {"default": 100}, "avg": {"default": 50}}

SSE 实时推送

GET {prefix}/sse

基于 Server-Sent Events 的实时统计推送，自动查询最近一小时的统计数据并按指定间隔推送。

查询参数：

属性名	类型	必填	默认值	说明
channels	string	是	-	数据通道(逗号分隔多个通道)
attributeNames	string	否	全部	属性名列表(逗号分隔)
aggregates	string	否	全部	聚合方法列表(逗号分隔): sum,avg,count,min,max
timezone	string	否	服务器时区	客户端时区(IANA格式)
includeChildren	boolean	否	false	是否包含子通道数据
interval	number	否	5	推送间隔秒数

响应格式：Content-Type: text/event-stream

事件类型	说明
data（无 event 字段）	正常数据推送，内容为查询结果的 JSON
timeout	连接超过 maxDuration 后自动断开通知
error	fetchData 出错时的错误事件
注释行（: heartbeat）	心跳保活

GET {prefix}/rebuild/sse

聚合重建进度 SSE（长任务，非轮询 query）。

查询参数：

属性名	类型	必填	默认值	说明
mode	string	否	aggregate-only	aggregate-only / reset-and-aggregate / repair
channelFilter	string	否	-	推断起始时间的 channel LIKE，如 interview:%
startTime	string	否	-	repair 或显式起始时间(ISO)
endTime	string	否	-	repair 或显式结束时间(ISO)
periods	string	否	h,d,w,m,q,y	聚合周期列表(逗号分隔)

SSE 事件：

事件	说明
progress	{ stage, message?, percent?, detail? }
done	{ success: true, mode, windowCounts, watermarks }
error	{ message, statusCode? }
注释行	心跳保活

宿主可通过插件选项 onRebuild.beforeAggregate / onRebuild.afterAggregate 注入采集与校验逻辑。

程序化 API

通过 fastify.statistics.services 访问：

通用方法

方法	说明
services.collect(data)	采集数据，同 /collect 接口逻辑
services.query(params)	查询统计，同 /query 接口逻辑

dataRecord 服务

方法	说明
services.dataRecord.collect(data)	同 services.collect
services.dataRecord.flush()	手动刷新缓冲区
services.dataRecord.cleanup()	清理过期的原始数据

periodStat 服务

方法	说明
services.periodStat.init()	初始化水位线并执行启动补偿（插件 onReady 自动调用）
services.periodStat.aggregate(period, opts)	手动触发指定周期的聚合。opts.startTime/opts.endTime 可选，默认聚合上一个时间窗口
services.periodStat.query(params)	同 services.query
services.periodStat.queryFlat(params)	保留策略感知查询 + 扁平 records
services.periodStat.queryTotals(params)	在 queryFlat 之上汇总 totals / totalsByChannel / maxByChannel
services.periodStat.rebuild(opts)	聚合重建（支持 onProgress）
services.periodStat.clearAll(opts)	清空统计表（flush → destroy）
services.periodStat.getRetentionPolicy()	读取保留策略
services.periodStat.getWatermark(period)	读取聚合水位线
services.periodStat.setWatermark(period, nextTime)	设置聚合水位线
services.periodStat.isRebuilding()	是否正在 rebuild
services.periodStat.isRebuildInProgress()	是否有 rebuild 任务 Promise
services.periodStat.isCompensating()	当前是否正在执行补偿聚合
services.periodStat.invalidateQueryCache(channels?)	使查询缓存失效。传入 channels 时只失效相关通道，不传则失效全部
services.periodStat.cleanupOldPeriodStats()	清理过期的周期统计数据
services.periodStat.resetPeriodStats(period, opts)	重置指定周期的数据和水位线，详见下方

resetPeriodStats 参数

属性名	类型	必填	默认值	说明
period	string	是	-	周期类型: h/d/w/m/q/y
opts.startTime	Date	否	当前截断时间	重置起始时间（水位线将设为此值）
opts.endTime	Date	否	全部	仅删除此时间范围内的 period-stat 数据
opts.cascade	boolean	否	false	是否级联重置下游周期

返回值：{ period, deletedCount, nextTime, cascade_h?, cascade_d?, ... }

queryFlat 参数

与 query 相同，另支持 channelScope、maxDepth。

返回值：{ channelMetas, records: FlatRecord[], meta: { channelScope, isRealtime, windowsUsed } }

queryTotals 参数

与 queryFlat 相同，另支持 includeRecords（默认 false）。

返回值：

{
  meta: { retentionPolicy, windowsUsed, isRealtime, channelScope },
  totals: { [attributeName]: number },
  totalsByChannel: { [channel]: { [attributeName]: number } },
  maxByChannel: { [channel]: { [attributeName]: number } },
  attrStats: { [attributeName]: { sum, max, count } },
  channelMetas: {},
  records?: FlatRecord[]
}

rebuild 参数

属性名	类型	必填	默认值	说明
mode	string	否	aggregate-only	aggregate-only / reset-and-aggregate / repair
startTime	Date	否	MIN(data_record.time)	聚合起始（truncate 到 h）
endTime	Date	否	now	聚合结束
periods	string[]	否	h,d,w,m,q,y	要重建的周期
channelFilter	string	否	-	推断 startTime 的 LIKE 过滤
maxWindows	number	否	Infinity	单周期最大窗口数
onProgress	function	否	-	进度回调
beforeAggregate	function	否	-	清库后、聚合前（宿主采集 hook）
afterAggregate	function	否	-	聚合后（宿主校验 hook）

返回值：{ success, mode, windowCounts, watermarks, skipped? }

并发：重复调用抛 409 Rebuild already in progress。

详见 summary.md 中「数据保留策略」「查询辅助 API」「重置、重建与修复」章节。

channelMeta 服务

方法	说明
services.channelMeta.detail({ channel })	查询通道元数据
services.channelMeta.list({ filter? })	列出元数据，filter.channel 可按通道筛选
services.channelMeta.save({ channel, title?, description? })	修改元数据

sseStream 服务

方法	说明
services.sseStream.send(reply, config)	发送 SSE 实时数据流（轮询 query）
services.sseStream.runTask(reply, config)	发送 SSE 长任务进度流

send 配置：

config 属性	类型	必填	默认值	说明
name	string	是	-	缓存键名称标识
params	object	是	-	传递给 fetchData 的参数
fetchData	function	是	-	异步函数 (params) => data
interval	number	否	5	推送间隔秒数
heartbeatInterval	number	否	30000	心跳间隔(ms)
maxDuration	number	否	1800000	最大连接时长(ms)

返回值：SSE 上下文对象

方法	说明
isConnected()	返回当前连接状态
close()	手动关闭 SSE 连接
onClose(callback)	注册连接关闭回调，若已断开则立即执行

runTask 配置：

config 属性	类型	必填	默认值	说明
name	string	是	-	任务名称（日志）
task	function	是	-	异步 ({ emit }) => result，emit(event, data)
heartbeatInterval	number	否	15000	心跳间隔(ms)
maxDuration	number	否	1800000	最大连接时长(ms)

任务完成后自动发送 done 事件；异常时发送 error 事件。

数据模型

data-record（数据采集记录）

属性名	类型	说明
channel	STRING	数据通道(必填)
attributeName	STRING	属性名(默认 default)
data	DECIMAL(16,2)	数据值(必填)
time	DATE	采集时间(必填)
unit	STRING	数据单位

索引：channel、time、[channel, time]、[channel, attributeName, time]、attributeName

period-stat（周期统计）

属性名	类型	说明
period	STRING	统计周期(必填): h/d/w/m/q/y
time	DATE	统计时间(必填)
channel	STRING	数据通道(必填)
attributeName	STRING	属性名(默认 default)
aggregate	ENUM	聚合方法(必填): sum/avg/count/min/max
data	DECIMAL(16,2)	统计数据值(必填)
unit	STRING	数据单位

唯一约束：(period, channel, attributeName, aggregate, time)

索引：[channel, attributeName, time]、[period, time]、attributeName

channel-meta（通道元数据）

属性名	类型	说明
channel	STRING	数据通道(唯一键)
title	STRING	标题(必填)
description	TEXT	描述

唯一约束：channel

说明：按 root channel（一级通道）唯一存储，所有子通道共享同一份元数据。

aggregation-watermark（聚合水位线）

属性名	类型	说明
period	STRING	统计周期(唯一键): h/d/w/m/q/y
nextTime	DATE	下一个待聚合时间

唯一约束：period

说明：水位线记录各周期下一次应聚合的时间起点，用于补偿聚合逻辑。首次聚合时，根据原始数据或上游周期统计的最早时间自动初始化。

统计周期

周期	key	Cron 表达式	数据来源
时	h	1 * * * *	原始数据(data-record)
日	d	1 0 * * *	小时统计(period-stat)
周	w	1 0 * * 1	日统计(period-stat)
月	m	1 0 1 * *	日统计(period-stat)
季	q	1 0 1 1,4,7,10 *	月统计(period-stat)
年	y	1 0 1 1 *	季统计(period-stat)

聚合方法

方法	key	说明
合计	sum	数值求和
平均	avg	数值平均(由sum/count计算)
计数	count	记录计数
最小	min	最小值
最大	max	最大值

补偿聚合机制

插件启动时自动执行补偿聚合（可通过 compensationEnabled: false 关闭）：

• 水位线记录各周期下一次应聚合的时间起点
• 补偿时从水位线开始逐步向前聚合，直到追上当前时间
• 每次最多处理 compensationBatchSize 个时间窗口
• 上游周期未完成时自动先补偿上游
• 每个周期有独立的补偿锁，防止并发补偿
• 连续失败 maxCompensationFailCount 次（默认 3）后停止，下次 Cron 继续
• 补偿聚合通过 Cron 定时触发，启动时也会执行一次

查询缓存

特性	说明
无外部缓存	使用内存 LRU 缓存，最大 queryCacheMaxEntries 条
有外部缓存	查询结果存入外部缓存，支持 TTL 和版本校验
版本校验	缓存条目记录写入时的通道版本号，读取时校验版本是否变化
TTL 策略	实时查询用 queryCacheTTL（30s），历史查询用 queryCacheHistoryTTL（3600s）
补偿期间	正在执行补偿聚合时查询不走缓存

数据保留策略

数据类型	保留策略	安全检查
data-record	dataRetentionDays 天（默认 7 天）	不超过 h 周期水位线
period-stat(h)	当月	不超过 d 水位线
period-stat(d)	当年	不超过 w、m 水位线
period-stat(w)	当年	无下游依赖
period-stat(m/q/y)	永久保留	-