fix: Phase 6联调 — 插件管理器端口修正 + 多模型配置系统整合 + 历史消息刷新修复
## 调试日志
### 1. 插件管理器启动失败
- **症状**: DevTools 显示插件管理器一直"已停止"，手动启动正常
- **排查**: 对比 process-manager.js 传入的环境变量 vs plugin-manager config.go 读取的变量
- **根因**: config.js 传入 PLUGIN_MANAGER_PORT=8094，但 config.go 读取 os.Getenv("PORT")，env 名不匹配。且 process.env 中 PORT 泄露时被误读为 9090，与 DevTools 端口冲突
- **修复**: config.js 将 PLUGIN_MANAGER_PORT → PORT，使 env 名与代码一致 (c3055f4)
### 2. 历史消息刷新后消失
- **症状**: 浏览器刷新后聊天历史清空
- **排查**: WebSocket history_response handler 中 if (msg.messages) 对空数组 [] 为 truthy
- **根因**: 后端返回空的 history_response (缓存为空) 时，空数组覆盖了 HTTP 已加载的消息
- **修复**: useWebSocket.ts 改为 if (msg.messages && msg.messages.length > 0)，空数组走 else-if 分支仅打日志，不覆盖已有消息
### 3. Phase 6 多模型配置系统
- Gateway: ModelsConfigStore (JSON文件持久化) + Admin CRUD API (providers/models/routing)
- ai-core: ModelSelector 支持按 purpose 选择 + fallback_chain，无配置时回退 .env
- DevTools: 模型配置管理面板 (Providers/Models/Routing 三Tab)、在线模型查询代理、路由表单 checkbox 多选、关键词搜索过滤
- .gitignore: models.json + platform_configs.json
### 4. 多端客户端追踪
- Hub 新增 knownClients 映射 (clientID → KnownClient)，在线/离线状态追踪
- 客户端备注持久化到 PostgreSQL
- DevTools 客户端管理面板
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -20,12 +20,28 @@ import (
|
||||
"github.com/yourname/cyrene-ai/platform-bridge/internal/bridge"
|
||||
"github.com/yourname/cyrene-ai/platform-bridge/internal/config"
|
||||
"github.com/yourname/cyrene-ai/platform-bridge/internal/handler"
|
||||
"github.com/yourname/cyrene-ai/platform-bridge/internal/logging"
|
||||
"github.com/yourname/cyrene-ai/platform-bridge/internal/permissions"
|
||||
)
|
||||
|
||||
func main() {
|
||||
cfg := config.Load()
|
||||
|
||||
// Config store for platform adapter configs.
|
||||
configStore, err := config.NewStore("platform_configs.json")
|
||||
if err != nil {
|
||||
fmt.Printf("FATAL: config store: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
// Message logger.
|
||||
msgLogger, err := logging.NewLogger("logs")
|
||||
if err != nil {
|
||||
fmt.Printf("FATAL: logger: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
defer msgLogger.Close()
|
||||
|
||||
// Core components.
|
||||
mapper := bridge.NewIdentityMapper()
|
||||
checker := permissions.NewChecker()
|
||||
@@ -34,23 +50,58 @@ func main() {
|
||||
// Seed default identities from environment.
|
||||
seedIdentities(mapper)
|
||||
|
||||
// Register platform adapters.
|
||||
adapters := []bridge.PlatformAdapter{
|
||||
qqadapter.NewAdapter(cfg.QQBotPort),
|
||||
telegramadapter.NewAdapter(cfg.TelegramToken, cfg.TelegramWebhookURL),
|
||||
webhookadapter.NewAdapter("webhook"),
|
||||
wechatstub.NewAdapter(),
|
||||
feishustub.NewAdapter(),
|
||||
discordstub.NewAdapter(),
|
||||
}
|
||||
|
||||
// Register platform adapters based on stored configs or defaults.
|
||||
adapters := createAdapters(cfg, configStore)
|
||||
for _, a := range adapters {
|
||||
router.RegisterAdapter(a)
|
||||
}
|
||||
|
||||
// Set message handler: forward to AI-Core.
|
||||
// Set message handler with logging.
|
||||
router.SetMessageHandler(func(msg *bridge.UnifiedMessage) (*bridge.UnifiedResponse, error) {
|
||||
return forwardToAICore(cfg, msg)
|
||||
// Log incoming.
|
||||
msgLogger.Log(logging.LogEntry{
|
||||
Timestamp: time.Now(),
|
||||
Direction: "incoming",
|
||||
Platform: msg.Platform,
|
||||
ChannelID: msg.ChannelID,
|
||||
SenderID: msg.SenderID,
|
||||
SenderName: msg.SenderName,
|
||||
Content: msg.Content,
|
||||
ContentType: msg.ContentType,
|
||||
MessageID: msg.MessageID,
|
||||
Success: true,
|
||||
})
|
||||
|
||||
response, err := forwardToAICore(cfg, msg)
|
||||
if err != nil {
|
||||
msgLogger.Log(logging.LogEntry{
|
||||
Timestamp: time.Now(),
|
||||
Direction: "outgoing",
|
||||
Platform: msg.Platform,
|
||||
ChannelID: msg.ChannelID,
|
||||
SenderID: msg.SenderID,
|
||||
Success: false,
|
||||
Error: err.Error(),
|
||||
})
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Log outgoing.
|
||||
for _, rm := range response.Messages {
|
||||
msgLogger.Log(logging.LogEntry{
|
||||
Timestamp: time.Now(),
|
||||
Direction: "outgoing",
|
||||
Platform: msg.Platform,
|
||||
ChannelID: msg.ChannelID,
|
||||
SenderID: msg.SenderID,
|
||||
SenderName: "Cyrene",
|
||||
Content: rm.Content,
|
||||
ContentType: "text",
|
||||
Success: true,
|
||||
})
|
||||
}
|
||||
|
||||
return response, nil
|
||||
})
|
||||
|
||||
// Connect all adapters.
|
||||
@@ -68,6 +119,12 @@ func main() {
|
||||
bh := handler.NewBridgeHandler(router)
|
||||
bh.RegisterRoutes(mux)
|
||||
|
||||
// Config and log handlers.
|
||||
ch := handler.NewConfigHandler(configStore, router)
|
||||
ch.RegisterRoutes(mux)
|
||||
lh := handler.NewLogHandler(msgLogger)
|
||||
lh.RegisterRoutes(mux)
|
||||
|
||||
// Start QQ message reader loop.
|
||||
qq, _ := router.GetAdapter("qq")
|
||||
if qqa, ok := qq.(*qqadapter.Adapter); ok {
|
||||
@@ -80,13 +137,12 @@ func main() {
|
||||
fmt.Printf("[qq] route error: %v\n", err)
|
||||
continue
|
||||
}
|
||||
// Send response back through QQ adapter.
|
||||
msgs, err := router.SendResponse(response)
|
||||
if err != nil {
|
||||
fmt.Printf("[qq] send error: %v\n", err)
|
||||
continue
|
||||
}
|
||||
_ = msgs // QQ adapter handles sending via WebSocket
|
||||
_ = msgs
|
||||
}
|
||||
}()
|
||||
}
|
||||
@@ -115,6 +171,76 @@ func main() {
|
||||
fmt.Println("Platform Bridge stopped")
|
||||
}
|
||||
|
||||
// createAdapters builds platform adapters, preferring stored configs over defaults.
|
||||
func createAdapters(cfg *config.Config, store *config.Store) []bridge.PlatformAdapter {
|
||||
allNames := []string{"qq", "telegram", "webhook", "wechat", "feishu", "discord"}
|
||||
var adapters []bridge.PlatformAdapter
|
||||
|
||||
for _, name := range allNames {
|
||||
stored, _ := store.Get(name)
|
||||
if stored != nil && !stored.Enabled {
|
||||
fmt.Printf("Platform %s is disabled in config, skipping\n", name)
|
||||
continue
|
||||
}
|
||||
|
||||
var a bridge.PlatformAdapter
|
||||
fields := mergeFields(cfg, name, stored)
|
||||
|
||||
switch name {
|
||||
case "qq":
|
||||
port := cfg.QQBotPort
|
||||
if p, ok := fields["bot_port"]; ok && p != "" {
|
||||
port = p
|
||||
}
|
||||
a = qqadapter.NewAdapter(port)
|
||||
case "telegram":
|
||||
token := cfg.TelegramToken
|
||||
if t, ok := fields["bot_token"]; ok && t != "" {
|
||||
token = t
|
||||
}
|
||||
webhookURL := cfg.TelegramWebhookURL
|
||||
if w, ok := fields["webhook_url"]; ok && w != "" {
|
||||
webhookURL = w
|
||||
}
|
||||
a = telegramadapter.NewAdapter(token, webhookURL)
|
||||
case "webhook":
|
||||
a = webhookadapter.NewAdapter("webhook")
|
||||
case "wechat":
|
||||
a = wechatstub.NewAdapter()
|
||||
case "feishu":
|
||||
a = feishustub.NewAdapter()
|
||||
case "discord":
|
||||
a = discordstub.NewAdapter()
|
||||
}
|
||||
|
||||
if a != nil {
|
||||
adapters = append(adapters, a)
|
||||
}
|
||||
}
|
||||
return adapters
|
||||
}
|
||||
|
||||
// mergeFields returns fields from stored config, falling back to env defaults.
|
||||
func mergeFields(cfg *config.Config, name string, stored *config.PlatformConfig) map[string]string {
|
||||
fields := make(map[string]string)
|
||||
if stored != nil {
|
||||
for k, v := range stored.Fields {
|
||||
fields[k] = v
|
||||
}
|
||||
}
|
||||
// Apply env var defaults if fields are missing.
|
||||
if fields["bot_token"] == "" && cfg.TelegramToken != "" && name == "telegram" {
|
||||
fields["bot_token"] = cfg.TelegramToken
|
||||
}
|
||||
if fields["webhook_url"] == "" && cfg.TelegramWebhookURL != "" && name == "telegram" {
|
||||
fields["webhook_url"] = cfg.TelegramWebhookURL
|
||||
}
|
||||
if fields["bot_port"] == "" && cfg.QQBotPort != "" && name == "qq" {
|
||||
fields["bot_port"] = cfg.QQBotPort
|
||||
}
|
||||
return fields
|
||||
}
|
||||
|
||||
// forwardToAICore sends a unified message to AI-Core's chat endpoint and returns the response.
|
||||
func forwardToAICore(cfg *config.Config, msg *bridge.UnifiedMessage) (*bridge.UnifiedResponse, error) {
|
||||
reqBody, _ := json.Marshal(map[string]interface{}{
|
||||
@@ -145,13 +271,11 @@ func forwardToAICore(cfg *config.Config, msg *bridge.UnifiedMessage) (*bridge.Un
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
// For simplicity, collect full text from SSE stream.
|
||||
var result struct {
|
||||
Content string `json:"content"`
|
||||
Error string `json:"error"`
|
||||
}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
|
||||
// Non-JSON response — read as raw text.
|
||||
buf := new(bytes.Buffer)
|
||||
buf.ReadFrom(resp.Body)
|
||||
return &bridge.UnifiedResponse{
|
||||
@@ -181,7 +305,6 @@ func forwardToAICore(cfg *config.Config, msg *bridge.UnifiedMessage) (*bridge.Un
|
||||
|
||||
// seedIdentities loads default identity mappings.
|
||||
func seedIdentities(m *bridge.IdentityMapper) {
|
||||
// Admin on QQ.
|
||||
if qqAdmin := os.Getenv("QQ_ADMIN_UID"); qqAdmin != "" {
|
||||
m.Register(permissions.PlatformIdentity{
|
||||
Platform: "qq",
|
||||
@@ -191,7 +314,6 @@ func seedIdentities(m *bridge.IdentityMapper) {
|
||||
PermissionLevel: "admin",
|
||||
})
|
||||
}
|
||||
// Admin on Telegram.
|
||||
if tgAdmin := os.Getenv("TELEGRAM_ADMIN_UID"); tgAdmin != "" {
|
||||
m.Register(permissions.PlatformIdentity{
|
||||
Platform: "telegram",
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// PlatformConfig holds persistent configuration for one platform adapter.
|
||||
type PlatformConfig struct {
|
||||
Name string `json:"name"`
|
||||
Enabled bool `json:"enabled"`
|
||||
Label string `json:"label"`
|
||||
Fields map[string]string `json:"fields"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
|
||||
// Store manages persistence of platform configs to a JSON file.
|
||||
type Store struct {
|
||||
mu sync.RWMutex
|
||||
path string
|
||||
configs map[string]*PlatformConfig
|
||||
}
|
||||
|
||||
// NewStore creates a Store, creating the config file if it doesn't exist.
|
||||
func NewStore(path string) (*Store, error) {
|
||||
s := &Store{
|
||||
path: path,
|
||||
configs: make(map[string]*PlatformConfig),
|
||||
}
|
||||
if err := s.load(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (s *Store) load() error {
|
||||
data, err := os.ReadFile(s.path)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
// Initialize empty file.
|
||||
return s.save()
|
||||
}
|
||||
return fmt.Errorf("read config file: %w", err)
|
||||
}
|
||||
if len(data) == 0 {
|
||||
return nil
|
||||
}
|
||||
if err := json.Unmarshal(data, &s.configs); err != nil {
|
||||
return fmt.Errorf("parse config file: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Store) save() error {
|
||||
data, err := json.MarshalIndent(s.configs, "", " ")
|
||||
if err != nil {
|
||||
return fmt.Errorf("marshal configs: %w", err)
|
||||
}
|
||||
tmpPath := s.path + ".tmp"
|
||||
if err := os.WriteFile(tmpPath, data, 0640); err != nil {
|
||||
return fmt.Errorf("write config file: %w", err)
|
||||
}
|
||||
return os.Rename(tmpPath, s.path)
|
||||
}
|
||||
|
||||
// List returns all platform configs.
|
||||
func (s *Store) List() []PlatformConfig {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
result := make([]PlatformConfig, 0, len(s.configs))
|
||||
for _, c := range s.configs {
|
||||
result = append(result, *c)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// Get returns a single platform config.
|
||||
func (s *Store) Get(name string) (*PlatformConfig, error) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
c, ok := s.configs[name]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("config not found: %s", name)
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// Set upserts a platform config and persists.
|
||||
func (s *Store) Set(cfg PlatformConfig) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if cfg.Fields == nil {
|
||||
cfg.Fields = make(map[string]string)
|
||||
}
|
||||
cfg.UpdatedAt = time.Now()
|
||||
s.configs[cfg.Name] = &cfg
|
||||
return s.save()
|
||||
}
|
||||
|
||||
// Delete removes a platform config and persists.
|
||||
func (s *Store) Delete(name string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if _, ok := s.configs[name]; !ok {
|
||||
return fmt.Errorf("config not found: %s", name)
|
||||
}
|
||||
delete(s.configs, name)
|
||||
return s.save()
|
||||
}
|
||||
|
||||
// HasConfig checks if a config exists for the given platform.
|
||||
func (s *Store) HasConfig(name string) bool {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
_, ok := s.configs[name]
|
||||
return ok
|
||||
}
|
||||
@@ -0,0 +1,179 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
|
||||
"github.com/yourname/cyrene-ai/platform-bridge/internal/bridge"
|
||||
"github.com/yourname/cyrene-ai/platform-bridge/internal/config"
|
||||
)
|
||||
|
||||
var knownPlatforms = map[string]bool{
|
||||
"qq": true, "telegram": true, "webhook": true,
|
||||
"wechat": true, "feishu": true, "discord": true,
|
||||
}
|
||||
|
||||
// ConfigHandler exposes CRUD endpoints for platform configs.
|
||||
type ConfigHandler struct {
|
||||
store *config.Store
|
||||
router *bridge.PlatformRouter
|
||||
}
|
||||
|
||||
func NewConfigHandler(store *config.Store, router *bridge.PlatformRouter) *ConfigHandler {
|
||||
return &ConfigHandler{store: store, router: router}
|
||||
}
|
||||
|
||||
func (h *ConfigHandler) RegisterRoutes(mux *http.ServeMux) {
|
||||
mux.HandleFunc("/api/v1/configs", h.listConfigs)
|
||||
mux.HandleFunc("/api/v1/configs/", h.handleConfig)
|
||||
}
|
||||
|
||||
func (h *ConfigHandler) listConfigs(w http.ResponseWriter, r *http.Request) {
|
||||
configs := h.store.List()
|
||||
|
||||
type configSummary struct {
|
||||
Name string `json:"name"`
|
||||
Enabled bool `json:"enabled"`
|
||||
Label string `json:"label,omitempty"`
|
||||
Fields map[string]string `json:"fields"`
|
||||
UpdatedAt string `json:"updated_at"`
|
||||
Connected bool `json:"connected"`
|
||||
}
|
||||
|
||||
var result []configSummary
|
||||
for _, c := range configs {
|
||||
connected := false
|
||||
if a, err := h.router.GetAdapter(c.Name); err == nil {
|
||||
connected = a.IsConnected()
|
||||
}
|
||||
result = append(result, configSummary{
|
||||
Name: c.Name,
|
||||
Enabled: c.Enabled,
|
||||
Label: c.Label,
|
||||
Fields: c.Fields,
|
||||
UpdatedAt: c.UpdatedAt.Format("2006-01-02T15:04:05Z07:00"),
|
||||
Connected: connected,
|
||||
})
|
||||
}
|
||||
// Also include platforms that exist as adapters but have no config yet.
|
||||
for _, name := range h.router.ListAdapters() {
|
||||
found := false
|
||||
for _, c := range configs {
|
||||
if c.Name == name {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
connected := false
|
||||
if a, err := h.router.GetAdapter(name); err == nil {
|
||||
connected = a.IsConnected()
|
||||
}
|
||||
result = append(result, configSummary{
|
||||
Name: name,
|
||||
Enabled: false,
|
||||
Fields: map[string]string{},
|
||||
Connected: connected,
|
||||
})
|
||||
}
|
||||
}
|
||||
if result == nil {
|
||||
result = []configSummary{}
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]interface{}{
|
||||
"configs": result,
|
||||
"total": len(result),
|
||||
})
|
||||
}
|
||||
|
||||
func (h *ConfigHandler) handleConfig(w http.ResponseWriter, r *http.Request) {
|
||||
name := r.URL.Path[len("/api/v1/configs/"):]
|
||||
if name == "" {
|
||||
writeJSON(w, http.StatusBadRequest, errResp("missing config name"))
|
||||
return
|
||||
}
|
||||
if !knownPlatforms[name] {
|
||||
writeJSON(w, http.StatusBadRequest, errResp("unknown platform: "+name))
|
||||
return
|
||||
}
|
||||
|
||||
switch r.Method {
|
||||
case "GET":
|
||||
h.getConfig(w, r, name)
|
||||
case "POST", "PUT":
|
||||
h.saveConfig(w, r, name)
|
||||
case "DELETE":
|
||||
h.deleteConfig(w, r, name)
|
||||
default:
|
||||
writeJSON(w, http.StatusMethodNotAllowed, errResp("method not allowed"))
|
||||
}
|
||||
}
|
||||
|
||||
func (h *ConfigHandler) getConfig(w http.ResponseWriter, r *http.Request, name string) {
|
||||
cfg, err := h.store.Get(name)
|
||||
if err != nil {
|
||||
writeJSON(w, http.StatusNotFound, errResp(err.Error()))
|
||||
return
|
||||
}
|
||||
connected := false
|
||||
if a, err := h.router.GetAdapter(name); err == nil {
|
||||
connected = a.IsConnected()
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]interface{}{
|
||||
"name": cfg.Name,
|
||||
"enabled": cfg.Enabled,
|
||||
"label": cfg.Label,
|
||||
"fields": cfg.Fields,
|
||||
"updated_at": cfg.UpdatedAt.Format("2006-01-02T15:04:05Z07:00"),
|
||||
"connected": connected,
|
||||
})
|
||||
}
|
||||
|
||||
func (h *ConfigHandler) saveConfig(w http.ResponseWriter, r *http.Request, name string) {
|
||||
var body struct {
|
||||
Enabled *bool `json:"enabled"`
|
||||
Label string `json:"label"`
|
||||
Fields map[string]string `json:"fields"`
|
||||
}
|
||||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
||||
writeJSON(w, http.StatusBadRequest, errResp("invalid JSON: "+err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
enabled := true
|
||||
if body.Enabled != nil {
|
||||
enabled = *body.Enabled
|
||||
}
|
||||
|
||||
fields := body.Fields
|
||||
if fields == nil {
|
||||
fields = make(map[string]string)
|
||||
}
|
||||
|
||||
cfg := config.PlatformConfig{
|
||||
Name: name,
|
||||
Enabled: enabled,
|
||||
Label: body.Label,
|
||||
Fields: fields,
|
||||
}
|
||||
if err := h.store.Set(cfg); err != nil {
|
||||
writeJSON(w, http.StatusInternalServerError, errResp(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusOK, map[string]interface{}{
|
||||
"name": name,
|
||||
"enabled": enabled,
|
||||
"label": body.Label,
|
||||
"fields": fields,
|
||||
"status": "saved",
|
||||
})
|
||||
}
|
||||
|
||||
func (h *ConfigHandler) deleteConfig(w http.ResponseWriter, r *http.Request, name string) {
|
||||
if err := h.store.Delete(name); err != nil {
|
||||
writeJSON(w, http.StatusNotFound, errResp(err.Error()))
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "deleted", "name": name})
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"strconv"
|
||||
|
||||
"github.com/yourname/cyrene-ai/platform-bridge/internal/logging"
|
||||
)
|
||||
|
||||
// LogHandler exposes message log retrieval endpoints.
|
||||
type LogHandler struct {
|
||||
logger *logging.Logger
|
||||
}
|
||||
|
||||
func NewLogHandler(logger *logging.Logger) *LogHandler {
|
||||
return &LogHandler{logger: logger}
|
||||
}
|
||||
|
||||
func (h *LogHandler) RegisterRoutes(mux *http.ServeMux) {
|
||||
mux.HandleFunc("/api/v1/logs/", h.handleLogs)
|
||||
}
|
||||
|
||||
func (h *LogHandler) handleLogs(w http.ResponseWriter, r *http.Request) {
|
||||
name := r.URL.Path[len("/api/v1/logs/"):]
|
||||
if name == "" {
|
||||
writeJSON(w, http.StatusBadRequest, errResp("missing platform name in path"))
|
||||
return
|
||||
}
|
||||
|
||||
limit := 100
|
||||
if l := r.URL.Query().Get("limit"); l != "" {
|
||||
if n, err := strconv.Atoi(l); err == nil && n > 0 && n <= 1000 {
|
||||
limit = n
|
||||
}
|
||||
}
|
||||
|
||||
entries, err := h.logger.ReadLogs(name, limit)
|
||||
if err != nil {
|
||||
writeJSON(w, http.StatusInternalServerError, errResp(err.Error()))
|
||||
return
|
||||
}
|
||||
if entries == nil {
|
||||
entries = []logging.LogEntry{}
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]interface{}{
|
||||
"platform": name,
|
||||
"total": len(entries),
|
||||
"logs": entries,
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,149 @@
|
||||
package logging
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// LogEntry represents one message log record.
|
||||
type LogEntry struct {
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
Direction string `json:"direction"` // "incoming" or "outgoing"
|
||||
Platform string `json:"platform"`
|
||||
ChannelID string `json:"channel_id"`
|
||||
SenderID string `json:"sender_id"`
|
||||
SenderName string `json:"sender_name"`
|
||||
Content string `json:"content"`
|
||||
ContentType string `json:"content_type"`
|
||||
MessageID string `json:"message_id,omitempty"`
|
||||
Success bool `json:"success"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// Logger writes message logs to per-platform JSONL files.
|
||||
type Logger struct {
|
||||
mu sync.Mutex
|
||||
dir string
|
||||
files map[string]*os.File
|
||||
}
|
||||
|
||||
// NewLogger creates a Logger, ensuring the log directory exists.
|
||||
func NewLogger(dir string) (*Logger, error) {
|
||||
if err := os.MkdirAll(dir, 0750); err != nil {
|
||||
return nil, fmt.Errorf("create log dir: %w", err)
|
||||
}
|
||||
return &Logger{
|
||||
dir: dir,
|
||||
files: make(map[string]*os.File),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Log writes a log entry to the appropriate platform log file.
|
||||
func (l *Logger) Log(entry LogEntry) error {
|
||||
if entry.Timestamp.IsZero() {
|
||||
entry.Timestamp = time.Now()
|
||||
}
|
||||
|
||||
f, err := l.getOrCreateFile(entry.Platform)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
data, err := json.Marshal(entry)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marshal log entry: %w", err)
|
||||
}
|
||||
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
|
||||
if _, err := f.Write(append(data, '\n')); err != nil {
|
||||
return fmt.Errorf("write log: %w", err)
|
||||
}
|
||||
return f.Sync()
|
||||
}
|
||||
|
||||
// ReadLogs reads the last N log entries for a platform, newest first.
|
||||
func (l *Logger) ReadLogs(platform string, limit int) ([]LogEntry, error) {
|
||||
if limit <= 0 || limit > 1000 {
|
||||
limit = 1000
|
||||
}
|
||||
|
||||
l.mu.Lock()
|
||||
// Flush any pending writes to the file before reading.
|
||||
if f, ok := l.files[platform]; ok {
|
||||
f.Sync()
|
||||
}
|
||||
l.mu.Unlock()
|
||||
|
||||
path := filepath.Join(l.dir, platform+".log")
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return []LogEntry{}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("open log file: %w", err)
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
// Read all lines, keep only the last `limit`.
|
||||
var lines []string
|
||||
scanner := bufio.NewScanner(f)
|
||||
scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
||||
for scanner.Scan() {
|
||||
lines = append(lines, scanner.Text())
|
||||
}
|
||||
if err := scanner.Err(); err != nil {
|
||||
return nil, fmt.Errorf("read log file: %w", err)
|
||||
}
|
||||
|
||||
// Take last N lines and reverse.
|
||||
start := len(lines) - limit
|
||||
if start < 0 {
|
||||
start = 0
|
||||
}
|
||||
lines = lines[start:]
|
||||
|
||||
entries := make([]LogEntry, 0, len(lines))
|
||||
for i := len(lines) - 1; i >= 0; i-- {
|
||||
var entry LogEntry
|
||||
if err := json.Unmarshal([]byte(lines[i]), &entry); err != nil {
|
||||
continue // Skip corrupted lines.
|
||||
}
|
||||
entries = append(entries, entry)
|
||||
}
|
||||
return entries, nil
|
||||
}
|
||||
|
||||
// Close closes all open log file handles.
|
||||
func (l *Logger) Close() error {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
for _, f := range l.files {
|
||||
f.Close()
|
||||
}
|
||||
l.files = make(map[string]*os.File)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *Logger) getOrCreateFile(platform string) (*os.File, error) {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
|
||||
if f, ok := l.files[platform]; ok {
|
||||
return f, nil
|
||||
}
|
||||
|
||||
path := filepath.Join(l.dir, platform+".log")
|
||||
f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0640)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("open log file %s: %w", path, err)
|
||||
}
|
||||
l.files[platform] = f
|
||||
return f, nil
|
||||
}
|
||||
Reference in New Issue
Block a user