Files

marsal 758ed73f9d fix: 渠道账号群聊同步缓存 key 及成员名称兜底；统一 Docker/Maven 项目约定

2026-07-10 15:55:30 +08:00

64 KiB

Raw Blame History

channel-service 与 mci-server 双向集成 Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 在 channel-service（Python）中实现本地 SQLite 持久化的事件上报道具，并新增 mci-server（Java）的 /api/v1/channel-account/event 接收端点，完成心跳、登录/在线状态、发送结果的双向 HTTP 集成。

Architecture: channel-service 所有状态变更与发送操作先写本地 outbound_events（SQLite），再由单线程后台 worker 按指数退避推送到 mci-server；mci-server 用 event_id 幂等并更新 mci_channel_account / mci_send_record。

Tech Stack: Python 3.11 + FastAPI + SQLite + httpx；Java 21 + Spring Boot 4.1.x + MyBatis Plus + MySQL 8。

0. 文件结构总览

channel-service（Python）

文件	操作	说明
`channel-service/app/models/event.py`	新建	事件枚举、数据类、payload schema
`channel-service/app/repositories/event_store.py`	新建	SQLite `outbound_events` 表及 CRUD
`channel-service/app/clients/mci_server_client.py`	新建	HTTP 客户端，POST 事件到 mci-server
`channel-service/app/services/event_publisher.py`	新建	单线程后台 worker、重试、死信
`channel-service/app/config.py`	修改	新增事件存储/重试/清理配置
`channel-service/app/clients/pywechat_client.py`	修改	增加登录/在线状态变更辅助方法
`channel-service/app/services/heartbeat_service.py`	修改	心跳时写事件
`channel-service/app/services/login_service.py`	修改	登录流程写事件
`channel-service/app/services/send_service.py`	修改	发送改为异步受理并写 send_result 事件
`channel-service/app/api/wechat_personal.py`	修改	send 请求增加 `record_id`，新增 expire 路由
`channel-service/app/api/monitor.py`	修改	heartbeat 路由透传 `tenant_code`
`channel-service/app/main.py`	修改	启动/停止 EventPublisher worker
`channel-service/tests/test_event_store.py`	新建	事件存储测试
`channel-service/tests/test_event_publisher.py`	新建	worker 重试/死信测试
`channel-service/tests/test_send_async.py`	新建	发送异步化测试

mci-server（Java）

文件	操作	说明
`backend/mci-server/src/main/java/com/sino/mci/channel/entity/ChannelEventRecord.java`	新建	幂等去重表实体
`backend/mci-server/src/main/java/com/sino/mci/channel/repository/ChannelEventRecordMapper.java`	新建	MyBatis Plus Mapper
`backend/mci-server/src/main/java/com/sino/mci/channel/dto/ChannelAccountEventRequest.java`	新建	事件接收 DTO
`backend/mci-server/src/main/java/com/sino/mci/channel/service/ChannelAccountEventService.java`	新建	事件处理逻辑
`backend/mci-server/src/main/java/com/sino/mci/channel/controller/ChannelAccountEventController.java`	新建	接收端点
`backend/mci-server/src/main/java/com/sino/mci/send/service/SendService.java`	修改	dispatchRecord 时把 `record_id` 放入 extra
`backend/mci-channel-wechat-personal/.../WechatPersonalChannelAdapter.java`	修改	把 `record_id` 传给 channel-service
`backend/mci-server/src/main/resources/db/migration/V22__add_channel_event_record.sql`	新建	幂等表 DDL
`backend/mci-server/src/test/java/com/sino/mci/channel/controller/ChannelAccountEventControllerTest.java`	新建	控制器测试
`backend/mci-channel-wechat-personal/.../WechatPersonalChannelAdapterTest.java`	修改	验证 record_id 透传

Phase 1: channel-service 事件基础设施

Task 1: 创建事件模型 `app/models/event.py`

Files:

Create: channel-service/app/models/event.py
Test: channel-service/tests/test_event_store.py（后续步骤编写）
Step 1: 编写事件枚举与数据类

from __future__ import annotations

import uuid
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Optional


class EventType(str, Enum):
    HEARTBEAT = "heartbeat"
    LOGIN_STATUS_CHANGED = "login_status_changed"
    ONLINE_STATUS_CHANGED = "online_status_changed"
    SEND_RESULT = "send_result"
    ERROR = "error"


class EventStatus(str, Enum):
    PENDING = "pending"
    RETRYING = "retrying"
    DELIVERED = "delivered"
    DEAD_LETTER = "dead_letter"


@dataclass
class OutboundEvent:
    event_type: EventType
    account_id: str
    tenant_code: str
    data: dict[str, Any]
    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    def to_payload(self) -> dict[str, Any]:
        return {
            "event_id": self.event_id,
            "event_type": self.event_type.value,
            "account_id": self.account_id,
            "tenant_code": self.tenant_code,
            "timestamp": self.timestamp.isoformat(),
            "data": self.data,
        }

Step 2: Commit

cd /Users/marsal/Projects/sino-project/sino-mci/channel-service
git add app/models/event.py
git commit -m "feat(event): add outbound event models"

Task 2: 创建 SQLite 事件存储 `app/repositories/event_store.py`

Files:

Create: channel-service/app/repositories/event_store.py
Test: channel-service/tests/test_event_store.py
Step 1: 编写测试 — 验证建表与保存事件

import os
import tempfile

import pytest

from app.models.event import EventType, OutboundEvent
from app.repositories.event_store import EventStore


@pytest.fixture
def store():
    fd, path = tempfile.mkstemp(suffix=".db")
    os.close(fd)
    store = EventStore(path)
    yield store
    os.unlink(path)


def test_save_and_fetch_pending(store):
    event = OutboundEvent(
        event_type=EventType.HEARTBEAT,
        account_id="bot-001",
        tenant_code="rescue",
        data={"online_status": "online"},
    )
    store.save(event)
    pending = store.fetch_pending(limit=10)
    assert len(pending) == 1
    assert pending[0].event_id == event.event_id


def test_mark_delivered(store):
    event = OutboundEvent(
        event_type=EventType.HEARTBEAT,
        account_id="bot-001",
        tenant_code="rescue",
        data={},
    )
    store.save(event)
    store.mark_delivered(event.event_id)
    pending = store.fetch_pending(limit=10)
    assert len(pending) == 0

Step 2: 运行测试，确认失败

cd /Users/marsal/Projects/sino-project/sino-mci/channel-service
pytest tests/test_event_store.py -v

Expected: ModuleNotFoundError 或 ImportError。

Step 3: 实现 EventStore

from __future__ import annotations

import json
import sqlite3
import threading
from datetime import datetime, timedelta, timezone
from typing import Optional

from app.models.event import EventStatus, EventType, OutboundEvent


class EventStore:
    def __init__(self, db_path: str):
        self.db_path = db_path
        self._local = threading.local()
        self._init_db()

    def _now(self) -> datetime:
        return datetime.now(timezone.utc).replace(tzinfo=None)

    def _connection(self) -> sqlite3.Connection:
        if not hasattr(self._local, "conn") or self._local.conn is None:
            self._local.conn = sqlite3.connect(self.db_path, check_same_thread=False)
            self._local.conn.row_factory = sqlite3.Row
        return self._local.conn

    def _init_db(self) -> None:
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS outbound_events (
                    id TEXT PRIMARY KEY,
                    event_type TEXT NOT NULL,
                    payload TEXT NOT NULL,
                    status TEXT NOT NULL DEFAULT 'pending',
                    attempt_count INTEGER NOT NULL DEFAULT 0,
                    next_retry_at TIMESTAMP NOT NULL,
                    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
                    last_error TEXT
                )
                """
            )
            conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_status_next_retry ON outbound_events(status, next_retry_at)"
            )
            conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_created_at ON outbound_events(created_at)"
            )

    def save(self, event: OutboundEvent) -> None:
        now = self._now()
        with self._connection() as conn:
            conn.execute(
                """
                INSERT INTO outbound_events
                (id, event_type, payload, status, attempt_count, next_retry_at, created_at, updated_at, last_error)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    event.event_id,
                    event.event_type.value,
                    json.dumps(event.to_payload(), default=str),
                    EventStatus.PENDING.value,
                    0,
                    now,
                    now,
                    now,
                    None,
                ),
            )

    def fetch_pending(self, limit: int = 100) -> list[OutboundEvent]:
        now = self._now()
        cursor = self._connection().execute(
            """
            SELECT id, event_type, payload, status, attempt_count
            FROM outbound_events
            WHERE status IN ('pending', 'retrying') AND next_retry_at <= ?
            ORDER BY next_retry_at ASC
            LIMIT ?
            """,
            (now, limit),
        )
        return [self._row_to_event(row) for row in cursor.fetchall()]

    def _row_to_event(self, row: sqlite3.Row) -> OutboundEvent:
        payload = json.loads(row["payload"])
        return OutboundEvent(
            event_id=payload["event_id"],
            event_type=EventType(payload["event_type"]),
            account_id=payload["account_id"],
            tenant_code=payload["tenant_code"],
            data=payload["data"],
            status=EventStatus(row["status"]),
            attempt_count=row["attempt_count"],
        )

    def mark_delivered(self, event_id: str) -> None:
        now = self._now()
        with self._connection() as conn:
            conn.execute(
                "UPDATE outbound_events SET status = 'delivered', updated_at = ? WHERE id = ?",
                (now, event_id),
            )

    def mark_retry(self, event_id: str, status: EventStatus, attempt_count: int, next_retry_at: datetime, last_error: Optional[str]) -> None:
        now = self._now()
        with self._connection() as conn:
            conn.execute(
                """
                UPDATE outbound_events
                SET status = ?, attempt_count = ?, next_retry_at = ?, updated_at = ?, last_error = ?
                WHERE id = ?
                """,
                (status.value, attempt_count, next_retry_at, now, last_error, event_id),
            )

    def cleanup(self, retention_days: int) -> int:
        threshold = self._now() - timedelta(days=retention_days)
        with self._connection() as conn:
            cursor = conn.execute(
                """
                DELETE FROM outbound_events
                WHERE status IN ('delivered', 'dead_letter') AND created_at < ?
                """,
                (threshold,),
            )
            return cursor.rowcount

Step 4: 运行测试，确认通过

pytest tests/test_event_store.py -v

Expected: PASS。

Step 5: Commit

git add app/repositories/event_store.py tests/test_event_store.py
git commit -m "feat(event): add SQLite outbound event store"

Task 3: 创建 mci-server HTTP 客户端 `app/clients/mci_server_client.py`

Files:

Create: channel-service/app/clients/mci_server_client.py
Test: channel-service/tests/test_event_publisher.py（后续步骤）
Step 1: 实现客户端

from __future__ import annotations

import httpx

from app.models.event import OutboundEvent


class MciServerClient:
    def __init__(self, base_url: str, timeout: float = 10.0):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout

    async def post_event(self, event: OutboundEvent) -> None:
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.post(
                f"{self.base_url}/api/v1/channel-account/event",
                json=event.to_payload(),
            )
            response.raise_for_status()

Step 2: Commit

git add app/clients/mci_server_client.py
git commit -m "feat(client): add mci-server event poster"

Task 4: 创建单线程事件发布 worker `app/services/event_publisher.py`

Files:

Create: channel-service/app/services/event_publisher.py
Test: channel-service/tests/test_event_publisher.py
Step 1: 安装测试依赖 respx

在 channel-service/requirements.txt 中追加：

respx==0.22.0

然后安装：

cd /Users/marsal/Projects/sino-project/sino-mci/channel-service
pip install -r requirements.txt

Step 2: 编写测试 — 验证成功推送与重试

import asyncio
from datetime import datetime, timezone

import httpx
import pytest
import respx
from httpx import Response

from app.clients.mci_server_client import MciServerClient
from app.models.event import EventType, OutboundEvent
from app.repositories.event_store import EventStore
from app.services.event_publisher import EventPublisher


@pytest.fixture
def publisher(tmp_path):
    db_path = tmp_path / "events.db"
    store = EventStore(str(db_path))
    client = MciServerClient("http://mock-mci-server/msg-platform")
    pub = EventPublisher(store, client, max_retry=2, base_delay_seconds=1, interval_seconds=0.1)
    yield pub


@respx.mock
def test_publish_delivers_event(publisher):
    route = respx.post("http://mock-mci-server/msg-platform/api/v1/channel-account/event").mock(return_value=Response(200))
    event = OutboundEvent(EventType.HEARTBEAT, "bot-001", "rescue", {"online_status": "online"})
    publisher.store.save(event)

    async def run():
        await publisher._tick()

    asyncio.run(run())
    assert route.called
    pending = publisher.store.fetch_pending(limit=10)
    assert len(pending) == 0


@respx.mock
def test_publish_retries_then_dead_letter(publisher):
    route = respx.post("http://mock-mci-server/msg-platform/api/v1/channel-account/event").mock(return_value=Response(500))
    event = OutboundEvent(EventType.HEARTBEAT, "bot-001", "rescue", {})
    publisher.store.save(event)

    async def run():
        await publisher._tick()

    asyncio.run(run())
    assert route.call_count == 1
    updated = publisher.store.fetch_pending(limit=10)
    assert len(updated) == 1
    assert updated[0].status == EventStatus.RETRYING

注意：EventStore._row_to_event 目前未保留 status，需要给 OutboundEvent 增加 status 字段或在 store 返回元组。为保持简单，可以在 fetch_pending 返回 (OutboundEvent, status, attempt_count) 元组，或在 OutboundEvent 上增加 status 与 attempt_count 字段。下面选择给 OutboundEvent 增加字段。

Step 2: 运行测试，确认失败

pytest tests/test_event_publisher.py -v

Expected: ImportError / AttributeError。

Step 3: 修改 app/models/event.py，给 OutboundEvent 增加状态字段

@dataclass
class OutboundEvent:
    event_type: EventType
    account_id: str
    tenant_code: str
    data: dict[str, Any]
    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    status: EventStatus = EventStatus.PENDING
    attempt_count: int = 0

Step 4: 修改 app/repositories/event_store.py 的 _row_to_event，回填 status 与 attempt_count

    def _row_to_event(self, row: sqlite3.Row) -> OutboundEvent:
        payload = json.loads(row["payload"])
        return OutboundEvent(
            event_id=payload["event_id"],
            event_type=EventType(payload["event_type"]),
            account_id=payload["account_id"],
            tenant_code=payload["tenant_code"],
            data=payload["data"],
            status=EventStatus(row["status"]),
            attempt_count=row["attempt_count"],
        )

Step 5: 实现 EventPublisher

from __future__ import annotations

import asyncio
import math
from datetime import datetime, timedelta, timezone
from typing import Optional

from app.clients.mci_server_client import MciServerClient
from app.models.event import EventStatus, OutboundEvent
from app.repositories.event_store import EventStore


class EventPublisher:
    def __init__(
        self,
        store: EventStore,
        client: MciServerClient,
        max_retry: int = 5,
        base_delay_seconds: int = 5,
        interval_seconds: float = 5.0,
        batch_size: int = 100,
    ):
        self.store = store
        self.client = client
        self.max_retry = min(max_retry, 10)
        self.base_delay_seconds = base_delay_seconds
        self.interval_seconds = interval_seconds
        self.batch_size = batch_size
        self._task: Optional[asyncio.Task] = None
        self._stop_event = asyncio.Event()

    def start(self) -> None:
        self._task = asyncio.create_task(self._loop())

    async def stop(self) -> None:
        self._stop_event.set()
        if self._task:
            try:
                await asyncio.wait_for(self._task, timeout=5.0)
            except asyncio.TimeoutError:
                self._task.cancel()

    async def _loop(self) -> None:
        while not self._stop_event.is_set():
            try:
                await self._tick()
            except Exception:
                pass
            try:
                await asyncio.wait_for(self._stop_event.wait(), timeout=self.interval_seconds)
            except asyncio.TimeoutError:
                pass

    async def _tick(self) -> None:
        events = self.store.fetch_pending(limit=self.batch_size)
        for event in events:
            try:
                await self.client.post_event(event)
                self.store.mark_delivered(event.event_id)
            except Exception as e:
                next_attempt = event.attempt_count + 1
                delay = self.base_delay_seconds * (2 ** (next_attempt - 1))
                next_retry_at = datetime.now(timezone.utc).replace(tzinfo=None) + timedelta(seconds=delay)
                status = EventStatus.DEAD_LETTER if next_attempt >= self.max_retry else EventStatus.RETRYING
                self.store.mark_retry(event.event_id, status, next_attempt, next_retry_at, str(e)[:500])

    async def cleanup_old_events(self, retention_days: int) -> int:
        return self.store.cleanup(retention_days)

Step 6: 运行测试，确认通过

pytest tests/test_event_publisher.py -v

Expected: PASS。

Step 7: Commit

git add app/models/event.py app/repositories/event_store.py app/services/event_publisher.py tests/test_event_publisher.py
git commit -m "feat(event): add single-thread event publisher with retry"

Task 5: 扩展配置 `app/config.py`

Files:

Modify: channel-service/app/config.py
Step 1: 添加配置项

class Settings(BaseSettings):
    app_name: str = "sino-mci-channel-service"
    debug: bool = os.getenv("DEBUG", "false").lower() == "true"
    host: str = "0.0.0.0"
    port: int = int(os.getenv("PORT", "8000"))
    mci_server_url: str = os.getenv("MCI_SERVER_URL", "http://localhost:8080/msg-platform")
    heartbeat_interval_seconds: int = int(os.getenv("HEARTBEAT_INTERVAL_SECONDS", "30"))
    pywechat_mock_mode: bool = os.getenv("PYWECHAT_MOCK_MODE", "true").lower() == "true"

    event_store_path: str = os.getenv("EVENT_STORE_PATH", "./data/events.db")
    max_retry: int = int(os.getenv("MAX_RETRY", "5"))
    base_retry_delay_seconds: int = int(os.getenv("BASE_RETRY_DELAY_SECONDS", "5"))
    event_retention_days: int = int(os.getenv("EVENT_RETENTION_DAYS", "10"))
    event_publish_interval_seconds: float = float(os.getenv("EVENT_PUBLISH_INTERVAL_SECONDS", "5"))

Step 2: Commit

git add app/config.py
git commit -m "feat(config): add event publisher settings"

Phase 2: channel-service 业务服务接入事件

Task 6: 改造心跳服务，发送 heartbeat 事件

Files:

Modify: channel-service/app/services/heartbeat_service.py
Modify: channel-service/app/api/monitor.py
Test: channel-service/tests/test_api.py
Step 1: 修改 HeartbeatRequest，增加 tenant_code

在 app/api/monitor.py 中：

class HeartbeatRequest(BaseModel):
    account_id: str = Field(..., description="渠道账号唯一标识")
    account_name: str | None = Field(None)
    tenant_code: str = Field("default", description="业务系统租户编码")
    risk_level: int = Field(0, ge=0, le=100)
    sent_count_today: int = Field(0, ge=0)

Step 2: 修改 HeartbeatService，注入 EventStore 并写事件

from app.models.event import EventType, OutboundEvent
from app.repositories.event_store import EventStore


class HeartbeatService:
    def __init__(self, registry: ClientRegistry, store: EventStore):
        self.registry = registry
        self.store = store

    def heartbeat(self, account_id: str, payload: dict) -> dict:
        client = self.registry.get_or_create(account_id, payload.get("account_name"))
        client.heartbeat()
        if "risk_level" in payload:
            client.state.risk_level = payload["risk_level"]
        if "sent_count_today" in payload:
            client.state.sent_count_today = payload["sent_count_today"]

        event = OutboundEvent(
            event_type=EventType.HEARTBEAT,
            account_id=account_id,
            tenant_code=payload.get("tenant_code", "default"),
            data={
                "login_status": client.state.login_status.value,
                "online_status": client.state.online_status.value,
                "risk_level": client.state.risk_level,
                "sent_count_today": client.state.sent_count_today,
            },
        )
        self.store.save(event)
        return {"success": True, "account_id": account_id}

Step 3: 修改 monitor.py heartbeat 路由，传入 tenant_code

@router.post("/v1/heartbeat")
def heartbeat(request: HeartbeatRequest):
    return heartbeat_service.heartbeat(request.account_id, request.model_dump(exclude_none=True))

Step 4: 修改 app/services/__init__.py 或 app/main.py 中初始化方式，注入 EventStore 单例

先创建全局 event_store：

# app/repositories/event_store.py 底部追加
event_store = EventStore(settings.event_store_path)

然后修改 app/services/heartbeat_service.py 底部：

from app.repositories.event_store import event_store
heartbeat_service = HeartbeatService(registry, event_store)

Step 5: 更新测试 tests/test_api.py 中的 heartbeat 测试，验证事件写入

def test_heartbeat_creates_event():
    from app.repositories.event_store import event_store
    event_store.cleanup(0)
    client.post("/v1/heartbeat", json={
        "account_id": "bot-heartbeat",
        "tenant_code": "rescue",
        "risk_level": 10,
        "sent_count_today": 5,
    })
    pending = event_store.fetch_pending(limit=10)
    assert any(e.event_type == EventType.HEARTBEAT and e.tenant_code == "rescue" for e in pending)

Step 6: 运行 channel-service 全部测试

pytest -v

Expected: PASS（可能需要调整旧测试）。

Step 7: Commit

git add app/services/heartbeat_service.py app/api/monitor.py app/repositories/event_store.py tests/test_api.py
git commit -m "feat(heartbeat): emit heartbeat event to SQLite"

Task 7: 改造登录服务，发送 login/online 状态事件

Files:

Modify: channel-service/app/services/login_service.py
Modify: channel-service/app/clients/pywechat_client.py
Test: channel-service/tests/test_api.py
Step 1: 给 PyWeChatClient 增加状态变更辅助方法

在 app/clients/pywechat_client.py 中：

    def set_login_status(self, status: LoginStatus, error_message: Optional[str] = None) -> None:
        old = self.state.login_status
        self.state.login_status = status
        if status == LoginStatus.SUCCESS:
            self.state.online_status = OnlineStatus.ONLINE
            self.state.last_heartbeat_at = time.time()
        elif status in (LoginStatus.EXPIRED, LoginStatus.FAILED):
            self.state.online_status = OnlineStatus.OFFLINE
        if error_message:
            self.state.error_message = error_message

    def set_online_status(self, status: OnlineStatus) -> None:
        self.state.online_status = status

Step 2: 修改 LoginService，注入 EventStore 并写事件

from app.models.event import EventType, OutboundEvent
from app.repositories.event_store import EventStore


class LoginService:
    def __init__(self, registry: ClientRegistry, store: EventStore):
        self.registry = registry
        self.store = store

    def start_login(self, account_id: str, account_name: Optional[str] = None, tenant_code: str = "default") -> dict:
        client = self.registry.get_or_create(account_id, account_name)
        login_id = client.start_login()
        self._emit_login_changed(account_id, tenant_code, LoginStatus.PENDING, client.state.login_status, client.state.qr_code_url)
        return {
            "login_id": login_id,
            "qr_code_url": client.state.qr_code_url,
            "status": client.state.login_status.value,
        }

    def confirm_login(self, account_id: str, tenant_code: str = "default") -> dict:
        client = self.registry.get(account_id)
        if client is None:
            return {"success": False, "error": "account not found"}
        previous_login = client.state.login_status
        previous_online = client.state.online_status
        client.set_login_status(LoginStatus.SUCCESS)
        self._emit_login_changed(account_id, tenant_code, previous_login, client.state.login_status)
        self._emit_online_changed(account_id, tenant_code, previous_online, client.state.online_status)
        return {"success": True, "status": client.state.login_status.value}

    def expire_login(self, account_id: str, tenant_code: str = "default") -> dict:
        client = self.registry.get(account_id)
        if client is None:
            return {"success": False, "error": "account not found"}
        previous = client.state.login_status
        client.set_login_status(LoginStatus.EXPIRED)
        self._emit_login_changed(account_id, tenant_code, previous, client.state.login_status)
        return {"success": True, "status": client.state.login_status.value}

    def _emit_login_changed(self, account_id: str, tenant_code: str, previous: LoginStatus, current: LoginStatus, qr_code_url: Optional[str] = None) -> None:
        event = OutboundEvent(
            event_type=EventType.LOGIN_STATUS_CHANGED,
            account_id=account_id,
            tenant_code=tenant_code,
            data={
                "previous": previous.value,
                "current": current.value,
                "qr_code_url": qr_code_url,
            },
        )
        self.store.save(event)

    def _emit_online_changed(self, account_id: str, tenant_code: str, previous: OnlineStatus, current: OnlineStatus) -> None:
        event = OutboundEvent(
            event_type=EventType.ONLINE_STATUS_CHANGED,
            account_id=account_id,
            tenant_code=tenant_code,
            data={
                "previous": previous.value,
                "current": current.value,
            },
        )
        self.store.save(event)

Step 3: 修改 API 路由传入 tenant_code

在 app/api/wechat_personal.py 中：

class StartLoginRequest(BaseModel):
    account_id: str = Field(..., description="渠道账号唯一标识")
    account_name: Optional[str] = Field(None, description="账号显示名称")
    tenant_code: str = Field("default", description="业务系统租户编码")


@router.post("/login/start")
def start_login(request: StartLoginRequest):
    return login_service.start_login(request.account_id, request.account_name, request.tenant_code)


class LoginConfirmRequest(BaseModel):
    tenant_code: str = Field("default", description="业务系统租户编码")


@router.post("/login/{account_id}/confirm")
def confirm_login(account_id: str, request: LoginConfirmRequest):
    return login_service.confirm_login(account_id, request.tenant_code)

Step 4: 更新 login_service 单例注入 EventStore

from app.repositories.event_store import event_store
login_service = LoginService(registry, event_store)

Step 5: 运行测试并 Commit

pytest -v
git add app/clients/pywechat_client.py app/services/login_service.py app/api/wechat_personal.py tests/test_api.py
git commit -m "feat(login): emit login and online status events"

Task 8: 改造发送服务，改为异步受理并发送 send_result 事件

Files:

Modify: channel-service/app/services/send_service.py
Modify: channel-service/app/api/wechat_personal.py
Test: channel-service/tests/test_send_async.py
Step 1: 修改 SendTextRequest / SendImageRequest 增加 record_id 与 tenant_code

class SendTextRequest(BaseModel):
    account_id: str
    conversation_id: str = Field(..., description="目标会话/群 ID")
    text: str = Field(..., min_length=1, description="消息内容")
    record_id: str = Field(..., description="mci-server 发送记录 ID")
    tenant_code: str = Field("default", description="业务系统租户编码")


class SendImageRequest(BaseModel):
    account_id: str
    conversation_id: str
    image_url: str
    record_id: str = Field(..., description="mci-server 发送记录 ID")
    tenant_code: str = Field("default", description="业务系统租户编码")

Step 2: 修改 SendService 为异步受理并写 send_result 事件

import uuid

from app.models.event import EventType, OutboundEvent
from app.repositories.event_store import EventStore


class SendService:
    def __init__(self, registry: ClientRegistry, store: EventStore):
        self.registry = registry
        self.store = store

    def send_text(self, account_id: str, conversation_id: str, text: str, record_id: str, tenant_code: str = "default") -> dict:
        client = self.registry.get(account_id)
        if client is None:
            return self._fail(account_id, record_id, tenant_code, conversation_id, f"account {account_id} not registered")
        if client.state.online_status != OnlineStatus.ONLINE:
            return self._fail(account_id, record_id, tenant_code, conversation_id, "account not online")
        if client.state.risk_level >= 80:
            return self._fail(account_id, record_id, tenant_code, conversation_id, "account risk level too high", error_code="RISK_CONTROL_BLOCKED")

        message_id = f"mock-msg-{uuid.uuid4().hex[:16]}"
        result = client.send_text(conversation_id, text)
        self._emit_send_result(account_id, record_id, tenant_code, conversation_id, True, message_id, result)
        return {"accepted": True, "message_id": message_id}

    def send_image(self, account_id: str, conversation_id: str, image_url: str, record_id: str, tenant_code: str = "default") -> dict:
        client = self.registry.get(account_id)
        if client is None:
            return self._fail(account_id, record_id, tenant_code, conversation_id, f"account {account_id} not registered")
        if client.state.online_status != OnlineStatus.ONLINE:
            return self._fail(account_id, record_id, tenant_code, conversation_id, "account not online")

        message_id = f"mock-img-{image_url.split('/')[-1]}"
        self._emit_send_result(account_id, record_id, tenant_code, conversation_id, True, message_id, {"image_url": image_url})
        return {"accepted": True, "message_id": message_id}

    def _fail(self, account_id: str, record_id: str, tenant_code: str, conversation_id: str, error: str, error_code: str = "SEND_FAILED") -> dict:
        self._emit_send_result(account_id, record_id, tenant_code, conversation_id, False, None, {"error": error, "error_code": error_code})
        return {"accepted": False, "error": error}

    def _emit_send_result(self, account_id: str, record_id: str, tenant_code: str, conversation_id: str, success: bool, message_id: Optional[str], channel_result: dict) -> None:
        event = OutboundEvent(
            event_type=EventType.SEND_RESULT,
            account_id=account_id,
            tenant_code=tenant_code,
            data={
                "record_id": record_id,
                "conversation_id": conversation_id,
                "success": success,
                "message_id": message_id,
                "channel_result": channel_result,
                "error": channel_result.get("error"),
            },
        )
        self.store.save(event)

Step 3: 修改 API 路由

@router.post("/send/text")
def send_text(request: SendTextRequest):
    return send_service.send_text(
        request.account_id,
        request.conversation_id,
        request.text,
        request.record_id,
        request.tenant_code,
    )


@router.post("/send/image")
def send_image(request: SendImageRequest):
    return send_service.send_image(
        request.account_id,
        request.conversation_id,
        request.image_url,
        request.record_id,
        request.tenant_code,
    )

Step 4: 更新 send_service 单例注入 EventStore

from app.repositories.event_store import event_store
send_service = SendService(registry, event_store)

Step 5: 更新旧测试

tests/test_api.py 中 test_send_after_login 和 test_send_without_login_fails 需要增加 record_id：

{"account_id": "bot-002", "conversation_id": "room-001", "text": "hello", "record_id": "rec-001"}

并把 assert r.json()["success"] is True 改为 assert r.json()["accepted"] is True。

Step 6: 新增 tests/test_send_async.py

from fastapi.testclient import TestClient

from app.main import app
from app.models.event import EventType
from app.repositories.event_store import event_store

client = TestClient(app)


def test_send_text_creates_send_result_event():
    event_store.cleanup(0)
    client.post("/v1/wechat-personal/login/start", json={"account_id": "send-bot", "tenant_code": "rescue"})
    client.post("/v1/wechat-personal/login/send-bot/confirm", json={"tenant_code": "rescue"})

    r = client.post("/v1/wechat-personal/send/text", json={
        "account_id": "send-bot",
        "conversation_id": "room-001",
        "text": "hello",
        "record_id": "rec-123",
        "tenant_code": "rescue",
    })
    assert r.status_code == 200
    assert r.json()["accepted"] is True

    pending = event_store.fetch_pending(limit=10)
    assert any(e.event_type == EventType.SEND_RESULT and e.data.get("record_id") == "rec-123" for e in pending)

Step 7: 运行测试并 Commit

pytest -v
git add app/services/send_service.py app/api/wechat_personal.py tests/test_api.py tests/test_send_async.py
git commit -m "feat(send): async acceptance and send_result event"

Task 9: 在 `app/main.py` 启动/停止 EventPublisher worker

Files:

Modify: channel-service/app/main.py
Modify: channel-service/app/clients/mci_server_client.py（可选，创建单例）
Step 1: 在 app/repositories/event_store.py 创建全局 event_store 单例（已完成）
Step 2: 修改 app/main.py

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.api import health, monitor, wechat_personal
from app.clients.mci_server_client import MciServerClient
from app.config import settings
from app.repositories.event_store import event_store
from app.services.event_publisher import EventPublisher


@asynccontextmanager
async def lifespan(app: FastAPI):
    client = MciServerClient(settings.mci_server_url)
    publisher = EventPublisher(
        event_store,
        client,
        max_retry=settings.max_retry,
        base_delay_seconds=settings.base_retry_delay_seconds,
        interval_seconds=settings.event_publish_interval_seconds,
    )
    publisher.start()
    yield
    await publisher.stop()


app = FastAPI(title=settings.app_name, debug=settings.debug, lifespan=lifespan)

app.include_router(health.router)
app.include_router(wechat_personal.router)
app.include_router(monitor.router)


if __name__ == "__main__":
    import uvicorn
    uvicorn.run("app.main:app", host=settings.host, port=settings.port, reload=settings.debug)

Step 3: 运行应用并验证启动无报错

python -c "from app.main import app; print('ok')"

Expected: 输出 ok。

Step 4: Commit

git add app/main.py
git commit -m "feat(main): start and stop event publisher worker"

Phase 3: channel-service 补充测试与清理任务

Task 10: 添加事件清理测试与每日清理触发

Files:

Modify: channel-service/app/services/event_publisher.py
Test: channel-service/tests/test_event_store.py
Step 1: 给 EventPublisher 增加定时清理

在 _loop 中每 24 小时调用一次 cleanup_old_events：

    async def _loop(self) -> None:
        cleanup_counter = 0
        while not self._stop_event.is_set():
            try:
                await self._tick()
            except Exception:
                pass
            cleanup_counter += 1
            if cleanup_counter >= 17280:  # 5s * 17280 = 24h
                try:
                    await self.cleanup_old_events(10)
                except Exception:
                    pass
                cleanup_counter = 0
            try:
                await asyncio.wait_for(self._stop_event.wait(), timeout=self.interval_seconds)
            except asyncio.TimeoutError:
                pass

Step 2: 补充清理测试

from datetime import datetime, timedelta, timezone

from app.models.event import EventType, OutboundEvent


def test_cleanup_old_delivered_events(store):
    old_event = OutboundEvent(EventType.HEARTBEAT, "bot-001", "rescue", {})
    store.save(old_event)
    store.mark_delivered(old_event.event_id)

    # 修改 created_at 为 11 天前
    with store._connection() as conn:
        conn.execute(
            "UPDATE outbound_events SET created_at = ? WHERE id = ?",
            (datetime.now(timezone.utc) - timedelta(days=11), old_event.event_id),
        )

    deleted = store.cleanup(10)
    assert deleted == 1

Step 3: 运行测试并 Commit

pytest -v
git add app/services/event_publisher.py tests/test_event_store.py
git commit -m "feat(event): add daily cleanup for delivered/dead-letter events"

Phase 4: mci-server 事件接收端

Task 11: 新增幂等去重表 `mci_channel_event_record`

Files:

Create: backend/mci-server/src/main/resources/db/migration/V22__add_channel_event_record.sql
Create: backend/mci-server/src/main/java/com/sino/mci/channel/entity/ChannelEventRecord.java
Create: backend/mci-server/src/main/java/com/sino/mci/channel/repository/ChannelEventRecordMapper.java
Step 1: 编写 Flyway 脚本

CREATE TABLE IF NOT EXISTS mci_channel_event_record (
    event_id VARCHAR(64) PRIMARY KEY,
    account_id VARCHAR(64) NOT NULL,
    event_type VARCHAR(32) NOT NULL,
    create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_channel_event_account_time ON mci_channel_event_record(account_id, create_time);

Step 2: 创建实体类

package com.sino.mci.channel.entity;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;

import java.time.LocalDateTime;

@Data
@TableName("mci_channel_event_record")
public class ChannelEventRecord {

    @TableId(type = IdType.INPUT)
    private String eventId;

    private String accountId;

    private String eventType;

    private LocalDateTime createTime;
}

Step 3: 创建 Mapper

package com.sino.mci.channel.repository;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.sino.mci.channel.entity.ChannelEventRecord;
import org.apache.ibatis.annotations.Mapper;

@Mapper
public interface ChannelEventRecordMapper extends BaseMapper<ChannelEventRecord> {
}

Step 4: Commit

cd /Users/marsal/Projects/sino-project/sino-mci
git add backend/mci-server/src/main/resources/db/migration/V22__add_channel_event_record.sql \
        backend/mci-server/src/main/java/com/sino/mci/channel/entity/ChannelEventRecord.java \
        backend/mci-server/src/main/java/com/sino/mci/channel/repository/ChannelEventRecordMapper.java
git commit -m "feat(db): add channel event record idempotency table"

Task 12: 创建事件处理服务 `ChannelAccountEventService`

Files:

Create: backend/mci-server/src/main/java/com/sino/mci/channel/dto/ChannelAccountEventRequest.java
Create: backend/mci-server/src/main/java/com/sino/mci/channel/service/ChannelAccountEventService.java
Step 1: 创建请求 DTO

package com.sino.mci.channel.dto;

import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;
import lombok.Data;

import java.util.Map;

@Data
public class ChannelAccountEventRequest {

    @NotBlank
    private String eventId;

    @NotBlank
    private String eventType;

    @NotBlank
    private String accountId;

    @NotBlank
    private String tenantCode;

    private String timestamp;

    @NotNull
    private Map<String, Object> data;
}

Step 2: 创建服务类

package com.sino.mci.channel.service;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.sino.mci.channel.dto.ChannelAccountEventRequest;
import com.sino.mci.channel.entity.ChannelAccount;
import com.sino.mci.channel.entity.ChannelEventRecord;
import com.sino.mci.channel.repository.ChannelAccountMapper;
import com.sino.mci.channel.repository.ChannelEventRecordMapper;
import com.sino.mci.exception.BusinessException;
import com.sino.mci.send.entity.SendRecord;
import com.sino.mci.send.repository.SendRecordMapper;
import com.sino.mci.send.service.SendCallbackService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.Map;

@Slf4j
@Service
@RequiredArgsConstructor
public class ChannelAccountEventService {

    private static final int LOGIN_STATUS_PENDING = 0;
    private static final int LOGIN_STATUS_SUCCESS = 1;
    private static final int LOGIN_STATUS_EXPIRED = 2;
    private static final int LOGIN_STATUS_FAILED = 3;

    private static final int ONLINE_STATUS_OFFLINE = 0;
    private static final int ONLINE_STATUS_ONLINE = 1;
    private static final int ONLINE_STATUS_ERROR = 2;

    private static final int SEND_STATUS_SUCCESS = 2;
    private static final int SEND_STATUS_FAILED = 3;

    private final ChannelAccountMapper accountMapper;
    private final ChannelEventRecordMapper eventRecordMapper;
    private final SendRecordMapper sendRecordMapper;
    private final SendCallbackService sendCallbackService;

    @Transactional
    public void handleEvent(ChannelAccountEventRequest request) {
        if (isDuplicate(request.getEventId())) {
            log.debug("重复事件已忽略: eventId={}", request.getEventId());
            return;
        }

        ChannelAccount account = findAccount(request.getTenantCode(), request.getAccountId());

        switch (request.getEventType()) {
            case "heartbeat":
                handleHeartbeat(account, request.getData());
                break;
            case "login_status_changed":
                handleLoginStatusChanged(account, request.getData());
                break;
            case "online_status_changed":
                handleOnlineStatusChanged(account, request.getData());
                break;
            case "send_result":
                handleSendResult(request.getData());
                break;
            case "error":
                log.warn("channel-service 上报错误: accountId={}, data={}", request.getAccountId(), request.getData());
                break;
            default:
                log.warn("未知事件类型: {}", request.getEventType());
        }

        saveEventRecord(request);
    }

    private boolean isDuplicate(String eventId) {
        return eventRecordMapper.selectById(eventId) != null;
    }

    private void saveEventRecord(ChannelAccountEventRequest request) {
        ChannelEventRecord record = new ChannelEventRecord();
        record.setEventId(request.getEventId());
        record.setAccountId(request.getAccountId());
        record.setEventType(request.getEventType());
        record.setCreateTime(LocalDateTime.now());
        eventRecordMapper.insert(record);
    }

    private ChannelAccount findAccount(String tenantCode, String accountId) {
        ChannelAccount account = accountMapper.selectOne(
                new LambdaQueryWrapper<ChannelAccount>()
                        .eq(ChannelAccount::getTenantCode, tenantCode)
                        .eq(ChannelAccount::getAccountId, accountId));
        if (account == null) {
            throw new BusinessException("渠道账号不存在: " + accountId);
        }
        return account;
    }

    private void handleHeartbeat(ChannelAccount account, Map<String, Object> data) {
        account.setLoginStatus(parseLoginStatus(getString(data, "login_status")));
        account.setOnlineStatus(parseOnlineStatus(getString(data, "online_status")));
        account.setLastHeartbeat(LocalDateTime.now());
        Integer riskLevel = getInteger(data, "risk_level");
        if (riskLevel != null) {
            account.setRiskLevel(riskLevel);
        }
        accountMapper.updateById(account);
    }

    private void handleLoginStatusChanged(ChannelAccount account, Map<String, Object> data) {
        String current = getString(data, "current");
        account.setLoginStatus(parseLoginStatus(current));
        if ("success".equalsIgnoreCase(current)) {
            account.setOnlineStatus(ONLINE_STATUS_ONLINE);
        }
        accountMapper.updateById(account);
    }

    private void handleOnlineStatusChanged(ChannelAccount account, Map<String, Object> data) {
        account.setOnlineStatus(parseOnlineStatus(getString(data, "current")));
        accountMapper.updateById(account);
    }

    private void handleSendResult(Map<String, Object> data) {
        Long recordId = getLong(data, "record_id");
        if (recordId == null) {
            log.warn("send_result 事件缺少 record_id");
            return;
        }
        SendRecord record = sendRecordMapper.selectById(recordId);
        if (record == null) {
            log.warn("发送记录不存在: recordId={}", recordId);
            return;
        }

        Boolean success = getBoolean(data, "success");
        record.setSendStatus(Boolean.TRUE.equals(success) ? SEND_STATUS_SUCCESS : SEND_STATUS_FAILED);
        record.setFinishTime(LocalDateTime.now());
        record.setChannelResult(toJson(data.get("channel_result")));
        sendRecordMapper.updateById(record);

        sendCallbackService.sendCallback(record);
    }

    private int parseLoginStatus(String status) {
        if ("success".equalsIgnoreCase(status)) {
            return LOGIN_STATUS_SUCCESS;
        }
        if ("expired".equalsIgnoreCase(status)) {
            return LOGIN_STATUS_EXPIRED;
        }
        if ("failed".equalsIgnoreCase(status)) {
            return LOGIN_STATUS_FAILED;
        }
        return LOGIN_STATUS_PENDING;
    }

    private int parseOnlineStatus(String status) {
        if ("online".equalsIgnoreCase(status)) {
            return ONLINE_STATUS_ONLINE;
        }
        if ("error".equalsIgnoreCase(status)) {
            return ONLINE_STATUS_ERROR;
        }
        return ONLINE_STATUS_OFFLINE;
    }

    private String getString(Map<String, Object> map, String key) {
        Object value = map.get(key);
        return value == null ? null : value.toString();
    }

    private Integer getInteger(Map<String, Object> map, String key) {
        Object value = map.get(key);
        if (value instanceof Number n) {
            return n.intValue();
        }
        return null;
    }

    private Long getLong(Map<String, Object> map, String key) {
        Object value = map.get(key);
        if (value instanceof Number n) {
            return n.longValue();
        }
        if (value instanceof String s) {
            try {
                return Long.parseLong(s);
            } catch (NumberFormatException e) {
                return null;
            }
        }
        return null;
    }

    private Boolean getBoolean(Map<String, Object> map, String key) {
        Object value = map.get(key);
        if (value instanceof Boolean b) {
            return b;
        }
        return null;
    }

    private String toJson(Object value) {
        if (value == null) {
            return null;
        }
        try {
            return new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(value);
        } catch (Exception e) {
            return value.toString();
        }
    }
}

Step 3: Commit

git add backend/mci-server/src/main/java/com/sino/mci/channel/dto/ChannelAccountEventRequest.java \
        backend/mci-server/src/main/java/com/sino/mci/channel/service/ChannelAccountEventService.java
git commit -m "feat(channel): add event handling service"

Task 13: 创建事件接收 Controller

Files:

Create: backend/mci-server/src/main/java/com/sino/mci/channel/controller/ChannelAccountEventController.java
Test: backend/mci-server/src/test/java/com/sino/mci/channel/controller/ChannelAccountEventControllerTest.java
Step 1: 创建 Controller

package com.sino.mci.channel.controller;

import com.sino.mci.channel.dto.ChannelAccountEventRequest;
import com.sino.mci.channel.service.ChannelAccountEventService;
import com.sino.mci.shared.web.ApiResponse;
import jakarta.validation.Valid;
import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/api/v1/channel-account")
@RequiredArgsConstructor
public class ChannelAccountEventController {

    private final ChannelAccountEventService eventService;

    @PostMapping("/event")
    public ApiResponse<?> receiveEvent(@Valid @RequestBody ChannelAccountEventRequest request) {
        eventService.handleEvent(request);
        return ApiResponse.ok(null);
    }
}

Step 2: 编写 Controller 测试

package com.sino.mci.channel.controller;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.sino.mci.channel.dto.ChannelAccountEventRequest;
import com.sino.mci.channel.entity.ChannelAccount;
import com.sino.mci.channel.repository.ChannelAccountMapper;
import com.sino.mci.channel.repository.ChannelEventRecordMapper;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.http.MediaType;
import org.springframework.test.context.jdbc.Sql;
import org.springframework.test.web.servlet.MockMvc;

import java.util.Map;

import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;

@SpringBootTest
@AutoConfigureMockMvc
@Sql(scripts = "/sql/test-channel-account.sql", executionPhase = Sql.ExecutionPhase.BEFORE_TEST_METHOD)
class ChannelAccountEventControllerTest {

    @Autowired
    private MockMvc mockMvc;

    @Autowired
    private ObjectMapper objectMapper;

    @Autowired
    private ChannelAccountMapper accountMapper;

    @Autowired
    private ChannelEventRecordMapper eventRecordMapper;

    @Test
    void shouldUpdateAccountOnHeartbeat() throws Exception {
        ChannelAccountEventRequest request = new ChannelAccountEventRequest();
        request.setEventId("evt-001");
        request.setEventType("heartbeat");
        request.setAccountId("wx-bot-01");
        request.setTenantCode("rescue");
        request.setData(Map.of(
                "login_status", "success",
                "online_status", "online",
                "risk_level", 10));

        mockMvc.perform(post("/api/v1/channel-account/event")
                        .contentType(MediaType.APPLICATION_JSON)
                        .content(objectMapper.writeValueAsString(request)))
                .andExpect(status().isOk())
                .andExpect(jsonPath("$.code").value(0));

        ChannelAccount account = accountMapper.selectById(1L);
        assertThat(account.getOnlineStatus()).isEqualTo(1);
        assertThat(account.getRiskLevel()).isEqualTo(10);
        assertThat(eventRecordMapper.selectById("evt-001")).isNotNull();
    }

    @Test
    void shouldIgnoreDuplicateEvent() throws Exception {
        ChannelAccountEventRequest request = new ChannelAccountEventRequest();
        request.setEventId("evt-dup-001");
        request.setEventType("heartbeat");
        request.setAccountId("wx-bot-01");
        request.setTenantCode("rescue");
        request.setData(Map.of("online_status", "online"));

        mockMvc.perform(post("/api/v1/channel-account/event")
                        .contentType(MediaType.APPLICATION_JSON)
                        .content(objectMapper.writeValueAsString(request)))
                .andExpect(status().isOk());

        mockMvc.perform(post("/api/v1/channel-account/event")
                        .contentType(MediaType.APPLICATION_JSON)
                        .content(objectMapper.writeValueAsString(request)))
                .andExpect(status().isOk());

        assertThat(eventRecordMapper.selectCount(null)).isEqualTo(1);
    }
}

Step 3: 创建测试 SQL 文件 backend/mci-server/src/test/resources/sql/test-channel-account.sql

DELETE FROM mci_channel_event_record;
DELETE FROM mci_channel_account;
DELETE FROM mci_tenant;

INSERT INTO mci_tenant (id, tenant_code, tenant_name, status, create_time)
VALUES (1, 'rescue', '救援', 1, NOW());

INSERT INTO mci_channel_account (id, tenant_code, account_type, channel_type, account_id, account_name, login_status, online_status, risk_level, status, create_time)
VALUES (1, 'rescue', 3, 'wechat_personal', 'wx-bot-01', '测试号', 0, 0, 0, 1, NOW());

Step 4: 运行 Java 测试

cd /Users/marsal/Projects/sino-project/sino-mci/backend/mci-server
./mvnw test -Dtest=ChannelAccountEventControllerTest

Expected: BUILD SUCCESS / tests passed。

Step 5: Commit

git add backend/mci-server/src/main/java/com/sino/mci/channel/controller/ChannelAccountEventController.java \
        backend/mci-server/src/test/java/com/sino/mci/channel/controller/ChannelAccountEventControllerTest.java \
        backend/mci-server/src/test/resources/sql/test-channel-account.sql
git commit -m "feat(channel): add channel account event receiver endpoint"

Task 14: mci-server 发送流程传入 record_id

Files:

Modify: backend/mci-server/src/main/java/com/sino/mci/send/service/SendService.java
Modify: backend/mci-channel-wechat-personal/.../WechatPersonalChannelAdapter.java
Test: backend/mci-channel-wechat-personal/.../WechatPersonalChannelAdapterTest.java
Step 1: 在 SendService.dispatchRecord 中把 record_id 与 tenant_code 放入 extra

找到 dispatchRecord 方法中构造 SendMessageRequest 的位置，已有类似代码：

extra.put("account_id", account.getAccountId());
messageRequest.setExtra(extra);

改为：

extra.put("account_id", account.getAccountId());
extra.put("record_id", record.getId());
extra.put("tenant_code", record.getTenantCode());
messageRequest.setExtra(extra);

Step 2: 修改 WechatPersonalChannelAdapter.sendMessage 传入 record_id

    private Map<String, Object> sendText(String accountId, String conversationId, String recordId, String tenantCode, Map<String, Object> content) {
        String text = getString(content, "text");
        if (isBlank(text)) {
            throw new IllegalArgumentException("文本内容为空");
        }
        Map<String, Object> body = new HashMap<>();
        body.put("account_id", accountId);
        body.put("conversation_id", conversationId);
        body.put("text", text);
        body.put("record_id", recordId);
        body.put("tenant_code", tenantCode);
        return post("/v1/wechat-personal/send/text", body);
    }

    private Map<String, Object> sendImage(String accountId, String conversationId, String recordId, String tenantCode, Map<String, Object> content) {
        String imageUrl = getString(content, "image_url");
        if (isBlank(imageUrl)) {
            throw new IllegalArgumentException("图片地址为空");
        }
        Map<String, Object> body = new HashMap<>();
        body.put("account_id", accountId);
        body.put("conversation_id", conversationId);
        body.put("image_url", imageUrl);
        body.put("record_id", recordId);
        body.put("tenant_code", tenantCode);
        return post("/v1/wechat-personal/send/image", body);
    }

并在 sendMessage 中读取 recordId 与 tenantCode：

        String recordId = getString(request.getExtra(), "record_id");
        String tenantCode = defaultString(getString(request.getExtra(), "tenant_code"));
        ...
        if ("image".equals(contentType)) {
            response = sendImage(accountId, conversationId, recordId, tenantCode, request.getContent());
        } else {
            response = sendText(accountId, conversationId, recordId, tenantCode, request.getContent());
        }

Step 3: 修改响应解析，兼容 accepted 字段

channel-service 异步受理后返回 {"accepted": true, "message_id": "..."} 而不是 {"success": true, ...}。在 sendMessage 中把成功判断改为：

        Boolean success = getBoolean(response, "success");
        Boolean accepted = getBoolean(response, "accepted");
        if (!Boolean.TRUE.equals(success) && !Boolean.TRUE.equals(accepted)) {
            result.setSuccess(false);
            result.setMessage("个人微信发送失败：" + getString(response, "error"));
            return result;
        }

        result.setSuccess(true);
        result.setChannelMessageId(getString(response, "message_id"));

Step 4: 更新 WechatPersonalChannelAdapterTest

在 shouldSendTextMessage 和 shouldSendImageMessage 中：

给 extra 增加 record_id 与 tenant_code：

extra.put("record_id", "1001");
extra.put("tenant_code", "rescue");

把 mock 响应从 success 改为 accepted：

.setBody(toJson(Map.of(
        "accepted", true,
        "message_id", "msg-123")))

断言请求体包含 record_id 与 tenant_code。

Step 5: 运行相关 Java 测试

cd /Users/marsal/Projects/sino-project/sino-mci
./mvnw test -pl backend/mci-channel-wechat-personal -Dtest=WechatPersonalChannelAdapterTest
./mvnw test -pl backend/mci-server -Dtest=SendServiceResolveTargetTest,SendEndToEndTest

Expected: PASS。

Step 6: Commit

git add backend/mci-server/src/main/java/com/sino/mci/send/service/SendService.java \
        backend/mci-channel-wechat-personal/src/main/java/com/sino/mci/channel/wechat/personal/WechatPersonalChannelAdapter.java \
        backend/mci-channel-wechat-personal/src/test/java/com/sino/mci/channel/wechat/personal/WechatPersonalChannelAdapterTest.java
git commit -m "feat(send): pass record_id and tenant_code, accept async response"

Phase 5: 集成与联调

Task 15: 更新 docker-compose 环境变量

Files:

Modify: docker-compose.yml
Step 1: 给 channel-service service 增加环境变量与数据卷

  channel-service:
    build: ./channel-service
    environment:
      - MCI_SERVER_URL=http://mci-server:8080/msg-platform
      - EVENT_STORE_PATH=/app/data/events.db
      - MAX_RETRY=5
      - BASE_RETRY_DELAY_SECONDS=5
      - EVENT_RETENTION_DAYS=10
      - EVENT_PUBLISH_INTERVAL_SECONDS=5
    volumes:
      - channel-service-data:/app/data

Step 2: 新增顶层 volume

volumes:
  channel-service-data:

Step 3: Commit

git add docker-compose.yml
git commit -m "chore(docker): add channel-service event store config and volume"

Task 16: 端到端冒烟测试

Step 1: 启动基础设施与两个服务

cd /Users/marsal/Projects/sino-project/sino-mci
docker-compose up -d mysql redis rabbitmq mci-server channel-service

Step 2: 在 mci-server 创建渠道账号并初始化登录

curl -X POST http://localhost:8080/msg-platform/api/v1/channel-account \
  -H "X-Tenant-Code: rescue" \
  -H "Content-Type: application/json" \
  -d '{"accountType":3,"channelType":"wechat_personal","accountId":"wx-bot-01","accountName":"测试号"}'

curl -X POST http://localhost:8080/msg-platform/api/v1/channel-account/init \
  -H "X-Tenant-Code: rescue" \
  -H "Content-Type: application/json" \
  -d '{"channelType":"wechat_personal","accountType":3,"config":{"account_id":"wx-bot-01"}}'

Step 3: 调用发送接口

curl -X POST http://localhost:8080/msg-platform/api/v1/send \
  -H "X-Tenant-Code: rescue" \
  -H "Content-Type: application/json" \
  -d '{
    "target":{"type":"conversation","conversationId":1},
    "content":{"type":"text","body":{"text":"hello"}},
    "channelStrategy":{"priority":["wechat_personal"]},
    "callbackUrl":"http://host.docker.internal:9999/callback"
  }'

Step 4: 验证 channel-service SQLite 中存在 send_result 事件，且 mci-server SendRecord 状态更新为成功

docker exec -it sino-mci-channel-service-1 sqlite3 /app/data/events.db "SELECT event_type, status FROM outbound_events;"

Expected: 至少一条 send_result / delivered。

Step 5: 提交联调记录（可选）

若测试通过，无需代码变更；否则修复问题后提交。

Self-Review Checklist

Spec Coverage

Spec 要求	对应 Task
channel-service SQLite 事件持久化	Task 2
单线程后台 worker	Task 4
指数退避重试 + 死信	Task 4
心跳/登录/在线/发送结果/错误事件	Task 6, 7, 8
发送改为异步受理	Task 8
10 天数据清理	Task 10
mci-server `/event` 接收端点	Task 13
event_id 幂等	Task 12, 13
record_id 透传	Task 14
状态字段映射	Task 12
风控处理	Task 8
docker-compose 配置	Task 15

Placeholder Scan

无 "TBD" / "TODO"。
每个代码步骤包含完整代码。
每个测试步骤包含可运行命令与期望输出。

Type Consistency

OutboundEvent.status 与 attempt_count 在 Task 3/4 中加入后，后续 store/publisher 均一致使用。
record_id 在 SendTextRequest / SendImageRequest、SendService、API 路由、WechatPersonalChannelAdapter 中一致使用。
ChannelAccountEventRequest 字段名与服务中 getString/getInteger/getLong 一致。

执行方式

Plan complete and saved to docs/superpowers/plans/2026-07-10-channel-service-mci-server-integration-plan.md.

Two execution options:

1. Subagent-Driven (recommended) - I dispatch a fresh subagent per task, review between tasks, fast iteration.

2. Inline Execution - Execute tasks in this session using executing-plans, batch execution with checkpoints.

Which approach?

64 KiB Raw Blame History Unescape Escape

channel-service 与 mci-server 双向集成 Implementation Plan

0. 文件结构总览

channel-service（Python）

mci-server（Java）

Phase 1: channel-service 事件基础设施

Task 1: 创建事件模型 app/models/event.py

Task 2: 创建 SQLite 事件存储 app/repositories/event_store.py

Task 3: 创建 mci-server HTTP 客户端 app/clients/mci_server_client.py

Task 4: 创建单线程事件发布 worker app/services/event_publisher.py

Task 5: 扩展配置 app/config.py

Phase 2: channel-service 业务服务接入事件

Task 6: 改造心跳服务，发送 heartbeat 事件

Task 7: 改造登录服务，发送 login/online 状态事件

Task 8: 改造发送服务，改为异步受理并发送 send_result 事件

Task 9: 在 app/main.py 启动/停止 EventPublisher worker

Phase 3: channel-service 补充测试与清理任务

Task 10: 添加事件清理测试与每日清理触发

Phase 4: mci-server 事件接收端

Task 11: 新增幂等去重表 mci_channel_event_record

Task 12: 创建事件处理服务 ChannelAccountEventService

Task 13: 创建事件接收 Controller

Task 14: mci-server 发送流程传入 record_id

Phase 5: 集成与联调

Task 15: 更新 docker-compose 环境变量

Task 16: 端到端冒烟测试

Self-Review Checklist

Spec Coverage

Placeholder Scan

Type Consistency

执行方式

64 KiB

Raw Blame History

Task 1: 创建事件模型 `app/models/event.py`

Task 2: 创建 SQLite 事件存储 `app/repositories/event_store.py`

Task 3: 创建 mci-server HTTP 客户端 `app/clients/mci_server_client.py`

Task 4: 创建单线程事件发布 worker `app/services/event_publisher.py`

Task 5: 扩展配置 `app/config.py`

Task 9: 在 `app/main.py` 启动/停止 EventPublisher worker

Task 11: 新增幂等去重表 `mci_channel_event_record`

Task 12: 创建事件处理服务 `ChannelAccountEventService`