diff --git a/go.mod b/go.mod index fcfdcb269..074d6975a 100644 --- a/go.mod +++ b/go.mod @@ -38,6 +38,7 @@ require ( github.com/gobwas/glob v0.2.3 github.com/golang-jwt/jwt/v5 v5.3.1 github.com/google/uuid v1.6.0 + github.com/gorilla/websocket v1.5.3 github.com/larksuite/oapi-sdk-go/v3 v3.5.3 github.com/mark3labs/mcp-go v0.47.1 github.com/mattn/go-sqlite3 v1.14.34 @@ -59,6 +60,7 @@ require ( go.opentelemetry.io/otel/sdk v1.42.0 go.opentelemetry.io/otel/sdk/metric v1.42.0 go.opentelemetry.io/otel/trace v1.42.0 + go.uber.org/goleak v1.3.0 gopkg.in/natefinch/lumberjack.v2 v2.2.1 gopkg.in/yaml.v3 v3.0.1 gorm.io/driver/sqlite v1.6.0 @@ -117,7 +119,6 @@ require ( github.com/google/s2a-go v0.1.9 // indirect github.com/googleapis/enterprise-certificate-proxy v0.3.14 // indirect github.com/googleapis/gax-go/v2 v2.18.0 // indirect - github.com/gorilla/websocket v1.5.3 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/jackmordaunt/icns/v3 v3.0.1 // indirect diff --git a/imbot/go.mod b/imbot/go.mod index 623707b6b..55a066718 100644 --- a/imbot/go.mod +++ b/imbot/go.mod @@ -7,6 +7,7 @@ go 1.25.6 require ( github.com/bwmarrin/discordgo v0.29.0 github.com/go-telegram/bot v1.20.0 + github.com/gorilla/websocket v1.5.3 github.com/larksuite/oapi-sdk-go/v3 v3.5.3 github.com/open-dingtalk/dingtalk-stream-sdk-go v0.9.1 github.com/slack-go/slack v0.19.0 @@ -20,7 +21,6 @@ require ( github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/gogo/protobuf v1.3.2 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/gorilla/websocket v1.5.3 // indirect github.com/kr/pretty v0.3.1 // indirect github.com/mdp/qrterminal/v3 v3.2.1 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect diff --git a/imbot/go.sum b/imbot/go.sum index 50bc42acd..c85c06489 100644 --- a/imbot/go.sum +++ b/imbot/go.sum @@ -51,6 +51,7 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4= golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI= +golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= @@ -59,6 +60,7 @@ golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLL golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA= +golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= @@ -67,8 +69,10 @@ golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI= +golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.42.0 h1:UiKe+zDFmJobeJ5ggPwOshJIVt6/Ft0rcfrXZDLWAWY= +golang.org/x/term v0.42.0/go.mod h1:Dq/D+snpsbazcBG5+F9Q1n2rXV8Ma+71xEjTRufARgY= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= diff --git a/imbot/platform/tingly/wire/wire.go b/imbot/platform/tingly/wire/wire.go new file mode 100644 index 000000000..8a13eae0a --- /dev/null +++ b/imbot/platform/tingly/wire/wire.go @@ -0,0 +1,166 @@ +// Package wire defines the JSON-over-WebSocket protocol used between tingly +// platform participants: the tingly server, bots, and chat clients. +// +// The protocol is intentionally small. Every transmission is a Frame whose +// Kind discriminates the payload, plus a few routing fields lifted out of the +// payload for cheap dispatch (Bot, Chat). +// +// Direction matrix: +// +// bot → server: bot.send / bot.edit / bot.delete / bot.react +// chat → server: chat.send / chat.callback +// server→ bot: chat.send / chat.callback (forwarded from chats) +// server→ chat: bot.send / bot.edit / bot.delete / bot.react (forwarded) +// server→ either: welcome / ack / error +// +// All clients open with a Hello frame; the server replies with Welcome. +// +// Wire types live in imbot rather than the top-level tingly module because +// the bot transport must construct them and imbot is its own Go module. +package wire + +import ( + "encoding/json" + + "github.com/tingly-dev/tingly-box/imbot/core" +) + +// Version is bumped on incompatible wire changes. +const Version = 1 + +// Role identifies the connecting client's role in the Hello frame. +type Role string + +const ( + RoleBot Role = "bot" + RoleChat Role = "chat" +) + +// Kind discriminates the payload carried in a Frame. +type Kind string + +const ( + KindHello Kind = "hello" + KindWelcome Kind = "welcome" + KindAck Kind = "ack" + KindError Kind = "error" + + KindBotSend Kind = "bot.send" + KindBotEdit Kind = "bot.edit" + KindBotDelete Kind = "bot.delete" + KindBotReact Kind = "bot.react" + + KindChatSend Kind = "chat.send" + KindChatCallback Kind = "chat.callback" +) + +// Frame is the on-wire envelope. Data is decoded according to Kind. +type Frame struct { + Kind Kind `json:"kind"` + ID string `json:"id,omitempty"` // optional client request id; echoed in ack + Bot string `json:"bot,omitempty"` // bot UUID (always present after Hello) + Chat string `json:"chat,omitempty"` // chat ID (present on chat-scoped frames) + Data json.RawMessage `json:"data,omitempty"` +} + +// Hello is the first frame sent by every client. +type Hello struct { + Version int `json:"version"` + Role Role `json:"role"` + BotID string `json:"botId"` + ChatID string `json:"chatId,omitempty"` // chat clients only + Token string `json:"token,omitempty"` + Sender *core.Sender `json:"sender,omitempty"` // chat clients only + // HistoryLimit asks the server to include up to N recent messages in + // Welcome. 0 means none. + HistoryLimit int `json:"historyLimit,omitempty"` +} + +// Welcome is the server's response to Hello. +type Welcome struct { + Version int `json:"version"` + ServerID string `json:"serverId,omitempty"` + History []HistoryEntry `json:"history,omitempty"` + Now int64 `json:"now,omitempty"` // server unix seconds +} + +// HistoryEntry is a stored event in chronological order. Frame is the +// rebuilt frame as it would appear on the wire (chat.send or bot.send/edit/ +// delete/react). +type HistoryEntry struct { + Frame Frame `json:"frame"` +} + +// Ack is sent by the server once a client frame is accepted. For bot/chat +// sends it carries the assigned MessageID and timestamp. For edits/deletes/ +// reactions only the ID echo and timestamp are meaningful. +type Ack struct { + MessageID string `json:"messageId,omitempty"` + Timestamp int64 `json:"timestamp"` +} + +// ErrorPayload is the payload of a KindError frame. +type ErrorPayload struct { + Code string `json:"code"` + Message string `json:"message"` +} + +// BotSend is the payload for KindBotSend. Mirrors core.SendMessageOptions +// but kept independent so the wire shape is stable across imbot versions. +type BotSend struct { + Text string `json:"text,omitempty"` + Media []core.MediaAttachment `json:"media,omitempty"` + ParseMode core.ParseMode `json:"parseMode,omitempty"` + ReplyTo string `json:"replyTo,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +// BotEdit payload. +type BotEdit struct { + MessageID string `json:"messageId"` + Text string `json:"text"` +} + +// BotDelete payload. +type BotDelete struct { + MessageID string `json:"messageId"` +} + +// BotReact payload. +type BotReact struct { + MessageID string `json:"messageId"` + Emoji string `json:"emoji"` +} + +// ChatSend is the payload for KindChatSend (text or media from a chat). +type ChatSend struct { + Text string `json:"text,omitempty"` + Media []core.MediaAttachment `json:"media,omitempty"` + ChatType core.ChatType `json:"chatType,omitempty"` + Sender core.Sender `json:"sender"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +// ChatCallback is the payload for KindChatCallback (button click etc.). +type ChatCallback struct { + Sender core.Sender `json:"sender"` + CallbackData string `json:"callbackData"` + ChatType core.ChatType `json:"chatType,omitempty"` +} + +// EncodeData JSON-encodes a payload struct into a Frame.Data slice. Errors +// are surfaced because malformed payloads here always indicate a bug. +func EncodeData(v any) (json.RawMessage, error) { + if v == nil { + return nil, nil + } + return json.Marshal(v) +} + +// DecodeData JSON-decodes Frame.Data into the provided pointer. +func DecodeData(raw json.RawMessage, out any) error { + if len(raw) == 0 { + return nil + } + return json.Unmarshal(raw, out) +} diff --git a/imbot/platform/tingly/ws_transport.go b/imbot/platform/tingly/ws_transport.go new file mode 100644 index 000000000..080765780 --- /dev/null +++ b/imbot/platform/tingly/ws_transport.go @@ -0,0 +1,488 @@ +package tingly + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/url" + "sync" + "sync/atomic" + "time" + + "github.com/gorilla/websocket" + "github.com/tingly-dev/tingly-box/imbot/core" + "github.com/tingly-dev/tingly-box/imbot/platform/tingly/wire" +) + +// WSTransportConfig configures a WSTransport. +type WSTransportConfig struct { + // URL is the tingly server's WS endpoint, e.g. ws://host:port/tingly/ws. + URL string + // BotID identifies this bot to the server. Required. + BotID string + // Token is an optional auth token forwarded in the Hello frame. + Token string + // HandshakeTimeout caps the WebSocket dial. + HandshakeTimeout time.Duration + // ReconnectInitial is the first reconnect backoff. Defaults to 1s. + ReconnectInitial time.Duration + // ReconnectMax caps reconnect backoff. Defaults to 30s. + ReconnectMax time.Duration +} + +// WSTransport implements Transport by talking to a remote tingly server over +// WebSocket. It is the production sibling of InProcessTransport. +// +// The transport keeps a single connection: on disconnect it reconnects with +// exponential backoff and re-sends the Hello frame. Pending Send calls block +// until the connection is up (bounded by the caller's context). +type WSTransport struct { + cfg WSTransportConfig + + mu sync.Mutex + conn *websocket.Conn + listener MessageHandler + ready chan struct{} // closed each time a connection is up + + writeMu sync.Mutex // serializes WriteJSON + + pending map[string]chan ackResult // request id → ack waiter + + closed atomic.Bool + idSeq atomic.Int64 + + // stop signals the read loop / reconnect loop to exit. + stop chan struct{} + done chan struct{} +} + +type ackResult struct { + messageID string + timestamp int64 + err error +} + +// NewWSTransport constructs a WSTransport. Call Start to dial; SendMessage +// etc. block until the first connection is established. +func NewWSTransport(cfg WSTransportConfig) (*WSTransport, error) { + if cfg.URL == "" { + return nil, fmt.Errorf("tingly: WSTransport URL is required") + } + if cfg.BotID == "" { + return nil, fmt.Errorf("tingly: WSTransport BotID is required") + } + if cfg.HandshakeTimeout == 0 { + cfg.HandshakeTimeout = 10 * time.Second + } + if cfg.ReconnectInitial == 0 { + cfg.ReconnectInitial = time.Second + } + if cfg.ReconnectMax == 0 { + cfg.ReconnectMax = 30 * time.Second + } + return &WSTransport{ + cfg: cfg, + ready: make(chan struct{}), + pending: make(map[string]chan ackResult), + stop: make(chan struct{}), + done: make(chan struct{}), + }, nil +} + +// Start launches the connect/read/reconnect loop in the background. It +// returns immediately; the first successful Hello unblocks waiters. +func (t *WSTransport) Start(ctx context.Context) { + go t.run(ctx) +} + +func (t *WSTransport) run(ctx context.Context) { + defer close(t.done) + backoff := t.cfg.ReconnectInitial + for { + select { + case <-t.stop: + return + case <-ctx.Done(): + return + default: + } + + err := t.connectAndServe(ctx) + if t.closed.Load() { + return + } + if err == nil { + backoff = t.cfg.ReconnectInitial + continue + } + + // Backoff and retry. + timer := time.NewTimer(backoff) + select { + case <-t.stop: + timer.Stop() + return + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + backoff *= 2 + if backoff > t.cfg.ReconnectMax { + backoff = t.cfg.ReconnectMax + } + } +} + +func (t *WSTransport) connectAndServe(ctx context.Context) error { + u, err := url.Parse(t.cfg.URL) + if err != nil { + return fmt.Errorf("tingly: invalid URL: %w", err) + } + dialer := &websocket.Dialer{HandshakeTimeout: t.cfg.HandshakeTimeout} + conn, _, err := dialer.DialContext(ctx, u.String(), http.Header{}) + if err != nil { + return fmt.Errorf("tingly: dial: %w", err) + } + + // Send Hello. + hello := wire.Hello{ + Version: wire.Version, + Role: wire.RoleBot, + BotID: t.cfg.BotID, + Token: t.cfg.Token, + } + helloData, _ := wire.EncodeData(hello) + if err := conn.WriteJSON(wire.Frame{Kind: wire.KindHello, Bot: t.cfg.BotID, Data: helloData}); err != nil { + conn.Close() + return fmt.Errorf("tingly: write hello: %w", err) + } + + // Read Welcome. + var welcome wire.Frame + if err := conn.ReadJSON(&welcome); err != nil { + conn.Close() + return fmt.Errorf("tingly: read welcome: %w", err) + } + if welcome.Kind != wire.KindWelcome { + conn.Close() + return fmt.Errorf("tingly: expected welcome, got %s", welcome.Kind) + } + + // Mark ready and serve. Closing the current ready channel wakes any + // waiters in waitReady; they re-check t.conn under the mutex. + t.mu.Lock() + t.conn = conn + prev := t.ready + t.mu.Unlock() + close(prev) + + readErr := t.readLoop(conn) + + // Tear down: fail any pending acks and reset readiness. + t.mu.Lock() + if t.conn == conn { + t.conn = nil + } + t.ready = make(chan struct{}) + for id, ch := range t.pending { + select { + case ch <- ackResult{err: fmt.Errorf("tingly: connection lost")}: + default: + } + delete(t.pending, id) + } + t.mu.Unlock() + conn.Close() + return readErr +} + +func (t *WSTransport) readLoop(conn *websocket.Conn) error { + for { + var f wire.Frame + if err := conn.ReadJSON(&f); err != nil { + return err + } + t.dispatch(f) + } +} + +func (t *WSTransport) dispatch(f wire.Frame) { + switch f.Kind { + case wire.KindAck: + var ack wire.Ack + _ = wire.DecodeData(f.Data, &ack) + t.mu.Lock() + ch, ok := t.pending[f.ID] + if ok { + delete(t.pending, f.ID) + } + t.mu.Unlock() + if ok { + select { + case ch <- ackResult{messageID: ack.MessageID, timestamp: ack.Timestamp}: + default: + } + } + case wire.KindError: + var ep wire.ErrorPayload + _ = wire.DecodeData(f.Data, &ep) + t.mu.Lock() + ch, ok := t.pending[f.ID] + if ok { + delete(t.pending, f.ID) + } + listener := t.listener + t.mu.Unlock() + if ok { + select { + case ch <- ackResult{err: fmt.Errorf("tingly: %s: %s", ep.Code, ep.Message)}: + default: + } + } else if listener != nil { + // orphan error: surface as a system message (best effort). + } + case wire.KindChatSend: + var cs wire.ChatSend + if err := wire.DecodeData(f.Data, &cs); err != nil { + return + } + msg := chatSendToMessage(f, cs) + t.deliver(msg) + case wire.KindChatCallback: + var cb wire.ChatCallback + if err := wire.DecodeData(f.Data, &cb); err != nil { + return + } + msg := chatCallbackToMessage(f, cb) + t.deliver(msg) + } +} + +func (t *WSTransport) deliver(msg core.Message) { + t.mu.Lock() + listener := t.listener + t.mu.Unlock() + if listener != nil { + listener(msg) + } +} + +func (t *WSTransport) waitReady(ctx context.Context) (*websocket.Conn, error) { + for { + t.mu.Lock() + conn := t.conn + ready := t.ready + t.mu.Unlock() + if conn != nil { + return conn, nil + } + select { + case <-ready: + // loop and recheck + case <-ctx.Done(): + return nil, ctx.Err() + case <-t.stop: + return nil, fmt.Errorf("tingly: transport closed") + } + } +} + +func (t *WSTransport) nextID() string { + return fmt.Sprintf("r%d", t.idSeq.Add(1)) +} + +func (t *WSTransport) sendFrame(ctx context.Context, f wire.Frame) (wire.Ack, error) { + if t.closed.Load() { + return wire.Ack{}, core.NewBotError(core.ErrConnectionFailed, "tingly transport closed", false) + } + conn, err := t.waitReady(ctx) + if err != nil { + return wire.Ack{}, err + } + if f.ID == "" { + f.ID = t.nextID() + } + if f.Bot == "" { + f.Bot = t.cfg.BotID + } + ch := make(chan ackResult, 1) + t.mu.Lock() + t.pending[f.ID] = ch + t.mu.Unlock() + + t.writeMu.Lock() + err = conn.WriteJSON(f) + t.writeMu.Unlock() + if err != nil { + t.mu.Lock() + delete(t.pending, f.ID) + t.mu.Unlock() + return wire.Ack{}, fmt.Errorf("tingly: write: %w", err) + } + + select { + case res := <-ch: + if res.err != nil { + return wire.Ack{}, res.err + } + return wire.Ack{MessageID: res.messageID, Timestamp: res.timestamp}, nil + case <-ctx.Done(): + t.mu.Lock() + delete(t.pending, f.ID) + t.mu.Unlock() + return wire.Ack{}, ctx.Err() + } +} + +// Send implements Transport. +func (t *WSTransport) Send(ctx context.Context, target string, opts *core.SendMessageOptions) (*core.SendResult, error) { + bs := wire.BotSend{} + if opts != nil { + bs.Text = opts.Text + bs.Media = opts.Media + bs.ParseMode = opts.ParseMode + bs.ReplyTo = opts.ReplyTo + bs.Metadata = sanitizeMetadata(opts.Metadata) + } + data, err := wire.EncodeData(bs) + if err != nil { + return nil, err + } + ack, err := t.sendFrame(ctx, wire.Frame{Kind: wire.KindBotSend, Chat: target, Data: data}) + if err != nil { + return nil, err + } + return &core.SendResult{MessageID: ack.MessageID, Timestamp: ack.Timestamp}, nil +} + +// SendMedia implements Transport. +func (t *WSTransport) SendMedia(ctx context.Context, target string, media []core.MediaAttachment) (*core.SendResult, error) { + return t.Send(ctx, target, &core.SendMessageOptions{Media: media}) +} + +// Edit implements Transport. +func (t *WSTransport) Edit(ctx context.Context, messageID, text string) error { + data, _ := wire.EncodeData(wire.BotEdit{MessageID: messageID, Text: text}) + _, err := t.sendFrame(ctx, wire.Frame{Kind: wire.KindBotEdit, Data: data}) + return err +} + +// Delete implements Transport. +func (t *WSTransport) Delete(ctx context.Context, messageID string) error { + data, _ := wire.EncodeData(wire.BotDelete{MessageID: messageID}) + _, err := t.sendFrame(ctx, wire.Frame{Kind: wire.KindBotDelete, Data: data}) + return err +} + +// React implements Transport. +func (t *WSTransport) React(ctx context.Context, messageID, emoji string) error { + data, _ := wire.EncodeData(wire.BotReact{MessageID: messageID, Emoji: emoji}) + _, err := t.sendFrame(ctx, wire.Frame{Kind: wire.KindBotReact, Data: data}) + return err +} + +// Subscribe implements Transport. +func (t *WSTransport) Subscribe(handler MessageHandler) { + t.mu.Lock() + t.listener = handler + t.mu.Unlock() +} + +// Close implements Transport. Idempotent. +func (t *WSTransport) Close() error { + if !t.closed.CompareAndSwap(false, true) { + return nil + } + close(t.stop) + t.mu.Lock() + conn := t.conn + t.conn = nil + t.mu.Unlock() + if conn != nil { + _ = conn.Close() + } + return nil +} + +// chatSendToMessage rebuilds a core.Message from a chat.send frame. +func chatSendToMessage(f wire.Frame, cs wire.ChatSend) core.Message { + msg := core.Message{ + ID: f.ID, + Platform: core.PlatformTingly, + Timestamp: time.Now().Unix(), + Sender: cs.Sender, + Recipient: core.Recipient{ + ID: f.Chat, + Type: recipientTypeFromChat(cs.ChatType), + }, + ChatType: cs.ChatType, + Metadata: copyMetadata(cs.Metadata), + } + if len(cs.Media) > 0 { + msg.Content = core.NewMediaContent(cs.Media, cs.Text) + } else { + msg.Content = core.NewTextContent(cs.Text) + } + return msg +} + +// chatCallbackToMessage rebuilds a callback core.Message from a frame. +func chatCallbackToMessage(f wire.Frame, cb wire.ChatCallback) core.Message { + return core.Message{ + ID: f.ID, + Platform: core.PlatformTingly, + Timestamp: time.Now().Unix(), + Sender: cb.Sender, + Recipient: core.Recipient{ + ID: f.Chat, + Type: recipientTypeFromChat(cb.ChatType), + }, + Content: core.NewTextContent(""), + ChatType: cb.ChatType, + Metadata: map[string]interface{}{ + "is_callback": true, + "callback_data": cb.CallbackData, + "callback_query_id": f.ID, + }, + } +} + +// sanitizeMetadata copies opts.Metadata while dropping non-JSON-encodable +// values (e.g. *itx.InlineKeyboardMarkup retained by reference). Keyboard +// payloads are reduced to their JSON shape via Marshal/Unmarshal so the wire +// stays compact. +func sanitizeMetadata(meta map[string]interface{}) map[string]any { + if meta == nil { + return nil + } + out := make(map[string]any, len(meta)) + for k, v := range meta { + if v == nil { + continue + } + // Try a roundtrip; drop if the value can't be JSON-encoded. + b, err := json.Marshal(v) + if err != nil { + continue + } + var any2 any + if err := json.Unmarshal(b, &any2); err != nil { + continue + } + out[k] = any2 + } + return out +} + +func copyMetadata(in map[string]any) map[string]interface{} { + if in == nil { + return nil + } + out := make(map[string]interface{}, len(in)) + for k, v := range in { + out[k] = v + } + return out +} diff --git a/tingly/chatclient/chatclient.go b/tingly/chatclient/chatclient.go new file mode 100644 index 000000000..5244e186a --- /dev/null +++ b/tingly/chatclient/chatclient.go @@ -0,0 +1,310 @@ +// Package chatclient is the client SDK for chat-side participants on the +// tingly platform. A chat client represents a user (or system) talking to a +// bot; it sends inbound messages and receives outbound bot frames in +// response. +// +// This is intentionally a low-ceremony client: it owns one WebSocket, no +// reconnect (a chat session is short-lived in typical use), and exposes +// callbacks for incoming bot frames. +package chatclient + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "sync" + "sync/atomic" + "time" + + "github.com/gorilla/websocket" + "github.com/tingly-dev/tingly-box/imbot/core" + "github.com/tingly-dev/tingly-box/imbot/platform/tingly/wire" +) + +// Config configures a chat client. +type Config struct { + // URL is the tingly server's WS endpoint, e.g. ws://host:port/tingly/ws. + URL string + // BotID is the bot the chat is talking to. Required. + BotID string + // ChatID identifies this chat / conversation. Required. + ChatID string + // Sender describes the chat participant. Required. + Sender core.Sender + // Token is forwarded in the Hello frame if non-empty. + Token string + // HistoryLimit asks for up to N recent messages on connect. + HistoryLimit int + // HandshakeTimeout caps the dial. + HandshakeTimeout time.Duration +} + +// BotEvent is a forwarded frame from the bot received by the chat. The +// concrete payload depends on Kind; the most common is a text message +// (KindBotSend with non-empty Text). +type BotEvent struct { + Kind wire.Kind + MessageID string + Text string + Media []core.MediaAttachment + ParseMode core.ParseMode + ReplyTo string + Emoji string // for KindBotReact + Metadata map[string]any // for KindBotSend +} + +// Client is a connected chat-side WebSocket client. +type Client struct { + cfg Config + + mu sync.Mutex + ws *websocket.Conn + pending map[string]chan ackResult + handler func(BotEvent) + + writeMu sync.Mutex + + idSeq atomic.Int64 + closed atomic.Bool + + // History is the snapshot returned at connect time. Populated by Connect. + History []wire.HistoryEntry +} + +type ackResult struct { + messageID string + timestamp int64 + err error +} + +// New constructs a Client. Call Connect to establish the WebSocket. +func New(cfg Config) (*Client, error) { + if cfg.URL == "" { + return nil, fmt.Errorf("chatclient: URL is required") + } + if cfg.BotID == "" || cfg.ChatID == "" { + return nil, fmt.Errorf("chatclient: BotID and ChatID are required") + } + if cfg.HandshakeTimeout == 0 { + cfg.HandshakeTimeout = 10 * time.Second + } + return &Client{cfg: cfg, pending: make(map[string]chan ackResult)}, nil +} + +// OnBotEvent registers the handler invoked for every forwarded bot frame. +// Setting it to nil drops events. +func (c *Client) OnBotEvent(h func(BotEvent)) { + c.mu.Lock() + c.handler = h + c.mu.Unlock() +} + +// Connect dials the server, performs the Hello/Welcome handshake, and +// starts the read loop. The returned History is the recent-message snapshot +// from the server (only populated when Config.HistoryLimit > 0). +func (c *Client) Connect(ctx context.Context) error { + dialer := &websocket.Dialer{HandshakeTimeout: c.cfg.HandshakeTimeout} + ws, _, err := dialer.DialContext(ctx, c.cfg.URL, http.Header{}) + if err != nil { + return fmt.Errorf("chatclient: dial: %w", err) + } + hello := wire.Hello{ + Version: wire.Version, + Role: wire.RoleChat, + BotID: c.cfg.BotID, + ChatID: c.cfg.ChatID, + Token: c.cfg.Token, + Sender: &c.cfg.Sender, + HistoryLimit: c.cfg.HistoryLimit, + } + hd, _ := wire.EncodeData(hello) + if err := ws.WriteJSON(wire.Frame{Kind: wire.KindHello, Bot: c.cfg.BotID, Chat: c.cfg.ChatID, Data: hd}); err != nil { + ws.Close() + return fmt.Errorf("chatclient: write hello: %w", err) + } + var welcome wire.Frame + if err := ws.ReadJSON(&welcome); err != nil { + ws.Close() + return fmt.Errorf("chatclient: read welcome: %w", err) + } + if welcome.Kind == wire.KindError { + var ep wire.ErrorPayload + _ = wire.DecodeData(welcome.Data, &ep) + ws.Close() + return fmt.Errorf("chatclient: server rejected hello: %s: %s", ep.Code, ep.Message) + } + if welcome.Kind != wire.KindWelcome { + ws.Close() + return fmt.Errorf("chatclient: expected welcome, got %s", welcome.Kind) + } + var w wire.Welcome + if err := wire.DecodeData(welcome.Data, &w); err == nil { + c.History = w.History + } + c.mu.Lock() + c.ws = ws + c.mu.Unlock() + go c.readLoop(ws) + return nil +} + +// Close closes the WebSocket. Idempotent. +func (c *Client) Close() error { + if !c.closed.CompareAndSwap(false, true) { + return nil + } + c.mu.Lock() + ws := c.ws + c.ws = nil + for id, ch := range c.pending { + select { + case ch <- ackResult{err: fmt.Errorf("chatclient: closed")}: + default: + } + delete(c.pending, id) + } + c.mu.Unlock() + if ws != nil { + _ = ws.Close() + } + return nil +} + +func (c *Client) readLoop(ws *websocket.Conn) { + for { + var f wire.Frame + if err := ws.ReadJSON(&f); err != nil { + return + } + c.dispatch(f) + } +} + +func (c *Client) dispatch(f wire.Frame) { + switch f.Kind { + case wire.KindAck: + var ack wire.Ack + _ = wire.DecodeData(f.Data, &ack) + c.completeAck(f.ID, ackResult{messageID: ack.MessageID, timestamp: ack.Timestamp}) + case wire.KindError: + var ep wire.ErrorPayload + _ = wire.DecodeData(f.Data, &ep) + c.completeAck(f.ID, ackResult{err: fmt.Errorf("%s: %s", ep.Code, ep.Message)}) + case wire.KindBotSend: + var bs wire.BotSend + _ = wire.DecodeData(f.Data, &bs) + c.fire(BotEvent{ + Kind: f.Kind, + MessageID: f.ID, + Text: bs.Text, + Media: bs.Media, + ParseMode: bs.ParseMode, + ReplyTo: bs.ReplyTo, + Metadata: bs.Metadata, + }) + case wire.KindBotEdit: + var be wire.BotEdit + _ = wire.DecodeData(f.Data, &be) + c.fire(BotEvent{Kind: f.Kind, MessageID: be.MessageID, Text: be.Text}) + case wire.KindBotDelete: + var bd wire.BotDelete + _ = wire.DecodeData(f.Data, &bd) + c.fire(BotEvent{Kind: f.Kind, MessageID: bd.MessageID}) + case wire.KindBotReact: + var br wire.BotReact + _ = wire.DecodeData(f.Data, &br) + c.fire(BotEvent{Kind: f.Kind, MessageID: br.MessageID, Emoji: br.Emoji}) + } +} + +func (c *Client) fire(ev BotEvent) { + c.mu.Lock() + h := c.handler + c.mu.Unlock() + if h != nil { + h(ev) + } +} + +func (c *Client) completeAck(id string, res ackResult) { + c.mu.Lock() + ch, ok := c.pending[id] + if ok { + delete(c.pending, id) + } + c.mu.Unlock() + if ok { + select { + case ch <- res: + default: + } + } +} + +func (c *Client) nextID() string { + return fmt.Sprintf("c%d", c.idSeq.Add(1)) +} + +// SendText sends a text message from the chat to the bot. It blocks until +// the server acknowledges it (returning the assigned message id) or ctx is +// canceled. +func (c *Client) SendText(ctx context.Context, text string, chatType core.ChatType) (string, error) { + cs := wire.ChatSend{Text: text, ChatType: chatType, Sender: c.cfg.Sender} + return c.send(ctx, wire.KindChatSend, cs) +} + +// SendCallback sends a button-click style callback from the chat. +func (c *Client) SendCallback(ctx context.Context, data string, chatType core.ChatType) (string, error) { + cb := wire.ChatCallback{Sender: c.cfg.Sender, CallbackData: data, ChatType: chatType} + return c.send(ctx, wire.KindChatCallback, cb) +} + +func (c *Client) send(ctx context.Context, kind wire.Kind, payload any) (string, error) { + if c.closed.Load() { + return "", fmt.Errorf("chatclient: closed") + } + c.mu.Lock() + ws := c.ws + c.mu.Unlock() + if ws == nil { + return "", fmt.Errorf("chatclient: not connected") + } + id := c.nextID() + data, err := json.Marshal(payload) + if err != nil { + return "", err + } + ch := make(chan ackResult, 1) + c.mu.Lock() + c.pending[id] = ch + c.mu.Unlock() + + c.writeMu.Lock() + err = ws.WriteJSON(wire.Frame{ + Kind: kind, + ID: id, + Bot: c.cfg.BotID, + Chat: c.cfg.ChatID, + Data: data, + }) + c.writeMu.Unlock() + if err != nil { + c.mu.Lock() + delete(c.pending, id) + c.mu.Unlock() + return "", err + } + select { + case res := <-ch: + if res.err != nil { + return "", res.err + } + return res.messageID, nil + case <-ctx.Done(): + c.mu.Lock() + delete(c.pending, id) + c.mu.Unlock() + return "", ctx.Err() + } +} diff --git a/tingly/server/conn.go b/tingly/server/conn.go new file mode 100644 index 000000000..c7623d9ee --- /dev/null +++ b/tingly/server/conn.go @@ -0,0 +1,72 @@ +package server + +import ( + "sync" + "time" + + "github.com/gorilla/websocket" + "github.com/tingly-dev/tingly-box/imbot/platform/tingly/wire" +) + +// conn is one live WebSocket connection wired into the hub. +type conn struct { + ws *websocket.Conn + writeTimeout time.Duration + + role wire.Role + botID string + chatID string + + writeMu sync.Mutex + closed bool +} + +func newConn(ws *websocket.Conn, writeTimeout time.Duration) *conn { + return &conn{ws: ws, writeTimeout: writeTimeout} +} + +// write serializes WriteJSON across goroutines (the read loop drives ack + +// fanout writes; pingLoop drives ping writes). +func (c *conn) write(f wire.Frame) error { + c.writeMu.Lock() + defer c.writeMu.Unlock() + if c.closed { + return websocket.ErrCloseSent + } + if c.writeTimeout > 0 { + _ = c.ws.SetWriteDeadline(time.Now().Add(c.writeTimeout)) + } + return c.ws.WriteJSON(f) +} + +func (c *conn) writeAck(reqID, messageID string) error { + data, _ := wire.EncodeData(wire.Ack{MessageID: messageID, Timestamp: time.Now().Unix()}) + return c.write(wire.Frame{Kind: wire.KindAck, ID: reqID, Bot: c.botID, Chat: c.chatID, Data: data}) +} + +func (c *conn) writeError(reqID, code, msg string) error { + data, _ := wire.EncodeData(wire.ErrorPayload{Code: code, Message: msg}) + return c.write(wire.Frame{Kind: wire.KindError, ID: reqID, Bot: c.botID, Chat: c.chatID, Data: data}) +} + +func (c *conn) ping() error { + c.writeMu.Lock() + defer c.writeMu.Unlock() + if c.closed { + return websocket.ErrCloseSent + } + if c.writeTimeout > 0 { + _ = c.ws.SetWriteDeadline(time.Now().Add(c.writeTimeout)) + } + return c.ws.WriteMessage(websocket.PingMessage, nil) +} + +func (c *conn) close() { + c.writeMu.Lock() + defer c.writeMu.Unlock() + if c.closed { + return + } + c.closed = true + _ = c.ws.Close() +} diff --git a/tingly/server/e2e_test.go b/tingly/server/e2e_test.go new file mode 100644 index 000000000..f994f58b8 --- /dev/null +++ b/tingly/server/e2e_test.go @@ -0,0 +1,189 @@ +package server_test + +import ( + "context" + "net" + "net/http/httptest" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/tingly-dev/tingly-box/imbot/core" + imtingly "github.com/tingly-dev/tingly-box/imbot/platform/tingly" + "github.com/tingly-dev/tingly-box/imbot/platform/tingly/wire" + "github.com/tingly-dev/tingly-box/tingly/chatclient" + "github.com/tingly-dev/tingly-box/tingly/server" +) + +// startTestServer brings up a tingly server on a random port and returns a +// ws:// URL pointing at the WS endpoint. +func startTestServer(t *testing.T) (string, *server.Server) { + t.Helper() + storePath := filepath.Join(t.TempDir(), "store.json") + srv, err := server.New(server.Config{ + StorePath: storePath, + Path: "/tingly/ws", + PingInterval: 5 * time.Second, + }) + require.NoError(t, err) + + ts := httptest.NewServer(srv.Handler()) + t.Cleanup(func() { + ts.Close() + _ = srv.Shutdown(context.Background()) + }) + wsURL := "ws" + strings.TrimPrefix(ts.URL, "http") + "/tingly/ws" + return wsURL, srv +} + +func TestE2E_ChatToBotAndBack(t *testing.T) { + wsURL, _ := startTestServer(t) + + // --- Bot side: WSTransport wired into a tingly.Bot via NewBot. --- + botID := "bot-test-1" + tr, err := imtingly.NewWSTransport(imtingly.WSTransportConfig{ + URL: wsURL, + BotID: botID, + }) + require.NoError(t, err) + tr.Start(context.Background()) + + bot, err := imtingly.NewBot(&core.Config{UUID: botID}, tr) + require.NoError(t, err) + + var ( + mu sync.Mutex + received []core.Message + ) + bot.OnMessage(func(m core.Message) { + mu.Lock() + received = append(received, m) + mu.Unlock() + }) + require.NoError(t, bot.Connect(context.Background())) + t.Cleanup(func() { _ = bot.Disconnect(context.Background()) }) + + // --- Chat side: connect a chat client. --- + chatID := "chat-1" + chat, err := chatclient.New(chatclient.Config{ + URL: wsURL, + BotID: botID, + ChatID: chatID, + Sender: core.Sender{ID: "u-1", DisplayName: "Alice"}, + }) + require.NoError(t, err) + + var ( + botEvents []chatclient.BotEvent + emu sync.Mutex + ) + chat.OnBotEvent(func(ev chatclient.BotEvent) { + emu.Lock() + botEvents = append(botEvents, ev) + emu.Unlock() + }) + + dialCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + require.NoError(t, chat.Connect(dialCtx)) + t.Cleanup(func() { _ = chat.Close() }) + + // --- 1) Chat sends a text → bot receives it as a core.Message. --- + sendCtx, cancel2 := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel2() + chatMsgID, err := chat.SendText(sendCtx, "hello bot", core.ChatTypeDirect) + require.NoError(t, err) + require.NotEmpty(t, chatMsgID) + + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(received) >= 1 + }, 3*time.Second, 20*time.Millisecond, "bot did not receive chat message") + + mu.Lock() + got := received[0] + mu.Unlock() + require.Equal(t, core.PlatformTingly, got.Platform) + require.Equal(t, chatID, got.Recipient.ID) + require.Equal(t, "hello bot", got.GetText()) + require.Equal(t, "u-1", got.Sender.ID) + require.Equal(t, core.ChatTypeDirect, got.ChatType) + + // --- 2) Bot replies → chat receives the bot.send event. --- + res, err := bot.SendText(sendCtx, chatID, "hi alice") + require.NoError(t, err) + require.NotEmpty(t, res.MessageID) + + require.Eventually(t, func() bool { + emu.Lock() + defer emu.Unlock() + for _, ev := range botEvents { + if ev.Kind == wire.KindBotSend && ev.Text == "hi alice" { + return true + } + } + return false + }, 3*time.Second, 20*time.Millisecond, "chat did not receive bot reply") + + // --- 3) Bot edits its message → chat sees bot.edit. --- + require.NoError(t, bot.EditMessage(sendCtx, res.MessageID, "hi alice (edited)")) + require.Eventually(t, func() bool { + emu.Lock() + defer emu.Unlock() + for _, ev := range botEvents { + if ev.Kind == wire.KindBotEdit && ev.MessageID == res.MessageID && ev.Text == "hi alice (edited)" { + return true + } + } + return false + }, 3*time.Second, 20*time.Millisecond, "chat did not receive edit") +} + +func TestE2E_HistoryReplay(t *testing.T) { + wsURL, _ := startTestServer(t) + + botID := "bot-hist-1" + chatID := "chat-hist-1" + + // Bot connects, sends a couple of messages, disconnects. + tr, err := imtingly.NewWSTransport(imtingly.WSTransportConfig{URL: wsURL, BotID: botID}) + require.NoError(t, err) + tr.Start(context.Background()) + bot, err := imtingly.NewBot(&core.Config{UUID: botID}, tr) + require.NoError(t, err) + require.NoError(t, bot.Connect(context.Background())) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, err = bot.SendText(ctx, chatID, "first") + require.NoError(t, err) + _, err = bot.SendText(ctx, chatID, "second") + require.NoError(t, err) + + // Give the server a moment to persist before tearing down. + time.Sleep(50 * time.Millisecond) + require.NoError(t, bot.Disconnect(context.Background())) + + // Now connect a chat client asking for history. + chat, err := chatclient.New(chatclient.Config{ + URL: wsURL, + BotID: botID, + ChatID: chatID, + Sender: core.Sender{ID: "u-2"}, + HistoryLimit: 10, + }) + require.NoError(t, err) + require.NoError(t, chat.Connect(ctx)) + defer chat.Close() + + require.Len(t, chat.History, 2) + require.Equal(t, wire.KindBotSend, chat.History[0].Frame.Kind) +} + +// startTestServerForListener is unused but kept as a useful reference for +// future tests that bring up a real net.Listener. +var _ = func() net.Listener { return nil } diff --git a/tingly/server/hub.go b/tingly/server/hub.go new file mode 100644 index 000000000..b3713f96f --- /dev/null +++ b/tingly/server/hub.go @@ -0,0 +1,127 @@ +package server + +import ( + "sync" + + "github.com/tingly-dev/tingly-box/imbot/platform/tingly/wire" +) + +// hub is the in-memory routing table for live WebSocket connections. +// +// Bots are looked up by botID; multiple bot connections for the same UUID +// (e.g. failover) are supported and all receive forwarded frames. +// +// Chat connections are keyed by (botID, chatID). Multiple chat clients can +// be open on the same chat (e.g. user has the chat open on phone + desktop). +type hub struct { + mu sync.RWMutex + + bots map[string]map[*conn]struct{} // botID → set of bot connections + chats map[string]map[string]map[*conn]struct{} // botID → chatID → set of chat connections +} + +func newHub() *hub { + return &hub{ + bots: make(map[string]map[*conn]struct{}), + chats: make(map[string]map[string]map[*conn]struct{}), + } +} + +func (h *hub) addBot(c *conn) { + h.mu.Lock() + defer h.mu.Unlock() + set := h.bots[c.botID] + if set == nil { + set = make(map[*conn]struct{}) + h.bots[c.botID] = set + } + set[c] = struct{}{} +} + +func (h *hub) removeBot(c *conn) { + h.mu.Lock() + defer h.mu.Unlock() + if set := h.bots[c.botID]; set != nil { + delete(set, c) + if len(set) == 0 { + delete(h.bots, c.botID) + } + } +} + +func (h *hub) addChat(c *conn) { + h.mu.Lock() + defer h.mu.Unlock() + byBot := h.chats[c.botID] + if byBot == nil { + byBot = make(map[string]map[*conn]struct{}) + h.chats[c.botID] = byBot + } + set := byBot[c.chatID] + if set == nil { + set = make(map[*conn]struct{}) + byBot[c.chatID] = set + } + set[c] = struct{}{} +} + +func (h *hub) removeChat(c *conn) { + h.mu.Lock() + defer h.mu.Unlock() + byBot := h.chats[c.botID] + if byBot == nil { + return + } + if set := byBot[c.chatID]; set != nil { + delete(set, c) + if len(set) == 0 { + delete(byBot, c.chatID) + } + } + if len(byBot) == 0 { + delete(h.chats, c.botID) + } +} + +// botConns snapshots the live bot connections for a given bot ID. +func (h *hub) botConns(botID string) []*conn { + h.mu.RLock() + defer h.mu.RUnlock() + set := h.bots[botID] + if len(set) == 0 { + return nil + } + out := make([]*conn, 0, len(set)) + for c := range set { + out = append(out, c) + } + return out +} + +// chatConns snapshots the live chat connections for (botID, chatID). +func (h *hub) chatConns(botID, chatID string) []*conn { + h.mu.RLock() + defer h.mu.RUnlock() + byBot := h.chats[botID] + if byBot == nil { + return nil + } + set := byBot[chatID] + if len(set) == 0 { + return nil + } + out := make([]*conn, 0, len(set)) + for c := range set { + out = append(out, c) + } + return out +} + +// fanout sends f to every conn in conns. A failed write closes the conn so +// the read loop notices and unregisters it. Errors are intentionally +// swallowed — the read side is the source of truth for connection health. +func fanout(conns []*conn, f wire.Frame) { + for _, c := range conns { + _ = c.write(f) + } +} diff --git a/tingly/server/server.go b/tingly/server/server.go new file mode 100644 index 000000000..83d4e2ea8 --- /dev/null +++ b/tingly/server/server.go @@ -0,0 +1,348 @@ +package server + +import ( + "context" + "errors" + "net" + "net/http" + "sync" + "time" + + "github.com/gorilla/websocket" + "github.com/tingly-dev/tingly-box/imbot/platform/tingly/wire" +) + +// Config configures a tingly server. +type Config struct { + // Addr is the TCP listen address (e.g. ":12581"). Required for ListenAndServe. + Addr string + // Path is the WS endpoint path. Defaults to "/tingly/ws". + Path string + // StorePath is the jsonstore file path for message persistence. Required. + StorePath string + // Token, if non-empty, is required from clients in their Hello frame. + Token string + // AllowOrigin, if non-nil, is consulted to allow cross-origin upgrades. + // When nil, all origins are allowed (suitable for local same-machine + // deployments where bot and chat client live next to the server). + AllowOrigin func(r *http.Request) bool + // HandshakeTimeout caps the WebSocket upgrade. Defaults to 10s. + HandshakeTimeout time.Duration + // WriteTimeout caps individual frame writes. Defaults to 10s. + WriteTimeout time.Duration + // PingInterval controls server→client ping cadence. Defaults to 30s. + PingInterval time.Duration +} + +// Server is the tingly platform service. It exposes a WebSocket endpoint +// where bots and chat clients connect, and routes frames between them. +type Server struct { + cfg Config + hub *hub + store *Store + up websocket.Upgrader + + mu sync.Mutex + httpSrv *http.Server + listener net.Listener +} + +// New constructs a Server. Call ListenAndServe (or Serve with a custom +// listener, via Handler) to start. +func New(cfg Config) (*Server, error) { + if cfg.Path == "" { + cfg.Path = "/tingly/ws" + } + if cfg.HandshakeTimeout == 0 { + cfg.HandshakeTimeout = 10 * time.Second + } + if cfg.WriteTimeout == 0 { + cfg.WriteTimeout = 10 * time.Second + } + if cfg.PingInterval == 0 { + cfg.PingInterval = 30 * time.Second + } + store, err := NewStore(cfg.StorePath) + if err != nil { + return nil, err + } + s := &Server{ + cfg: cfg, + hub: newHub(), + store: store, + up: websocket.Upgrader{ + HandshakeTimeout: cfg.HandshakeTimeout, + CheckOrigin: func(r *http.Request) bool { + if cfg.AllowOrigin != nil { + return cfg.AllowOrigin(r) + } + return true + }, + }, + } + return s, nil +} + +// Handler returns an http.Handler that serves the WS endpoint at cfg.Path. +// Useful for embedding into an existing http.ServeMux. +func (s *Server) Handler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc(s.cfg.Path, s.serveWS) + return mux +} + +// ListenAndServe starts an HTTP server bound to cfg.Addr. Blocks until the +// server stops; returns http.ErrServerClosed on graceful Shutdown. +func (s *Server) ListenAndServe() error { + if s.cfg.Addr == "" { + return errors.New("tingly server: Addr is required for ListenAndServe") + } + ln, err := net.Listen("tcp", s.cfg.Addr) + if err != nil { + return err + } + return s.Serve(ln) +} + +// Serve runs the HTTP server on the given listener. +func (s *Server) Serve(ln net.Listener) error { + srv := &http.Server{ + Handler: s.Handler(), + ReadTimeout: 0, // WS streams must not be capped + WriteTimeout: 0, + } + s.mu.Lock() + s.httpSrv = srv + s.listener = ln + s.mu.Unlock() + return srv.Serve(ln) +} + +// Addr returns the bound listener address (useful when Addr was ":0"). +func (s *Server) Addr() net.Addr { + s.mu.Lock() + defer s.mu.Unlock() + if s.listener == nil { + return nil + } + return s.listener.Addr() +} + +// Shutdown gracefully stops the server. +func (s *Server) Shutdown(ctx context.Context) error { + s.mu.Lock() + srv := s.httpSrv + s.mu.Unlock() + if srv != nil { + _ = srv.Shutdown(ctx) + } + return s.store.Close() +} + +func (s *Server) serveWS(w http.ResponseWriter, r *http.Request) { + wsConn, err := s.up.Upgrade(w, r, nil) + if err != nil { + // Upgrader has already written an error response. + return + } + c := newConn(wsConn, s.cfg.WriteTimeout) + defer c.close() + + // Expect Hello as first frame. + var f wire.Frame + if err := wsConn.ReadJSON(&f); err != nil { + _ = c.writeError("", "BAD_HELLO", "failed to read hello: "+err.Error()) + return + } + if f.Kind != wire.KindHello { + _ = c.writeError(f.ID, "BAD_HELLO", "first frame must be hello") + return + } + var hello wire.Hello + if err := wire.DecodeData(f.Data, &hello); err != nil { + _ = c.writeError(f.ID, "BAD_HELLO", "invalid hello payload") + return + } + if hello.Version != wire.Version { + _ = c.writeError(f.ID, "VERSION", "unsupported wire version") + return + } + if s.cfg.Token != "" && hello.Token != s.cfg.Token { + _ = c.writeError(f.ID, "AUTH_FAILED", "token rejected") + return + } + if hello.BotID == "" { + _ = c.writeError(f.ID, "BAD_HELLO", "botId is required") + return + } + c.botID = hello.BotID + + switch hello.Role { + case wire.RoleBot: + s.hub.addBot(c) + defer s.hub.removeBot(c) + case wire.RoleChat: + if hello.ChatID == "" { + _ = c.writeError(f.ID, "BAD_HELLO", "chatId is required for chat role") + return + } + c.chatID = hello.ChatID + c.role = wire.RoleChat + s.hub.addChat(c) + defer s.hub.removeChat(c) + default: + _ = c.writeError(f.ID, "BAD_HELLO", "unknown role") + return + } + c.role = hello.Role + + // Send Welcome (with history for chat clients that asked). + welcome := wire.Welcome{Version: wire.Version, Now: time.Now().Unix()} + if hello.Role == wire.RoleChat && hello.HistoryLimit > 0 { + welcome.History = s.store.History(c.botID, c.chatID, hello.HistoryLimit) + } + wd, _ := wire.EncodeData(welcome) + if err := c.write(wire.Frame{Kind: wire.KindWelcome, ID: f.ID, Bot: c.botID, Chat: c.chatID, Data: wd}); err != nil { + return + } + + // Start ping loop and read frames. + stopPing := make(chan struct{}) + go s.pingLoop(c, stopPing) + defer close(stopPing) + + s.readLoop(c) +} + +func (s *Server) pingLoop(c *conn, stop <-chan struct{}) { + t := time.NewTicker(s.cfg.PingInterval) + defer t.Stop() + for { + select { + case <-stop: + return + case <-t.C: + if err := c.ping(); err != nil { + return + } + } + } +} + +func (s *Server) readLoop(c *conn) { + for { + var f wire.Frame + if err := c.ws.ReadJSON(&f); err != nil { + return + } + s.handleFrame(c, f) + } +} + +func (s *Server) handleFrame(c *conn, f wire.Frame) { + // Force the routing fields to match the connection identity to prevent + // a misbehaving client from forging frames for other bots/chats. + f.Bot = c.botID + switch c.role { + case wire.RoleBot: + s.handleBotFrame(c, f) + case wire.RoleChat: + f.Chat = c.chatID + s.handleChatFrame(c, f) + default: + _ = c.writeError(f.ID, "BAD_STATE", "no role") + } +} + +func (s *Server) handleBotFrame(c *conn, f wire.Frame) { + switch f.Kind { + case wire.KindBotSend: + if f.Chat == "" { + _ = c.writeError(f.ID, "BAD_FRAME", "bot.send requires chat") + return + } + messageID := s.store.nextMessageID() + // Stamp the assigned message id back into the frame before persisting + // so history entries carry a stable id. + stored := f + stored.ID = messageID + _ = s.store.Append(stored) + _ = c.writeAck(f.ID, messageID) + // Forward to chat clients. + fanout(s.hub.chatConns(c.botID, f.Chat), stored) + + case wire.KindBotEdit, wire.KindBotDelete, wire.KindBotReact: + // These reference an existing message id inside their payload. We + // can't easily learn the chat from the payload without decoding, + // but the frame must specify it (clients always set it via + // chatIDForMessage on their side, but the tingly bot transport + // today doesn't track chat id for outbound ops — see + // imbot/platform/tingly/transport.go chatIDForMessage). The server + // resolves the chat by scanning store entries. + chatID := f.Chat + if chatID == "" { + chatID = s.resolveChatForOp(c.botID, f) + } + if chatID == "" { + _ = c.writeError(f.ID, "NOT_FOUND", "message not found for op") + return + } + f.Chat = chatID + _ = s.store.Append(f) + _ = c.writeAck(f.ID, "") + fanout(s.hub.chatConns(c.botID, chatID), f) + + default: + _ = c.writeError(f.ID, "BAD_FRAME", "unsupported bot kind: "+string(f.Kind)) + } +} + +func (s *Server) handleChatFrame(c *conn, f wire.Frame) { + switch f.Kind { + case wire.KindChatSend, wire.KindChatCallback: + messageID := s.store.nextMessageID() + stored := f + stored.ID = messageID + _ = s.store.Append(stored) + _ = c.writeAck(f.ID, messageID) + fanout(s.hub.botConns(c.botID), stored) + default: + _ = c.writeError(f.ID, "BAD_FRAME", "unsupported chat kind: "+string(f.Kind)) + } +} + +// resolveChatForOp finds the chat that owns the message id referenced by +// edit/delete/react. It scans persisted history newest-first. +func (s *Server) resolveChatForOp(botID string, f wire.Frame) string { + var msgID string + switch f.Kind { + case wire.KindBotEdit: + var p wire.BotEdit + _ = wire.DecodeData(f.Data, &p) + msgID = p.MessageID + case wire.KindBotDelete: + var p wire.BotDelete + _ = wire.DecodeData(f.Data, &p) + msgID = p.MessageID + case wire.KindBotReact: + var p wire.BotReact + _ = wire.DecodeData(f.Data, &p) + msgID = p.MessageID + } + if msgID == "" { + return "" + } + for _, key := range s.store.AllChats() { + // key is "{botID}/{chatID}" + if len(key) <= len(botID)+1 || key[:len(botID)] != botID || key[len(botID)] != '/' { + continue + } + chatID := key[len(botID)+1:] + for _, e := range s.store.History(botID, chatID, 0) { + if e.Frame.ID == msgID { + return chatID + } + } + } + return "" +} diff --git a/tingly/server/store.go b/tingly/server/store.go new file mode 100644 index 000000000..4e086bf73 --- /dev/null +++ b/tingly/server/store.go @@ -0,0 +1,109 @@ +// Package server implements the tingly platform: a small WebSocket router +// that connects bots and chat clients, persists message history per (bot, +// chat) pair, and forwards events between them. +// +// The server is intentionally minimal. It reuses pkg/jsonstore for +// persistence and gorilla/websocket for transport, and depends on the imbot +// wire package for protocol types. +package server + +import ( + "fmt" + "sync" + "sync/atomic" + "time" + + "github.com/tingly-dev/tingly-box/imbot/platform/tingly/wire" + "github.com/tingly-dev/tingly-box/pkg/jsonstore" +) + +// chatLog is the persisted record for a single (bot, chat) conversation. +// Entries are stored in chronological order; edits/deletes/reactions append +// new entries rather than mutating prior ones, preserving an audit trail. +type chatLog struct { + BotID string `json:"botId"` + ChatID string `json:"chatId"` + Entries []wire.HistoryEntry `json:"entries"` +} + +// Store is the persistence layer for tingly conversations. Backed by a +// single jsonstore keyed by "{botID}/{chatID}", it is safe for concurrent +// callers. +type Store struct { + js *jsonstore.Store[chatLog] + mu sync.Mutex // serializes read-modify-write on a key + idSeq atomic.Int64 +} + +// NewStore opens (or creates) the on-disk store at filePath. The directory +// is created if missing. Tests typically pass a t.TempDir()-derived path. +func NewStore(filePath string) (*Store, error) { + if filePath == "" { + return nil, fmt.Errorf("tingly store: filePath is required") + } + js, err := jsonstore.New[chatLog](filePath) + if err != nil { + return nil, fmt.Errorf("tingly store: %w", err) + } + return &Store{js: js}, nil +} + +// Close flushes and releases the store. +func (s *Store) Close() error { + if s == nil || s.js == nil { + return nil + } + return s.js.Close() +} + +// nextMessageID mints a globally unique synthetic message id. +func (s *Store) nextMessageID() string { + n := s.idSeq.Add(1) + return fmt.Sprintf("ty-%d-%d", time.Now().Unix(), n) +} + +// key builds the jsonstore key for a (bot, chat) pair. +func storeKey(botID, chatID string) string { + return botID + "/" + chatID +} + +// Append records a frame into the chat log and persists synchronously. The +// caller is responsible for having already populated f.Bot, f.Chat, f.ID, +// and f.Data. +func (s *Store) Append(f wire.Frame) error { + s.mu.Lock() + defer s.mu.Unlock() + key := storeKey(f.Bot, f.Chat) + cl := s.js.Get(key) + if cl == nil { + cl = &chatLog{BotID: f.Bot, ChatID: f.Chat} + } + cl.Entries = append(cl.Entries, wire.HistoryEntry{Frame: f}) + if err := s.js.Set(key, cl); err != nil { + return err + } + return s.js.ForceSave() +} + +// History returns up to limit recent history entries for (bot, chat). When +// limit <= 0, all entries are returned. +func (s *Store) History(botID, chatID string, limit int) []wire.HistoryEntry { + cl := s.js.Get(storeKey(botID, chatID)) + if cl == nil || len(cl.Entries) == 0 { + return nil + } + if limit <= 0 || limit >= len(cl.Entries) { + out := make([]wire.HistoryEntry, len(cl.Entries)) + copy(out, cl.Entries) + return out + } + out := make([]wire.HistoryEntry, limit) + copy(out, cl.Entries[len(cl.Entries)-limit:]) + return out +} + +// AllChats returns all (bot, chat) keys known to the store. +func (s *Store) AllChats() []string { + return s.js.Keys() +} +