# 设计文档：小程序核心业务模块（miniapp-core-business）

## 概述

本设计在 P1（miniapp-db-foundation）、P2（etl-dws-miniapp-extensions）、P3（miniapp-auth-system）基础上，实现小程序的核心业务逻辑：

1. **助教任务系统**：基于 WBI/NCI/RS 指数自动生成 4 种类型任务，支持状态流转（active → inactive/completed/abandoned）、类型变更追溯、48 小时回访滞留机制
2. **备注系统**：统一备注 CRUD，支持普通备注/回访备注/放弃原因三种类型，含星星评分（再次服务意愿 + 再来店可能性，各 1-5）
3. **召回完成检测与备注回溯**：ETL 数据更新后自动检测助教服务记录，匹配活跃任务标记完成，回溯重分类普通备注为回访备注
4. **触发器调度框架**：统一的 cron/interval/event 三种触发方式调度引擎，驱动任务生成、有效期检查、召回检测、备注回溯

**环境变量依赖**：

| 环境变量 | 用途 | 来源 |
|---------|------|------|
| `APP_DB_DSN` / `DB_HOST` 等 | 业务库连接（`test_zqyy_app`） | 根 `.env` |
| `PG_DSN` / `ETL_DB_HOST` 等 | ETL 库连接（FDW 读取指数） | 根 `.env` |
| `JWT_SECRET_KEY` | JWT 签名密钥 | `.env.local` |

**整体数据流向**：

```
ETL 库（test_etl_feiqiu）                    业务库（test_zqyy_app）
┌──────────────────────────┐                ┌──────────────────────────────┐
│ dws.dws_member_winback_  │                │ auth Schema                  │
│     index (WBI)          │                │   └ user_assistant_binding   │
│ dws.dws_member_newconv_  │                │                              │
│     index (NCI)          │   FDW 读取     │ biz Schema                   │
│ dws.dws_member_assistant_│ ◄────────────► │   ├ coach_tasks              │
│     relation_index (RS)  │                │   ├ coach_task_history        │
│ dwd.dwd_assistant_       │                │   ├ notes                     │
│     service_log          │                │   └ trigger_jobs              │
│                          │                │                              │
│ app Schema (RLS 视图)     │                │ fdw_etl Schema (外部表)       │
│   └ v_dws_*, v_dwd_*    │                │   └ v_dws_*, v_dwd_*         │
└──────────────────────────┘                │                              │
                                            │ public Schema                │
                                            │   └ member_retention_clue    │
                                            └──────────────────────────────┘
```

## 架构

### 分层架构

```mermaid
graph TB
    subgraph "小程序端"
        MP["微信小程序<br/>任务列表 / 备注 / 评分"]
    end

    subgraph "FastAPI 后端（apps/backend/）"
        subgraph "路由层"
            XCX_TASK["routers/xcx_tasks.py<br/>任务列表 / 置顶 / 放弃"]
            XCX_NOTE["routers/xcx_notes.py<br/>备注 CRUD / 星星评分"]
        end

        subgraph "中间件层"
            PERM_MW["middleware/permission.py<br/>require_approved()"]
        end

        subgraph "服务层"
            TASK_GEN["services/task_generator.py<br/>任务生成器"]
            TASK_MGR["services/task_manager.py<br/>任务 CRUD + 状态流转"]
            EXPIRY["services/task_expiry.py<br/>有效期轮询"]
            RECALL["services/recall_detector.py<br/>召回完成检测"]
            RECLASS["services/note_reclassifier.py<br/>备注回溯重分类"]
            NOTE_SVC["services/note_service.py<br/>备注 CRUD"]
            TRIGGER["services/trigger_scheduler.py<br/>触发器调度框架"]
        end

        DB["database.py<br/>get_connection() / get_etl_readonly_connection()"]
    end

    subgraph "数据库"
        BIZ["biz Schema<br/>coach_tasks / notes / trigger_jobs"]
        AUTH["auth Schema<br/>user_assistant_binding"]
        FDW["fdw_etl Schema<br/>v_dws_* / v_dwd_*（只读）"]
        PUB["public Schema<br/>member_retention_clue"]
    end

    MP --> XCX_TASK
    MP --> XCX_NOTE
    XCX_TASK --> PERM_MW
    XCX_NOTE --> PERM_MW
    PERM_MW --> TASK_MGR
    PERM_MW --> NOTE_SVC
    TRIGGER --> TASK_GEN
    TRIGGER --> EXPIRY
    TRIGGER --> RECALL
    RECALL --> RECLASS
    TASK_GEN --> DB
    TASK_MGR --> DB
    EXPIRY --> DB
    RECALL --> DB
    RECLASS --> DB
    NOTE_SVC --> DB
    DB --> BIZ
    DB --> AUTH
    DB --> FDW
    DB --> PUB
```

### 触发器调度流程

```mermaid
sequenceDiagram
    participant TS as TriggerScheduler
    participant DB as PostgreSQL
    participant TG as TaskGenerator
    participant EC as TaskExpiryChecker
    participant RD as RecallDetector
    participant NR as NoteReclassifier

    Note over TS: 每 30 秒轮询 trigger_jobs

    TS->>DB: SELECT * FROM biz.trigger_jobs<br/>WHERE status='enabled' AND next_run_at <= NOW()

    alt cron: task_generator（每日 04:00）
        TS->>TG: run()
        TG->>DB: 读取 FDW 指数数据（WBI/NCI/RS）
        TG->>DB: 读取 auth.user_assistant_binding
        TG->>DB: INSERT/UPDATE biz.coach_tasks
        TG->>DB: INSERT biz.coach_task_history
        TG->>DB: UPDATE trigger_jobs.last_run_at
    end

    alt interval: task_expiry_check（每小时）
        TS->>EC: run()
        EC->>DB: SELECT coach_tasks WHERE expires_at < NOW()
        EC->>DB: UPDATE status='inactive'
        EC->>DB: UPDATE trigger_jobs.last_run_at
    end

    alt event: recall_completion_check（ETL 更新后）
        TS->>RD: run()
        RD->>DB: 读取 FDW dwd_assistant_service_log
        RD->>DB: 匹配 active 任务 → completed
        RD->>DB: fire_event('recall_completed', payload)
    end

    alt event: note_reclassify_backfill（召回完成时）
        TS->>NR: run(payload)
        NR->>DB: 查找 normal 备注 → 更新为 follow_up
        NR->>DB: 触发 AI 应用 6 接口（P5 实现）
    end
```

### 任务状态机

```mermaid
stateDiagram-v2
    [*] --> active: 任务生成器创建
    active --> inactive: 类型变更（旧任务关闭）
    active --> inactive: expires_at 到期（轮询检查）
    active --> completed: 召回完成检测
    active --> abandoned: 助教放弃（需填原因）
    abandoned --> active: 助教取消放弃
    active --> active: 置顶/取消置顶（is_pinned 变更）

    note right of inactive: expires_at 机制：\n生成时 NULL（无限期）\n条件不满足时填充 created_at+48h\n轮询检查超期后标记 inactive
```


## 组件与接口

### 组件 1：触发器调度框架（services/trigger_scheduler.py）

**职责**：统一管理 cron/interval/event 三种触发方式，驱动后台任务执行。

**设计决策**：
- 复用现有 `Scheduler` 的轮询模式（每 30 秒检查 `trigger_jobs` 表），但 `trigger_jobs` 是独立于 `scheduled_tasks` 的新表，专门服务于业务触发器
- cron/interval 类型通过轮询 `next_run_at` 触发；event 类型通过 `fire_event()` 方法直接触发
- 每个 job 对应一个 Python 可调用对象，通过 `job_type` 映射到具体的服务方法

```python
import logging
from datetime import datetime, timezone
from typing import Any, Callable

from app.database import get_connection

logger = logging.getLogger(__name__)

# job_type → 执行函数的注册表
_JOB_REGISTRY: dict[str, Callable] = {}


def register_job(job_type: str, handler: Callable) -> None:
    """注册 job_type 对应的执行函数。"""
    _JOB_REGISTRY[job_type] = handler


def fire_event(event_name: str, payload: dict[str, Any] | None = None) -> int:
    """
    触发事件驱动型任务。

    查找 trigger_condition='event' 且 trigger_config.event_name 匹配的 enabled job，
    立即执行对应的 handler。

    返回: 执行的 job 数量
    """
    conn = get_connection()
    executed = 0
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT id, job_type, job_name
                FROM biz.trigger_jobs
                WHERE status = 'enabled'
                  AND trigger_condition = 'event'
                  AND trigger_config->>'event_name' = %s
                """,
                (event_name,),
            )
            rows = cur.fetchall()

        for job_id, job_type, job_name in rows:
            handler = _JOB_REGISTRY.get(job_type)
            if not handler:
                logger.warning("未注册的 job_type: %s (job_name=%s)", job_type, job_name)
                continue
            try:
                handler(payload=payload)
                executed += 1
                # 更新 last_run_at
                with conn.cursor() as cur:
                    cur.execute(
                        "UPDATE biz.trigger_jobs SET last_run_at = NOW() WHERE id = %s",
                        (job_id,),
                    )
                conn.commit()
            except Exception:
                logger.exception("触发器 %s 执行失败", job_name)
                conn.rollback()

    finally:
        conn.close()
    return executed


def check_scheduled_jobs() -> int:
    """
    检查 cron/interval 类型的到期 job 并执行。

    由 Scheduler 后台循环调用。
    返回: 执行的 job 数量
    """
    conn = get_connection()
    executed = 0
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT id, job_type, job_name, trigger_condition, trigger_config
                FROM biz.trigger_jobs
                WHERE status = 'enabled'
                  AND trigger_condition IN ('cron', 'interval')
                  AND (next_run_at IS NULL OR next_run_at <= NOW())
                ORDER BY next_run_at ASC NULLS FIRST
                """,
            )
            rows = cur.fetchall()

        for job_id, job_type, job_name, trigger_condition, trigger_config in rows:
            handler = _JOB_REGISTRY.get(job_type)
            if not handler:
                logger.warning("未注册的 job_type: %s", job_type)
                continue
            try:
                handler()
                executed += 1
                # 计算 next_run_at 并更新
                next_run = _calculate_next_run(trigger_condition, trigger_config)
                with conn.cursor() as cur:
                    cur.execute(
                        """
                        UPDATE biz.trigger_jobs
                        SET last_run_at = NOW(), next_run_at = %s
                        WHERE id = %s
                        """,
                        (next_run, job_id),
                    )
                conn.commit()
            except Exception:
                logger.exception("触发器 %s 执行失败", job_name)
                conn.rollback()

    finally:
        conn.close()
    return executed


def _calculate_next_run(trigger_condition: str, trigger_config: dict) -> datetime | None:
    """根据触发条件和配置计算下次运行时间。"""
    now = datetime.now(timezone.utc)
    if trigger_condition == "interval":
        seconds = trigger_config.get("interval_seconds", 3600)
        from datetime import timedelta
        return now + timedelta(seconds=seconds)
    elif trigger_condition == "cron":
        # 复用现有 scheduler._parse_simple_cron
        from app.services.scheduler import _parse_simple_cron
        return _parse_simple_cron(trigger_config.get("cron_expression", "0 4 * * *"), now)
    return None  # event 类型无 next_run_at
```

### 组件 2：任务生成器（services/task_generator.py）

**职责**：每日 4:00 运行，基于 WBI/NCI/RS 指数为每个助教生成/更新任务。

**核心逻辑**（纯函数，可独立测试）：

```python
from decimal import Decimal
from dataclasses import dataclass
from enum import IntEnum


class TaskPriority(IntEnum):
    """任务类型优先级，数值越小优先级越高。"""
    HIGH_PRIORITY_RECALL = 0
    PRIORITY_RECALL = 0
    FOLLOW_UP_VISIT = 1
    RELATIONSHIP_BUILDING = 2


TASK_TYPE_PRIORITY = {
    "high_priority_recall": TaskPriority.HIGH_PRIORITY_RECALL,
    "priority_recall": TaskPriority.PRIORITY_RECALL,
    "follow_up_visit": TaskPriority.FOLLOW_UP_VISIT,
    "relationship_building": TaskPriority.RELATIONSHIP_BUILDING,
}


@dataclass
class IndexData:
    """某客户-助教对的指数数据。"""
    site_id: int
    assistant_id: int
    member_id: int
    wbi: Decimal          # 流失回赢指数
    nci: Decimal          # 新客转化指数
    rs: Decimal           # 关系强度指数
    has_active_recall: bool       # 是否有活跃召回任务
    has_follow_up_note: bool      # 召回完成后是否有回访备注


def determine_task_type(index_data: IndexData) -> str | None:
    """
    根据指数数据确定应生成的任务类型。

    优先级规则（高 → 低）：
    1. max(WBI, NCI) > 7 → high_priority_recall
    2. max(WBI, NCI) > 5 → priority_recall
    3. 完成召回后无回访备注 → follow_up_visit
    4. RS < 6 → relationship_building
    5. 不满足任何条件 → None（不生成任务）

    返回: task_type 字符串或 None
    """
    priority_score = max(index_data.wbi, index_data.nci)

    if priority_score > 7:
        return "high_priority_recall"
    if priority_score > 5:
        return "priority_recall"
    if index_data.has_active_recall is False and index_data.has_follow_up_note is False:
        # 完成召回后无回访备注 → 回访任务
        # 注意：此条件需要外部传入是否"已完成召回"的状态
        pass
    if index_data.rs < 6:
        return "relationship_building"
    return None


def should_replace_task(
    existing_type: str, new_type: str
) -> bool:
    """
    判断新任务类型是否应替换现有任务类型。

    规则：高优先级任务覆盖低优先级任务。
    同优先级（如 high_priority_recall 和 priority_recall 都是 0）时，
    类型不同则替换。
    """
    if existing_type == new_type:
        return False
    return True  # 类型不同即替换
```

**执行流程**：

```python
def run(self) -> dict:
    """
    任务生成器主流程。

    1. 通过 auth.user_assistant_binding 获取所有已绑定助教
    2. 对每个助教，通过 FDW 读取 WBI/NCI/RS 指数
    3. 调用 determine_task_type() 确定任务类型
    4. 检查是否已存在相同 (site_id, assistant_id, member_id, task_type) 的 active 任务
       - 存在 → 跳过
    5. 检查是否已存在相同 (site_id, assistant_id, member_id) 但不同 task_type 的 active 任务
       - 存在 → 关闭旧任务 + 创建新任务 + 记录 history
    6. 不存在 → 创建新任务
    7. 更新 trigger_jobs 时间戳

    返回: {"created": int, "replaced": int, "skipped": int}
    """
    ...
```

### 组件 3：任务管理服务（services/task_manager.py）

**职责**：任务 CRUD、置顶、放弃、取消操作。

```python
async def get_task_list(
    user_id: int, site_id: int
) -> list[dict]:
    """
    获取助教的活跃任务列表。

    1. 通过 auth.user_assistant_binding 获取 assistant_id
    2. 查询 biz.coach_tasks WHERE site_id=? AND assistant_id=? AND status='active'
    3. JOIN fdw_etl.v_dim_member 获取客户基本信息
    4. JOIN fdw_etl.v_dws_member_assistant_relation_index 获取 RS 指数
    5. 计算爱心 icon 档位：💖>8.5 / 🧡>7 / 💛>5 / 💙<5
    6. 排序：is_pinned DESC, priority_score DESC, created_at ASC

    注意：FDW 查询需要 SET LOCAL app.current_site_id。
    """
    ...


async def pin_task(task_id: int, user_id: int, site_id: int) -> dict:
    """置顶任务。验证任务归属后设置 is_pinned=TRUE，记录 history。"""
    ...


async def unpin_task(task_id: int, user_id: int, site_id: int) -> dict:
    """取消置顶。验证任务归属后设置 is_pinned=FALSE。"""
    ...


async def abandon_task(
    task_id: int, user_id: int, site_id: int, reason: str
) -> dict:
    """
    放弃任务。

    1. 验证 reason 非空（空则 422）
    2. 验证任务归属和 status='active'
    3. 设置 status='abandoned', abandon_reason=reason
    4. 记录 coach_task_history
    """
    ...


async def cancel_abandon(task_id: int, user_id: int, site_id: int) -> dict:
    """
    取消放弃。

    1. 验证任务归属和 status='abandoned'
    2. 恢复 status='active', 清空 abandon_reason
    3. 记录 coach_task_history
    """
    ...


def _record_history(
    conn, task_id: int, action: str,
    old_status: str = None, new_status: str = None,
    old_task_type: str = None, new_task_type: str = None,
    detail: dict = None,
) -> None:
    """在 coach_task_history 中记录变更。"""
    ...
```

### 组件 4：有效期轮询器（services/task_expiry.py）

**职责**：每小时检查 `expires_at` 不为 NULL 且已过期的任务，标记为 inactive。

```python
def run() -> dict:
    """
    有效期轮询主流程。

    1. SELECT id FROM biz.coach_tasks
       WHERE expires_at IS NOT NULL AND expires_at < NOW() AND status = 'active'
    2. UPDATE status = 'inactive'
    3. INSERT coach_task_history (action='expired')

    返回: {"expired_count": int}
    """
    ...
```

### 组件 5：召回完成检测器（services/recall_detector.py）

**职责**：ETL 数据更新后，检测助教服务记录并匹配活跃任务。

```python
def run(payload: dict | None = None) -> dict:
    """
    召回完成检测主流程。

    1. 通过 FDW 读取 fdw_etl.v_dwd_assistant_service_log 中的新增服务记录
       （基于 last_run_at 过滤增量）
    2. 对每条服务记录，查找 biz.coach_tasks 中匹配的
       (site_id, assistant_id, member_id) 且 status='active' 的任务
    3. 将匹配任务标记为 completed：
       - status = 'completed'
       - completed_at = 服务时间
       - completed_task_type = 当前 task_type（快照）
    4. 记录 coach_task_history
    5. 触发 fire_event('recall_completed', {site_id, assistant_id, member_id, service_time})

    返回: {"completed_count": int}
    """
    ...
```

### 组件 6：备注回溯重分类器（services/note_reclassifier.py）

**职责**：召回完成后，回溯检查是否有普通备注需重分类为回访备注。

```python
def run(payload: dict | None = None) -> dict:
    """
    备注回溯主流程。

    payload 包含: {site_id, assistant_id, member_id, service_time}

    1. 查找 biz.notes 中该 (site_id, target_type='member', target_id=member_id)
       在 service_time 之后提交的第一条 type='normal' 的备注
    2. 将该备注 type 从 'normal' 更新为 'follow_up'
    3. 触发 AI 应用 6 接口（P5 实现，本 SPEC 仅定义触发接口）：
       - 调用 ai_analyze_note(note_id) → 返回 ai_score
    4. 若 ai_score >= 6：
       - 生成 follow_up_visit 任务，status='completed'（回溯完成）
    5. 若 ai_score < 6：
       - 生成 follow_up_visit 任务，status='active'（需助教重新备注）

    返回: {"reclassified_count": int, "tasks_created": int}
    """
    ...


def ai_analyze_note(note_id: int) -> int | None:
    """
    AI 应用 6 备注分析接口（占位）。

    P5 AI 集成层实现后替换此占位函数。
    当前返回 None 表示 AI 未就绪，跳过评分逻辑。
    """
    return None
```

### 组件 7：备注服务（services/note_service.py）

**职责**：备注 CRUD、星星评分存储与读取。

```python
async def create_note(
    site_id: int,
    user_id: int,
    target_type: str,       # 'member'
    target_id: int,         # member_id
    content: str,
    task_id: int | None = None,
    rating_service_willingness: int | None = None,   # 1-5
    rating_revisit_likelihood: int | None = None,     # 1-5
) -> dict:
    """
    创建备注。

    1. 验证评分范围（1-5 或 NULL），不合法则 422
    2. 确定 note type：
       - 若 task_id 关联的任务 task_type='follow_up_visit' → type='follow_up'
       - 否则 → type='normal'
    3. INSERT INTO biz.notes
    4. 若 type='follow_up'：
       - 触发 AI 应用 6 分析（P5 实现）
       - 若 ai_score >= 6 且关联任务 status='active' → 标记任务 completed
    5. 返回创建的备注记录

    注意：星星评分不参与回访完成判定，不参与 AI 分析，仅存储。
    """
    ...


async def get_notes(
    site_id: int, target_type: str, target_id: int
) -> list[dict]:
    """
    查询某目标的备注列表。

    按 created_at DESC 排序，包含星星评分和 AI 评分。
    """
    ...


async def delete_note(note_id: int, user_id: int, site_id: int) -> dict:
    """
    删除备注。

    验证备注归属后执行硬删除。
    """
    ...
```

### 组件 8：路由端点

#### 8.1 小程序任务路由（routers/xcx_tasks.py）

| 方法 | 路径 | 说明 | 认证要求 |
|------|------|------|---------|
| GET | `/api/xcx/tasks` | 获取任务列表 | JWT（approved） |
| POST | `/api/xcx/tasks/{id}/pin` | 置顶任务 | JWT（approved） |
| POST | `/api/xcx/tasks/{id}/unpin` | 取消置顶 | JWT（approved） |
| POST | `/api/xcx/tasks/{id}/abandon` | 放弃任务 | JWT（approved） |
| POST | `/api/xcx/tasks/{id}/cancel-abandon` | 取消放弃 | JWT（approved） |

#### 8.2 小程序备注路由（routers/xcx_notes.py）

| 方法 | 路径 | 说明 | 认证要求 |
|------|------|------|---------|
| POST | `/api/xcx/notes` | 创建备注 | JWT（approved） |
| GET | `/api/xcx/notes` | 查询备注列表（query: target_type, target_id） | JWT（approved） |
| DELETE | `/api/xcx/notes/{id}` | 删除备注 | JWT（approved） |

### 组件 9：Pydantic 模型

```python
# schemas/xcx_tasks.py
from pydantic import BaseModel, Field

class TaskListItem(BaseModel):
    id: int
    task_type: str
    status: str
    priority_score: float | None
    is_pinned: bool
    expires_at: str | None
    created_at: str
    # 客户信息（FDW 读取）
    member_id: int
    member_name: str | None
    member_phone: str | None
    # RS 指数 + 爱心 icon
    rs_score: float | None
    heart_icon: str  # 💖 / 🧡 / 💛 / 💙

class AbandonRequest(BaseModel):
    reason: str = Field(..., min_length=1, description="放弃原因（必填）")


# schemas/xcx_notes.py
from pydantic import BaseModel, Field

class NoteCreateRequest(BaseModel):
    target_type: str = Field(default="member")
    target_id: int
    content: str = Field(..., min_length=1)
    task_id: int | None = None
    rating_service_willingness: int | None = Field(None, ge=1, le=5)
    rating_revisit_likelihood: int | None = Field(None, ge=1, le=5)

class NoteOut(BaseModel):
    id: int
    type: str
    content: str
    rating_service_willingness: int | None
    rating_revisit_likelihood: int | None
    ai_score: int | None
    ai_analysis: str | None
    task_id: int | None
    created_at: str
    updated_at: str
```


## 数据模型

### ER 图

```mermaid
erDiagram
    coach_tasks {
        bigserial id PK
        bigint site_id "NOT NULL"
        bigint assistant_id "NOT NULL"
        bigint member_id "NOT NULL"
        varchar task_type "NOT NULL (4种)"
        varchar status "NOT NULL DEFAULT 'active'"
        numeric_5_2 priority_score "max(WBI,NCI) 快照"
        timestamptz expires_at "可空，有效期"
        boolean is_pinned "DEFAULT FALSE"
        text abandon_reason "可空"
        timestamptz completed_at "可空"
        varchar completed_task_type "可空，完成时类型快照"
        bigint parent_task_id "可空，FK → coach_tasks"
        timestamptz created_at "DEFAULT NOW()"
        timestamptz updated_at "DEFAULT NOW()"
    }

    coach_task_history {
        bigserial id PK
        bigint task_id "FK → coach_tasks"
        varchar action "NOT NULL"
        varchar old_status "可空"
        varchar new_status "可空"
        varchar old_task_type "可空"
        varchar new_task_type "可空"
        jsonb detail "可空"
        timestamptz created_at "DEFAULT NOW()"
    }

    notes {
        bigserial id PK
        bigint site_id "NOT NULL"
        integer user_id "NOT NULL"
        varchar target_type "NOT NULL"
        bigint target_id "NOT NULL"
        varchar type "NOT NULL DEFAULT 'normal'"
        text content "NOT NULL"
        smallint rating_service_willingness "可空 CHECK 1-5"
        smallint rating_revisit_likelihood "可空 CHECK 1-5"
        bigint task_id "可空 FK → coach_tasks"
        smallint ai_score "可空"
        text ai_analysis "可空"
        timestamptz created_at "DEFAULT NOW()"
        timestamptz updated_at "DEFAULT NOW()"
    }

    trigger_jobs {
        serial id PK
        varchar job_type "NOT NULL"
        varchar job_name "NOT NULL UNIQUE"
        varchar trigger_condition "NOT NULL (cron/interval/event)"
        jsonb trigger_config "NOT NULL"
        timestamptz last_run_at "可空"
        timestamptz next_run_at "可空"
        varchar status "NOT NULL DEFAULT 'enabled'"
        timestamptz created_at "DEFAULT NOW()"
    }

    coach_tasks ||--o{ coach_task_history : "变更记录"
    coach_tasks ||--o{ notes : "关联备注"
    coach_tasks ||--o| coach_tasks : "parent_task_id"
```

### 表 DDL

所有表在 `biz` Schema 下，迁移脚本位于 `db/zqyy_app/migrations/`。

#### biz.coach_tasks

```sql
CREATE TABLE IF NOT EXISTS biz.coach_tasks (
    id                  BIGSERIAL PRIMARY KEY,
    site_id             BIGINT NOT NULL,
    assistant_id        BIGINT NOT NULL,
    member_id           BIGINT NOT NULL,
    task_type           VARCHAR(50) NOT NULL,
    status              VARCHAR(20) NOT NULL DEFAULT 'active',
    priority_score      NUMERIC(5,2),
    expires_at          TIMESTAMPTZ,
    is_pinned           BOOLEAN DEFAULT FALSE,
    abandon_reason      TEXT,
    completed_at        TIMESTAMPTZ,
    completed_task_type VARCHAR(50),
    parent_task_id      BIGINT REFERENCES biz.coach_tasks(id),
    created_at          TIMESTAMPTZ DEFAULT NOW(),
    updated_at          TIMESTAMPTZ DEFAULT NOW()
);

-- 部分唯一索引：同一 (site_id, assistant_id, member_id, task_type) 下 active 任务最多一条
CREATE UNIQUE INDEX IF NOT EXISTS idx_coach_tasks_site_assistant_member_type
    ON biz.coach_tasks (site_id, assistant_id, member_id, task_type)
    WHERE status = 'active';

-- 助教任务列表查询索引
CREATE INDEX IF NOT EXISTS idx_coach_tasks_assistant_status
    ON biz.coach_tasks (site_id, assistant_id, status);
```

#### biz.coach_task_history

```sql
CREATE TABLE IF NOT EXISTS biz.coach_task_history (
    id              BIGSERIAL PRIMARY KEY,
    task_id         BIGINT NOT NULL REFERENCES biz.coach_tasks(id),
    action          VARCHAR(50) NOT NULL,
    old_status      VARCHAR(20),
    new_status      VARCHAR(20),
    old_task_type   VARCHAR(50),
    new_task_type   VARCHAR(50),
    detail          JSONB,
    created_at      TIMESTAMPTZ DEFAULT NOW()
);
```

#### biz.notes

```sql
CREATE TABLE IF NOT EXISTS biz.notes (
    id                          BIGSERIAL PRIMARY KEY,
    site_id                     BIGINT NOT NULL,
    user_id                     INTEGER NOT NULL,
    target_type                 VARCHAR(50) NOT NULL,
    target_id                   BIGINT NOT NULL,
    type                        VARCHAR(20) NOT NULL DEFAULT 'normal',
    content                     TEXT NOT NULL,
    rating_service_willingness  SMALLINT CHECK (rating_service_willingness BETWEEN 1 AND 5),
    rating_revisit_likelihood   SMALLINT CHECK (rating_revisit_likelihood BETWEEN 1 AND 5),
    task_id                     BIGINT REFERENCES biz.coach_tasks(id),
    ai_score                    SMALLINT,
    ai_analysis                 TEXT,
    created_at                  TIMESTAMPTZ DEFAULT NOW(),
    updated_at                  TIMESTAMPTZ DEFAULT NOW()
);

-- 按目标查询备注索引
CREATE INDEX IF NOT EXISTS idx_notes_target
    ON biz.notes (site_id, target_type, target_id);
```

#### biz.trigger_jobs

```sql
CREATE TABLE IF NOT EXISTS biz.trigger_jobs (
    id                  SERIAL PRIMARY KEY,
    job_type            VARCHAR(100) NOT NULL,
    job_name            VARCHAR(100) NOT NULL UNIQUE,
    trigger_condition   VARCHAR(20) NOT NULL,
    trigger_config      JSONB NOT NULL,
    last_run_at         TIMESTAMPTZ,
    next_run_at         TIMESTAMPTZ,
    status              VARCHAR(20) NOT NULL DEFAULT 'enabled',
    created_at          TIMESTAMPTZ DEFAULT NOW()
);
```

### 种子数据

```sql
-- 预置触发器配置
INSERT INTO biz.trigger_jobs (job_type, job_name, trigger_condition, trigger_config, next_run_at)
VALUES
    ('task_generator', 'task_generator', 'cron',
     '{"cron_expression": "0 4 * * *"}',
     (CURRENT_DATE + 1) + INTERVAL '4 hours'),

    ('task_expiry_check', 'task_expiry_check', 'interval',
     '{"interval_seconds": 3600}',
     NOW() + INTERVAL '1 hour'),

    ('recall_completion_check', 'recall_completion_check', 'event',
     '{"event_name": "etl_data_updated"}',
     NULL),

    ('note_reclassify_backfill', 'note_reclassify_backfill', 'event',
     '{"event_name": "recall_completed"}',
     NULL)
ON CONFLICT (job_name) DO NOTHING;
```

### 迁移脚本清单

| 序号 | 文件名 | 内容 |
|------|--------|------|
| 1 | `YYYY-MM-DD__p4_create_biz_tables.sql` | 创建 coach_tasks + coach_task_history + notes + trigger_jobs 表及索引 |
| 2 | `YYYY-MM-DD__p4_seed_trigger_jobs.sql` | 预置 4 条触发器配置种子数据 |

### FDW 数据依赖

任务生成器和召回检测器通过 `fdw_etl` Schema 读取以下外部表（P1 已建立映射）：

| 外部表 | 用途 |
|--------|------|
| `fdw_etl.v_dws_member_winback_index` | WBI 流失回赢指数 |
| `fdw_etl.v_dws_member_newconv_index` | NCI 新客转化指数 |
| `fdw_etl.v_dws_member_assistant_relation_index` | RS 关系强度指数 |
| `fdw_etl.v_dwd_assistant_service_log` | 助教服务记录（召回检测） |
| `fdw_etl.v_dim_member` | 客户基本信息（任务列表展示） |

**FDW 查询模式**：所有 FDW 查询通过业务库连接（`get_connection()`），在事务中 `SET LOCAL app.current_site_id = %s` 设置 RLS 隔离后查询 `fdw_etl.*` 外部表。


## 正确性属性（Correctness Properties）

*属性是系统在所有有效执行中都应保持为真的特征或行为——本质上是关于系统应该做什么的形式化陈述。属性是人类可读规格与机器可验证正确性保证之间的桥梁。*

### Property 1：任务类型确定正确性

*For any* 指数数据组合（WBI ≥ 0, NCI ≥ 0, RS ≥ 0），`determine_task_type()` 应满足：
- 当 `max(WBI, NCI) > 7` 时返回 `high_priority_recall`
- 当 `5 < max(WBI, NCI) ≤ 7` 时返回 `priority_recall`
- 当 `RS < 6` 且不满足上述条件时返回 `relationship_building`
- 且 `priority_score` 始终等于 `max(WBI, NCI)`

**Validates: Requirements 3.1, 3.2, 3.3, 3.5**

### Property 2：活跃任务唯一性不变量

*For any* `(site_id, assistant_id, member_id, task_type)` 组合，`biz.coach_tasks` 中 `status = 'active'` 的记录最多只有一条。任何创建或状态变更操作都不应违反此约束。

**Validates: Requirements 1.5, 3.6, 14.1**

### Property 3：任务类型变更状态机

*For any* 已存在 `(site_id, assistant_id, member_id)` 的 active 任务，当新任务类型与现有类型不同时，执行类型变更后：旧任务 `status` 变为 `inactive`，新任务 `status` 为 `active`，且 `coach_task_history` 中存在对应的变更记录（包含 `old_task_type` 和 `new_task_type`）。

**Validates: Requirements 3.7, 5.1, 5.4, 14.2**

### Property 4：48 小时滞留机制

*For any* `follow_up_visit` 类型任务：
- 生成时 `expires_at` 为 NULL，`status` 为 `active`
- 当触发条件不再满足时，`expires_at` 被填充为 `created_at + 48 小时`
- 当 `expires_at` 不为 NULL 且当前时间超过 `expires_at` 时，轮询后 `status` 变为 `inactive`
- 当新 `follow_up_visit` 任务顶替有 `expires_at` 的旧任务时，旧任务变为 `inactive`，新任务 `expires_at` 为 NULL

**Validates: Requirements 4.1, 4.2, 4.3, 4.4, 14.3**

### Property 5：放弃与取消放弃往返

*For any* `status = 'active'` 的任务和非空的放弃原因字符串：
- 放弃操作后 `status = 'abandoned'` 且 `abandon_reason` 等于提供的原因
- 取消放弃后 `status = 'active'` 且 `abandon_reason` 为空
- 放弃时若 `abandon_reason` 为空字符串或纯空白，应返回 422 错误且任务状态不变

**Validates: Requirements 8.4, 8.6, 8.7, 14.4**

### Property 6：召回完成检测与类型快照

*For any* `(site_id, assistant_id, member_id)` 组合，当检测到助教为该客户提供了服务时，所有匹配的 `status = 'active'` 任务应变为 `completed`，且 `completed_at` 记录服务时间，`completed_task_type` 记录完成时的 `task_type` 值（快照不变量）。

**Validates: Requirements 6.2, 6.3, 14.6**

### Property 7：备注回溯重分类

*For any* 召回完成事件（包含 site_id, assistant_id, member_id, service_time），若在 service_time 之后存在该客户的 `type = 'normal'` 备注，回溯操作后该备注的 `type` 应变为 `follow_up`。若不存在符合条件的备注，则不做任何修改。

**Validates: Requirements 7.1, 7.2, 14.7**

### Property 8：备注类型自动设置

*For any* 备注创建操作：
- 若关联的 `task_id` 对应任务的 `task_type = 'follow_up_visit'`，则备注 `type` 应为 `follow_up`
- 若关联的 `task_id` 对应任务的 `task_type` 不是 `follow_up_visit`，则备注 `type` 应为 `normal`
- 若未关联 `task_id`，则备注 `type` 应为 `normal`

**Validates: Requirements 9.2, 9.3**

### Property 9：星星评分范围约束

*For any* 备注创建操作，`rating_service_willingness` 和 `rating_revisit_likelihood` 的值应为 NULL 或在 1-5 范围内。范围外的值应被拒绝（422 错误），且备注不被创建。

**Validates: Requirements 9.8, 14.5**

### Property 10：任务列表排序正确性

*For any* 助教的活跃任务列表，返回结果应按 `is_pinned DESC, priority_score DESC, created_at ASC` 排序。即：置顶任务在前，同置顶状态下高优先级在前，同优先级下先创建的在前。

**Validates: Requirements 8.1**

### Property 11：爱心 icon 档位计算

*For any* RS 指数值，爱心 icon 应满足：
- RS > 8.5 → 💖
- 7 < RS ≤ 8.5 → 🧡
- 5 < RS ≤ 7 → 💛
- RS ≤ 5 → 💙

**Validates: Requirements 8.2**

### Property 12：触发器 next_run_at 计算

*For any* cron 类型触发器配置和当前时间，计算的 `next_run_at` 应大于当前时间。*For any* interval 类型触发器配置（interval_seconds > 0），计算的 `next_run_at` 应等于当前时间加上 interval_seconds 秒。

**Validates: Requirements 10.1, 10.2**

### Property 13：迁移脚本幂等性

*For any* 本次新增的迁移脚本（DDL + 种子数据），连续执行两次的结果应与执行一次相同——第二次执行不应产生错误，且数据库状态不变。

**Validates: Requirements 1.8, 2.5, 11.4, 11.5**

### Property 14：AI 评分驱动的任务完成判定

*For any* `type = 'follow_up'` 的备注和关联的 `follow_up_visit` 任务：
- 当 AI 应用 6 返回评分 ≥ 6 且任务 `status = 'active'` 时，任务应标记为 `completed`
- 当 AI 应用 6 返回评分 < 6 时，任务 `status` 保持 `active`

**Validates: Requirements 7.4, 7.5, 9.5**

### Property 15：状态变更历史完整性

*For any* 任务状态变更操作（类型变更、放弃、取消放弃、完成、过期），`coach_task_history` 中应存在对应记录，包含正确的 `action`、`old_status`、`new_status` 字段。

**Validates: Requirements 5.4, 8.3**


## 错误处理

### API 错误码规范

| HTTP 状态码 | 场景 | 响应体 |
|------------|------|--------|
| 401 | JWT 无效/过期 | `{"detail": "无效的令牌"}` |
| 403 | 用户未 approved、操作非自己的任务 | `{"detail": "权限不足"}` |
| 404 | 任务/备注不存在 | `{"detail": "资源不存在"}` |
| 409 | 任务状态不允许操作（如对 inactive 任务置顶） | `{"detail": "任务状态不允许此操作"}` |
| 422 | 放弃原因为空、星星评分超范围、请求体校验失败 | Pydantic 标准错误格式 |
| 500 | 数据库连接失败、FDW 查询异常 | `{"detail": "服务器内部错误"}` |

### 后台任务错误处理

| 场景 | 处理方式 |
|------|---------|
| FDW 查询失败（ETL 库不可用） | 记录错误日志，跳过本次执行，不影响其他触发器 |
| 单个任务生成失败 | 捕获异常，记录日志，继续处理下一个助教-客户对 |
| 触发器执行异常 | 记录错误日志，不中断其他触发器执行（需求 11.7） |
| 数据库写入冲突（唯一索引） | 捕获 `UniqueViolation`，视为"已存在"跳过 |
| AI 应用 6 接口不可用 | 跳过评分逻辑，备注正常创建，ai_score 保持 NULL |
| expires_at 计算溢出 | 防御性检查，确保 expires_at > created_at |

### 数据库事务策略

| 操作 | 事务范围 |
|------|---------|
| 任务生成器（批量） | 每个助教-客户对独立事务，失败不影响其他 |
| 任务状态变更 | 单任务事务：UPDATE task + INSERT history 在同一事务 |
| 备注创建 + 任务完成 | 同一事务：INSERT note + UPDATE task（如触发完成） |
| 触发器调度 | 每个 job 独立事务 |
| FDW 查询 | 只读事务，`SET LOCAL app.current_site_id` 设置 RLS |

### 环境变量缺失处理

| 变量 | 缺失时行为 |
|------|-----------|
| `DB_HOST` 等数据库参数 | 数据库连接失败，返回 500 |
| `ETL_DB_HOST` 等 ETL 参数 | FDW 查询失败，任务生成器/召回检测器跳过，记录日志 |
| `JWT_SECRET_KEY` | JWT 签发/验证使用空密钥（仅开发环境） |

## 测试策略

### 属性测试（Property-Based Testing）

使用 Python `hypothesis` 框架，测试目录：`tests/`（Monorepo 级属性测试目录）。

每个属性测试至少运行 100 次迭代。每个测试用注释标注对应的设计属性编号。

标注格式：`# Feature: 04-miniapp-core-business, Property N: <属性标题>`

**核心纯函数（可直接属性测试，不依赖数据库）**：

| 属性 | 测试文件 | 测试方法 | 生成器 |
|------|---------|---------|--------|
| P1 任务类型确定 | `tests/test_core_business_properties.py` | 生成随机 WBI/NCI/RS 值，验证 `determine_task_type()` 返回值 | `hypothesis.strategies.decimals(min_value=0, max_value=10, places=2)` |
| P9 评分范围约束 | `tests/test_core_business_properties.py` | 生成随机整数，验证 Pydantic 模型校验 | `hypothesis.strategies.integers(min_value=-100, max_value=100)` |
| P10 任务列表排序 | `tests/test_core_business_properties.py` | 生成随机任务列表（不同 is_pinned/priority_score/created_at），验证排序 | 自定义 strategy 生成任务列表 |
| P11 爱心 icon 档位 | `tests/test_core_business_properties.py` | 生成随机 RS 值，验证 icon 映射 | `hypothesis.strategies.decimals(min_value=0, max_value=10, places=1)` |
| P12 next_run_at 计算 | `tests/test_core_business_properties.py` | 生成随机 cron/interval 配置和当前时间，验证计算结果 | 自定义 strategy 生成触发器配置 |

**状态机属性（需要 FakeDB 模拟数据库状态）**：

| 属性 | 测试文件 | 测试方法 | 生成器 |
|------|---------|---------|--------|
| P2 唯一性不变量 | `tests/test_core_business_properties.py` | 生成随机 (site_id, assistant_id, member_id, task_type) 组合，模拟插入，验证唯一性 | 自定义 strategy 生成任务组合 |
| P3 类型变更状态机 | `tests/test_core_business_properties.py` | 生成随机现有任务+新任务类型，执行变更，验证旧任务 inactive + 新任务 active + history | 自定义 strategy |
| P4 48h 滞留机制 | `tests/test_core_business_properties.py` | 生成随机 follow_up_visit 任务+时间偏移，验证 expires_at 填充和过期逻辑 | `hypothesis.strategies.datetimes` + `timedeltas` |
| P5 放弃往返 | `tests/test_core_business_properties.py` | 生成随机任务+放弃原因，执行放弃→取消放弃，验证状态恢复 | `hypothesis.strategies.text(min_size=1)` |
| P6 召回完成+快照 | `tests/test_core_business_properties.py` | 生成随机 active 任务+服务记录，执行完成检测，验证 completed_task_type | 自定义 strategy |
| P7 备注回溯 | `tests/test_core_business_properties.py` | 生成随机备注列表+service_time，执行回溯，验证 type 变更 | 自定义 strategy |
| P8 备注类型自动设置 | `tests/test_core_business_properties.py` | 生成随机 task_type + 备注创建，验证 note.type | `hypothesis.strategies.sampled_from(task_types)` |
| P14 AI 评分判定 | `tests/test_core_business_properties.py` | 生成随机 ai_score + 任务状态，验证完成判定 | `hypothesis.strategies.integers(min_value=0, max_value=10)` |
| P15 历史完整性 | `tests/test_core_business_properties.py` | 生成随机状态变更操作序列，验证 history 记录数量和内容 | 自定义 strategy |

**注意**：P13（迁移幂等性）作为集成测试在测试库中执行，不使用 hypothesis。

### 单元测试

单元测试位于 `apps/backend/tests/`，聚焦于：

- `test_task_generator.py`：任务生成器的边界情况（无指数数据、全部跳过、全部替换）
- `test_task_manager.py`：任务 CRUD 的边界情况（操作非自己的任务、状态不允许的操作）
- `test_note_service.py`：备注服务的边界情况（无关联任务、AI 接口不可用）
- `test_task_expiry.py`：有效期轮询的边界情况（无过期任务、批量过期）
- `test_recall_detector.py`：召回检测的边界情况（无匹配任务、多任务匹配）
- `test_note_reclassifier.py`：备注回溯的边界情况（无符合条件备注、多条备注取第一条）
- `test_trigger_scheduler.py`：触发器调度的边界情况（disabled job、未注册 handler）

### 集成测试

集成测试通过以下方式验证：

1. **迁移脚本幂等性**：在 `test_zqyy_app` 中连续执行两次迁移脚本，验证无错误
2. **种子数据完整性**：验证 4 条触发器配置正确插入
3. **端到端流程**：任务生成 → 任务列表 → 置顶/放弃 → 备注创建 → 召回完成 → 备注回溯

### 测试配置

- 属性测试：`cd C:\Project\NeoZQYY && pytest tests/test_core_business_properties.py -v`
- 后端单元测试：`cd apps/backend && pytest tests/ -v`
- 每个属性测试标注 `@settings(max_examples=200)`
- 每个属性测试注释引用设计文档 Property 编号
- 属性测试库：`hypothesis`（已在项目依赖中）
- 数据库测试使用 `test_zqyy_app`（禁止连正式库）
