feat(gateway): wire real services into Gateway API handlers
- Router accepts full Dependencies (memory/voice/orchestrator/WShub) - Memory API: CRUD + search endpoints with PostgreSQL persistence - Voice API: STT/TTS endpoints - Admin dashboard: real uptime + service status - WebSocket chat: routes to AI-Core orchestrator pipeline - APIAdapter bridges memory domain model to REST interface - Smoke test: memory CRUD + session creation verified
This commit is contained in:
Aska committed
1 parent
4a71965686
commit
6811480786
5 files changed
+448
-111
No files matched your search
@@ -1,13 +1,16 @@
|
||||
// Package ws provides WebSocket connection management for Cyrene.
|
||||
// Package ws provides WebSocket connection management.
|
||||
package ws
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log"
|
||||
"net/http"
|
||||
"sync"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/model"
|
||||
)
|
||||
|
||||
var upgrader = websocket.Upgrader{
|
||||
@@ -16,42 +19,50 @@ var upgrader = websocket.Upgrader{
|
||||
CheckOrigin: func(r *http.Request) bool { return true },
|
||||
}
|
||||
|
||||
// Orchestrator is the interface for the AI dialogue pipeline.
|
||||
type Orchestrator interface {
|
||||
ProcessInput(ctx context.Context, sessionID, userMessage string) (<-chan model.StreamToken, *model.Intent, error)
|
||||
}
|
||||
|
||||
// Hub manages all active WebSocket connections.
|
||||
type Hub struct {
|
||||
mu sync.RWMutex
|
||||
clients map[*Client]bool
|
||||
|
||||
register chan *Client
|
||||
unregister chan *Client
|
||||
broadcast chan []byte
|
||||
mu sync.RWMutex
|
||||
clients map[*Client]bool
|
||||
register chan *Client
|
||||
unregister chan *Client
|
||||
broadcast chan []byte
|
||||
orchestrator Orchestrator
|
||||
}
|
||||
|
||||
// Client represents a single WebSocket connection.
|
||||
type Client struct {
|
||||
hub *Hub
|
||||
conn *websocket.Conn
|
||||
send chan []byte
|
||||
UserID string
|
||||
closed bool
|
||||
mu sync.Mutex
|
||||
hub *Hub
|
||||
conn *websocket.Conn
|
||||
send chan []byte
|
||||
UserID string
|
||||
SessionID string
|
||||
closed bool
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
// Message represents a WebSocket message envelope.
|
||||
type Message struct {
|
||||
// WSMessage represents a client-server message.
|
||||
type WSMessage struct {
|
||||
Type string `json:"type"`
|
||||
Content string `json:"content,omitempty"`
|
||||
SessionID string `json:"session_id,omitempty"`
|
||||
Timestamp int64 `json:"timestamp,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
Devices []any `json:"devices,omitempty"`
|
||||
}
|
||||
|
||||
// NewHub creates and starts a new WebSocket hub.
|
||||
func NewHub() *Hub {
|
||||
// NewHub creates a WebSocket hub with an optional orchestrator.
|
||||
func NewHub(orch Orchestrator) *Hub {
|
||||
h := &Hub{
|
||||
clients: make(map[*Client]bool),
|
||||
register: make(chan *Client),
|
||||
unregister: make(chan *Client),
|
||||
broadcast: make(chan []byte, 256),
|
||||
clients: make(map[*Client]bool),
|
||||
register: make(chan *Client),
|
||||
unregister: make(chan *Client),
|
||||
broadcast: make(chan []byte, 256),
|
||||
orchestrator: orch,
|
||||
}
|
||||
go h.run()
|
||||
return h
|
||||
@@ -60,28 +71,27 @@ func NewHub() *Hub {
|
||||
func (h *Hub) run() {
|
||||
for {
|
||||
select {
|
||||
case client := <-h.register:
|
||||
case c := <-h.register:
|
||||
h.mu.Lock()
|
||||
h.clients[client] = true
|
||||
h.clients[c] = true
|
||||
n := len(h.clients)
|
||||
h.mu.Unlock()
|
||||
log.Printf("[ws] client connected (%d total)", len(h.clients))
|
||||
log.Printf("[ws] client connected (total=%d)", n)
|
||||
|
||||
case client := <-h.unregister:
|
||||
case c := <-h.unregister:
|
||||
h.mu.Lock()
|
||||
if _, ok := h.clients[client]; ok {
|
||||
delete(h.clients, client)
|
||||
close(client.send)
|
||||
if _, ok := h.clients[c]; ok {
|
||||
delete(h.clients, c)
|
||||
close(c.send)
|
||||
}
|
||||
h.mu.Unlock()
|
||||
log.Printf("[ws] client disconnected (%d remaining)", len(h.clients))
|
||||
|
||||
case message := <-h.broadcast:
|
||||
case msg := <-h.broadcast:
|
||||
h.mu.RLock()
|
||||
for client := range h.clients {
|
||||
for c := range h.clients {
|
||||
select {
|
||||
case client.send <- message:
|
||||
case c.send <- msg:
|
||||
default:
|
||||
// Client too slow, skip
|
||||
}
|
||||
}
|
||||
h.mu.RUnlock()
|
||||
@@ -97,20 +107,16 @@ func (h *Hub) HandleUpgrade(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
userID := r.URL.Query().Get("user_id")
|
||||
if userID == "" {
|
||||
userID = "anonymous"
|
||||
}
|
||||
uid := r.URL.Query().Get("user_id")
|
||||
if uid == "" { uid = "anonymous" }
|
||||
|
||||
client := &Client{
|
||||
hub: h,
|
||||
conn: conn,
|
||||
send: make(chan []byte, 64),
|
||||
UserID: userID,
|
||||
send: make(chan []byte, 128),
|
||||
UserID: uid,
|
||||
}
|
||||
|
||||
h.register <- client
|
||||
|
||||
go client.writePump()
|
||||
go client.readPump()
|
||||
}
|
||||
@@ -122,38 +128,77 @@ func (c *Client) readPump() {
|
||||
}()
|
||||
|
||||
for {
|
||||
_, msgBytes, err := c.conn.ReadMessage()
|
||||
_, raw, err := c.conn.ReadMessage()
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
|
||||
var msg Message
|
||||
if err := json.Unmarshal(msgBytes, &msg); err != nil {
|
||||
var msg WSMessage
|
||||
if err := json.Unmarshal(raw, &msg); err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
// Echo back for now (will route to AI-Core later)
|
||||
reply := Message{
|
||||
Type: "response",
|
||||
Content: "echo: " + msg.Content,
|
||||
SessionID: msg.SessionID,
|
||||
switch msg.Type {
|
||||
case "message":
|
||||
c.handleChatMessage(msg)
|
||||
default:
|
||||
c.sendJSON(WSMessage{Type: "response", Content: "unknown message type: " + msg.Type})
|
||||
}
|
||||
replyBytes, _ := json.Marshal(reply)
|
||||
c.send <- replyBytes
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) handleChatMessage(msg WSMessage) {
|
||||
c.SessionID = msg.SessionID
|
||||
|
||||
// If orchestrator not available, echo
|
||||
if c.hub.orchestrator == nil {
|
||||
c.sendJSON(WSMessage{Type: "response", Content: "[系统] AI 核心未配置，当前为 echo 模式: " + msg.Content, SessionID: msg.SessionID})
|
||||
return
|
||||
}
|
||||
|
||||
// Call orchestrator
|
||||
tokenCh, intent, err := c.hub.orchestrator.ProcessInput(context.Background(), msg.SessionID, msg.Content)
|
||||
if err != nil {
|
||||
c.sendJSON(WSMessage{Type: "error", Error: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
// Send intent info
|
||||
if intent != nil {
|
||||
c.sendJSON(WSMessage{Type: "response", Content: "[Intent: " + intent.Type + "]", SessionID: msg.SessionID})
|
||||
}
|
||||
|
||||
// Stream tokens
|
||||
for token := range tokenCh {
|
||||
if token.Done {
|
||||
c.sendJSON(WSMessage{Type: "stream_end", SessionID: msg.SessionID})
|
||||
break
|
||||
}
|
||||
c.sendJSON(WSMessage{Type: "stream_chunk", Content: token.Content, SessionID: msg.SessionID})
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) writePump() {
|
||||
defer c.conn.Close()
|
||||
for msg := range c.send {
|
||||
if err := c.conn.WriteMessage(websocket.TextMessage, msg); err != nil {
|
||||
c.mu.Lock()
|
||||
err := c.conn.WriteMessage(websocket.TextMessage, msg)
|
||||
c.mu.Unlock()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) sendJSON(v any) {
|
||||
data, _ := json.Marshal(v)
|
||||
select {
|
||||
case c.send <- data:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// Broadcast sends a message to all connected clients.
|
||||
func (h *Hub) Broadcast(msg Message) error {
|
||||
func (h *Hub) Broadcast(msg any) error {
|
||||
data, err := json.Marshal(msg)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
Reference in new issue
Block a user