Skip to content

Feat: Add ChatID for threads #10

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
RussellLuo merged 1 commit into from
May 27, 2026
Merged

Feat: Add ChatID for threads #10

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
122 changes: 106 additions & 16 deletions pkg/channels/csgclaw/csgclaw.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,13 +44,23 @@
}

type eventPayload struct {
MessageID string `json:"message_id"`
RoomID string `json:"room_id"`
ChatType string `json:"chat_type"`
Sender sender `json:"sender"`
Text string `json:"text"`
Timestamp string `json:"timestamp"`
Mentions []string `json:"mentions"`
MessageID string `json:"message_id"`
RoomID string `json:"room_id"`
ChatID string `json:"chat_id"`
ChatType string `json:"chat_type"`
ThreadRootID string `json:"thread_root_id"`
Sender sender `json:"sender"`
Text string `json:"text"`
Timestamp string `json:"timestamp"`
Mentions []string `json:"mentions"`
Context eventContext `json:"context"`
}

type eventContext struct {
Channel string `json:"channel"`
ChatID string `json:"chat_id"`
ChatType string `json:"chat_type"`
TopicID string `json:"topic_id"`
}

type sender struct {
Expand All @@ -60,8 +70,16 @@
}

type sendRequest struct {
RoomID string `json:"room_id"`
Text string `json:"text"`
RoomID string `json:"room_id"`
Text string `json:"text"`
TopicID string `json:"topic_id,omitempty"`
Context *sendContext `json:"context,omitempty"`
}

type sendContext struct {
Channel string `json:"channel,omitempty"`
ChatID string `json:"chat_id,omitempty"`
TopicID string `json:"topic_id,omitempty"`
}

func NewChannel(cfg config.CSGClawConfig, messageBus *bus.MessageBus) (*Channel, error) {
Expand Down Expand Up @@ -125,19 +143,28 @@
if !c.IsRunning() {
return channels.ErrNotRunning
}
fmt.Printf("received msg: %+v\n", msg)
roomID := msg.ChatID
roomID, topicID := splitTopicChatID(msg.ChatID)
if strings.TrimSpace(roomID) == "" {
return fmt.Errorf("csgclaw chat ID is empty: %w", channels.ErrSendFailed)
}
if strings.TrimSpace(msg.Content) == "" {
return fmt.Errorf("csgclaw content is empty: %w", channels.ErrSendFailed)
}

body, err := json.Marshal(sendRequest{
payload := sendRequest{
RoomID: roomID,
Text: msg.Content,
})
}
if topicID != "" {
payload.TopicID = topicID
payload.Context = &sendContext{
Channel: "csgclaw",
ChatID: roomID,
TopicID: topicID,
}
}

body, err := json.Marshal(payload)
if err != nil {
return fmt.Errorf("csgclaw marshal send payload: %w", channels.ErrSendFailed)
}
Expand All @@ -151,6 +178,7 @@

logger.InfoCF("csgclaw", "Sending outbound message", map[string]any{
"room_id": roomID,
"topic_id": topicID,
"content_len": len(msg.Content),
"endpoint_url": c.sendURL(),
})
Expand Down Expand Up @@ -206,7 +234,7 @@
}
if !strings.Contains(resp.Header.Get("Content-Type"), "text/event-stream") {
defer resp.Body.Close()
return nil, fmt.Errorf("csgclaw events endpoint returned non-SSE content type %q", resp.Header.Get("Content-Type"))

Check failure on line 237 in pkg/channels/csgclaw/csgclaw.go

View workflow job for this annotation

GitHub Actions / Linter 

File is not properly formatted (golines)
}
return resp, nil
}
Expand All @@ -219,7 +247,7 @@
return
}

resp, err := c.openEventStream(c.ctx)

Check failure on line 250 in pkg/channels/csgclaw/csgclaw.go

View workflow job for this annotation

GitHub Actions / Linter 

response body must be closed (bodyclose)
if err != nil {
if c.ctx.Err() != nil {
return
Expand Down Expand Up @@ -324,7 +352,11 @@
"event": evt,
})

if strings.TrimSpace(evt.RoomID) == "" || strings.TrimSpace(evt.Sender.ID) == "" {
roomID := resolvedRoomID(evt)
topicID := resolvedTopicID(evt)
chatID := topicChatID(roomID, topicID)

if strings.TrimSpace(roomID) == "" || strings.TrimSpace(evt.Sender.ID) == "" {
return
}

Expand All @@ -333,6 +365,9 @@
if strings.EqualFold(evt.ChatType, "group") {
peerKind = "group"
isMentioned := hasInboundBotAtMention(content, c.config.BotID)
if !isMentioned && hasInboundAtMention(content) {
return
}
content = normalizeInboundAtMentions(content)
shouldRespond, normalized := c.ShouldRespondInGroup(isMentioned, content)
if !shouldRespond {
Expand All @@ -352,24 +387,71 @@
metadata := map[string]string{
"timestamp": evt.Timestamp,
"chat_type": evt.ChatType,
"room_id": roomID,
}
if len(evt.Mentions) > 0 {
metadata["mentions"] = strings.Join(evt.Mentions, ",")
}
if topicID != "" {
metadata["topic_id"] = topicID
metadata["parent_peer_kind"] = "topic"
metadata["parent_peer_id"] = topicID
}
if strings.TrimSpace(evt.ThreadRootID) != "" {
metadata["thread_root_id"] = strings.TrimSpace(evt.ThreadRootID)
}

c.HandleMessage(
c.ctx,
bus.Peer{Kind: peerKind, ID: evt.RoomID},
bus.Peer{Kind: peerKind, ID: chatID},
evt.MessageID,
evt.Sender.ID,
evt.RoomID,
chatID,
content,
nil,
metadata,
senderInfo,
)
}

func resolvedRoomID(evt eventPayload) string {
if roomID := strings.TrimSpace(evt.RoomID); roomID != "" {
return roomID
}
if chatID := strings.TrimSpace(evt.ChatID); chatID != "" {
return chatID
}
return strings.TrimSpace(evt.Context.ChatID)
}

func resolvedTopicID(evt eventPayload) string {
if topicID := strings.TrimSpace(evt.Context.TopicID); topicID != "" {
return topicID
}
return strings.TrimSpace(evt.ThreadRootID)
}

func topicChatID(roomID, topicID string) string {
roomID = strings.TrimSpace(roomID)
topicID = strings.TrimSpace(topicID)
if roomID == "" || topicID == "" {
return roomID
}
return roomID + "/" + topicID
}

func splitTopicChatID(chatID string) (roomID, topicID string) {
chatID = strings.TrimSpace(chatID)
if chatID == "" {
return "", ""
}
idx := strings.LastIndex(chatID, "/")
if idx <= 0 || idx == len(chatID)-1 {
return chatID, ""
}
return strings.TrimSpace(chatID[:idx]), strings.TrimSpace(chatID[idx+1:])
}

func hasInboundBotAtMention(content, botID string) bool {
content = strings.TrimSpace(content)
botID = strings.TrimSpace(botID)
Expand All @@ -396,6 +478,14 @@
}
}

func hasInboundAtMention(content string) bool {
content = strings.TrimSpace(content)
if content == "" {
return false
}
return strings.Contains(content, `<at user_id="`)
}

func normalizeInboundAtMentions(content string) string {
if content == "" {
return content
Expand Down
108 changes: 108 additions & 0 deletions pkg/channels/csgclaw/csgclaw_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,9 @@

import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/http/httptest"
"sync/atomic"
Expand Down Expand Up @@ -40,7 +42,7 @@
t.Fatal("response writer does not implement http.Flusher")
}

payload := fmt.Sprintf(`{"message_id":"msg-%d","room_id":"room-1","chat_type":"direct","sender":{"id":"user-1","username":"alice","display_name":"Alice"},"text":"@test-bot hello-%d","timestamp":"2026-03-26T00:00:00Z"}`, attempt, attempt)

Check failure on line 45 in pkg/channels/csgclaw/csgclaw_test.go

View workflow job for this annotation

GitHub Actions / Linter 

File is not properly formatted (golines)
_, _ = fmt.Fprintf(w, "event: message\n")
_, _ = fmt.Fprintf(w, "data: %s\n\n", payload)
flusher.Flush()
Expand Down Expand Up @@ -253,6 +255,112 @@
}
}

func TestHandleInboundEventThreadUsesTopicChatID(t *testing.T) {
mb := bus.NewMessageBus()
defer mb.Close()

ch, err := NewChannel(config.CSGClawConfig{
BaseURL: "http://127.0.0.1:18080",
BotID: "u-manager",
AccessToken: "secret",
}, mb)
if err != nil {
t.Fatalf("NewChannel() error = %v", err)
}

ctx, cancel := context.WithCancel(context.Background())
defer cancel()
ch.ctx = ctx

ch.dispatchEvent("message", `{"message_id":"msg-reply","room_id":"room-1","chat_type":"group","thread_root_id":"msg-root","sender":{"id":"user-1"},"text":"<at user_id=\"u-manager\">manager</at> hi","context":{"topic_id":"msg-root"}}`)

select {
case msg := <-mb.InboundChan():
if msg.ChatID != "room-1/msg-root" {
t.Fatalf("inbound chat ID = %q, want %q", msg.ChatID, "room-1/msg-root")
}
if msg.Peer.Kind != "group" {
t.Fatalf("peer kind = %q, want group", msg.Peer.Kind)
}
if msg.Peer.ID != "room-1/msg-root" {
t.Fatalf("peer ID = %q, want %q", msg.Peer.ID, "room-1/msg-root")
}
if msg.Metadata["room_id"] != "room-1" {
t.Fatalf("metadata room_id = %q, want room-1", msg.Metadata["room_id"])
}
if msg.Metadata["topic_id"] != "msg-root" {
t.Fatalf("metadata topic_id = %q, want msg-root", msg.Metadata["topic_id"])
}
if msg.Metadata["parent_peer_kind"] != "topic" {
t.Fatalf("metadata parent_peer_kind = %q, want topic", msg.Metadata["parent_peer_kind"])
}
if msg.Metadata["parent_peer_id"] != "msg-root" {
t.Fatalf("metadata parent_peer_id = %q, want msg-root", msg.Metadata["parent_peer_id"])
}
case <-time.After(50 * time.Millisecond):
t.Fatal("timed out waiting for inbound message")
}
}

func TestSendThreadMessageIncludesTopicContext(t *testing.T) {
var got map[string]any
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/api/bots/u-manager/messages/send" {
http.NotFound(w, r)
return
}
body, err := io.ReadAll(r.Body)
if err != nil {
t.Fatalf("ReadAll() error = %v", err)
}
if err := json.Unmarshal(body, &got); err != nil {
t.Fatalf("Unmarshal(%s) error = %v", string(body), err)
}
w.WriteHeader(http.StatusOK)
}))
defer server.Close()

mb := bus.NewMessageBus()
defer mb.Close()

ch, err := NewChannel(config.CSGClawConfig{
BaseURL: server.URL,
BotID: "u-manager",
AccessToken: "secret",
}, mb)
if err != nil {
t.Fatalf("NewChannel() error = %v", err)
}
ch.SetRunning(true)

if err := ch.Send(context.Background(), bus.OutboundMessage{
ChatID: "room-1/msg-root",
Content: "thread answer",
}); err != nil {
t.Fatalf("Send() error = %v", err)
}

if got["room_id"] != "room-1" {
t.Fatalf("room_id = %v, want room-1; payload=%v", got["room_id"], got)
}
if got["text"] != "thread answer" {
t.Fatalf("text = %v, want thread answer; payload=%v", got["text"], got)
}
if got["topic_id"] != "msg-root" {
t.Fatalf("topic_id = %v, want msg-root; payload=%v", got["topic_id"], got)
}
contextPayload, ok := got["context"].(map[string]any)
if !ok {
t.Fatalf("context = %T, want object; payload=%v", got["context"], got)
}
if contextPayload["chat_id"] != "room-1" {
t.Fatalf("context.chat_id = %v, want room-1; payload=%v", contextPayload["chat_id"], got)
}
if contextPayload["topic_id"] != "msg-root" {
t.Fatalf("context.topic_id = %v, want msg-root; payload=%v", contextPayload["topic_id"], got)
}
}

func TestHandleInboundEventConsumesRoomIDPayload(t *testing.T) {
mb := bus.NewMessageBus()
defer mb.Close()
Expand Down
Loading