Civitai Event Engine 通用层源码解析：缓存体系、Meilisearch Feed 与指标服务的实战指南-北京尧图网络

Civitai Event Engine 通用层源码解析缓存体系、Meilisearch Feed 与指标服务的实战指南【免费下载链接】civitaiA repository of models, textual inversions, and more项目地址: https://gitcode.com/GitHub_Trending/ci/civitai导读本文以 apps/event-engine/src/common/README.md 为骨架深入剖析 Civitai 事件引擎Event Engine中通用服务与工具Common Services Utilities一层的设计与实现。这一层承载了指标事件监听应用与 Feed 系统的共享能力基于 Redis 的分布式缓存含防缓存击穿的分布式锁、stale-while-revalidate、TTL 滑动等机制、基于 ClickHouse Redis 的实体指标服务、基于 PostgreSQL 的 Outbox 可靠事件追踪以及基于 Meilisearch 的 Feed 数据管道。读完本文你将掌握如何在本仓库中新增一个缓存、注册一个 Feed并理解其底层调用链与真实工程坑位如全数字用户名的类型推断事故 civitai#4768。目录结构一次看懂通用层布局common/目录是整个事件引擎与 Feed 系统的公共底座布局如下与 README 保持一致并对照实际源码树apps/event-engine/src/common/ ├── caches/ # 缓存实现 │ ├── base.ts # createCache 工厂与核心逻辑 │ ├── value-codec.ts # 缓存字段类型声明与反序列化类型安全关键 │ ├── index.ts # 缓存注册表barrel export │ ├── userData.cache.ts # 示例缓存用户数据 │ ├── modelData.cache.ts # 模型数据缓存 │ └── imageData.cache.ts # 图片标签/标签/装扮/头像等缓存 ├── feeds/ # Feed 实现 │ ├── base.ts # createFeed 工厂与核心逻辑 │ ├── types.ts # Feed 类型定义Schema/Context/配置 │ ├── index.ts # Feed 注册表 │ └── images.feed.ts # 真实示例图片 Feed1286 行含完整 NSFW/排序逻辑 ├── services/ # 核心服务 │ ├── cache.ts # 集中式缓存服务 CacheService │ ├── metrics.ts # 指标抓取服务 MetricService │ ├── outbox.ts # Outbox 模式服务 │ └── signals.ts # 信号相关服务 ├── types/ # 类型定义 │ ├── metric-types.ts # 各实体指标类型生成文件 │ ├── database.ts # 数据库类型 │ ├── image-feed-types.ts # 图片 Feed 类型 │ └── meilisearch/ # Meilisearch 文档/索引/输入类型 └── utils/ # 工具函数 ├── basic.ts # chunk / sleep ├── async-utils.ts # createAsyncBatcher / runWithConcurrency ├── cache-keys.ts # 统一的 Redis key 生成 ├── query-utils.ts # 数据库客户端包装 ├── meilisearch-helpers.ts # Meilisearch 索引初始化 └── redis-packer.ts # Redis 数据打包MessagePack从实际源码看index.ts是公共层的统一出口同时导出了缓存注册表与全部服务类型便于上层以import * as common from ../common的方式引用。核心服务CacheService / MetricService / OutboxServiceCacheService类型安全的集中式缓存入口CacheServiceservices/cache.ts是所有缓存访问的统一门面解决了散落各处、类型混乱的问题类型安全访问fetchK(name, ids)中name的类型是keyof typeof caches返回值类型通过AwaitedReturnType(typeof caches)[K][fetch]自动推断——写cacheService.fetch(userData, [1,2,3])就能得到Recordnumber, UserCacheData不需要手动标注泛型。集中上下文管理构造时统一接收redis、pg、chClickHouse三个客户端并包装为统一的CacheContext若传入可选的packer如 MessagePack 打包器Redis 客户端会被withRedisPacking包装。统一操作接口fetch/bust/refresh三个方法分别对应缓存的读、失效、刷新。额外的 Redis 直通能力mGet/set/sAdd三个方法在提供packer的前提下开放打包后的 Redis 操作供 Feed 做自定义缓存如图片 Feed 的排序分页缓存。const cacheService new CacheService(redis, pg, ch, msgpackPacker); // Fetch user data - return type is automatically inferred const users await cacheService.fetch(userData, [1, 2, 3]); // Returns: Recordnumber, UserCacheData // Bust cache for updated users await cacheService.bust(userData, [1, 2, 3]); // Refresh cache with fresh data from database await cacheService.refresh(modelData, [100, 101]);若传入不存在的缓存名三个方法都会抛出Cache named ... could not be found。MetricServiceClickHouse 指标 Redis 读穿透缓存MetricServiceservices/metrics.ts负责从 ClickHouse 聚合表entityMetricDailyAgg抓取实体指标并用 Redis 做读穿透缓存。它的核心设计点两级 TTL命中指标缓存 24 小时CACHE_TTL未命中结果notFound: 1只缓存 5 分钟MISS_CACHE_TTL避免对不存在的实体反复打库。防击穿分布式锁缓存未命中时通过setNxKeepTtlWithEx(lockKey, 1, 2)抢锁锁 key 为metrics:lock:${entityType}:${id}见 utils/cache-keys.ts抢到锁的进程去 ClickHouse 拉数据并回填抢不到锁的进程以 200ms 间隔最多重试 10 次从缓存重新读取。热键 TTL 滑动每次访问时有 10% 概率对该键执行expire(key, CACHE_TTL)让高频访问的指标常驻。按实体类型推断返回类型fetchT extends EntityType(entityType: T, ids)的返回类型由EntityMetricMap[T]决定例如fetch(Article, [...])返回Recordnumber, ArticleMetricsENTITY_METRIC_TYPES定义在 types/metric-types.ts该文件为生成文件由npm run generate-types生成。零值补齐返回结果会对每个请求 ID 补齐该实体类型的全部指标字段并填 0保证调用方拿到的永远是完整形状。fetchTimeframes则用 ClickHouse 的sumIf一次性算出 Day / Week / Month / Year / AllTime 五个时间窗的指标分别对应today()、subtractWeeks(today(), 1)、subtractMonths(today(), 1)、subtractYears(today(), 1)并按 1000 个 ID 一批分批查询SELECT entityId, metricType, sumIf(total, day today()) AS Day, sumIf(total, day subtractWeeks(today(), 1)) AS Week, sumIf(total, day subtractMonths(today(), 1)) AS Month, sumIf(total, day subtractYears(today(), 1)) AS Year, sum(total) AS AllTime FROM entityMetricDailyAgg WHERE entityType ... AND entityId IN (...) AND metricType IN (...) GROUP BY entityId, metricTypeconst metricService new MetricService(clickhouse, redis); // Fetch article metrics const articleMetrics await metricService.fetch(Article, [1, 2, 3]); // Returns: Recordnumber, ArticleMetrics // Fetch metrics with timeframes const timeframeMetrics await metricService.fetchTimeframes(Image, [4, 5, 6]); // Bust cache for updated entities await metricService.bustCache(Model, [7, 8]);OutboxService基于 PostgreSQL 的可靠事件追踪OutboxServiceservices/outbox.ts实现 Outbox 模式业务变更先落 PostgreSQL 的Outbox表event / entityType / entityId再由后续消费者处理从而保证业务操作与领域事件的最终一致。目前支持Article | Image | Model | Post | ModelVersion五类实体的事件记录如 PUBLISHED、UNPUBLISHED、DELETED、UPDATED。const outboxService new OutboxService(pgClient); // Add an event to the outbox await outboxService.add({ event: OutboxEvent.PUBLISHED, entityType: Article, entityId: 123 }); // Delete processed event await outboxService.delete(recordId);add对应INSERT INTO Outbox (event, entityType, entityId) VALUES ($1, $2, $3)delete对应按主键DELETE实现极为精简把可靠性的语义留给了上层调度器。缓存系统createCache 工厂的完整工作机制缓存架构与特性每个缓存都由 caches/base.ts 的createCacheT()工厂创建并在 caches/index.ts 中注册。工厂内置以下高级特性对应源码中的具体常量特性说明源码常量/默认值分布式锁防止热门 key 缓存击穿stampedeLOCK_DURATION 2秒LOCK_RETRY_DELAY 200msLOCK_MAX_RETRIES 10Stale-while-revalidate过期数据先返回后台异步刷新默认开启刷新期 TTL 翻倍为ttl * 2TTL 滑动热条目访问时 10% 概率续期CACHE_SLIDE_CHANCE 0.1Debounce 处理写入传播期内避免读到只读副本的旧数据debounceTime默认 10 秒未命中缓存不存在的记录以短 TTL 缓存降低 DB 压力MISS_CACHE_TTL 5分钟批量操作高效批量 fetch / bust / refreshRedis 查询按 200 key/批DB 抓取按 10000 id/批Fetch 流程读路径源码中fetch()的执行步骤可归纳为Redis 批量查询将所有redisKey:id拼成 key按 200 个一批并发hGetAll。结果分类命中且为notFound 标记仅notFound: 1一个字段→ 跳过视为未找到命中但带debounce: 1且cachedAt在 debounce 窗口内 → 视为未命中数据可能未传播到读副本命中且未过期 → 直接返回并有 10% 概率加入 TTL 滑动列表命中但已过期cachedAt Date.now() - ttl * 1000→ 进入待重校验集合先返回旧数据完全未命中 → 进入 miss 列表。TTL 滑动对滑动的 key 批量expire(key, ttl)失败仅打印错误不影响主流程。重校验锁对待重校验的 key 抢lock:redisKey:idSET NX EX 2抢到锁的进入 miss 列表重新抓取抢不到的返回旧数据。miss 处理再次抢锁抢到锁的按 10000 id/批调用config.fetch()从数据库取数回填缓存命中数据用staleWhileRevalidate ? ttl * 2 : ttl作为过期时间未命中写notFound: 1并给 5 分钟 TTL最后释放锁没抢到锁的进程每 200ms 重试读缓存最多 10 次。Bust 流程失效路径bust()并非直接删除 key而是写入debounce: 1 当前cachedAt并以debounceTime默认 10 秒作为过期时间。这样在主库已写、只读副本尚未同步的窗口期内下一次 fetch 会跳过缓存直查数据库从而避免读到旧数据。Refresh 流程刷新路径refresh()直接调用config.fetch()取最新数据并整体覆写缓存对于数据库中已不存在的 ID则删除对应缓存 key保证缓存与数据库最终一致。手把手新增一个缓存Step 1在 caches/ 下新建缓存文件以userData.cache.ts为模板import { createCache, CacheContext } from ./base; export type UserCacheData { userId: number; // Must match idKey below username: string; image?: string; deletedAt?: Date | null; }; export const userData createCacheUserCacheData({ redisKey: user:data, // Redis key prefix idKey: userId, // Property to use as ID // Fetch function returns array of items async fetch({ pg }: CacheContext, ids: number[]) { const users await pg.queryUserCacheData( SELECT id as userId, username, image, deletedAt FROM User WHERE id ANY($1), [ids] ); return users; }, // Optional configuration ttl: 60 * 60 * 24, // 24 hours (default) debounceTime: 10, // 10 seconds (default) cacheNotFound: true, // Cache misses (default: true) staleWhileRevalidate: true, // Background refresh (default: true) // Optional: Skip caching for specific data dontCacheFn: (data) data.deletedAt ! null, });Step 2在 caches/index.ts 中导出export { userData } from ./userData.cache; export type { UserCacheData } from ./userData.cache;Step 3通过 CacheService 使用推荐或在 Feed 上下文中直接使用ctx.cache.fetch// Via CacheService (recommended) const users await cacheService.fetch(userData, [1, 2, 3]); // Or directly in feed context const users await ctx.cache.fetch(userData, [1, 2, 3]);配置选项完整说明{ redisKey: string; // Redis key prefix (e.g., user:data) idKey: keyof T; // Property name used as the ID fetch: (ctx, ids) T[]; // Function to fetch data from database ttl?: number; // Cache TTL in seconds (default: 24 hours) debounceTime?: number; // Write propagation delay (default: 10s) cacheNotFound?: boolean; // Cache not found results (default: true) dontCacheFn?: (data) boolean; // Skip caching certain records staleWhileRevalidate?: boolean; // Background refresh (default: true) }关键工程细节value-codec 与字段类型声明这是 caches/base.ts 中最值得注意的约束CacheConfig的fieldTypes字段是必填且必须完整非Partial的记录类型。原因是 Redis Hash 只能存字符串反序列化时必须知道每个字段的真实类型。此前代码通过isNaN(Number(value)) ? value : Number(value)猜测类型导致了一系列真实事故civitai#4768 / civitai/cli#513全数字用户名0222被读成数字222前导零被破坏查询不到任何账户2428023993被读成 JSON 数字API 返回username:2428023993未加引号所有类型化客户端解码失败被Number() 0变成0false被当成 truthy 字符串用户名叫null时deletedAt被误判。修复方案见 caches/value-codec.ts是为每个字段显式声明类型而不是改存储格式——因为存量 Redis 数据格式本身是正确的改格式会引入新旧版本进程混读的窗口期。CacheFieldType支持string | string? | number | number? | boolean | boolean? | date | date? | json | json?其中带?表示可空可把存储文本null正确解码为null。decodeCacheFields是唯一的解码入口普通读取与锁竞争重试两条路径都走它杜绝了两套解码逻辑漂移cachedAt、notFound、debounce三个保留字段永远不参与实体解码。这一点同样体现在真实的 userData.cache.ts 中username必须声明为string且故意不声明为可空——因为User.username列是NOT NULL而null本身是合法用户名usernameSchema为/^[A-Za-z0-9_]*$/。Feed 系统createFeed 工厂与 Meilisearch 数据管道Feed 架构与特性Feed 系统feeds/base.ts feeds/types.ts提供以 Meilisearch 为搜索引擎、集成指标与缓存的文档管道Schema 驱动用FeedSchema声明文档结构自动推导出 TypeScript 类型InferSchemaType并在configure()时自动把sortable/filterable字段同步为 Meilisearch 属性集成指标通过ctx.metric.fetch()内置从 ClickHouse 抓取指标并写入文档缓存集成通过ctx.cache.fetch()访问所有注册缓存用于文档填充如取用户名批量操作upsert支持可配置并发与双层批量fetchBatchSize默认 1000 /upsertBatchSize默认 100000游标分页query内置offset|timestamp游标返回{ items, nextCursor }懒配置构造时只做只读的getIndex写操作upsert/delete前才configure()且configure()幂等——索引不存在时createIndex属性有变化时才updateSortableAttributes/updateFilterableAttributes避免无谓地反复打 Meilisearch。Feed 方法一览方法说明upsert(ids, type?: full \| metrics)插入/更新 Meilisearch 文档full全量重建metrics仅更新指标字段轻量delete(ids)从 Meilisearch 删除文档任务入队即返回不等完成query(input)带过滤与分页的查询返回{ items, nextCursor }populate(docs, input)用关联数据用户、URL 等丰富文档返回TPop[]populatedQuery(input)querypopulate一步完成query的实现细节会请求limit 1条结果来判断是否还有下一页若有多余结果则从最后一个返回项取时间戳优先getCursor()默认sortAtUnix/sortAt/id拼接出nextCursor ${offset limit}|${timestamp}。手把手新增一个 FeedStep 1在 feeds/ 下新建 Feed 文件以 README 中的image.feed.ts示例为模板import { createFeed } from ./base; import type { ImageMetrics } from ../types/metric-types; // Define schema for Meilisearch const schema { id: { type: number as const, primary: true, filterable: true }, userId: { type: number as const, filterable: true }, nsfw: { type: boolean as const, filterable: true }, createdAt: { type: Date as const, sortable: true }, heartCount: { type: number as const, sortable: true }, likeCount: { type: number as const, sortable: true }, } as const; // Document stored in Meilisearch type ImageDocument { id: number; userId: number; nsfw: boolean; createdAt: Date; heartCount: number; likeCount: number; }; // Query input (filters) type ImageQueryInput { nsfw?: boolean; userId?: number; }; // Populated document returned to API type PopulatedImage ImageDocument { url: string; username: string; }; export const ImageFeed createFeed({ entityType: Image as const, name: images, connection: { host: process.env.MEILISEARCH_IMAGE_INDEX_URL!, apiKey: process.env.MEILISEARCH_API_KEY!, }, schema, // Create documents from entity IDs async createDocuments(ctx, ids, type full) { const images await ctx.pg.queryOmitImageDocument, heartCount | likeCount( SELECT id, userId, nsfw, createdAt FROM Image WHERE id ANY($1), [ids] ); // Fetch metrics const metrics await ctx.metric.fetch(images.map(i i.id)); // Combine return images.map((img): ImageDocument ({ ...img, heartCount: metrics[img.id]?.Heart ?? 0, likeCount: metrics[img.id]?.Like ?? 0, })); }, // Query documents from Meilisearch async queryDocuments(ctx, input: ImageQueryInput): PromiseImageDocument[] { const { limit, cursor } ctx.pagination; const { nsfw, userId } input; // Build filters const filters: string[] []; if (nsfw ! undefined) filters.push(nsfw ${nsfw}); if (userId) filters.push(userId ${userId}); // Handle cursor pagination let cursorFilter ; if (cursor) { const [createdAt, id] cursor.split(:); cursorFilter createdAt ${createdAt} OR (createdAt ${createdAt} AND id ${id}); if (filters.length) cursorFilter AND (${cursorFilter}); } // Search const result await ctx.index.searchImageDocument(null, { filter: filters.join( AND ) cursorFilter, sort: [createdAt:desc, id:desc], limit, }); return result.hits; }, // Populate documents with related data async populateDocuments(ctx, documents) { const userIds [...new Set(documents.map(d d.userId))]; const users await ctx.cache.fetch(userData, userIds); return documents.map((doc): PopulatedImage ({ ...doc, url: https://image.civitai.com/${doc.id}, username: users[doc.userId]?.username ?? unknown, })); }, // Optional: Advanced options options: { fetchBatchSize: 1000, // Batch size for createDocuments upsertBatchSize: 100000, // Batch size for Meilisearch updates createConcurrency: 2, // Concurrent createDocuments calls }, });Step 2在 feeds/index.ts 中导出export { ImageFeed } from ./image.feed;Step 3在应用中实例化并使用// Initialize feed const imageFeed new ImageFeed(meilisearch, ch, pg, metricService, cacheService); // Upsert documents await imageFeed.upsert([1, 2, 3], full); // Query documents const images await imageFeed.query({ nsfw: false, userId: 123, limit: 20, cursor: 2024-01-15T10:00:00.000Z:456 }); // Query and populate in one call const populatedImages await imageFeed.populatedQuery({ nsfw: false, limit: 20 });注意createFeed返回的是类构造器实例化时第一个参数是 Meilisearch 初始化函数未在connection配置时会回退到process.env.FEED_API_KEY/process.env.FEED_HOST默认http://localhost:7700见 feeds/base.ts。Schema 字段类型{ type: string | number | Date | boolean | array; arrayType?: string | number | boolean; // For array types sortable?: boolean; // Make field sortable in Meilisearch filterable?: boolean; // Make field filterable in Meilisearch primary?: boolean; // Mark as primary key }type: array时可用arrayType声明数组元素类型InferSchemaType会据此推导出string[] | number[] | boolean[]的 TS 类型。真实世界的复杂示例见 feeds/images.feed.ts 的 schema它定义了 40 个字段包括可排序的sortAt/sortAtUnix、数组类型的modelVersionIds/tagIds/toolIds/techniqueIds、NSFW 相关的nsfwLevel/combinedNsfwLevel/availability/blockedFor/poi/minor以及既 sortable 又 filterable 的指标字段reactionCount/commentCount/collectedCount同时支持排序与游标过滤。Feed Context所有 Feed 回调函数收到的ctx参数定义见 feeds/types.ts{ pg: { queryT(sql, params?) PromiseT[] }, // PostgreSQL返回 rows ch: { queryT(sql, params?) PromiseT[] }, // ClickHouseJSONEachRow自动做 $1 参数替换 cache: { fetch(name, ids) PromiseRecord... }, // 缓存访问含 mGet / set / sAdd metric: { fetch(ids) PromiseRecord... }, // 指标访问已绑定当前实体类型 index: IMeilisearchIndex, // Meilisearch 索引就绪后可用 pagination: { limit: number, cursor?: string, offset?: number }, // 分页参数 }注意ch.query会把$1、$2之类的占位符替换为实际参数再以JSONEachRow格式查询并json()解析index是惰性 getter索引未就绪时访问会抛Index not ready。实际调用前应await feed.ready()。工具函数与依赖Utils 一览utils/basic.tschunk(array, size)分块、sleep(ms)异步休眠——它们是缓存批量操作与锁重试的基础原语utils/async-utils.tscreateAsyncBatcher(batchSize, flushFn)创建异步批处理器enqueue累积、达到阈值自动 flush、flush收尾runWithConcurrency(tasks, concurrency)以固定并发数执行任务数组Feed.upsert正是用这两个工具实现并发建文档批量写 Meilisearchutils/cache-keys.ts统一生成 Redis key——指标metrics:${entityType}:${id}、指标锁metrics:lock:${entityType}:${id}保证全仓 key 命名一致utils/meilisearch-helpers.tsgetMeilisearchFeed(config)初始化 Feed 用的 Meilisearch 索引其他query-utils.ts数据库客户端包装SimpleClickhouse/withRedisHelpers等、redis-packer.tsRedis 打包供 MessagePack 压缩、nsfw-utils.ts/s3-utils.ts/logger.ts。依赖关系所有服务与工具共同依赖以下模块utils/query-utils.ts —— 数据库客户端包装带辅助方法types/metric-types.ts —— 实体与指标类型定义生成文件含EntityMetricMap/ENTITY_METRIC_TYPEStypes/package-stubs.ts —— 外部包接口定义IRedisClient/IDbClient/IClickhouseClient/IPgClient/IDataPacker等保证类型不依赖具体 SDK 实现utils/cache-keys.ts —— 统一的 Redis key 生成utils/basic.ts —— 基础工具函数chunk、sleep 等。小结apps/event-engine/src/common是 Civitai 事件引擎的可复用底座其价值集中在三个层面类型安全缓存按名推断、Feed 由 Schema 推断文档类型、指标按实体类型推断返回类型、高并发稳健性分布式锁防击穿、stale-while-revalidate、debounce 防读到只读副本旧数据、TTL 滑动保热键、工程事故沉淀value-codec的字段类型声明机制直接修复了全数字用户名被错误反序列化的线上事故。后续接入新的实体 Feed 或缓存时只需遵循建文件 → 注册 barrel → 通过 Service 使用三步模式即可复用整套分布式缓存与 Meilisearch 管道能力。【免费下载链接】civitaiA repository of models, textual inversions, and more项目地址: https://gitcode.com/GitHub_Trending/ci/civitai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

Civitai Event Engine 通用层源码解析：缓存体系、Meilisearch Feed 与指标服务的实战指南

相关推荐

CocosCreator3.8.2多分辨率适配与横竖屏自动切换完整方案

MMPose for AIGC：生成姿态骨架图，为姿态引导式 AI 图像生成提供条件输入

低功耗策略实战：从功耗预算到系统平衡的完整设计指南

更多资讯

SLG服务器开发：菱形瓦片与笛卡尔坐标互转原理与落地

FckSignups 的 PWA 化改造思路：让工具导航可以离线安装（附完整步骤）

远程设备运维实战：不靠机票搞定几百里外的Bug

kube-state-metrics 第三方依赖管理策略详解：从 docs/dependencies-policy.md 到 go.mod 与 CI 强制校验

Electron+Vue驱动的电力缺陷知识图谱问答系统前端实践

职场人如何高效利用碎片时间与应急技巧

今日更新

AWS无服务器应用开发指南：从Lambda到SAM的架构与实践

SpringBoot+Vue人事系统实战：解决HR数据断点与事务一致性

MATLAB手写紧束缚模型计算石墨烯能带结构

本周精选

AI驱动的自动化专利分析平台技术解析

LeetCode-Go 题解：1252. Cells with Odd Values in a Matrix（奇数值单元格计数）双解法剖析

在线答疑系统Java毕设实战：Spring Boot前后端分离与状态流转

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？