diff --git a/docs/ai-brand-monitoring-tech-design-v4.md b/docs/ai-brand-monitoring-tech-design-v4.md new file mode 100644 index 0000000..3ed82b4 --- /dev/null +++ b/docs/ai-brand-monitoring-tech-design-v4.md @@ -0,0 +1,1408 @@ +# AI 品牌曝光监测系统技术方案 V4 + +## 1. 文档信息 + +| 项目 | 内容 | +| --- | --- | +| 文档名称 | AI 品牌曝光监测系统技术方案 V4（插件采集 + 配额约束） | +| 文档版本 | V4.0 | +| 文档状态 | 待评审 | +| 创建日期 | 2026-04-08 | +| 基线文档 | `docs/ai-brand-monitoring-tech-design-v3.md` | +| 适用范围 | 数据追踪模块全部页面 | +| 关联文档 | `docs/geo-platform-prd-v1.md` (PRD 8.4 数据追踪) | +| 关联文档 | `docs/ai-brand-monitoring-collection-feasibility-v1.md` (采集可行性) | +| 容量目标 | 5 万用户（B2B 共享租户，~3,000 租户） | + +## 2. V4 修订范围 + +V4 在 V3 基础上做两项根本性变更： + +1. **采集方式**：从服务端 AI API 调用改为 **浏览器插件采集**，消除 AI API 成本。 +2. **配额约束**：引入租户级品牌/关键词/问题配额，控制采集规模。 + +### 2.1 与 V3 的关系 + +| V3 章节 | V4 处理 | 原因 | +| --- | --- | --- | +| 2.1 数据口径分层 | 保留，`run_mode` 值调整为 `plugin_*` | 口径逻辑不变，采集来源变了 | +| 2.2 Schema Delta | 保留 + 扩展 | 新增配额表、任务分配字段 | +| 3. 读流量与缓存 | **完全保留** | 读侧与采集方式无关 | +| 4. 采集流水线 | **全部替换** | 插件采集替代服务端 Collector | +| 5. 部署拓扑 | 精简 | 去掉 Collector Pod、Queue Redis | +| 6. 降级与容灾 | 保留读侧 + 新增插件降级 | 新增采集不足的降级策略 | +| 7. 压测验收 | 保留读侧 + 新增插件压测 | 新增插件并发场景 | +| 8. 实施计划 | 重写 | 工期和顺序均有变化 | + +V3 其他章节（页面需求、前端组件、API 响应结构、指标定义）保持不变。 + +--- + +## 3. 用户配额体系 + +### 3.1 配额规则 + +| 配额项 | 普通租户（Free） | 高级租户（Pro） | 企业租户（Enterprise） | +| --- | --- | --- | --- | +| 最大品牌数 | 1 | 3 | 自定义 | +| 每品牌关键词数 | 8 | 8 | 自定义 | +| 每关键词问题数 | 5 | 5 | 自定义 | +| **每品牌最大问题数** | **40** | **40** | **自定义** | +| 采集频率 | 每 3 天 | 每天 | 每天 + 按需触发 | +| AI 平台数 | 3（首批） | 6（全部） | 6 + 自定义 | +| 搜索增强采集 | 不支持 | 支持 | 支持 | + +### 3.2 配额表 Schema + +```sql +CREATE TABLE tenant_monitoring_quotas ( + tenant_id UUID PRIMARY KEY REFERENCES tenants(id), + max_brands INT NOT NULL DEFAULT 1, + max_keywords_per_brand INT NOT NULL DEFAULT 8, + max_questions_per_keyword INT NOT NULL DEFAULT 5, + collect_frequency VARCHAR(20) NOT NULL DEFAULT 'every_3_days', + enabled_platforms JSONB NOT NULL DEFAULT '["deepseek","qwen","doubao"]', + search_enhanced BOOLEAN NOT NULL DEFAULT false, + plan_tier VARCHAR(20) NOT NULL DEFAULT 'free', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +COMMENT ON COLUMN tenant_monitoring_quotas.collect_frequency IS 'daily / every_3_days / weekly'; +COMMENT ON COLUMN tenant_monitoring_quotas.plan_tier IS 'free / pro / enterprise'; +``` + +### 3.3 采集规模估算 + +| 参数 | 估值 | +| --- | --- | +| 总用户 | 50,000 | +| 租户数 | ~3,000（平均 15-20 人/租户） | +| 高级租户（10%） | 300，品牌数 900 | +| 普通租户（90%） | 2,700，品牌数 2,700 | +| 品牌实例总数（最大） | 3,600 | +| 监测采用率（30%） | 活跃监测品牌 ~1,080 | +| 实际问题数/品牌（70% 使用率） | ~28 | + +每日采集任务量（中等场景）： + +``` +标准模式: + 1,080 品牌 × 28 问题 × 6 平台 = 181,440 任务/天 + +考虑采集频率分层（Free 每 3 天）: + 高级 (30%): 324 品牌 × 28q × 6 平台 = 54,432/天 + 普通 (70%): 756 品牌 × 28q × 6 平台 ÷ 3 = 42,336/天 + 合计 ≈ 96,768 任务/天（均摊） +``` + +--- + +## 4. 插件采集架构 + +### 4.1 现有插件能力复用 + +当前 GEO Publisher 插件（`apps/browser-extension/`）已具备完整基础设施： + +| 现有能力 | 文件位置 | 监测复用方式 | +| --- | --- | --- | +| 平台适配器模式 | `src/adapters/` (13 个) | 新增 AI 平台 adapter | +| 登录状态检测 | 各 adapter 的 `detect()` | 检测 AI 平台登录态 | +| 后台 Service Worker | `entrypoints/background.ts` | 定时触发采集任务 | +| 页面 ↔ 插件通信 | `entrypoints/content.ts` (postMessage) | 前端可触发手动采集（辅助通道） | +| 后端回调 | `src/runtime.ts` (HTTP POST) | 采集结果回传后端 | +| 后端直连 | `installation_token` + `api_base_url` | **插件后台直接拉取任务 + 回传结果** | +| Chrome Storage | `src/storage.ts` | 存储采集队列和进度 | +| DNR 跨域规则 | `public/rules.json` (18 条) | 新增 AI 平台规则 | +| 安装认证 | `installation_token` + `installation_id` | 复用，校验采集权限 | + +### 4.2 整体采集流程 + +``` +┌──────────────────────────────────────────────────────────────────┐ +│ 后端调度器 │ +│ 1. 根据 tenant_monitoring_quotas 和 collect_frequency │ +│ 生成当日采集任务（品牌 × 去重问题 × 平台） │ +│ 2. 写入 monitoring_collect_tasks 表 │ +│ 3. 状态机: pending → assigned → completed / failed / expired │ +└───────────────────────────────┬──────────────────────────────────┘ + │ + 插件 Background SW 定时心跳（每 15 分钟） + ▼ +┌──────────────────────────────────────────────────────────────────┐ +│ 浏览器插件 Background Service Worker │ +│ │ +│ 主通道：插件后台直接拉取（不依赖前端页面） │ +│ ────────────────────────────────────────── │ +│ 4. 心跳触发 → GET /api/plugin/monitoring/tasks │ +│ (Header: X-Geo-Installation-Token) │ +│ → 后端根据 installation → tenant 映射，返回待采集任务 │ +│ 5. POST /api/plugin/monitoring/tasks/{id}/claim → 领取任务 │ +│ 6. 按 AI 平台分发到对应 adapter │ +│ 7. 检测用户是否登录该 AI 平台（Cookie 检测） │ +│ 8. 已登录 → 调用平台 Web API 创建对话 → 输入问题 → 等待回答 │ +│ 未登录 → 标记 skip │ +│ 9. 提取回答文本 + 引用信息（如有） │ +│ 10. POST /api/callback/plugin/monitor → 结果回传后端 │ +│ │ +│ 辅助通道：前端 postMessage（用户主动触发时） │ +│ ────────────────────────────────────────── │ +│ Admin Web 点击"立即采集" → postMessage → 插件立即执行 │ +│ (仅作为加速手段，不是必须路径) │ +└───────────────────────────────┬──────────────────────────────────┘ + │ + POST /api/callback/plugin/monitor + ▼ +┌──────────────────────────────────────────────────────────────────┐ +│ 后端采集服务 │ +│ 11. 校验 installation_token + 任务归属 │ +│ 12. 解析回答：品牌提及、位置、情感、引用 │ +│ 13. 写入 question_monitor_runs + question_monitor_parse_results │ +│ 14. 原始回答存 MinIO（对象存储） │ +│ 15. 标记任务 completed │ +└──────────────────────────────────────────────────────────────────┘ + + ↓ 每日 6:30 聚合 + +┌──────────────────────────────────────────────────────────────────┐ +│ Aggregator Job（不变） │ +│ 16. 按 business_date 聚合已完成的采集数据 │ +│ 17. 写入 10 张 monitoring_*_daily 汇总表 │ +│ 18. 预热热门品牌缓存 │ +└──────────────────────────────────────────────────────────────────┘ +``` + +**关键设计**：插件通过 Background SW 的定时心跳（每 15 分钟）**直接调用后端 API** 拉取和领取任务，不依赖 Admin Web 前端页面是否打开。前端 postMessage 仅作为用户主动触发"立即采集"的辅助通道。 + +后端需要新增 **插件专用 API**（不走 JWT 认证，走 installation_token 认证）： + +``` +GET /api/plugin/monitoring/tasks → 返回当前插件所属租户的 pending 任务 +POST /api/plugin/monitoring/tasks/{id}/claim → 领取任务（分布式锁） +POST /api/callback/plugin/monitor → 回传采集结果（已有回调模式） +POST /api/callback/plugin/monitor/batch-skip → 批量跳过未登录平台的任务 +``` + +### 4.3 AI 平台适配器 + +#### 4.3.1 适配器接口 + +```typescript +// apps/browser-extension/src/adapters/ai/types.ts + +export interface AIMonitorAdapter { + /** 平台标识 */ + platformId: AIPlatformId; + + /** 检测用户是否已登录该 AI 平台 */ + detect(): Promise; + + /** 向 AI 平台提问并获取回答 */ + ask(question: string, options?: AskOptions): Promise; +} + +export type AIPlatformId = 'deepseek' | 'qwen' | 'doubao' | 'kimi' | 'ernie' | 'hunyuan'; + +export interface AIMonitorPlatformState { + platformId: AIPlatformId; + connected: boolean; + userId?: string; + nickname?: string; + message?: string; +} + +export interface AskOptions { + /** 是否启用联网搜索 */ + searchEnabled?: boolean; + /** 任务超时（毫秒） */ + timeout?: number; +} + +export interface AIMonitorResult { + success: boolean; + question: string; + answer: string; + citations: Citation[]; + searchResults: SearchResult[]; + responseTime: number; + model: string; + timestamp: string; + rawHtml?: string; +} + +export interface Citation { + url: string; + title: string; + domain: string; + snippet?: string; +} + +export interface SearchResult { + url: string; + title: string; + snippet: string; +} +``` + +#### 4.3.2 各平台采集方式 + +| AI 平台 | Web 端 URL | 采集方式 | 登录检测 | 引用支持 | +| --- | --- | --- | --- | --- | +| DeepSeek | chat.deepseek.com | 调用 Web 端 chat API（SSE 流） → 拼接完整回答 | Cookie 检测 | 无原生引用 | +| 千问 | tongyi.aliyun.com | 调用 Web 端对话 API → 等待回答完成 | Cookie 检测 | 联网搜索模式支持 | +| 豆包 | doubao.com | 调用 Web 端对话 API → 等待回答 | Cookie 检测 | 联网搜索支持 | +| Kimi | kimi.moonshot.cn | 调用 Web 端对话 API → 等待回答 | Cookie 检测 | 联网搜索支持 | +| 文心一言 | yiyan.baidu.com | 调用 Web 端对话 API → 等待回答 | 百度账号 Cookie | 支持 | +| 混元 | hunyuan.tencent.com | 调用 Web 端对话 API → 等待回答 | Cookie 检测 | 搜索增强支持 | + +**核心实现模式**（以 DeepSeek 为例）： + +```typescript +// apps/browser-extension/src/adapters/ai/deepseek.ts + +import type { AIMonitorAdapter, AIMonitorPlatformState, AIMonitorResult, AskOptions } from './types'; + +const DEEPSEEK_CHAT_URL = 'https://chat.deepseek.com'; +const DEEPSEEK_API_BASE = 'https://chat.deepseek.com/api/v0'; + +export const deepseekAdapter: AIMonitorAdapter = { + platformId: 'deepseek', + + async detect(): Promise { + try { + // 通过 Cookie 判断登录状态，类似现有 zhihu/bilibili adapter + const cookies = await browser.cookies.getAll({ domain: '.deepseek.com' }); + const hasSession = cookies.some(c => c.name === 'ds_session' || c.name === 'token'); + + if (!hasSession) { + return { platformId: 'deepseek', connected: false, message: '未登录' }; + } + + // 验证 token 有效性 + const resp = await fetch(`${DEEPSEEK_API_BASE}/user/current`, { + credentials: 'include', + }); + + if (!resp.ok) { + return { platformId: 'deepseek', connected: false, message: '登录已过期' }; + } + + const data = await resp.json(); + return { + platformId: 'deepseek', + connected: true, + userId: data.data?.id, + nickname: data.data?.nickname, + }; + } catch { + return { platformId: 'deepseek', connected: false, message: '检测失败' }; + } + }, + + async ask(question: string, options?: AskOptions): Promise { + const startTime = Date.now(); + const timeout = options?.timeout ?? 60_000; + + try { + // 1. 创建新对话 + const createResp = await fetch(`${DEEPSEEK_API_BASE}/chat/create`, { + method: 'POST', + credentials: 'include', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({}), + }); + const { data: chat } = await createResp.json(); + + // 2. 发送问题并监听 SSE 流 + const answer = await streamChat(chat.id, question, timeout); + + return { + success: true, + question, + answer: answer.text, + citations: [], // DeepSeek 标准模式无引用 + searchResults: [], + responseTime: Date.now() - startTime, + model: answer.model ?? 'deepseek-chat', + timestamp: new Date().toISOString(), + }; + } catch (err: any) { + return { + success: false, + question, + answer: '', + citations: [], + searchResults: [], + responseTime: Date.now() - startTime, + model: '', + timestamp: new Date().toISOString(), + }; + } + }, +}; + +async function streamChat(chatId: string, question: string, timeout: number) { + // SSE 流式读取，拼接完整回答 + // 具体实现需根据 DeepSeek Web 端实际 API 格式调整 + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeout); + + try { + const resp = await fetch(`${DEEPSEEK_API_BASE}/chat/completion`, { + method: 'POST', + credentials: 'include', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + chat_id: chatId, + prompt: question, + stream: true, + }), + signal: controller.signal, + }); + + const reader = resp.body!.getReader(); + const decoder = new TextDecoder(); + let fullText = ''; + let model = ''; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + const chunk = decoder.decode(value, { stream: true }); + // 解析 SSE data 行 + for (const line of chunk.split('\n')) { + if (line.startsWith('data: ')) { + try { + const json = JSON.parse(line.slice(6)); + if (json.choices?.[0]?.delta?.content) { + fullText += json.choices[0].delta.content; + } + if (json.model) model = json.model; + } catch { /* 跳过非 JSON 行 */ } + } + } + } + return { text: fullText, model }; + } finally { + clearTimeout(timer); + } +} +``` + +#### 4.3.3 适配器注册表 + +```typescript +// apps/browser-extension/src/adapters/ai/index.ts + +import { deepseekAdapter } from './deepseek'; +import { qwenAdapter } from './qwen'; +import { doubaoAdapter } from './doubao'; +import { kimiAdapter } from './kimi'; +import { ernieAdapter } from './ernie'; +import { hunyuanAdapter } from './hunyuan'; +import type { AIMonitorAdapter, AIPlatformId } from './types'; + +export const aiAdapters: Record = { + deepseek: deepseekAdapter, + qwen: qwenAdapter, + doubao: doubaoAdapter, + kimi: kimiAdapter, + ernie: ernieAdapter, + hunyuan: hunyuanAdapter, +}; + +export function getAIAdapter(platformId: AIPlatformId): AIMonitorAdapter | undefined { + return aiAdapters[platformId]; +} +``` + +### 4.4 采集任务调度器（插件侧） + +```typescript +// apps/browser-extension/src/monitor/scheduler.ts + +import { getAIAdapter } from '../adapters/ai'; +import type { AIMonitorResult, AIPlatformId } from '../adapters/ai/types'; +import { getState } from '../storage'; + +/** 反检测配置 */ +const ANTI_DETECT = { + maxQueriesPerPlatformPerHour: 3, + minIntervalSec: 30, + maxIntervalSec: 120, + maxDailyQueries: 50, +}; + +interface CollectTask { + id: number; + question_text: string; + platform_id: AIPlatformId; + brand_id: number; + question_id: number; + search_enabled: boolean; +} + +/** 执行一批采集任务 */ +export async function executeCollectTasks(tasks: CollectTask[]): Promise { + const state = await getState(); + if (!state.installation_token || !state.api_base_url) return; + + // 按平台分组，串行执行（避免同时操作多个平台） + const byPlatform = groupBy(tasks, t => t.platform_id); + + for (const [platformId, platformTasks] of Object.entries(byPlatform)) { + const adapter = getAIAdapter(platformId as AIPlatformId); + if (!adapter) continue; + + // 检测登录状态 + const detection = await adapter.detect(); + if (!detection.connected) { + // 标记该平台全部任务为 skip + await reportSkipped(state, platformTasks, '未登录'); + continue; + } + + // 逐条执行，加入随机延迟 + for (const task of platformTasks) { + // 检查日限额 + if (await getDailyCount() >= ANTI_DETECT.maxDailyQueries) break; + // 检查平台小时限额 + if (await getPlatformHourlyCount(platformId) >= ANTI_DETECT.maxQueriesPerPlatformPerHour) break; + + const result = await adapter.ask(task.question_text, { + searchEnabled: task.search_enabled, + timeout: 60_000, + }); + + // 回传后端 + await reportResult(state, task, result); + await incrementCounters(platformId); + + // 随机延迟 + const delay = randomBetween(ANTI_DETECT.minIntervalSec, ANTI_DETECT.maxIntervalSec); + await sleep(delay * 1000); + } + } +} + +async function reportResult( + state: ExtensionStorageState, + task: CollectTask, + result: AIMonitorResult, +) { + await fetch(`${state.api_base_url}/api/callback/plugin/monitor`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Geo-Installation-Token': state.installation_token!, + 'X-Geo-Installation-Id': String(state.plugin_installation_id), + }, + body: JSON.stringify({ + task_id: task.id, + platform_id: task.platform_id, + brand_id: task.brand_id, + question_id: task.question_id, + success: result.success, + answer: result.answer, + citations: result.citations, + search_results: result.searchResults, + response_time: result.responseTime, + model: result.model, + timestamp: result.timestamp, + raw_html: result.rawHtml, + client_version: browser.runtime.getManifest().version, + }), + }); +} + +async function reportSkipped(state: ExtensionStorageState, tasks: CollectTask[], reason: string) { + await fetch(`${state.api_base_url}/api/callback/plugin/monitor/batch-skip`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Geo-Installation-Token': state.installation_token!, + 'X-Geo-Installation-Id': String(state.plugin_installation_id), + }, + body: JSON.stringify({ + task_ids: tasks.map(t => t.id), + reason, + }), + }); +} + +function randomBetween(min: number, max: number): number { + return Math.floor(Math.random() * (max - min + 1)) + min; +} + +function sleep(ms: number): Promise { + return new Promise(resolve => setTimeout(resolve, ms)); +} + +function groupBy(arr: T[], key: (item: T) => string): Record { + return arr.reduce((acc, item) => { + const k = key(item); + (acc[k] ??= []).push(item); + return acc; + }, {} as Record); +} +``` + +### 4.5 采集执行策略 + +采用**插件自主 + 前端辅助**模式： + +| 模式 | 触发方式 | 单次任务数 | 频率 | 依赖前端 | +| --- | --- | --- | --- | --- | +| **心跳自主采集（主通道）** | Background SW 定时心跳 | 3-5 个任务 | 每 15 分钟 | 否 | +| **用户主动采集（辅助）** | Dashboard 点击"立即采集" → postMessage | 批量 pending 任务 | 用户触发 | 是 | + +心跳自主采集保证：**只要用户安装了插件且浏览器在运行，采集就在持续进行**，无需用户打开 Admin Web。 + +每日限额：单用户最多 50 个采集任务。 + +### 4.6 Background SW 扩展 + +#### 4.6.1 新增采集心跳（15 分钟间隔） + +```typescript +// entrypoints/background.ts + +// 原有 60 分钟心跳保持不变（平台检测 + 状态同步） +// 新增 15 分钟采集心跳 +browser.alarms.create('monitor-heartbeat', { periodInMinutes: 15 }); + +browser.alarms.onAlarm.addListener(async (alarm) => { + if (alarm.name === 'heartbeat') { + // 原有逻辑：刷新内容平台登录状态 + await refreshPlatformState(); + } + + if (alarm.name === 'monitor-heartbeat') { + // 新增：插件后台直接拉取并执行采集任务 + await pullAndExecuteMonitorTasks(); + } +}); + +async function pullAndExecuteMonitorTasks() { + const state = await getState(); + if (!state.installation_token || !state.api_base_url) return; + + try { + // 1. 直接调后端 API 拉取待采集任务（不经过前端） + const resp = await fetch(`${state.api_base_url}/api/plugin/monitoring/tasks`, { + headers: { + 'X-Geo-Installation-Token': state.installation_token, + 'X-Geo-Installation-Id': String(state.plugin_installation_id), + }, + }); + if (!resp.ok) return; + + const { data: tasks } = await resp.json(); + if (!tasks?.length) return; + + // 2. 执行采集（串行，带反检测延迟） + await executeCollectTasks(tasks); + } catch { + // 静默失败，下次心跳再试 + } +} +``` + +#### 4.6.2 前端辅助通道（用户主动触发） + +```typescript +// 新增 action 处理（前端 postMessage 触发） +case 'collectMonitorData': + // 用户点击"立即采集"时，前端传入任务列表 + return await handleCollectMonitorData(payload); + +case 'detectAIPlatforms': + // 检测 AI 平台登录状态（供前端展示） + return await handleDetectAIPlatforms(); +``` + +### 4.7 分布式任务调度（后端侧） + +#### 4.7.1 任务生成 + +```go +// 每日凌晨由 Aggregator 或独立 CronJob 生成 +func (s *MonitoringScheduler) GenerateDailyTasks(ctx context.Context, businessDate time.Time) error { + // 1. 获取所有启用监测的租户 + tenants, _ := s.repo.ListMonitoringTenants(ctx) + + for _, tenant := range tenants { + quota, _ := s.repo.GetMonitoringQuota(ctx, tenant.ID) + + // 检查采集频率 + if !shouldCollectToday(quota.CollectFrequency, businessDate) { + continue + } + + // 2. 获取租户的品牌和问题 + brands, _ := s.repo.ListBrandsWithQuestions(ctx, tenant.ID, quota.MaxBrands) + + for _, brand := range brands { + for _, question := range brand.Questions { + for _, platformID := range quota.EnabledPlatforms { + // 3. 生成任务（幂等，按唯一键去重） + s.repo.UpsertCollectTask(ctx, MonitorCollectTask{ + TenantID: tenant.ID, + BrandID: brand.ID, + QuestionID: question.ID, + PlatformID: platformID, + BusinessDate: businessDate, + RunMode: "plugin_standard", + Status: "pending", + SearchEnabled: quota.SearchEnhanced && platformSupportsSearch(platformID), + }) + } + } + } + } + return nil +} +``` + +#### 4.7.2 任务领取（分布式锁） + +```go +// GET /api/tenant/monitoring/collect-tasks +func (s *MonitoringService) GetPendingTasks(ctx context.Context, tenantID uuid.UUID, limit int) ([]CollectTask, error) { + // 只返回 status=pending 的任务 + return s.repo.ListPendingCollectTasks(ctx, tenantID, limit) +} + +// POST /api/tenant/monitoring/collect-tasks/{id}/claim +func (s *MonitoringService) ClaimTask(ctx context.Context, taskID int64, installationID string) error { + // Redis 分布式锁，TTL 10 分钟 + lockKey := fmt.Sprintf("mon:task_lock:%d", taskID) + acquired, _ := s.cache.SetNX(ctx, lockKey, installationID, 10*time.Minute) + if !acquired { + return ErrTaskAlreadyClaimed + } + + // 更新任务状态 + return s.repo.UpdateCollectTaskStatus(ctx, taskID, "assigned", installationID) +} +``` + +#### 4.7.3 结果接收（HTTP → RabbitMQ 异步处理） + +插件回传的采集结果通过 RabbitMQ 异步处理，API 层只做校验和投递，**不阻塞插件等待解析完成**： + +```go +// POST /api/callback/plugin/monitor +func (h *MonitorCallbackHandler) HandleMonitorResult(c *gin.Context) { + var req MonitorResultRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(400, gin.H{"error": "invalid request"}) + return + } + + // 校验 installation_token + installationToken := c.GetHeader("X-Geo-Installation-Token") + if !h.service.ValidateInstallation(c, installationToken) { + c.JSON(401, gin.H{"error": "unauthorized"}) + return + } + + // 仅做两件事：标记任务 assigned → received，投递到 MQ + // 解析、入库、MinIO 存储全部由 Consumer 异步完成 + if err := h.service.MarkTaskReceived(c, req.TaskID); err != nil { + c.JSON(409, gin.H{"error": "task not found or already completed"}) + return + } + + // 投递到 RabbitMQ + if err := h.mq.Publish(c, "monitor.result", req); err != nil { + // MQ 投递失败 → 回退任务状态，插件下次心跳会重试 + h.service.MarkTaskPending(c, req.TaskID) + c.JSON(500, gin.H{"error": "queue unavailable"}) + return + } + + c.JSON(200, gin.H{"ok": true}) +} +``` + +### 4.8 RabbitMQ 异步处理架构 + +#### 4.8.1 队列拓扑 + +``` + 插件 HTTP 回传 + │ + ▼ +┌──────────────────────────────────────────────────┐ +│ tenant-api (HTTP 层) │ +│ 校验 token → 标记 received → publish to MQ │ +└───────────────────────┬──────────────────────────┘ + │ + ▼ +┌──────────────────────────────────────────────────┐ +│ RabbitMQ Broker │ +│ │ +│ Exchange: monitor.direct (direct) │ +│ ┌─────────────────────────────────────────────┐ │ +│ │ Queue: monitor.result.parse │ │ +│ │ routing_key: monitor.result │ │ +│ │ consumer: MonitorParseWorker × 3 │ │ +│ │ 作用: 解析回答 + 入库 + MinIO 存储 │ │ +│ └─────────────────────────────────────────────┘ │ +│ ┌─────────────────────────────────────────────┐ │ +│ │ Queue: monitor.task.generate │ │ +│ │ routing_key: monitor.task.generate │ │ +│ │ consumer: TaskGenerateWorker × 1 │ │ +│ │ 作用: 凌晨批量生成采集任务 │ │ +│ └─────────────────────────────────────────────┘ │ +│ ┌─────────────────────────────────────────────┐ │ +│ │ Queue: monitor.aggregate.trigger │ │ +│ │ routing_key: monitor.aggregate │ │ +│ │ consumer: AggregateWorker × 1 │ │ +│ │ 作用: 按品牌触发增量聚合 │ │ +│ └─────────────────────────────────────────────┘ │ +│ ┌─────────────────────────────────────────────┐ │ +│ │ Queue: monitor.result.dlq (死信队列) │ │ +│ │ routing_key: monitor.result.dlq │ │ +│ │ 作用: 处理失败的消息暂存，人工排查 │ │ +│ └─────────────────────────────────────────────┘ │ +└──────────────────────────────────────────────────┘ +``` + +#### 4.8.2 Parse Worker（核心消费者） + +```go +// server/internal/monitoring/worker/parse_worker.go + +type MonitorParseWorker struct { + repo *repository.MonitoringRepository + parser *MonitorAnswerParser + storage objectstorage.Client + mq *rabbitmq.Client +} + +func (w *MonitorParseWorker) Handle(ctx context.Context, msg MonitorResultMessage) error { + // 1. 存原始回答到 MinIO + objectKey := fmt.Sprintf("monitor/%s/%d/%d.json", + msg.Timestamp[:10], msg.TaskID, time.Now().UnixMilli()) + if err := w.storage.PutJSON(ctx, objectKey, msg); err != nil { + return fmt.Errorf("minio put failed: %w", err) + } + + // 2. 写 question_monitor_runs + run, err := w.repo.CreateMonitorRun(ctx, CreateMonitorRunInput{ + TaskID: msg.TaskID, + PlatformID: msg.PlatformID, + BrandID: msg.BrandID, + QuestionID: msg.QuestionID, + RunMode: "plugin_standard", + RawAnswer: msg.Answer, + ResponseTime: msg.ResponseTime, + Model: msg.Model, + ObjectKey: objectKey, + }) + if err != nil { + return fmt.Errorf("create run failed: %w", err) + } + + // 3. 解析回答（品牌提及、位置、情感、引用） + parseResults := w.parser.Parse(ctx, run, msg.Answer, msg.Citations) + if err := w.repo.SaveParseResults(ctx, run.ID, parseResults); err != nil { + return fmt.Errorf("save parse results failed: %w", err) + } + + // 4. 标记任务 completed + if err := w.repo.CompleteTask(ctx, msg.TaskID); err != nil { + return fmt.Errorf("complete task failed: %w", err) + } + + // 5. 检查该品牌当日任务是否全部完成 → 触发增量聚合 + allDone, _ := w.repo.IsBrandDailyTasksComplete(ctx, msg.BrandID, msg.BusinessDate) + if allDone { + w.mq.Publish(ctx, "monitor.aggregate", AggregateMessage{ + BrandID: msg.BrandID, + TenantID: msg.TenantID, + BusinessDate: msg.BusinessDate, + }) + } + + return nil +} +``` + +#### 4.8.3 Aggregate Worker（增量聚合） + +```go +// server/internal/monitoring/worker/aggregate_worker.go + +type AggregateWorker struct { + repo *repository.MonitoringRepository + cache cache.Cache +} + +func (w *AggregateWorker) Handle(ctx context.Context, msg AggregateMessage) error { + // 按品牌 + business_date 增量聚合（替代 V3 的全量定时聚合） + // 好处：品牌任务完成即聚合，不必等到每日 6:30 + if err := w.repo.AggregateBrandDaily(ctx, msg.TenantID, msg.BrandID, msg.BusinessDate); err != nil { + return err + } + + // 预热该品牌缓存 + w.cache.WarmBrandComposite(ctx, msg.TenantID, msg.BrandID) + return nil +} +``` + +#### 4.8.4 RabbitMQ 配置 + +```yaml +# server/configs/config.yaml 新增 +rabbitmq: + url: "amqp://guest:guest@localhost:5672/" + exchange: "monitor.direct" + queues: + result_parse: + name: "monitor.result.parse" + routing_key: "monitor.result" + concurrency: 3 # 3 个并发消费者 + prefetch: 5 # 每个消费者预取 5 条 + retry_max: 3 # 最大重试 3 次 + retry_delay: "5s" # 重试间隔 + dlq: "monitor.result.dlq" + task_generate: + name: "monitor.task.generate" + routing_key: "monitor.task.generate" + concurrency: 1 + aggregate: + name: "monitor.aggregate.trigger" + routing_key: "monitor.aggregate" + concurrency: 1 + prefetch: 1 +``` + +#### 4.8.5 Go 客户端初始化 + +```go +// server/internal/shared/mq/rabbitmq.go + +import amqp "github.com/rabbitmq/amqp091-go" + +type Client struct { + conn *amqp.Connection + channel *amqp.Channel + exchange string +} + +func NewClient(url, exchange string) (*Client, error) { + conn, err := amqp.Dial(url) + if err != nil { + return nil, err + } + ch, err := conn.Channel() + if err != nil { + return nil, err + } + + // 声明 exchange + err = ch.ExchangeDeclare(exchange, "direct", true, false, false, false, nil) + if err != nil { + return nil, err + } + + return &Client{conn: conn, channel: ch, exchange: exchange}, nil +} + +func (c *Client) Publish(ctx context.Context, routingKey string, body any) error { + data, _ := json.Marshal(body) + return c.channel.PublishWithContext(ctx, c.exchange, routingKey, false, false, amqp.Publishing{ + DeliveryMode: amqp.Persistent, + ContentType: "application/json", + Body: data, + MessageId: uuid.NewString(), + Timestamp: time.Now(), + }) +} + +func (c *Client) Consume(queueName string, prefetch int, handler func(amqp.Delivery)) error { + c.channel.Qos(prefetch, 0, false) + msgs, err := c.channel.Consume(queueName, "", false, false, false, false, nil) + if err != nil { + return err + } + for msg := range msgs { + handler(msg) + } + return nil +} +``` + +#### 4.8.6 异步化带来的架构优势 + +| 对比项 | 同步处理（V4 原方案） | RabbitMQ 异步 | +| --- | --- | --- | +| 插件回调延迟 | ~500ms（含解析+入库+MinIO） | ~10ms（仅校验+投递） | +| API Pod 负载 | 高（CPU 密集的解析在 API 进程） | 低（解析转移到 Worker） | +| 解析失败影响 | 插件收到 500，需重试整个采集 | 消息重试，不影响插件 | +| 聚合时机 | 每日 6:30 全量聚合 | **品牌级增量聚合**（任务完成即触发） | +| 数据可见延迟 | T+1（次日早上） | **准实时**（品牌任务完成后分钟级可见） | +| 峰值吸收 | API 直接承压 | MQ 削峰填谷 | +| 故障隔离 | 解析崩溃 → API 不可用 | Worker 崩溃 → 消息堆积 → 恢复后自动消化 | + +### 4.8 DNR 规则扩展 + +在 `apps/browser-extension/public/rules.json` 中新增 AI 平台的跨域规则： + +```json +[ + { + "id": 101, + "action": { + "type": "modifyHeaders", + "requestHeaders": [ + {"header": "Referer", "operation": "set", "value": "https://chat.deepseek.com"}, + {"header": "origin", "operation": "set", "value": "https://chat.deepseek.com"} + ] + }, + "condition": { + "urlFilter": "https://chat.deepseek.com/*", + "resourceTypes": ["xmlhttprequest"] + } + }, + { + "id": 102, + "action": { + "type": "modifyHeaders", + "requestHeaders": [ + {"header": "Referer", "operation": "set", "value": "https://tongyi.aliyun.com"}, + {"header": "origin", "operation": "set", "value": "https://tongyi.aliyun.com"} + ] + }, + "condition": { + "urlFilter": "https://tongyi.aliyun.com/*", + "resourceTypes": ["xmlhttprequest"] + } + }, + { + "id": 103, + "action": { + "type": "modifyHeaders", + "requestHeaders": [ + {"header": "Referer", "operation": "set", "value": "https://www.doubao.com"}, + {"header": "origin", "operation": "set", "value": "https://www.doubao.com"} + ] + }, + "condition": { + "urlFilter": "https://www.doubao.com/*", + "resourceTypes": ["xmlhttprequest"] + } + }, + { + "id": 104, + "action": { + "type": "modifyHeaders", + "requestHeaders": [ + {"header": "Referer", "operation": "set", "value": "https://kimi.moonshot.cn"}, + {"header": "origin", "operation": "set", "value": "https://kimi.moonshot.cn"} + ] + }, + "condition": { + "urlFilter": "https://kimi.moonshot.cn/*", + "resourceTypes": ["xmlhttprequest"] + } + }, + { + "id": 105, + "action": { + "type": "modifyHeaders", + "requestHeaders": [ + {"header": "Referer", "operation": "set", "value": "https://yiyan.baidu.com"}, + {"header": "origin", "operation": "set", "value": "https://yiyan.baidu.com"} + ] + }, + "condition": { + "urlFilter": "https://yiyan.baidu.com/*", + "resourceTypes": ["xmlhttprequest"] + } + }, + { + "id": 106, + "action": { + "type": "modifyHeaders", + "requestHeaders": [ + {"header": "Referer", "operation": "set", "value": "https://hunyuan.tencent.com"}, + {"header": "origin", "operation": "set", "value": "https://hunyuan.tencent.com"} + ] + }, + "condition": { + "urlFilter": "https://hunyuan.tencent.com/*", + "resourceTypes": ["xmlhttprequest"] + } + } +] +``` + +--- + +## 5. 数据口径修订（覆盖 V3 2.1） + +### 5.1 `run_mode` 调整 + +| `run_mode` | 含义 | 进入主统计 | 默认消费页面 | +| --- | --- | --- | --- | +| `plugin_standard` | 插件在 AI Web 端标准对话采集 | 是 | Dashboard、平台占比、竞争对手、业务主题、AI 对话问题、品牌印象 | +| `plugin_search` | 插件在 AI Web 端开启联网搜索后采集 | 是，单独统计 | AI 引用排名 | +| `api_standard` | 后端 API 直接调用（降级/验证用） | 是 | 同 plugin_standard | + +插件采集的回答更接近真实用户体验。采集可行性方案已确认 DeepSeek API 与 Web 版本不同，插件方案反而是优势。 + +### 5.2 Schema Delta 补充 + +在 V3 Schema Delta 基础上，新增/修改： + +#### 5.2.1 `monitoring_collect_tasks` 表扩展 + +```sql +CREATE TABLE monitoring_collect_tasks ( + id BIGSERIAL PRIMARY KEY, + tenant_id UUID NOT NULL REFERENCES tenants(id), + brand_id INT NOT NULL, + question_id INT NOT NULL, + question_version_id INT NOT NULL, + ai_platform_id VARCHAR(30) NOT NULL, + run_mode VARCHAR(30) NOT NULL DEFAULT 'plugin_standard', + business_date DATE NOT NULL, + search_enabled BOOLEAN NOT NULL DEFAULT false, + status VARCHAR(20) NOT NULL DEFAULT 'pending', + assigned_to VARCHAR(100), -- 插件 installation_id + assigned_at TIMESTAMPTZ, + completed_at TIMESTAMPTZ, + error_message TEXT, + retry_count INT NOT NULL DEFAULT 0, + priority INT NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + + CONSTRAINT uk_collect_task_idempotent + UNIQUE (tenant_id, brand_id, question_id, ai_platform_id, run_mode, business_date) +); + +CREATE INDEX idx_collect_tasks_pending + ON monitoring_collect_tasks(tenant_id, status, business_date) + WHERE status = 'pending'; + +COMMENT ON COLUMN monitoring_collect_tasks.status IS 'pending / assigned / completed / failed / expired / skipped'; +COMMENT ON COLUMN monitoring_collect_tasks.assigned_to IS '领取该任务的插件 installation_id'; +``` + +#### 5.2.2 `ai_platforms` 表 + +```sql +CREATE TABLE ai_platforms ( + id VARCHAR(30) PRIMARY KEY, + name VARCHAR(100) NOT NULL, + web_url VARCHAR(200) NOT NULL, + supports_standard BOOLEAN NOT NULL DEFAULT true, + supports_search BOOLEAN NOT NULL DEFAULT false, + supports_citation BOOLEAN NOT NULL DEFAULT false, + plugin_adapter_available BOOLEAN NOT NULL DEFAULT false, + display_order INT NOT NULL DEFAULT 0, + enabled BOOLEAN NOT NULL DEFAULT true, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- 初始数据 +INSERT INTO ai_platforms (id, name, web_url, supports_search, supports_citation, plugin_adapter_available, display_order) VALUES +('deepseek', 'DeepSeek', 'https://chat.deepseek.com', false, false, true, 1), +('qwen', '通义千问', 'https://tongyi.aliyun.com', true, true, true, 2), +('doubao', '豆包', 'https://www.doubao.com', true, true, true, 3), +('kimi', 'Kimi', 'https://kimi.moonshot.cn', true, true, false, 4), +('ernie', '文心一言', 'https://yiyan.baidu.com', true, true, false, 5), +('hunyuan', '混元', 'https://hunyuan.tencent.com', true, true, false, 6); +``` + +--- + +## 6. 读流量模型与缓存架构（保留 V3 不变） + +**完全保留 V3 第 3 章的设计**，包括： + +- 3.1 页面 Fan-out 矩阵 +- 3.2 QPS 模型（13,500 设计容量 vs 实际 ~900 QPS，绰绰有余） +- 3.3 BFF 聚合层（Dashboard composite 接口） +- 3.4 缓存防雪崩（singleflight + TTL 抖动 + stale-while-revalidate + L1） +- 3.5 缓存 Key 修订（`scope` 改为对应 `plugin_standard`/`plugin_search`） +- 3.6 缓存失效策略（渐进式失效 + 预热） +- 3.7 TTL 矩阵 + +--- + +## 7. 部署拓扑与资源（修订 V3 第 5 章） + +### 7.1 部署架构 + +``` + ┌─────────────────────────────┐ + │ CDN (静态资源) │ + └──────────────┬──────────────┘ + │ + ┌──────────────▼──────────────┐ + │ Nginx / SLB │ + │ 限流: 每 IP 200 req/s │ + └──────────┬───────────────────┘ + │ + ┌─────────────────────┼─────────────────────┐ + ▼ ▼ ▼ + ┌────────────┐ ┌────────────┐ ┌────────────┐ + │ tenant-api │ │ tenant-api │ │ (HPA 弹性) │ + │ Pod #1 │ │ Pod #2 │ │ Pod #3 │ + │ L1 cache │ │ L1 cache │ │ │ + └──┬────┬─────┘ └──┬────┬─────┘ └──┬────┬─────┘ + │ │ │ │ │ │ + 查询读取 │ │ 投递消息 │ │ │ │ + │ │ │ │ │ │ + ┌──▼────┴─────────────────▼────┴─────────────────▼────┴──┐ + │ Cache Redis (主从) │ + │ 用途: API 缓存 + JWT 会话 + 任务分布式锁 │ + └────────────────────────┬────────────────────────────────┘ + │ + ┌────────────────────────┼────────────────────────┬───────────────┐ + │ │ │ │ + ▼ ▼ ▼ ▼ +┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌──────────────┐ +│ PG Primary │────────▶│ PG Replica │ │ MinIO │ │ RabbitMQ │ +│ (读写) │ 流复制 │ (只读) │ │ (原始回答) │ │ (异步队列) │ +└─────────────┘ └─────────────┘ └─────────────┘ └──────┬───────┘ + ▲ ▲ │ + │ │ │ + │ │ 消费消息 │ + │ │ ▼ +┌──────┴────────────────────────┴────────────────────────────────────────────┐ +│ Worker 进程（与 API 独立部署） │ +│ │ +│ ┌─────────────────────────────────────────────────────────┐ │ +│ │ Monitor Worker (单进程，多 goroutine) │ │ +│ │ • ParseWorker × 3 ← consume: monitor.result.parse │ │ +│ │ • AggregateWorker × 1 ← consume: monitor.aggregate │ │ +│ │ • TaskGenWorker × 1 ← consume: monitor.task.generate │ │ +│ │ • 写 PG Primary / 读 PG Replica / 写 MinIO │ │ +│ └─────────────────────────────────────────────────────────┘ │ +╚═══════════════════════════════════════════════════════════════════════════╝ + + ┌─────────────────────────────────────┐ + │ 浏览器插件 × 30,000（用户端） │ + │ 替代 V3 的 Collector Pod × 3 │ + │ 采集结果通过 HTTP 回传 API │ + │ API 投递到 RabbitMQ 异步处理 │ + └─────────────────────────────────────┘ +``` + +**对比 V3 的变更**： +- 去掉 Collector Pod × 3（插件替代） +- 去掉 Queue Redis Sentinel → 替换为 **RabbitMQ**（功能更强：死信队列、重试、多队列路由） +- 新增 **Monitor Worker** 进程（消费 MQ 消息，执行解析/入库/聚合） +- V3 的 Aggregator 独立进程合并到 Monitor Worker 中（AggregateWorker 消费者） + +### 7.2 资源规格 + +| 组件 | 实例数 | 规格 | 月成本（阿里云） | +| --- | --- | --- | --- | +| **tenant-api** | 2（HPA 2-4） | 4C8G | ¥2,400 | +| **Cache Redis** | 1 主 1 从 | 4G 内存 | ¥1,500 | +| **RabbitMQ** | 1（镜像队列可选主从） | 2C4G | ¥800 | +| **PG Primary** | 1 | 4C16G RDS | ¥2,500 | +| **PG Replica** | 1 | 2C8G 只读 | ¥1,000 | +| **MinIO / OSS** | 1 | 50G | ¥200 | +| **Monitor Worker** | 1 | 2C4G | ¥600 | +| **SLB + CDN** | - | - | ¥500 | +| **合计** | | | **¥9,500/月** | + +### 7.3 连接池修订 + +```yaml +# 每个 API 实例 +database: + max_open_conns: 20 + +# 2 API 实例 → PG 总连接: 40 +# 1 Monitor Worker → PG 连接: 15 +# 合计: 55 连接 +# PostgreSQL: max_connections = 100 (留余量) +``` + +--- + +## 8. 降级与容灾（扩展 V3 第 6 章） + +### 8.1 采集侧降级 + +| 故障场景 | 降级措施 | 用户感知 | +| --- | --- | --- | +| 某 AI 平台改版导致 adapter 失效 | 该平台任务自动跳过，其他平台正常 | 该平台数据缺失，标记"暂不可用" | +| 插件安装率不足 / 用户不在线 | 未完成任务次日继续；Dashboard 展示已采集数据 | 部分平台数据延迟 1-2 天 | +| 采集完成率 < 70% | 告警通知；可选开启少量 API 采集补充 | 无感知（API 降级透明） | +| AI 平台风控封禁用户 | 降低该平台采集频率；通知用户重新登录 | 该平台数据暂停 | +| 全部 AI 平台不可用 | Dashboard 展示最后一次成功聚合数据 | 显示"数据更新时间" | + +### 8.2 API 采集降级通道（可选） + +保留 V3 的 API 采集能力作为降级通道。当插件采集完成率连续 3 天 < 70% 时，自动为该租户开启 API 采集： + +```go +// 降级判断（每日聚合后执行） +func (s *MonitoringService) CheckCollectionCoverage(ctx context.Context, tenantID uuid.UUID) { + rate := s.repo.GetCollectionCompletionRate(ctx, tenantID, 3) // 近 3 天 + if rate < 0.70 { + s.enableAPIFallback(ctx, tenantID) + } +} +``` + +### 8.3 读侧降级（保留 V3 不变） + +V3 第 6 章的 Cache Redis 故障、PG 故障、Aggregator 异常等降级策略完全保留。 + +### 8.4 监控告警 + +| 指标 | 告警阈值 | 说明 | +| --- | --- | --- | +| API P99 延迟 | > 2s 持续 5 分钟 | 同 V3 | +| Cache Redis 命中率 | < 80% | 同 V3 | +| PG 连接池使用率 | > 80% | 同 V3 | +| 日采集完成率 | < 70% | **新增**，按平台分别告警 | +| 单平台 adapter 连续失败 | > 50 次/小时 | **新增**，可能平台改版 | +| 聚合 Job 未在 8:00 前完成 | 超时 | 同 V3 | +| 待采集任务堆积 | > 50,000 pending | **新增**，可能插件采集能力不足 | + +--- + +## 9. 插件方案的风险与反检测 + +### 9.1 风险矩阵 + +| 风险 | 严重性 | 概率 | 对策 | +| --- | --- | --- | --- | +| AI 平台 Web 端改版 | 高 | 中 | adapter 独立维护，改版快速热更新；多平台冗余 | +| AI 平台检测自动化行为 | 高 | 中 | 控制频率 ≤ 3 次/平台/小时；模拟真实行为；随机延迟 | +| 用户未登录目标 AI 平台 | 中 | 高 | 跳过未登录平台；引导登录；多用户互补覆盖 | +| 采集数据质量不一致 | 中 | 中 | 同一问题多次采集取中位；异常过滤 | +| 采集延迟（依赖用户在线） | 中 | 中 | T+1 模型兼容延迟；次日继续 | + +### 9.2 反检测策略 + +```typescript +const ANTI_DETECT_CONFIG = { + // 频率控制 + maxQueriesPerPlatformPerHour: 3, + maxDailyQueries: 50, + + // 时间随机化 + minIntervalSec: 30, + maxIntervalSec: 120, + + // 行为模拟 + typeDelay: { min: 50, max: 150 }, // 逐字输入延迟 ms + postAnswerDelay: { min: 2000, max: 5000 }, // 回答后停留 ms + + // 请求特征 + randomizeUserAgent: false, // 使用浏览器真实 UA + addRandomHeaders: false, // 不添加额外头，避免特征 +}; +``` + +--- + +## 10. 压测验收标准（扩展 V3 第 7 章） + +### 10.1 读侧压测（保留 V3 不变） + +V3 第 7 章的 S1-S5 场景、SLA 指标完全保留。 + +### 10.2 新增：插件采集压测 + +| 场景 | 方法 | 目标 | +| --- | --- | --- | +| S7: 1000 并发插件实例同时回传结果 | 模拟 1000 个 POST /api/callback/plugin/monitor 并发 | P99 < 500ms，0 丢失 | +| S8: 任务领取竞争 | 500 个并发 claim 同一任务 | 只有 1 个成功，其余 409 | +| S9: 高频任务生成 | 生成 100K 采集任务 | 生成耗时 < 30s | + +### 10.3 数据质量验证 + +| 验证项 | 方法 | 目标 | +| --- | --- | --- | +| 回答完整性 | 同一问题 3 次采集，对比回答长度 | 长度标准差 < 30% | +| 品牌提及准确率 | 人工标注 100 条回答，对比解析结果 | 准确率 > 90% | +| 引用提取率 | 有引用的平台，对比插件提取 vs 手动检查 | 召回率 > 85% | + +--- + +## 11. 实施计划 + +### 11.1 Phase 分期 + +| Phase | 内容 | 估时 | 依赖 | +| --- | --- | --- | --- | +| A | 插件 AI 适配器框架 + 首批 3 平台（DeepSeek/千问/豆包） | 8 天 | 无 | +| B | 后端：Migration + 配额 + RabbitMQ 基础设施 + 采集回调 API + 插件认证 API | 8 天 | A（接口定义） | +| C | 后端：Monitor Worker（ParseWorker + AggregateWorker）+ 缓存层 + 查询 API | 8 天 | B | +| D | 前端：6 个数据追踪页面 + 采集控制面板 | 7 天 | C（API 就绪） | +| E | 第二批 3 个 AI 适配器（Kimi/文心/混元）+ 联调 | 5 天 | A + D | +| F | 压测 + 数据质量验证 + 验收 | 5 天 | E | +| **合计** | | **41 天** | | + +Phase B 细分： +- RabbitMQ 客户端封装 + Exchange/Queue 声明（1 天） +- Migration（配额表 + 任务表 + AI 平台表 + Schema Delta）（2 天） +- 插件认证 API（`/api/plugin/monitoring/tasks`、`/claim`）（2 天） +- 采集回调 API（接收 → 投递 MQ）（1 天） +- TaskGenerateWorker（凌晨批量生成任务）（2 天） + +Phase C 细分： +- ParseWorker（MQ 消费 → 解析 → 入库 → MinIO）（3 天） +- AggregateWorker（品牌级增量聚合）（2 天） +- 缓存层（singleflight + stale-while-revalidate + L1）（1 天） +- 查询 API（6 个页面 + composite）（2 天） + +双人并行方案（前后端各 1 人）： + +``` +Week 1-2: 后端 Phase B (MQ + Migration + API) | 插件 Phase A (适配器) +Week 3-4: 后端 Phase C (Worker + 缓存 + 查询) | 前端 Phase D (页面) +Week 5: 联调 Phase E | 第二批适配器 +Week 6: 压测 Phase F +总计: 28 天 +``` + +### 11.2 关键里程碑 + +| 时间点 | 里程碑 | 验收标准 | +| --- | --- | --- | +| Day 8 | 首个 AI 适配器可用 | DeepSeek 登录检测 + 提问 + 获取回答 E2E 通过 | +| Day 10 | RabbitMQ 基础设施就绪 | Exchange/Queue 声明完成，Publish/Consume 冒烟测试通过 | +| Day 16 | 后端采集链路通 | 插件回传 → MQ → ParseWorker → PG 入库全链路通过 | +| Day 24 | 查询 API + 增量聚合可用 | 品牌任务完成后分钟级数据可见 | +| Day 31 | 前端页面完成 | 6 个页面 + 采集面板可交互 | +| Day 36 | 全部 6 平台适配器可用 | 全平台检测 + 采集 E2E | +| Day 41 | 验收通过 | 读侧压测 SLA + MQ 吞吐压测 + 数据质量验证 | + +--- + +## 12. V3 → V4 变更汇总 + +| 变更编号 | V3 方案 | V4 方案 | 原因 | +| --- | --- | --- | --- | +| **C1** | 服务端 AI API 采集（30 worker × 3 Pod） | 浏览器插件采集（30,000 用户端节点） | 消除 AI API 成本 | +| **C2** | Queue Redis Stream 任务队列 | RabbitMQ 异步处理（解析/入库/聚合） + PG 任务表 + Redis 分布式锁 | 插件 HTTP 回传 → MQ 异步解耦，API 不阻塞 | +| **C3** | 100 品牌 × 100 问题 | 配额限制（Free 1 品牌 / Pro 3 品牌，每品牌 40 问题） | 控制采集规模 | +| **C4** | 全品牌每日采集 | Free 每 3 天 / Pro 每天 | 降低采集压力 | +| **C5** | `run_mode`: api_standard / api_search_grounded | `run_mode`: plugin_standard / plugin_search | 采集来源变更 | +| **C6** | 月成本 ~¥42K（含 API ¥22K） | 月成本 ~¥9.5K（无 API 费用） | 成本优化 77% | +| **C7** | 无配额机制 | `tenant_monitoring_quotas` 租户级配额 | 支撑多租户规模化 | +| **C8** | 3 个 Collector Pod + Queue Redis | Collector → 插件替代；Queue Redis → RabbitMQ + Monitor Worker | 采集去中心化，处理异步化 | +| **C9** | 数据一致性高（固定 prompt/温度） | 数据更接近真实用户（Web 端采集） | 采集可行性方案确认 API ≠ Web | +| **新增** | 无 | 反检测策略 | 插件采集必须的安全措施 | +| **新增** | 无 | 采集控制面板（前端） | 用户管理采集进度和平台登录 | +| **新增** | 无 | API 降级通道 | 插件采集不足时的兜底方案 | +| **新增** | 无 | RabbitMQ 异步处理架构 | 解析/入库/聚合与 API 层解耦，插件回调 ~10ms 返回 | +| **新增** | 每日 6:30 全量定时聚合 | 品牌级增量聚合（MQ 触发） | 数据可见延迟从 T+1 降至准实时（分钟级） | +| **新增** | 无 | Monitor Worker 独立进程 | ParseWorker × 3 + AggregateWorker + TaskGenWorker | +| **新增** | 无 | 死信队列 (DLQ) | 处理失败的消息暂存，不丢数据 | diff --git a/docs/ai-brand-monitoring-tech-design-v5.md b/docs/ai-brand-monitoring-tech-design-v5.md new file mode 100644 index 0000000..799a3af --- /dev/null +++ b/docs/ai-brand-monitoring-tech-design-v5.md @@ -0,0 +1,960 @@ +# AI 品牌曝光监测系统技术方案 V5 + +## 1. 文档信息 + +| 项目 | 内容 | +| --- | --- | +| 文档名称 | AI 品牌曝光监测系统技术方案 V5（采样型趋势监测 + 异步解耦） | +| 文档版本 | V5.1 | +| 文档状态 | 待评审 | +| 创建日期 | 2026-04-08 | +| 基线文档 | `docs/ai-brand-monitoring-tech-design-v4.md` | +| 适用范围 | 数据追踪模块全部页面 | +| 关联文档 | `docs/geo-platform-prd-v1.md`（PRD 8.4 数据追踪） | +| 关联文档 | `docs/question-driven-monitoring-design-v1.md`（问题驱动模型） | +| 核心约束 | 单租户仅 1 个监测插件实例，插件所在设备不保证全天在线 | + +## 2. V5 核心结论 + +V5 不再把该能力定义为“全量、精确、准实时监控”，而是明确收敛为： + +1. **采样型趋势监测系统**，不是审计系统。 +2. **单实例插件 + 非全天在线设备**前提下，只承诺趋势参考，不承诺全量覆盖。 +3. **页面百分比全部按实际采样成功样本计算**，不再暗示“全量真实占比”。 +4. **页面必须展示采样覆盖信息**，包括计划采样数、实际采样数、覆盖率、最近更新时间。 +5. **引入 RabbitMQ 和独立 Monitoring PostgreSQL 隔离写链路**，但聚合仍以日批为主，不承诺准实时展示。 + +### 2.1 相比 V4 的根本变化 + +| 维度 | V4 | V5 | +| --- | --- | --- | +| 产品定位 | 全量品牌曝光监测 | 采样型趋势参考 | +| 采集目标 | 尽量覆盖全部问题 × 全平台 | 每日仅覆盖活跃监测题池 | +| 插件节点假设 | 大量用户端节点可互补 | 单租户单主插件实例 | +| 数据时效 | 准实时 + T+1 | T+1 为主，允许延迟 | +| 数据精度承诺 | 接近真实全量 | 趋势可参考，精度受覆盖率影响 | +| 任务处理 | PG + Redis 锁 + RabbitMQ | Monitoring PG 任务表 + PG 租约状态机 + RabbitMQ 结果异步化 | +| 存储边界 | 单业务库承载配置与监测结果 | 主业务 PG 存配置，Monitoring PG 存监测数据 | +| 页面表达 | 指标卡直接展示百分比 | 指标卡 + 采样覆盖信息 + 置信度提示 | + +### 2.2 非目标 + +V5 明确不做以下承诺： + +1. 不承诺所有问题每天都有结果。 +2. 不承诺所有 AI 平台每天都有样本。 +3. 不承诺分钟级实时更新。 +4. 不承诺百分比可用于财务、法务、投放结算或精确 KPI 审计。 + +--- + +## 3. 产品口径与页面表达 + +### 3.1 页面定位 + +数据追踪页展示的是： + +- 在“当前监测题池”上的 AI 曝光趋势 +- 在“实际采样成功样本”上的品牌表现 +- 在“可归因引用样本”上的文章引用趋势 + +不是： + +- 品牌在全网 AI 场景中的绝对真实份额 +- 所有问题、所有平台、所有时间点的完整监控 + +### 3.2 页面必须新增的展示项 + +为避免误导，数据追踪页和 Dashboard composite 接口必须返回并展示： + +1. `planned_sample_count`：计划采样数 +2. `actual_sample_count`：实际采样成功数 +3. `coverage_rate`：覆盖率 = `actual / planned` +4. `last_sampled_at`：最近成功采样时间 +5. `confidence_level`：`high / medium / low` + +推荐前端呈现方式： + +- 顶部右侧信息条：`采样模式` +- 趋势图上方副文案：`已采样 18/30，覆盖率 60%，更新于 04-08 09:20` +- 覆盖率过低时展示 `样本不足，仅供参考` + +### 3.3 低覆盖场景的展示规则 + +| 覆盖率 | 置信度 | 页面表现 | +| --- | --- | --- | +| >= 70% | high | 正常展示百分比和趋势 | +| 40% - 69% | medium | 展示百分比，并提示“样本偏少” | +| < 40% | low | 保留原始计数，百分比弱化展示或标记“仅供参考” | + +当 `actual_sample_count < 5` 时： + +1. 不展示“精确到整数”的百分比变化解释。 +2. 趋势线可以显示，但需打低置信度标记。 +3. 引用排行允许为空，不用补零。 + +--- + +## 4. 采样配额与容量模型 + +### 4.1 套餐定义 + +V5 中配额控制的不是“题库总量”，而是“每日活跃监测问题数”。 + +| 配额项 | Free | Pro | Enterprise | +| --- | --- | --- | --- | +| 最大品牌数 | 1 | 3 | 自定义 | +| 题库总量 | 不限于监测能力，只做配置资产 | 不限于监测能力，只做配置资产 | 自定义 | +| **每日活跃监测问题数** | **3-5** | **5-10** | 自定义 | +| 采集频率 | 每天 | 每天 | 每天 / 自定义 | +| 平台范围 | 当前已接入且已登录的平台 | 当前已接入且已登录的平台 | 同左 | +| 数据承诺 | 趋势参考 | 趋势参考 | 趋势参考 / 专属增强 | + +### 4.2 活跃监测题池 + +系统将问题分成两层： + +1. **问题题库**：品牌下维护的全部问题，仅做配置和运营资产。 +2. **活跃监测题池**：每日实际参与采样的问题集合，数量受套餐限制。 + +推荐实现： + +- `brand_questions` 保留全部题库 +- 新增 `monitor_priority`、`monitor_enabled` +- 调度器每日只选 Top N 问题进入任务生成 + +### 4.3 单实例前提下的目标任务量 + +按首批 6 个平台测算： + +| 套餐 | 活跃问题数 | 平台数 | 目标计划任务/天 | +| --- | --- | --- | --- | +| Free | 3-5 | 6 | 18-30 | +| Pro | 5-10 | 6 | 30-60 | + +说明： + +1. 这是 **计划采样量**，不是保证完成量。 +2. 实际完成量取决于： + - 主监测插件是否在线 + - 插件机器当日在线时长 + - 对应 AI 平台是否已登录 + - 平台风控和采集成功率 + +### 4.4 核心调度原则 + +V5 的调度器不追求“补齐所有漏采任务”，而追求“稳定输出可比较样本”。 + +因此： + +1. 每天只生成当天计划任务，不补历史大积压。 +2. 超过 48 小时未完成的任务自动过期，不回补到新一天的分母中。 +3. 读侧按 `planned_sample_count` 和 `actual_sample_count` 共同解释结果。 + +### 4.5 主插件在线约束 + +单租户只指定 1 个 `primary_monitoring_installation` 参与自动监测。 + +调度前先判断该插件是否活跃： + +- `last_seen_at <= 24h`：生成当天任务 +- `last_seen_at > 24h`：不生成新任务，页面沿用最近一次成功数据并提示“近期无新采样” + +这条规则的目的： + +1. 防止设备长期离线时任务无限堆积 +2. 防止“未执行任务”污染覆盖率统计 +3. 让系统行为和单实例现实约束一致 + +--- + +## 5. 数据口径定义 + +### 5.1 顶部指标口径 + +所有百分比的分母统一为 **实际采样成功回答数**，不是计划任务数，也不是题库总数。 + +| 指标 | 定义 | +| --- | --- | +| 提及率 | `mentioned_count / actual_sample_count` | +| 首位提及率 | `top1_mentioned_count / actual_sample_count` | +| 首选推荐率 | `first_recommended_count / actual_sample_count` | +| 正面提及率 | `positive_mentioned_count / actual_sample_count` | +| 品牌可见率 | 与提及率同源，可作为 UI 别名展示 | +| 文章引用率 | `cited_answer_count / actual_sample_count` | + +### 5.2 趋势图口径 + +趋势图展示的是按天聚合后的样本表现，不是平台真实总体份额。 + +每个点必须同时具备： + +- `metric_value` +- `actual_sample_count` +- `coverage_rate` + +可选增强： + +- 默认展示 7 天滚动均值 +- 用户切换后可看原始日值 + +### 5.3 高频问题口径 + +高频问题区不再解释为“全站高频触发问题”，而是： + +> 在当前时间窗口内，被实际采样成功次数最多的问题 + +推荐展示字段： + +- 问题文案 +- 成功采样次数 +- 提及率 +- 最近采样时间 + +### 5.4 引用排行口径 + +引用排行按模型平台展示： + +- 有过可归因引用的回答数 +- 引用文章数 +- 引用率 + +若平台当日无样本或平台未登录： + +- 展示为 `未采样` +- 不强行按 `0` 并入分母 + +--- + +## 6. 数据模型修订 + +### 6.1 双 PostgreSQL 存储边界 + +V5.1 将数据按“业务配置”和“监测结果”拆到两个 PostgreSQL： + +**主业务 PostgreSQL** + +- `tenants` +- `brands` +- `brand_keywords` +- `brand_questions` +- `articles` +- `publish_records` +- `plugin_installations` +- `tenant_monitoring_quotas` + +**Monitoring PostgreSQL** + +- `monitoring_collect_tasks` +- `question_monitor_runs` +- `question_monitor_parse_results` +- `monitoring_citation_facts` +- `monitoring_article_url_aliases` +- `monitoring_*_daily` + +设计原则： + +1. 不做跨库物理外键，只保留逻辑 ID 引用。 +2. 监测链路的高频写入、重试、归档不影响主业务 PG。 +3. 读侧监测查询优先从 Monitoring PG 获取。 + +### 6.2 `question_version_id` 移除后的替代方案 + +V5 接受原系统不再维护 `question_version_id`，但仍然保留“输入快照不可变”原则。 + +替代方案： + +1. 在任务表和运行表写入 `question_text_snapshot` +2. 同时写入 `question_hash` +3. 历史汇总按 `question_hash` 区分口径 + +这样即使 `question_id` 不变、文案发生修改，历史数据也不会被新文案污染。 + +### 6.3 `tenant_monitoring_quotas` + +```sql +CREATE TABLE tenant_monitoring_quotas ( + tenant_id BIGINT PRIMARY KEY REFERENCES tenants(id), + max_brands INT NOT NULL DEFAULT 1, + daily_active_question_limit INT NOT NULL DEFAULT 5, + collect_frequency VARCHAR(20) NOT NULL DEFAULT 'daily', + enabled_platforms JSONB NOT NULL DEFAULT '["deepseek","qwen","doubao"]', + primary_installation_id BIGINT REFERENCES plugin_installations(id), + task_daily_hard_cap INT NOT NULL DEFAULT 36, + plan_tier VARCHAR(20) NOT NULL DEFAULT 'free', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +``` + +建议默认值： + +- Free：`daily_active_question_limit = 5`，`task_daily_hard_cap = 36` +- Pro：`daily_active_question_limit = 10`，`task_daily_hard_cap = 72` + +说明： + +1. `tenant_monitoring_quotas` 放在主业务 PostgreSQL。 +2. `primary_installation_id` 继续引用主业务库中的 `plugin_installations`。 + +### 6.4 `monitoring_collect_tasks` + +V5.1 使用 Monitoring PG 单表状态机进行任务租约，并用 RabbitMQ 解耦结果写入。 + +```sql +CREATE TABLE monitoring_collect_tasks ( + id BIGSERIAL PRIMARY KEY, + tenant_id BIGINT NOT NULL, + brand_id BIGINT NOT NULL, + question_id BIGINT NOT NULL, + question_text_snapshot TEXT NOT NULL, + question_hash VARCHAR(64) NOT NULL, + ai_platform_id VARCHAR(30) NOT NULL, + run_mode VARCHAR(30) NOT NULL DEFAULT 'plugin_standard', + business_date DATE NOT NULL, + planned_at TIMESTAMPTZ NOT NULL DEFAULT now(), + status VARCHAR(20) NOT NULL DEFAULT 'pending', + lease_token_hash VARCHAR(128), + leased_to_installation BIGINT, + leased_at TIMESTAMPTZ, + lease_expires_at TIMESTAMPTZ, + callback_received_at TIMESTAMPTZ, + completed_at TIMESTAMPTZ, + skip_reason VARCHAR(50), + error_message TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + + CONSTRAINT uk_collect_task_idempotent + UNIQUE (tenant_id, brand_id, question_id, question_hash, ai_platform_id, run_mode, business_date) +); + +CREATE INDEX idx_collect_tasks_lease + ON monitoring_collect_tasks(tenant_id, status, business_date, lease_expires_at); +``` + +状态流转： + +- `pending` +- `leased` +- `received` +- `completed` +- `skipped` +- `expired` +- `failed` + +说明： + +1. 任务租约状态在 Monitoring PG 中维护。 +2. `received` 表示 API 已接收插件结果并成功投递 MQ，但 worker 尚未完成最终入库。 + +### 6.5 `question_monitor_runs` + +在 V5 中至少补充： + +```sql +ALTER TABLE question_monitor_runs + ADD COLUMN IF NOT EXISTS tenant_id BIGINT, + ADD COLUMN IF NOT EXISTS business_date DATE, + ADD COLUMN IF NOT EXISTS run_mode VARCHAR(30), + ADD COLUMN IF NOT EXISTS question_text_snapshot TEXT, + ADD COLUMN IF NOT EXISTS question_hash VARCHAR(64), + ADD COLUMN IF NOT EXISTS provider_model VARCHAR(100), + ADD COLUMN IF NOT EXISTS request_id VARCHAR(100), + ADD COLUMN IF NOT EXISTS raw_answer_storage_key TEXT; +``` + +建议幂等键： + +```sql +CREATE UNIQUE INDEX uk_monitor_run_idempotent +ON question_monitor_runs(tenant_id, brand_id, question_id, question_hash, ai_platform_id, run_mode, business_date); +``` + +建议： + +1. `question_monitor_runs` 落在 Monitoring PostgreSQL。 +2. 若原始回答文本不超过阈值，可直接存库；超大原文再转对象存储。 + +### 6.6 文章 URL 归因表 + +为支撑“总引用数 / 文章引用率 / 引用文章数”，新增 URL 归因别名表： + +```sql +CREATE TABLE monitoring_article_url_aliases ( + id BIGSERIAL PRIMARY KEY, + tenant_id BIGINT NOT NULL, + article_id BIGINT NOT NULL, + publish_record_id BIGINT, + platform_id VARCHAR(64), + external_article_id VARCHAR(128), + original_url TEXT NOT NULL, + normalized_url TEXT NOT NULL, + last_path_segment VARCHAR(255), + confidence VARCHAR(20) NOT NULL DEFAULT 'high', + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE UNIQUE INDEX uk_article_url_alias +ON monitoring_article_url_aliases(tenant_id, normalized_url); +``` + +--- + +## 7. 解析标准化与 SoV 统计模型 + +### 7.1 解析分层 + +V5.1 的解析分为 3 层： + +1. **回答层**：品牌提及、首位提及、首选推荐、情感倾向 +2. **引用层**：从回答、citation JSON、search_results 中提取 URL 事实 +3. **来源层**：基于域名清洗和站点聚合生成 SoV 统计结果 + +### 7.2 基于 `tldextract` 的顶级域名清洗 + +为避免 `www`、二级域名、国家后缀等问题导致同一站点重复计数，引用归因与来源排行统一采用 **Public Suffix List + `tldextract`** 口径。 + +处理目标： + +1. 从原始 URL 中提取 `subdomain / domain / suffix` +2. 生成 `registrable_domain` +3. 生成统一站点主键 `site_key` +4. 为页面级和站点级聚合提供稳定维度 + +示例： + +| 原始 URL | `registrable_domain` | 说明 | +| --- | --- | --- | +| `https://www.zhihu.com/question/123` | `zhihu.com` | `www` 折叠 | +| `https://zhuanlan.zhihu.com/p/123` | `zhihu.com` | 子域归并到站点 | +| `https://mp.weixin.qq.com/s/abc` | `weixin.qq.com` | 基于 PSL 提取可注册域 | +| `https://finance.sina.com.cn/roll/123.html` | `sina.com.cn` | 正确处理 `com.cn` | +| `https://sub.example.co.uk/a` | `example.co.uk` | 正确处理 `co.uk` | + +推荐清洗流程： + +1. URL 小写化主机名 +2. 去掉默认端口 +3. 去掉 fragment +4. path 做基础规范化 +5. host 通过 `tldextract` 计算 `registrable_domain` +6. 保存： + - `normalized_url` + - `host` + - `registrable_domain` + - `subdomain` + - `suffix` + +### 7.3 `monitoring_citation_facts` 推荐字段 + +为支撑域名清洗和来源排行，建议引用事实表至少包含： + +```sql +CREATE TABLE monitoring_citation_facts ( + id BIGSERIAL PRIMARY KEY, + tenant_id BIGINT NOT NULL, + run_id BIGINT NOT NULL, + brand_id BIGINT NOT NULL, + ai_platform_id VARCHAR(30) NOT NULL, + business_date DATE NOT NULL, + cited_url TEXT NOT NULL, + normalized_url TEXT NOT NULL, + host VARCHAR(255) NOT NULL, + registrable_domain VARCHAR(255) NOT NULL, + subdomain VARCHAR(255), + suffix VARCHAR(50), + site_key VARCHAR(255) NOT NULL, + article_id BIGINT, + publish_record_id BIGINT, + resolution_status VARCHAR(20) NOT NULL DEFAULT 'resolved', + resolution_confidence VARCHAR(20) NOT NULL DEFAULT 'high', + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX idx_citation_facts_domain +ON monitoring_citation_facts(tenant_id, business_date, registrable_domain); +``` + +约定： + +1. `normalized_url` 用于页面级去重和文章归因 +2. `registrable_domain` 用于站点级来源排行 +3. `site_key` 默认等于 `registrable_domain` +4. 如业务需要区分站群，可对特定域名做覆写映射 + +### 7.4 `site_key` 映射规则 + +默认规则： + +- `site_key = registrable_domain` + +可选增强规则： + +1. 将 `mp.weixin.qq.com`、`weixin.qq.com` 统一到 `weixin.qq.com` +2. 将业务上需要区分的站群写入 `site_domain_mapping` +3. 对同集团但不同产品是否合并，必须通过配置显式决定，不靠临时规则猜测 + +推荐新增配置表： + +```sql +CREATE TABLE site_domain_mappings ( + id BIGSERIAL PRIMARY KEY, + registrable_domain VARCHAR(255) NOT NULL, + site_key VARCHAR(255) NOT NULL, + site_name VARCHAR(255) NOT NULL, + is_active BOOLEAN NOT NULL DEFAULT true, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +``` + +### 7.5 SoV 统计模型 + +V5.1 中的 SoV 统一解释为： + +> 在当前采样窗口内，某品牌、某站点或某文章在全部有效样本中的相对占比 + +必须明确分母，禁止不同口径混用。 + +### 7.6 品牌 SoV + +品牌 SoV 适用于品牌曝光类页面。 + +定义： + +```text +brand_sov = brand_mentioned_answers / actual_sample_count +``` + +其中： + +- `brand_mentioned_answers`：成功采样回答中，提及目标品牌的回答数 +- `actual_sample_count`：成功采样回答总数 + +这和页面上的“提及率”同源，可视为品牌在样本回答中的 `Share of Voice`。 + +### 7.7 引用来源 SoV + +引用来源 SoV 适用于“引用来源排行”“来源平台占比”。 + +定义： + +```text +source_sov = source_citation_count / total_resolved_citation_count +``` + +其中： + +- `source_citation_count`：某 `site_key` 在窗口内被引用的次数 +- `total_resolved_citation_count`：全部已成功解析并完成站点归类的引用次数 + +补充口径： + +- unresolved 引用不进入分子 +- unresolved 是否进入分母，要固定为“不进入” +- unresolved 单独展示占比，作为质量指标 + +### 7.8 文章 SoV + +文章 SoV 用于内部文章在可归因引用中的占比。 + +定义： + +```text +article_sov = article_citation_count / total_article_resolved_citation_count +``` + +其中： + +- `article_citation_count`：某篇内部文章被 AI 引用的次数 +- `total_article_resolved_citation_count`：全部成功归因到内部文章的引用次数 + +注意： + +1. 文章 SoV 不是品牌 SoV +2. 文章 SoV 的分母不能拿全部回答数 +3. 若用户看的是“文章引用率”，仍然应使用 `cited_answer_count / actual_sample_count` + +### 7.9 覆盖率与 SoV 的关系 + +覆盖率和 SoV 必须分开解释： + +1. `coverage_rate` 反映“今天采样够不够” +2. `SoV` 反映“在已采样样本里占比多少” + +因此页面解释顺序应为： + +1. 先看覆盖率 +2. 再看 SoV / 提及率 / 引用率 + +当 `coverage_rate < 40%` 时： + +- 允许展示 SoV +- 但必须附带低置信度提示 + +### 7.10 聚合输出建议 + +建议在 `monitoring_citation_source_daily` 和 `monitoring_citation_page_daily` 中增加： + +1. `citation_count` +2. `citation_sov` +3. `resolved_count` +4. `unresolved_count` +5. `confidence_level` + +这样前端可以同时展示： + +- 引用次数 +- SoV +- 数据质量 + +--- + +## 8. 任务协议与插件侧流程 + +### 8.1 统一为租约式协议 + +V5 删除 `GET tasks -> POST claim` 两步式协议，统一改为： + +1. 插件请求租约 +2. 服务端原子分配任务并返回 `lease_token` +3. 插件执行采集 +4. 插件带 `lease_token` 回传结果 +5. 服务端校验归属、先写任务状态为 `received` 并投递 RabbitMQ +6. Monitor Worker 异步写入运行结果、解析结果并完成任务 + +### 8.2 插件 API + +```text +POST /api/plugin/monitoring/tasks/lease +POST /api/plugin/monitoring/tasks/{id}/result +POST /api/plugin/monitoring/tasks/{id}/skip +``` + +请求头： + +- `X-Geo-Installation-Token` +- `X-Geo-Installation-Id` + +### 8.3 租约接口行为 + +租约接口建议按 PG 原子领取： + +```sql +SELECT id +FROM monitoring_collect_tasks +WHERE tenant_id = $1 + AND status = 'pending' + AND business_date = $2 +ORDER BY planned_at ASC, id ASC +FOR UPDATE SKIP LOCKED +LIMIT $3; +``` + +服务端完成以下动作： + +1. 选出任务 +2. 生成短期 `lease_token` +3. 更新 `status = leased` +4. 写入 `lease_expires_at = now() + interval '15 minutes'` + +### 8.4 插件执行策略 + +V5 继续复用现有插件基础设施： + +- [background.ts](/Users/liangxu/Documents/test/geo-rankly/apps/browser-extension/entrypoints/background.ts) +- [runtime.ts](/Users/liangxu/Documents/test/geo-rankly/apps/browser-extension/src/runtime.ts) +- [storage.ts](/Users/liangxu/Documents/test/geo-rankly/apps/browser-extension/src/storage.ts) + +但采样策略改为“小批量、固定预算”： + +| 项目 | 建议值 | +| --- | --- | +| 心跳频率 | 15 分钟 | +| 单次租约任务数 | 2-3 | +| 单任务超时 | 60 秒 | +| 租约时长 | 15 分钟 | +| 每日硬上限 | 按套餐 36 / 72 | + +### 8.5 任务过期与回收 + +定时任务每 30 分钟处理一次： + +1. `lease_expires_at < now()` 且状态仍为 `leased` 的任务改为 `expired` +2. `received` 状态超过阈值未被 worker 完成时，进入异常告警 +3. 同一 `business_date` 的 `expired` 任务最多重试 1 次 +4. 超过 `business_date + 2 days` 的未完成任务不再重试 + +--- + +## 9. 采集、解析、聚合架构 + +### 9.1 首期简化版链路 + +```text +每日 00:10 调度器生成当天任务 + ↓ +插件后台 15 分钟心跳租约任务 + ↓ +插件执行提问并回传结果 + ↓ +tenant-api：校验租约 -> 标记 received -> 投递 RabbitMQ + ↓ +Monitor Worker：消费结果 -> 写 Monitoring PG -> 解析结果 -> 完成任务 + ↓ +每日 06:30 聚合 Job 汇总到 monitoring_*_daily + ↓ +读侧 API + Redis 缓存 +``` + +### 9.2 RabbitMQ 的职责边界 + +V5.1 中 RabbitMQ 只承担“结果异步化”，不承担“任务分发”。 + +使用 RabbitMQ 的原因： + +1. 插件回调接口可以快速返回，降低超时和重试风险 +2. 解析、引用归因、写库失败不会直接影响插件侧体验 +3. 方便把高频写入隔离到独立 worker +4. 后续若企业版放大量级，不需要重做写链路 + +但 RabbitMQ 不参与以下事项： + +1. 不做任务租约 +2. 不做读侧查询 +3. 不强制引入准实时前端刷新 + +推荐队列： + +- `monitor.result.ingest` +- `monitor.result.dlq` +- `monitor.aggregate.trigger` + +### 9.3 Monitoring PostgreSQL 的职责 + +独立 Monitoring PG 的目的不是“多一套库”，而是隔离监测链路的写放大。 + +承载内容： + +1. 任务表 +2. 原始回答 +3. 解析结果 +4. 引用事实 +5. 日聚合结果 + +收益： + +1. 主业务 PG 不承受高频插入和大文本更新 +2. Monitoring PG 可独立做分区、归档、VACUUM 策略 +3. 监测表膨胀不会影响文章、品牌、发布主链路 + +### 9.4 原始回答存储 + +V5.1 中抓取结果优先进入 Monitoring PG： + +1. 原始回答文本 +2. 平台响应元数据 +3. 引用与搜索结果原始 JSON +4. 解析后的结构化字段 + +可选归档策略： + +1. 小于 64 KB 的原文直接存 Monitoring PG +2. 超大原文或完整调试负载转存对象存储 +3. Monitoring PG 仅保留摘要、关键字段和存储路径 + +--- + +## 10. 文章引用归因策略 + +### 10.1 为什么不能只看 URL 最后一段 + +仅按最后一段匹配存在以下问题： + +1. 不同平台可能出现相同 path segment +2. 同一平台 PC / H5 / 分享页 URL 可能不同 +3. AI 引用结果可能带参数、重定向或 canonical URL +4. 同一篇文章可能存在多个可访问别名 + +因此 V5 的归因策略是 **分层匹配 + 置信度标记**。 + +### 10.2 归因顺序 + +1. `normalized_url` 精确匹配 +2. `(platform_id, external_article_id)` 精确匹配 +3. `(domain, last_path_segment)` 唯一匹配 +4. 无法唯一命中时，记为 `unresolved` + +### 10.3 低置信度处理 + +当仅通过 `last_path_segment` 命中时： + +1. 给归因结果打 `low_confidence` +2. 可以进入趋势参考视图 +3. 不建议进入需要精确计费或精确归属的报表 + +这与 V5 的“趋势参考”定位是一致的。 + +--- + +## 11. 平台接入范围 + +### 11.1 V5 交付范围 + +数据模型支持 6 平台扩展，但首期交付只承诺首批 3 平台： + +1. DeepSeek +2. 通义千问 +3. 豆包 + +### 11.2 页面与套餐表达 + +页面和套餐都必须按“当前已接入且已登录的平台”解释： + +- 已接入且已登录：进入采样计划 +- 已接入未登录：展示 `未登录` +- 未接入：展示 `未接入` + +不得再用“全 6 平台稳定可用”作为首期对外承诺。 + +--- + +## 12. 读侧与汇总表 + +V5 读侧继续沿用 V3 的聚合表和缓存思路，但口径增加采样字段： + +1. `planned_sample_count` +2. `actual_sample_count` +3. `coverage_rate` +4. `confidence_level` + +推荐所有 `monitoring_*_daily` 汇总表统一增加以上字段。 + +### 12.1 Dashboard composite + +`/api/tenant/monitoring/dashboard/composite` 继续保留，但返回结构需扩展： + +```json +{ + "overview": { + "mention_rate": 0.42, + "citation_rate": 0.18, + "planned_sample_count": 30, + "actual_sample_count": 21, + "coverage_rate": 0.70, + "confidence_level": "high", + "last_sampled_at": "2026-04-08T09:20:00Z" + } +} +``` + +### 12.2 缓存策略 + +读侧缓存策略可继续沿用 V3： + +- composite 聚合缓存 +- singleflight +- TTL 抖动 +- stale-while-revalidate + +但缓存失效触发点改为： + +1. 每日聚合完成 +2. 用户切换品牌 / 时间窗口 + +不需要为单条采样结果做准实时逐条失效。 + +--- + +## 13. 降级、告警与风险 + +### 13.1 降级策略 + +| 场景 | V5 行为 | +| --- | --- | +| 插件 24h 未活跃 | 暂停生成新任务，页面展示上次数据 | +| 平台未登录 | 该平台不进入当天计划样本 | +| 当日覆盖率低 | 页面提示“样本不足，仅供参考” | +| 某平台 adapter 失效 | 该平台展示“暂不可用”，其余平台继续 | +| URL 无法归因 | 计入 unresolved，不强行挂到文章 | + +### 13.2 关键告警 + +| 指标 | 阈值 | +| --- | --- | +| 主插件离线时长 | > 24h | +| 日覆盖率 | < 40% | +| 单平台连续失败 | > 20 次 / 天 | +| unresolved 引用占比 | > 30% | +| 每日聚合延迟 | > 08:00 未完成 | + +### 13.3 风险接受说明 + +V5 明确接受以下风险： + +1. 单日样本偏少导致波动大 +2. 插件离线导致当天无新数据 +3. 部分 URL 只能低置信度归因 + +但不接受以下风险： + +1. 任务重复执行导致统计翻倍 +2. 输入口径变更污染历史数据 +3. 未授权回调写入虚假监测结果 + +--- + +## 14. 实施计划 + +### 14.1 Phase 分期 + +| Phase | 内容 | 估时 | +| --- | --- | --- | +| A | 双库方案落地：主业务 PG / Monitoring PG schema 与连接配置 | 4 天 | +| B | RabbitMQ 基础设施 + Monitor Worker 骨架 + DLQ | 4 天 | +| C | 插件：AI adapter 框架 + 3 平台 + 租约式任务执行 | 6 天 | +| D | 后端：租约 API、结果回调、MQ 消费、解析入库、日聚合 | 7 天 | +| E | 前端：数据追踪页补覆盖率、更新时间、置信度提示 | 4 天 | +| F | 联调、坏样本验证、口径验收 | 4 天 | +| **合计** | | **29 天** | + +### 14.2 验收标准 + +1. 页面可展示 `planned_sample_count / actual_sample_count / coverage_rate` +2. 页面可正常展示趋势图、高频问题、引用排行 +3. 当覆盖率不足时，页面有明确提示 +4. 同一任务不会被重复计入统计 +5. 问题文案修改后，历史趋势不被污染 +6. 引用文章数可通过 URL alias 规则稳定归因，无法归因时进入 unresolved + +--- + +## 15. V4 → V5 变更摘要 + +| 编号 | V4 | V5 | +| --- | --- | --- | +| C1 | 全量监测叙事 | 采样趋势叙事 | +| C2 | Free 1 品牌 / 40 问题 | Free 3-5 个每日活跃问题 | +| C3 | Pro 3 品牌 / 40 问题 | Pro 5-10 个每日活跃问题 | +| C4 | GET + claim 两步任务协议 | 一步租约式任务协议 | +| C5 | `question_version_id` | `question_hash + question_text_snapshot` | +| C6 | RabbitMQ + 增量聚合 | RabbitMQ 仅做结果异步化，不参与任务分发或前端准实时 | +| C7 | 准实时可见 | T+1 / 延迟可接受 | +| C8 | 仅展示百分比 | 百分比 + 采样覆盖信息 | +| C9 | 首期 6 平台叙事 | 首期只承诺 3 平台 | +| C10 | 单一业务库承载全部数据 | 主业务 PG + Monitoring PG 双库隔离 | + +## 16. 最终建议 + +V5 适合当前阶段的原因不是“它最强”，而是“它和当前约束一致”： + +1. 和单实例插件现实相符 +2. 和 PRD 的 V1 边界相符 +3. 和用户对“趋势参考”的接受度相符 + +如果未来要升级到 V6，再考虑： + +1. 专用在线采集节点 +2. 企业版更高采样量 +3. 更细粒度的增量聚合和近实时刷新 +4. 更强的 URL 归因与多次重复采样校准 diff --git a/packages/shared-types/src/index.ts b/packages/shared-types/src/index.ts index 5ba16cf..8ea3765 100644 --- a/packages/shared-types/src/index.ts +++ b/packages/shared-types/src/index.ts @@ -516,9 +516,7 @@ export interface Question { id: number; brand_id: number; keyword_id: number; - current_version_id: number | null; - question_text: string | null; - version_no: number | null; + question_text: string; status: string; created_at: string; } diff --git a/server/internal/tenant/app/brand_service.go b/server/internal/tenant/app/brand_service.go index 2bd0b45..abff050 100644 --- a/server/internal/tenant/app/brand_service.go +++ b/server/internal/tenant/app/brand_service.go @@ -2,7 +2,6 @@ package app import ( "context" - "crypto/sha256" "encoding/json" "fmt" "strings" @@ -276,7 +275,7 @@ func (s *BrandService) DeleteKeyword(ctx context.Context, brandID, keywordID int return nil } -// --- Question CRUD with versioning --- +// --- Question CRUD --- type QuestionRequest struct { KeywordID int64 `json:"keyword_id" binding:"required"` @@ -284,23 +283,19 @@ type QuestionRequest struct { } type QuestionResponse struct { - ID int64 `json:"id"` - BrandID int64 `json:"brand_id"` - KeywordID int64 `json:"keyword_id"` - CurrentVersionID *int64 `json:"current_version_id"` - QuestionText *string `json:"question_text"` - VersionNo *int `json:"version_no"` - Status string `json:"status"` - CreatedAt string `json:"created_at"` + ID int64 `json:"id"` + BrandID int64 `json:"brand_id"` + KeywordID int64 `json:"keyword_id"` + QuestionText string `json:"question_text"` + Status string `json:"status"` + CreatedAt string `json:"created_at"` } func (s *BrandService) ListQuestions(ctx context.Context, brandID int64, keywordID *int64) ([]QuestionResponse, error) { actor := auth.MustActor(ctx) query := ` - SELECT q.id, q.brand_id, q.keyword_id, q.current_version_id, q.status, q.created_at, - v.question_text, v.version_no + SELECT q.id, q.brand_id, q.keyword_id, q.question_text, q.status, q.created_at FROM brand_questions q - LEFT JOIN brand_question_versions v ON v.id = q.current_version_id WHERE q.brand_id = $1 AND q.tenant_id = $2 AND q.deleted_at IS NULL` args := []interface{}{brandID, actor.TenantID} @@ -320,7 +315,7 @@ func (s *BrandService) ListQuestions(ctx context.Context, brandID int64, keyword for rows.Next() { var q QuestionResponse var ca interface{} - if err := rows.Scan(&q.ID, &q.BrandID, &q.KeywordID, &q.CurrentVersionID, &q.Status, &ca, &q.QuestionText, &q.VersionNo); err != nil { + if err := rows.Scan(&q.ID, &q.BrandID, &q.KeywordID, &q.QuestionText, &q.Status, &ca); err != nil { return nil, response.ErrInternal(50010, "scan_failed", err.Error()) } q.CreatedAt = fmt.Sprintf("%v", ca) @@ -334,42 +329,29 @@ func (s *BrandService) ListQuestions(ctx context.Context, brandID int64, keyword func (s *BrandService) CreateQuestion(ctx context.Context, brandID int64, req QuestionRequest) (*QuestionResponse, error) { actor := auth.MustActor(ctx) - tx, err := s.pool.Begin(ctx) - if err != nil { - return nil, fmt.Errorf("begin tx: %w", err) + req.QuestionText = strings.TrimSpace(req.QuestionText) + if req.QuestionText == "" { + return nil, response.ErrBadRequest(40001, "invalid_params", "question_text is required") } - defer func() { - _ = tx.Rollback(ctx) - }() var questionID int64 - err = tx.QueryRow(ctx, ` - INSERT INTO brand_questions (tenant_id, brand_id, keyword_id, status) VALUES ($1, $2, $3, 'active') RETURNING id - `, actor.TenantID, brandID, req.KeywordID).Scan(&questionID) + var ca interface{} + err := s.pool.QueryRow(ctx, ` + INSERT INTO brand_questions (tenant_id, brand_id, keyword_id, question_text, status) + VALUES ($1, $2, $3, $4, 'active') + RETURNING id, created_at + `, actor.TenantID, brandID, req.KeywordID, req.QuestionText).Scan(&questionID, &ca) if err != nil { return nil, response.ErrInternal(50010, "create_failed", "failed to create question") } - hash := fmt.Sprintf("%x", sha256.Sum256([]byte(req.QuestionText))) - var versionID int64 - err = tx.QueryRow(ctx, ` - INSERT INTO brand_question_versions (question_id, question_text, question_hash, version_no, is_active) - VALUES ($1, $2, $3, 1, true) RETURNING id - `, questionID, req.QuestionText, hash).Scan(&versionID) - if err != nil { - return nil, response.ErrInternal(50010, "create_version_failed", "failed to create question version") - } - - _, _ = tx.Exec(ctx, `UPDATE brand_questions SET current_version_id = $1 WHERE id = $2`, versionID, questionID) - - if err := tx.Commit(ctx); err != nil { - return nil, fmt.Errorf("commit: %w", err) - } - - vno := 1 return &QuestionResponse{ - ID: questionID, BrandID: brandID, KeywordID: req.KeywordID, - CurrentVersionID: &versionID, QuestionText: &req.QuestionText, VersionNo: &vno, Status: "active", + ID: questionID, + BrandID: brandID, + KeywordID: req.KeywordID, + QuestionText: req.QuestionText, + Status: "active", + CreatedAt: fmt.Sprintf("%v", ca), }, nil } @@ -379,42 +361,19 @@ type UpdateQuestionRequest struct { func (s *BrandService) UpdateQuestion(ctx context.Context, brandID, questionID int64, req UpdateQuestionRequest) error { actor := auth.MustActor(ctx) - tx, err := s.pool.Begin(ctx) - if err != nil { - return fmt.Errorf("begin tx: %w", err) + req.QuestionText = strings.TrimSpace(req.QuestionText) + if req.QuestionText == "" { + return response.ErrBadRequest(40001, "invalid_params", "question_text is required") } - defer func() { - _ = tx.Rollback(ctx) - }() - // Verify ownership - var exists bool - _ = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM brand_questions WHERE id = $1 AND brand_id = $2 AND tenant_id = $3 AND deleted_at IS NULL)`, - questionID, brandID, actor.TenantID).Scan(&exists) - if !exists { + tag, err := s.pool.Exec(ctx, ` + UPDATE brand_questions SET question_text = $1, updated_at = NOW() + WHERE id = $2 AND brand_id = $3 AND tenant_id = $4 AND deleted_at IS NULL + `, req.QuestionText, questionID, brandID, actor.TenantID) + if err != nil || tag.RowsAffected() == 0 { return response.ErrNotFound(40422, "question_not_found", "question not found") } - - // Deactivate current version - _, _ = tx.Exec(ctx, `UPDATE brand_question_versions SET is_active = false WHERE question_id = $1 AND is_active = true`, questionID) - - // Get next version number - var maxVersion int - _ = tx.QueryRow(ctx, `SELECT COALESCE(MAX(version_no), 0) FROM brand_question_versions WHERE question_id = $1`, questionID).Scan(&maxVersion) - - hash := fmt.Sprintf("%x", sha256.Sum256([]byte(req.QuestionText))) - var versionID int64 - err = tx.QueryRow(ctx, ` - INSERT INTO brand_question_versions (question_id, question_text, question_hash, version_no, is_active) - VALUES ($1, $2, $3, $4, true) RETURNING id - `, questionID, req.QuestionText, hash, maxVersion+1).Scan(&versionID) - if err != nil { - return fmt.Errorf("create version: %w", err) - } - - _, _ = tx.Exec(ctx, `UPDATE brand_questions SET current_version_id = $1, updated_at = NOW() WHERE id = $2`, versionID, questionID) - - return tx.Commit(ctx) + return nil } func (s *BrandService) DeleteQuestion(ctx context.Context, brandID, questionID int64) error { diff --git a/server/internal/tenant/domain/brand_question.go b/server/internal/tenant/domain/brand_question.go index 907584a..f3cae7e 100644 --- a/server/internal/tenant/domain/brand_question.go +++ b/server/internal/tenant/domain/brand_question.go @@ -3,21 +3,11 @@ package domain import "time" type BrandQuestion struct { - ID int64 - TenantID int64 - BrandID int64 - KeywordID int64 - CurrentVersionID *int64 - Status string - CreatedAt time.Time -} - -type BrandQuestionVersion struct { ID int64 - QuestionID int64 + TenantID int64 + BrandID int64 + KeywordID int64 QuestionText string - QuestionHash string - VersionNo int - IsActive bool + Status string CreatedAt time.Time } diff --git a/server/internal/tenant/repository/common.go b/server/internal/tenant/repository/common.go index 8b2eac7..1aa46c8 100644 --- a/server/internal/tenant/repository/common.go +++ b/server/internal/tenant/repository/common.go @@ -20,6 +20,22 @@ func nullableText(value pgtype.Text) *string { return &text } +func nullableAnyText(value interface{}) *string { + switch typed := value.(type) { + case nil: + return nil + case string: + return &typed + case []byte: + text := string(typed) + return &text + case pgtype.Text: + return nullableText(typed) + default: + return nil + } +} + func nullableInt64(value pgtype.Int8) *int64 { if !value.Valid { return nil diff --git a/server/internal/tenant/repository/generated/brand.sql.go b/server/internal/tenant/repository/generated/brand.sql.go index e104478..d81f500 100644 --- a/server/internal/tenant/repository/generated/brand.sql.go +++ b/server/internal/tenant/repository/generated/brand.sql.go @@ -94,59 +94,30 @@ func (q *Queries) CreateKeyword(ctx context.Context, arg CreateKeywordParams) (C } const createQuestion = `-- name: CreateQuestion :one -INSERT INTO brand_questions (tenant_id, brand_id, keyword_id, status) -VALUES ($1, $2, $3, 'active') +INSERT INTO brand_questions (tenant_id, brand_id, keyword_id, question_text, status) +VALUES ($1, $2, $3, $4, 'active') RETURNING id ` type CreateQuestionParams struct { - TenantID int64 `json:"tenant_id"` - BrandID int64 `json:"brand_id"` - KeywordID int64 `json:"keyword_id"` + TenantID int64 `json:"tenant_id"` + BrandID int64 `json:"brand_id"` + KeywordID int64 `json:"keyword_id"` + QuestionText string `json:"question_text"` } func (q *Queries) CreateQuestion(ctx context.Context, arg CreateQuestionParams) (int64, error) { - row := q.db.QueryRow(ctx, createQuestion, arg.TenantID, arg.BrandID, arg.KeywordID) - var id int64 - err := row.Scan(&id) - return id, err -} - -const createQuestionVersion = `-- name: CreateQuestionVersion :one -INSERT INTO brand_question_versions (question_id, question_text, question_hash, version_no, is_active) -VALUES ($1, $2, $3, $4, true) -RETURNING id -` - -type CreateQuestionVersionParams struct { - QuestionID int64 `json:"question_id"` - QuestionText string `json:"question_text"` - QuestionHash string `json:"question_hash"` - VersionNo int32 `json:"version_no"` -} - -func (q *Queries) CreateQuestionVersion(ctx context.Context, arg CreateQuestionVersionParams) (int64, error) { - row := q.db.QueryRow(ctx, createQuestionVersion, - arg.QuestionID, + row := q.db.QueryRow(ctx, createQuestion, + arg.TenantID, + arg.BrandID, + arg.KeywordID, arg.QuestionText, - arg.QuestionHash, - arg.VersionNo, ) var id int64 err := row.Scan(&id) return id, err } -const deactivateQuestionVersion = `-- name: DeactivateQuestionVersion :exec -UPDATE brand_question_versions SET is_active = false -WHERE question_id = $1 AND is_active = true -` - -func (q *Queries) DeactivateQuestionVersion(ctx context.Context, questionID int64) error { - _, err := q.db.Exec(ctx, deactivateQuestionVersion, questionID) - return err -} - const getBrandByID = `-- name: GetBrandByID :one SELECT id, tenant_id, name, description, status, created_at, updated_at FROM brands @@ -183,19 +154,6 @@ func (q *Queries) GetBrandByID(ctx context.Context, arg GetBrandByIDParams) (Get return i, err } -const getLatestQuestionVersionNo = `-- name: GetLatestQuestionVersionNo :one -SELECT COALESCE(MAX(version_no), 0)::INT AS max_version -FROM brand_question_versions -WHERE question_id = $1 -` - -func (q *Queries) GetLatestQuestionVersionNo(ctx context.Context, questionID int64) (int32, error) { - row := q.db.QueryRow(ctx, getLatestQuestionVersionNo, questionID) - var max_version int32 - err := row.Scan(&max_version) - return max_version, err -} - const listBrands = `-- name: ListBrands :many SELECT id, tenant_id, name, description, status, created_at, updated_at FROM brands @@ -346,10 +304,8 @@ func (q *Queries) ListKeywords(ctx context.Context, arg ListKeywordsParams) ([]L } const listQuestions = `-- name: ListQuestions :many -SELECT q.id, q.tenant_id, q.brand_id, q.keyword_id, q.current_version_id, q.status, q.created_at, - v.question_text, v.version_no +SELECT q.id, q.tenant_id, q.brand_id, q.keyword_id, q.question_text, q.status, q.created_at FROM brand_questions q -LEFT JOIN brand_question_versions v ON v.id = q.current_version_id WHERE q.brand_id = $1 AND q.tenant_id = $2 AND q.deleted_at IS NULL AND ($3::bigint IS NULL OR q.keyword_id = $3) ORDER BY q.created_at DESC @@ -362,15 +318,13 @@ type ListQuestionsParams struct { } type ListQuestionsRow struct { - ID int64 `json:"id"` - TenantID int64 `json:"tenant_id"` - BrandID int64 `json:"brand_id"` - KeywordID int64 `json:"keyword_id"` - CurrentVersionID pgtype.Int8 `json:"current_version_id"` - Status string `json:"status"` - CreatedAt pgtype.Timestamptz `json:"created_at"` - QuestionText pgtype.Text `json:"question_text"` - VersionNo pgtype.Int4 `json:"version_no"` + ID int64 `json:"id"` + TenantID int64 `json:"tenant_id"` + BrandID int64 `json:"brand_id"` + KeywordID int64 `json:"keyword_id"` + QuestionText string `json:"question_text"` + Status string `json:"status"` + CreatedAt pgtype.Timestamptz `json:"created_at"` } func (q *Queries) ListQuestions(ctx context.Context, arg ListQuestionsParams) ([]ListQuestionsRow, error) { @@ -387,11 +341,9 @@ func (q *Queries) ListQuestions(ctx context.Context, arg ListQuestionsParams) ([ &i.TenantID, &i.BrandID, &i.KeywordID, - &i.CurrentVersionID, + &i.QuestionText, &i.Status, &i.CreatedAt, - &i.QuestionText, - &i.VersionNo, ); err != nil { return nil, err } @@ -584,18 +536,24 @@ func (q *Queries) UpdateKeyword(ctx context.Context, arg UpdateKeywordParams) er return err } -const updateQuestionCurrentVersion = `-- name: UpdateQuestionCurrentVersion :exec -UPDATE brand_questions SET current_version_id = $1, updated_at = NOW() -WHERE id = $2 AND tenant_id = $3 +const updateQuestion = `-- name: UpdateQuestion :exec +UPDATE brand_questions SET question_text = $1, updated_at = NOW() +WHERE id = $2 AND brand_id = $3 AND tenant_id = $4 AND deleted_at IS NULL ` -type UpdateQuestionCurrentVersionParams struct { - VersionID pgtype.Int8 `json:"version_id"` - ID int64 `json:"id"` - TenantID int64 `json:"tenant_id"` +type UpdateQuestionParams struct { + QuestionText string `json:"question_text"` + ID int64 `json:"id"` + BrandID int64 `json:"brand_id"` + TenantID int64 `json:"tenant_id"` } -func (q *Queries) UpdateQuestionCurrentVersion(ctx context.Context, arg UpdateQuestionCurrentVersionParams) error { - _, err := q.db.Exec(ctx, updateQuestionCurrentVersion, arg.VersionID, arg.ID, arg.TenantID) +func (q *Queries) UpdateQuestion(ctx context.Context, arg UpdateQuestionParams) error { + _, err := q.db.Exec(ctx, updateQuestion, + arg.QuestionText, + arg.ID, + arg.BrandID, + arg.TenantID, + ) return err } diff --git a/server/internal/tenant/repository/generated/models.go b/server/internal/tenant/repository/generated/models.go index 0740a84..e5aeef7 100644 --- a/server/internal/tenant/repository/generated/models.go +++ b/server/internal/tenant/repository/generated/models.go @@ -94,25 +94,15 @@ type BrandKeyword struct { } type BrandQuestion struct { - ID int64 `json:"id"` - TenantID int64 `json:"tenant_id"` - BrandID int64 `json:"brand_id"` - KeywordID int64 `json:"keyword_id"` - CurrentVersionID pgtype.Int8 `json:"current_version_id"` - Status string `json:"status"` - CreatedAt pgtype.Timestamptz `json:"created_at"` - UpdatedAt pgtype.Timestamptz `json:"updated_at"` - DeletedAt pgtype.Timestamptz `json:"deleted_at"` -} - -type BrandQuestionVersion struct { ID int64 `json:"id"` - QuestionID int64 `json:"question_id"` + TenantID int64 `json:"tenant_id"` + BrandID int64 `json:"brand_id"` + KeywordID int64 `json:"keyword_id"` QuestionText string `json:"question_text"` - QuestionHash string `json:"question_hash"` - VersionNo int32 `json:"version_no"` - IsActive bool `json:"is_active"` + Status string `json:"status"` CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` + DeletedAt pgtype.Timestamptz `json:"deleted_at"` } type Competitor struct { @@ -146,6 +136,77 @@ type GenerationTask struct { OperatorID pgtype.Int8 `json:"operator_id"` } +type KnowledgeChunksMetum struct { + ID int64 `json:"id"` + TenantID int64 `json:"tenant_id"` + KnowledgeItemID int64 `json:"knowledge_item_id"` + ChunkIndex int32 `json:"chunk_index"` + TokenCount int32 `json:"token_count"` + QdrantPointID string `json:"qdrant_point_id"` + ItemVersion int32 `json:"item_version"` + Status string `json:"status"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type KnowledgeGroup struct { + ID int64 `json:"id"` + TenantID int64 `json:"tenant_id"` + Name string `json:"name"` + ParentID pgtype.Int8 `json:"parent_id"` + SortOrder int32 `json:"sort_order"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` + DeletedAt pgtype.Timestamptz `json:"deleted_at"` +} + +type KnowledgeItem struct { + ID int64 `json:"id"` + TenantID int64 `json:"tenant_id"` + GroupID int64 `json:"group_id"` + SourceType string `json:"source_type"` + Name string `json:"name"` + SourceUri pgtype.Text `json:"source_uri"` + StorageKey string `json:"storage_key"` + ContentText pgtype.Text `json:"content_text"` + Status string `json:"status"` + SizeBytes int64 `json:"size_bytes"` + ItemVersion int32 `json:"item_version"` + ErrorMessage pgtype.Text `json:"error_message"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` + DeletedAt pgtype.Timestamptz `json:"deleted_at"` + MarkdownContent pgtype.Text `json:"markdown_content"` +} + +type KnowledgeParseTask struct { + ID int64 `json:"id"` + TenantID int64 `json:"tenant_id"` + KnowledgeItemID int64 `json:"knowledge_item_id"` + SourceType string `json:"source_type"` + Status string `json:"status"` + ErrorMessage pgtype.Text `json:"error_message"` + StartedAt pgtype.Timestamptz `json:"started_at"` + CompletedAt pgtype.Timestamptz `json:"completed_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type MediaPlatform struct { + ID int64 `json:"id"` + PlatformID string `json:"platform_id"` + Name string `json:"name"` + Category string `json:"category"` + ShortName string `json:"short_name"` + AccentColor string `json:"accent_color"` + LoginUrl pgtype.Text `json:"login_url"` + LogoUrl pgtype.Text `json:"logo_url"` + Status string `json:"status"` + SortOrder int32 `json:"sort_order"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + type Plan struct { ID int64 `json:"id"` PlanCode string `json:"plan_code"` @@ -157,6 +218,22 @@ type Plan struct { DeletedAt pgtype.Timestamptz `json:"deleted_at"` } +type PlatformAccount struct { + ID int64 `json:"id"` + TenantID int64 `json:"tenant_id"` + UserID int64 `json:"user_id"` + PlatformID string `json:"platform_id"` + PlatformUid string `json:"platform_uid"` + Nickname string `json:"nickname"` + AvatarUrl pgtype.Text `json:"avatar_url"` + Status string `json:"status"` + MetadataJson []byte `json:"metadata_json"` + LastCheckAt pgtype.Timestamptz `json:"last_check_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` + DeletedAt pgtype.Timestamptz `json:"deleted_at"` +} + type PlatformUserRole struct { ID int64 `json:"id"` UserID int64 `json:"user_id"` @@ -166,6 +243,40 @@ type PlatformUserRole struct { DeletedAt pgtype.Timestamptz `json:"deleted_at"` } +type PluginInstallation struct { + ID int64 `json:"id"` + TenantID int64 `json:"tenant_id"` + UserID int64 `json:"user_id"` + InstallationKey string `json:"installation_key"` + InstallationName string `json:"installation_name"` + BrowserName pgtype.Text `json:"browser_name"` + ClientVersion pgtype.Text `json:"client_version"` + InstallationTokenHash string `json:"installation_token_hash"` + Status string `json:"status"` + LastSeenAt pgtype.Timestamptz `json:"last_seen_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` + DeletedAt pgtype.Timestamptz `json:"deleted_at"` +} + +type PluginSession struct { + ID int64 `json:"id"` + TenantID int64 `json:"tenant_id"` + UserID int64 `json:"user_id"` + PluginInstallationID pgtype.Int8 `json:"plugin_installation_id"` + ActionType string `json:"action_type"` + ResourceType pgtype.Text `json:"resource_type"` + ResourceID pgtype.Int8 `json:"resource_id"` + PlatformAccountID pgtype.Int8 `json:"platform_account_id"` + PlatformID string `json:"platform_id"` + SessionToken string `json:"session_token"` + ClientVersion pgtype.Text `json:"client_version"` + Status string `json:"status"` + ExpireAt pgtype.Timestamptz `json:"expire_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + type PromptRule struct { ID int64 `json:"id"` TenantID int64 `json:"tenant_id"` @@ -191,6 +302,44 @@ type PromptRuleGroup struct { DeletedAt pgtype.Timestamptz `json:"deleted_at"` } +type PromptRuleKnowledgeGroup struct { + PromptRuleID int64 `json:"prompt_rule_id"` + KnowledgeGroupID int64 `json:"knowledge_group_id"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} + +type PublishBatch struct { + ID int64 `json:"id"` + TenantID int64 `json:"tenant_id"` + ArticleID int64 `json:"article_id"` + InitiatorUserID int64 `json:"initiator_user_id"` + Status string `json:"status"` + PublishType string `json:"publish_type"` + CoverAssetUrl pgtype.Text `json:"cover_asset_url"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type PublishRecord struct { + ID int64 `json:"id"` + TenantID int64 `json:"tenant_id"` + PublishBatchID int64 `json:"publish_batch_id"` + ArticleID int64 `json:"article_id"` + PlatformAccountID int64 `json:"platform_account_id"` + PlatformID string `json:"platform_id"` + Status string `json:"status"` + ExternalArticleID pgtype.Text `json:"external_article_id"` + ExternalArticleUrl pgtype.Text `json:"external_article_url"` + ExternalManageUrl pgtype.Text `json:"external_manage_url"` + PublishedAt pgtype.Timestamptz `json:"published_at"` + RequestPayloadJson []byte `json:"request_payload_json"` + ResponsePayloadJson []byte `json:"response_payload_json"` + ErrorMessage pgtype.Text `json:"error_message"` + RetryCount int32 `json:"retry_count"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + type QuotaReservation struct { ID int64 `json:"id"` TenantID int64 `json:"tenant_id"` diff --git a/server/internal/tenant/repository/generated/querier.go b/server/internal/tenant/repository/generated/querier.go index 0e93190..56af0c5 100644 --- a/server/internal/tenant/repository/generated/querier.go +++ b/server/internal/tenant/repository/generated/querier.go @@ -27,16 +27,13 @@ type Querier interface { CreatePromptRule(ctx context.Context, arg CreatePromptRuleParams) (CreatePromptRuleRow, error) CreatePromptRuleGroup(ctx context.Context, arg CreatePromptRuleGroupParams) (CreatePromptRuleGroupRow, error) CreateQuestion(ctx context.Context, arg CreateQuestionParams) (int64, error) - CreateQuestionVersion(ctx context.Context, arg CreateQuestionVersionParams) (int64, error) CreateQuotaReservation(ctx context.Context, arg CreateQuotaReservationParams) (int64, error) CreateScheduleTask(ctx context.Context, arg CreateScheduleTaskParams) (CreateScheduleTaskRow, error) - DeactivateQuestionVersion(ctx context.Context, questionID int64) error GetActivePlanForTenant(ctx context.Context, tenantID int64) (GetActivePlanForTenantRow, error) GetArticleByID(ctx context.Context, arg GetArticleByIDParams) (GetArticleByIDRow, error) GetArticleVersions(ctx context.Context, articleID int64) ([]GetArticleVersionsRow, error) GetBrandByID(ctx context.Context, arg GetBrandByIDParams) (GetBrandByIDRow, error) GetCurrentBalance(ctx context.Context, arg GetCurrentBalanceParams) (int32, error) - GetLatestQuestionVersionNo(ctx context.Context, questionID int64) (int32, error) GetPromptRuleByID(ctx context.Context, arg GetPromptRuleByIDParams) (GetPromptRuleByIDRow, error) GetQuotaSummary(ctx context.Context, tenantID int64) (int32, error) GetRecentArticles(ctx context.Context, tenantID int64) ([]GetRecentArticlesRow, error) @@ -85,7 +82,7 @@ type Querier interface { UpdatePromptRule(ctx context.Context, arg UpdatePromptRuleParams) error UpdatePromptRuleGroup(ctx context.Context, arg UpdatePromptRuleGroupParams) error UpdatePromptRuleStatus(ctx context.Context, arg UpdatePromptRuleStatusParams) error - UpdateQuestionCurrentVersion(ctx context.Context, arg UpdateQuestionCurrentVersionParams) error + UpdateQuestion(ctx context.Context, arg UpdateQuestionParams) error UpdateQuotaReservationResource(ctx context.Context, arg UpdateQuotaReservationResourceParams) error UpdateScheduleTask(ctx context.Context, arg UpdateScheduleTaskParams) error UpdateScheduleTaskStatus(ctx context.Context, arg UpdateScheduleTaskStatusParams) error diff --git a/server/internal/tenant/repository/generated/workspace.sql.go b/server/internal/tenant/repository/generated/workspace.sql.go index 1ffebdb..dbc9b4d 100644 --- a/server/internal/tenant/repository/generated/workspace.sql.go +++ b/server/internal/tenant/repository/generated/workspace.sql.go @@ -98,7 +98,7 @@ SELECT a.id, a.generate_status, a.publish_status, a.source_type, a.created_at, t.template_name, COALESCE( NULLIF(gt.input_params_json ->> 'generation_mode', ''), - CASE WHEN a.source_type = 'custom_generation' THEN 'instant' ELSE NULL END + CASE WHEN a.source_type = 'custom_generation' THEN 'instant'::TEXT ELSE NULL::TEXT END ) AS generation_mode FROM articles a LEFT JOIN article_versions av ON av.id = a.current_version_id @@ -126,7 +126,7 @@ type GetRecentArticlesRow struct { WordCount pgtype.Int4 `json:"word_count"` SourceLabel pgtype.Text `json:"source_label"` TemplateName pgtype.Text `json:"template_name"` - GenerationMode pgtype.Text `json:"generation_mode"` + GenerationMode interface{} `json:"generation_mode"` } func (q *Queries) GetRecentArticles(ctx context.Context, tenantID int64) ([]GetRecentArticlesRow, error) { diff --git a/server/internal/tenant/repository/queries/brand.sql b/server/internal/tenant/repository/queries/brand.sql index 0825979..fb46f89 100644 --- a/server/internal/tenant/repository/queries/brand.sql +++ b/server/internal/tenant/repository/queries/brand.sql @@ -46,36 +46,20 @@ UPDATE brand_keywords SET deleted_at = NOW(), updated_at = NOW() WHERE brand_id = sqlc.arg(brand_id) AND tenant_id = sqlc.arg(tenant_id) AND deleted_at IS NULL; -- name: ListQuestions :many -SELECT q.id, q.tenant_id, q.brand_id, q.keyword_id, q.current_version_id, q.status, q.created_at, - v.question_text, v.version_no +SELECT q.id, q.tenant_id, q.brand_id, q.keyword_id, q.question_text, q.status, q.created_at FROM brand_questions q -LEFT JOIN brand_question_versions v ON v.id = q.current_version_id WHERE q.brand_id = sqlc.arg(brand_id) AND q.tenant_id = sqlc.arg(tenant_id) AND q.deleted_at IS NULL AND (sqlc.narg(keyword_id)::bigint IS NULL OR q.keyword_id = sqlc.narg(keyword_id)) ORDER BY q.created_at DESC; -- name: CreateQuestion :one -INSERT INTO brand_questions (tenant_id, brand_id, keyword_id, status) -VALUES (sqlc.arg(tenant_id), sqlc.arg(brand_id), sqlc.arg(keyword_id), 'active') +INSERT INTO brand_questions (tenant_id, brand_id, keyword_id, question_text, status) +VALUES (sqlc.arg(tenant_id), sqlc.arg(brand_id), sqlc.arg(keyword_id), sqlc.arg(question_text), 'active') RETURNING id; --- name: CreateQuestionVersion :one -INSERT INTO brand_question_versions (question_id, question_text, question_hash, version_no, is_active) -VALUES (sqlc.arg(question_id), sqlc.arg(question_text), sqlc.arg(question_hash), sqlc.arg(version_no), true) -RETURNING id; - --- name: UpdateQuestionCurrentVersion :exec -UPDATE brand_questions SET current_version_id = sqlc.arg(version_id), updated_at = NOW() -WHERE id = sqlc.arg(id) AND tenant_id = sqlc.arg(tenant_id); - --- name: DeactivateQuestionVersion :exec -UPDATE brand_question_versions SET is_active = false -WHERE question_id = sqlc.arg(question_id) AND is_active = true; - --- name: GetLatestQuestionVersionNo :one -SELECT COALESCE(MAX(version_no), 0)::INT AS max_version -FROM brand_question_versions -WHERE question_id = sqlc.arg(question_id); +-- name: UpdateQuestion :exec +UPDATE brand_questions SET question_text = sqlc.arg(question_text), updated_at = NOW() +WHERE id = sqlc.arg(id) AND brand_id = sqlc.arg(brand_id) AND tenant_id = sqlc.arg(tenant_id) AND deleted_at IS NULL; -- name: SoftDeleteQuestion :exec UPDATE brand_questions SET deleted_at = NOW(), updated_at = NOW() diff --git a/server/internal/tenant/repository/queries/workspace.sql b/server/internal/tenant/repository/queries/workspace.sql index 00c0e17..56dd6ce 100644 --- a/server/internal/tenant/repository/queries/workspace.sql +++ b/server/internal/tenant/repository/queries/workspace.sql @@ -16,7 +16,7 @@ SELECT a.id, a.generate_status, a.publish_status, a.source_type, a.created_at, t.template_name, COALESCE( NULLIF(gt.input_params_json ->> 'generation_mode', ''), - CASE WHEN a.source_type = 'custom_generation' THEN 'instant' ELSE NULL END + CASE WHEN a.source_type = 'custom_generation' THEN 'instant'::TEXT ELSE NULL::TEXT END ) AS generation_mode FROM articles a LEFT JOIN article_versions av ON av.id = a.current_version_id diff --git a/server/internal/tenant/repository/workspace_repo.go b/server/internal/tenant/repository/workspace_repo.go index 96bd28a..993c52f 100644 --- a/server/internal/tenant/repository/workspace_repo.go +++ b/server/internal/tenant/repository/workspace_repo.go @@ -70,7 +70,7 @@ func (r *workspaceRepository) GetRecentArticles(ctx context.Context, tenantID in GenerateStatus: row.GenerateStatus, PublishStatus: row.PublishStatus, SourceType: row.SourceType, - GenerationMode: nullableText(row.GenerationMode), + GenerationMode: nullableAnyText(row.GenerationMode), CreatedAt: timeFromTimestamp(row.CreatedAt), Title: nullableText(row.Title), WordCount: intFromInt4(row.WordCount), diff --git a/server/migrations/20260331100022_create_brand_questions.up.sql b/server/migrations/20260331100022_create_brand_questions.up.sql index 1e90f36..645b249 100644 --- a/server/migrations/20260331100022_create_brand_questions.up.sql +++ b/server/migrations/20260331100022_create_brand_questions.up.sql @@ -1,12 +1,12 @@ CREATE TABLE brand_questions ( - id BIGSERIAL PRIMARY KEY, - tenant_id BIGINT NOT NULL REFERENCES tenants(id), - brand_id BIGINT NOT NULL REFERENCES brands(id), - keyword_id BIGINT NOT NULL REFERENCES brand_keywords(id), - current_version_id BIGINT, - status VARCHAR(20) NOT NULL DEFAULT 'active', - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - deleted_at TIMESTAMPTZ + id BIGSERIAL PRIMARY KEY, + tenant_id BIGINT NOT NULL REFERENCES tenants(id), + brand_id BIGINT NOT NULL REFERENCES brands(id), + keyword_id BIGINT NOT NULL REFERENCES brand_keywords(id), + question_text TEXT NOT NULL, + status VARCHAR(20) NOT NULL DEFAULT 'active', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + deleted_at TIMESTAMPTZ ); CREATE INDEX idx_brand_questions_keyword ON brand_questions(keyword_id); diff --git a/server/migrations/20260331100023_create_brand_question_versions.down.sql b/server/migrations/20260331100023_create_brand_question_versions.down.sql index d789915..1af4560 100644 --- a/server/migrations/20260331100023_create_brand_question_versions.down.sql +++ b/server/migrations/20260331100023_create_brand_question_versions.down.sql @@ -1 +1 @@ -DROP TABLE IF EXISTS brand_question_versions; +-- No-op: question versioning was removed during development. diff --git a/server/migrations/20260331100023_create_brand_question_versions.up.sql b/server/migrations/20260331100023_create_brand_question_versions.up.sql index 46cfb9c..1af4560 100644 --- a/server/migrations/20260331100023_create_brand_question_versions.up.sql +++ b/server/migrations/20260331100023_create_brand_question_versions.up.sql @@ -1,11 +1 @@ -CREATE TABLE brand_question_versions ( - id BIGSERIAL PRIMARY KEY, - question_id BIGINT NOT NULL REFERENCES brand_questions(id), - question_text TEXT NOT NULL, - question_hash VARCHAR(64) NOT NULL, - version_no INT NOT NULL, - is_active BOOLEAN NOT NULL DEFAULT true, - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - CONSTRAINT uk_question_version_no UNIQUE (question_id, version_no) -); -CREATE INDEX idx_question_hash ON brand_question_versions(question_hash); +-- No-op: question versioning was removed during development. diff --git a/server/migrations/20260408110000_flatten_brand_questions.down.sql b/server/migrations/20260408110000_flatten_brand_questions.down.sql new file mode 100644 index 0000000..cf56b69 --- /dev/null +++ b/server/migrations/20260408110000_flatten_brand_questions.down.sql @@ -0,0 +1,37 @@ +ALTER TABLE brand_questions + ADD COLUMN IF NOT EXISTS current_version_id BIGINT; + +CREATE TABLE IF NOT EXISTS brand_question_versions ( + id BIGSERIAL PRIMARY KEY, + question_id BIGINT NOT NULL REFERENCES brand_questions(id), + question_text TEXT NOT NULL, + question_hash VARCHAR(64) NOT NULL, + version_no INT NOT NULL, + is_active BOOLEAN NOT NULL DEFAULT true, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT uk_question_version_no UNIQUE (question_id, version_no) +); + +CREATE INDEX IF NOT EXISTS idx_question_hash ON brand_question_versions(question_hash); + +INSERT INTO brand_question_versions (question_id, question_text, question_hash, version_no, is_active) +SELECT q.id, + q.question_text, + md5(q.question_text), + 1, + true +FROM brand_questions q +WHERE NOT EXISTS ( + SELECT 1 + FROM brand_question_versions v + WHERE v.question_id = q.id +); + +UPDATE brand_questions q +SET current_version_id = v.id +FROM brand_question_versions v +WHERE v.question_id = q.id + AND v.version_no = 1; + +ALTER TABLE brand_questions + DROP COLUMN IF EXISTS question_text; diff --git a/server/migrations/20260408110000_flatten_brand_questions.up.sql b/server/migrations/20260408110000_flatten_brand_questions.up.sql new file mode 100644 index 0000000..707e458 --- /dev/null +++ b/server/migrations/20260408110000_flatten_brand_questions.up.sql @@ -0,0 +1,29 @@ +ALTER TABLE brand_questions + ADD COLUMN IF NOT EXISTS question_text TEXT; + +UPDATE brand_questions q +SET question_text = COALESCE( + ( + SELECT v.question_text + FROM brand_question_versions v + WHERE v.id = q.current_version_id + LIMIT 1 + ), + ( + SELECT v.question_text + FROM brand_question_versions v + WHERE v.question_id = q.id + ORDER BY v.version_no DESC, v.id DESC + LIMIT 1 + ), + '' +) +WHERE q.question_text IS NULL; + +ALTER TABLE brand_questions + ALTER COLUMN question_text SET NOT NULL; + +ALTER TABLE brand_questions + DROP COLUMN IF EXISTS current_version_id; + +DROP TABLE IF EXISTS brand_question_versions;