61db646805
后端： 1. LLM 独立服务与统一计费出口落地：新增 `cmd/llm`、`client/llm` 与 `services/llm/rpc`，补齐 BillingContext、CreditBalanceGuard、价格规则解析、stream usage 归集与 `credit.charge.requested` outbox 发布，active-scheduler / agent / course / memory / gateway fallback 全部改走 llm zrpc，不再各自本地初始化模型。 2. TokenStore 收口为 Credit 权威账本：新增 credit account / ledger / product / order / price-rule / reward-rule 能力与 Redis 快照缓存，扩展 tokenstore rpc/client 支撑余额快照、消耗看板、商品、订单、流水、价格规则和奖励规则，并接入 LLM charge 事件消费完成 Credit 扣费落账。 3. 计费旧链路下线与网关切口切换：`/token-store` 语义整体切到 `/credit-store`，agent chat 移除旧 TokenQuotaGuard，userauth 的 CheckTokenQuota / AdjustTokenUsage 改为废弃，聊天历史落库不再同步旧 token 额度账本，course 图片解析请求补 user_id 进入新计费口径。 前端： 4. 计划广场从 mock 数据切到真实接口：新增 forum api/types，首页支持真实列表、标签、搜索、防抖、点赞、导入和发布计划，详情页补齐帖子详情、评论树、回复和删除评论链路，同时补上“至少一个标签”的前后端约束与默认标签兜底。 5. 商店页切到 Credit 体系并重做展示：顶部改为余额 + Credit/Token 消耗看板，支持 24h/7d/30d/all 周期切换；套餐区展示原价与当前价；历史区改为当前用户 Credit 流水并支持查看更多，整体视觉和交互同步收口。 仓库： 6. 配置与本地启动体系补齐 llm / outbox 编排：`config.example.yaml` 增加 llm rpc 和统一 outbox service 配置，`dev-common.ps1` 把 llm 纳入多服务依赖并自动建 Kafka topic，`docker-compose.yml` 同步初始化 agent/task/memory/active-scheduler/notification/taskclass-forum/llm/token-store 全量 outbox topic。
100 lines
3.0 KiB
Go
100 lines
3.0 KiB
Go
package eventsvc
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"strconv"
|
|
"strings"
|
|
|
|
"github.com/LoveLosita/smartflow/backend/services/runtime/dao"
|
|
"github.com/LoveLosita/smartflow/backend/services/runtime/model"
|
|
kafkabus "github.com/LoveLosita/smartflow/backend/shared/infra/kafka"
|
|
outboxinfra "github.com/LoveLosita/smartflow/backend/shared/infra/outbox"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
const (
|
|
// EventTypeChatHistoryPersistRequested 是聊天消息持久化请求的业务事件类型。
|
|
EventTypeChatHistoryPersistRequested = "chat.history.persist.requested"
|
|
)
|
|
|
|
// RegisterChatHistoryPersistHandler 注册“聊天消息持久化”消费者。
|
|
// 职责边界：
|
|
// 1. 只处理聊天历史事件，不处理其它业务事件；
|
|
// 2. 只负责注册，不负责总线启动；
|
|
// 3. 先写本地 chat 相关表，不再把聊天 token 消耗同步到旧 userauth 额度账本；
|
|
// 4. 当前版本仅注册新路由键，不再注册旧兼容键。
|
|
func RegisterChatHistoryPersistHandler(
|
|
bus OutboxBus,
|
|
outboxRepo *outboxinfra.Repository,
|
|
repoManager *dao.RepoManager,
|
|
) error {
|
|
if bus == nil {
|
|
return errors.New("event bus is nil")
|
|
}
|
|
if outboxRepo == nil {
|
|
return errors.New("outbox repository is nil")
|
|
}
|
|
if repoManager == nil {
|
|
return errors.New("repo manager is nil")
|
|
}
|
|
|
|
eventOutboxRepo, err := scopedOutboxRepoForEvent(outboxRepo, EventTypeChatHistoryPersistRequested)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
handler := func(ctx context.Context, envelope kafkabus.Envelope) error {
|
|
var payload model.ChatHistoryPersistPayload
|
|
if unmarshalErr := json.Unmarshal(envelope.Payload, &payload); unmarshalErr != nil {
|
|
_ = eventOutboxRepo.MarkDead(ctx, envelope.OutboxID, "解析聊天持久化载荷失败: "+unmarshalErr.Error())
|
|
return nil
|
|
}
|
|
|
|
eventID := strings.TrimSpace(envelope.EventID)
|
|
if eventID == "" {
|
|
eventID = strconv.FormatInt(envelope.OutboxID, 10)
|
|
}
|
|
|
|
if err := eventOutboxRepo.ConsumeInTx(ctx, envelope.OutboxID, func(tx *gorm.DB) error {
|
|
txM := repoManager.WithTx(tx)
|
|
return txM.Agent.SaveChatHistoryInTx(
|
|
ctx,
|
|
payload.UserID,
|
|
payload.ConversationID,
|
|
payload.Role,
|
|
payload.Message,
|
|
payload.ReasoningContent,
|
|
payload.ReasoningDurationSeconds,
|
|
payload.TokensConsumed,
|
|
eventID,
|
|
)
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
|
|
return eventOutboxRepo.MarkConsumed(ctx, envelope.OutboxID)
|
|
}
|
|
|
|
return bus.RegisterEventHandler(EventTypeChatHistoryPersistRequested, handler)
|
|
}
|
|
|
|
// PublishChatHistoryPersistRequested 发布“聊天消息持久化请求”事件。
|
|
func PublishChatHistoryPersistRequested(
|
|
ctx context.Context,
|
|
publisher outboxinfra.EventPublisher,
|
|
payload model.ChatHistoryPersistPayload,
|
|
) error {
|
|
if publisher == nil {
|
|
return errors.New("event publisher is nil")
|
|
}
|
|
return publisher.Publish(ctx, outboxinfra.PublishRequest{
|
|
EventType: EventTypeChatHistoryPersistRequested,
|
|
EventVersion: outboxinfra.DefaultEventVersion,
|
|
MessageKey: payload.ConversationID,
|
|
AggregateID: payload.ConversationID,
|
|
Payload: payload,
|
|
})
|
|
}
|