import { Client } from "@elastic/elasticsearch"; import { env } from "./env.js"; import { logger } from "../shared/observability/logger.js"; /** * msg 服务 Elasticsearch 客户端。 * * 仲裁依据: * - ES_URL 可选,未配置时 esClient=null,降级到 MySQL LIKE 查询 * - 02-architecture-design.md §3.2.1:notifications 索引 mapping(ik_max_word 分词) */ export const esClient: Client | null = env.ES_URL ? new Client({ node: env.ES_URL }) : null; /** * notifications 索引 mapping(对齐 02-architecture-design.md §3.2.1)。 * * ik_max_word:索引时最大粒度分词;ik_smart:查询时智能分词。 * 若 ES 未安装 ik 分词器,会回退到 standard 分词器(不影响功能)。 */ const NOTIFICATIONS_INDEX_MAPPING = { mappings: { properties: { id: { type: "keyword" }, user_id: { type: "keyword" }, type: { type: "keyword" }, title: { type: "text", analyzer: "ik_max_word", search_analyzer: "ik_smart", }, content: { type: "text", analyzer: "ik_max_word", search_analyzer: "ik_smart", }, channel: { type: "keyword" }, status: { type: "keyword" }, group_id: { type: "keyword" }, related_entity_type: { type: "keyword" }, related_entity_id: { type: "keyword" }, sender_id: { type: "keyword" }, is_read: { type: "boolean" }, created_at: { type: "date" }, read_at: { type: "date" }, }, }, settings: { number_of_shards: 1, number_of_replicas: 1, }, } as const; export async function checkEsConnection(): Promise { if (!esClient) { logger.info("Elasticsearch disabled (ES_URL not set)"); return; } try { await esClient.ping(); await ensureNotificationsIndex(); logger.info("Elasticsearch connected, notifications index ensured"); } catch (err) { logger.error({ err }, "Elasticsearch connection failed"); } } /** * 幂等创建 notifications 索引。若索引已存在则跳过。 */ export async function ensureNotificationsIndex(): Promise { if (!esClient) return; try { const exists = await esClient.indices.exists({ index: "notifications" }); if (!exists) { await esClient.indices.create({ index: "notifications", ...NOTIFICATIONS_INDEX_MAPPING, }); logger.info("notifications index created"); } } catch (err) { logger.error({ err }, "ensureNotificationsIndex failed"); } } export async function closeEs(): Promise { if (esClient) { await esClient.close(); } } export interface IndexParams { index: string; id: string; document: Record; } export interface SearchHit { _id: string; _source: Record; } export interface SearchResult { hits: SearchHit[]; total: number; } export async function safeIndex(params: IndexParams): Promise { if (!esClient) return false; try { await esClient.index(params); return true; } catch (err) { logger.error( { err, index: params.index, id: params.id }, "Elasticsearch index failed", ); return false; } } export async function safeSearch( index: string, query: Record, from: number = 0, size: number = 20, ): Promise { if (!esClient) return { hits: [], total: 0 }; try { const result = await esClient.search({ index, query, from, size, }); const hits: SearchHit[] = []; for (const raw of result.hits.hits) { const source = raw._source; if ( raw._id && source && typeof source === "object" && !Array.isArray(source) ) { hits.push({ _id: raw._id, _source: source as Record, }); } } const total = typeof result.hits.total === "number" ? result.hits.total : (result.hits.total?.value ?? hits.length); return { hits, total }; } catch (err) { logger.error({ err, index }, "Elasticsearch search failed"); return { hits: [], total: 0 }; } } /** * 删除索引文档(撤回通知时同步删除 ES 索引)。 */ export async function safeDelete(index: string, id: string): Promise { if (!esClient) return false; try { await esClient.delete({ index, id }); return true; } catch (err) { logger.error({ err, index, id }, "Elasticsearch delete failed"); return false; } }