package main import ( "context" "fmt" "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge" "git.yeij.top/AskaEth/Cyrene_v2/internal/services/gateway" ) // capacityEmitter 是面向网关的 capacity_awaken 广播出口。 // *gateway.Server 实现它;测试用假实现替换以捕获入参。 type capacityEmitter interface { BroadcastCapacityAwaken(sessionID, fromState, toState, tool string) } // wireFrontend 用内嵌/外部 SPA 文件系统与 JWT/RBAC 认证器构建网关。 // Session/Persona/Memories 接到具体的管理器上,管理端/用户端的列表接口才会 // 返回真实数据。 func (a *app) wireFrontend() error { jwt := auth.New(a.cfg.Auth) apiKeys := auth.NewAPIKey(a.cfg.Auth) ragStore := knowledge.New(knowledge.Deps{ Projects: a.projStore, Retriever: a.retriever, DefaultPersona: a.defaultPersonaID(), VectorOn: a.vindex != nil, Log: a.log, }) auditStore, aerr := gateway.NewSQLiteAuditStore(context.Background(), a.db, a.log) if aerr != nil { a.log.Warn("audit store unavailable, falling back to in-memory", "error", aerr) auditStore = nil } // 不透明刷新令牌(B2)与主 SQLite 句柄共用同一连接。这里失败要优雅降级: // 登录只是不下发 refreshToken,/auth/refresh 报「未配置」,而不是让整个 // 启动失败。store 以接口形式持有,构造失败就保持 nil(若存成带类型的 nil // 指针,接口值非 nil,首次调用会 panic)。 var refreshStore gateway.RefreshGrantStore if store, rerr := auth.NewRefreshStore(context.Background(), a.db, a.log); rerr != nil { a.log.Warn("refresh token store unavailable", "error", rerr) } else { refreshStore = store } // a.history 在 wireAuxiliary 里构建,让渠道桥(wireChannels)与网关共用同一个 // 持久化存储。IM 通路之前 History 是个空转依赖,因此把它的创建顺序提前不会 // 影响任何下游。 gw, err := gateway.New(a.cfg, gateway.Deps{ Log: logger.WithSource(a.log, "gateway"), Auth: jwt, Bus: a.bus, Registry: a.caps, Loop: a.loop, Session: a.sessions, Personas: a.personas, Memories: a.memStore, History: a.history, IoT: a.iot, Scheduler: a.scheduler, Tasks: a.tasks, Automations: a.automation, Users: apiKeys, Refresh: refreshStore, Backthink: a.backthink, RAG: ragStore, LLM: a.llm, Channels: a.channels, Plugins: a.plugins, OwnerRegistrar: &ownerRegistrar{path: defaultOwnerPath, owner: a.owner, log: a.log}, AuditStore: auditStore, Streaming: true, Frontend: resolveFrontendFS(a.log), Webhooks: a.channelBridge.WebhookHandlers(), DatabasePath: a.dbPath, DataDir: resolveDataDir(a.cfg), }) if err != nil { return fmt.Errorf("gateway: %w", err) } gw.SetVoice(a.voice) a.gw = gw // 必须在下面接线 hook 之前赋值——那些 hook 捕获了 a.gw / gw(否则会 nil 接收者 panic) // 可观测性 trace 出口(design/19 §2):网关已存在,其环形缓冲必须由循环产出的 // span 喂养,否则管理端 trace/metrics 接口会一直是空的。 wireGatewayTraceSink(a, gw) if a.channelBridge != nil { // 把渠道桥中「主人私聊」持久化的消息接入编排器所用的同一个 // `message_persisted` 广播,让 Web 客户端用同一套统一帧来镜像 IM 消息 // (design/16 §4.4)。 a.channelBridge.SetOnPersisted(a.gw.BroadcastPersisted) // design/16 §4.6 §6:被平台白/黑名单(allow/deny)丢弃的消息会记入网关审计 // 轨迹,这样即使它从未到达循环,这次拒绝也「audit 可查」。 a.channelBridge.SetOnRejected(func(msg adapter.UnifiedMessage, reason string) { a.gw.AuditEvent("bridge:"+msg.Platform, "channel.reject", reason) }) } a.capacitySink = gw a.moodSink = gw // 网关 hub(活跃客户端通路)与渠道桥(IM 回落)都已在位,现在接线 backthink 的 // 主动推送出口。backthink.New 执行时它们都还不存在(hub 需要循环,桥需要循环与 // 主人档案,这些都更晚才组装),若不在此处补接线,推送器会解析出 RouteNone。 // 每个 setter 都在服务的生命周期锁下重建推送器。 if a.backthink != nil { a.backthink.SetActiveSink(gw) if a.channelBridge != nil { a.backthink.SetIMSink(a.channelBridge) } } return nil }