feat: filter receive messages by conversation and query

2026-07-10 01:22:17 +08:00
parent bbffa2189d
commit e7324d8aa6
9 changed files with 181 additions and 21 deletions
--- a/internal/isphere/source.go
+++ b/internal/isphere/source.go
@@ -23,7 +23,9 @@ type Message struct {
 }

 type ReceiveMessagesQuery struct {
-	Limit int
+	ConversationID string
+	Query          string
+	Limit          int
 }

 type ReceiveMessagesResult struct {
@@ -41,9 +43,6 @@ func (s EncryptedPacketLogSource) ReceiveMessages(ctx context.Context, query Rec
 	}
 	messages := make([]Message, 0, limit)
 	for index, line := range s.Lines {
-		if len(messages) == limit {
-			break
-		}
 		select {
 		case <-ctx.Done():
 			return ReceiveMessagesResult{}, ctx.Err()
@@ -57,11 +56,43 @@ func (s EncryptedPacketLogSource) ReceiveMessages(ctx context.Context, query Rec
 		if err != nil {
 			return ReceiveMessagesResult{}, fmt.Errorf("parse packet log line %d: %w", index, err)
 		}
-		messages = append(messages, normalizePacketMessage(parsed))
+		message := normalizePacketMessage(parsed)
+		if !receiveMessageMatchesQuery(message, query) {
+			continue
+		}
+		messages = append(messages, message)
+		if len(messages) == limit {
+			break
+		}
 	}
 	return ReceiveMessagesResult{Messages: messages}, nil
 }

+func receiveMessageMatchesQuery(message Message, query ReceiveMessagesQuery) bool {
+	if strings.TrimSpace(query.ConversationID) != "" && message.ConversationID != strings.TrimSpace(query.ConversationID) {
+		return false
+	}
+	keyword := strings.ToLower(strings.TrimSpace(query.Query))
+	if keyword == "" {
+		return true
+	}
+	values := []string{
+		message.ID,
+		message.ConversationID,
+		message.SenderID,
+		message.ReceiverID,
+		message.Text,
+		message.Subject,
+		message.ReceiptID,
+	}
+	for _, value := range values {
+		if strings.Contains(strings.ToLower(value), keyword) {
+			return true
+		}
+	}
+	return false
+}
+
 func normalizePacketMessage(msg packetlog.Message) Message {
 	sender := bareJID(msg.From)
 	receiver := bareJID(msg.To)
--- a/internal/isphere/source_test.go
+++ b/internal/isphere/source_test.go
@@ -42,6 +42,58 @@ func TestEncryptedPacketLogSourceReturnsNormalizedMessages(t *testing.T) {
 	}
 }

+func TestEncryptedPacketLogSourceFiltersByConversationAndQuery(t *testing.T) {
+	first := `--------------------------------------------------------------------------------------------------------------------------------------------
+2026/7/7 15:30:07
+<message id="msg-alpha" from="sender@imopenfire1-lanzhou/imp_pc_4.1.2.6842" to="receiver@imopenfire1-lanzhou" type="chat">
+  <body>alpha status update</body>
+  <received xmlns="urn:xmpp:receipts" id="receipt-alpha" type="1" stamp="" />
+  <subject>TASK_UPDATE</subject>
+  <isphere xmlns="isphere.im" type="1001" sendtime="1783423807000" version="1" />
+  <readed />
+</message>`
+	second := `--------------------------------------------------------------------------------------------------------------------------------------------
+2026/7/7 15:31:07
+<message id="msg-beta" from="project-room@conference.imopenfire1-lanzhou/imp_pc_4.1.2.6842" to="sender@imopenfire1-lanzhou" type="groupchat">
+  <body>beta budget redacted-report.docx</body>
+  <received xmlns="urn:xmpp:receipts" id="receipt-beta" type="1" stamp="" />
+  <subject>FILE_TRANSFER_CANCEL</subject>
+  <isphere xmlns="isphere.im" type="1001" sendtime="1783423867000" version="1" />
+  <readed />
+</message>`
+	source := EncryptedPacketLogSource{Lines: []string{
+		encryptPacketLineForTest(t, first),
+		encryptPacketLineForTest(t, second),
+	}}
+
+	got, err := source.ReceiveMessages(context.Background(), ReceiveMessagesQuery{
+		ConversationID: "project-room@conference.imopenfire1-lanzhou|sender@imopenfire1-lanzhou",
+		Query:          "BUDGET",
+		Limit:          10,
+	})
+	if err != nil {
+		t.Fatalf("ReceiveMessages returned error: %v", err)
+	}
+	if len(got.Messages) != 1 {
+		t.Fatalf("message count = %d, want 1: %+v", len(got.Messages), got.Messages)
+	}
+	if got.Messages[0].ID != "msg-beta" {
+		t.Fatalf("message id = %q, want msg-beta", got.Messages[0].ID)
+	}
+
+	empty, err := source.ReceiveMessages(context.Background(), ReceiveMessagesQuery{
+		ConversationID: "project-room@conference.imopenfire1-lanzhou|sender@imopenfire1-lanzhou",
+		Query:          "alpha",
+		Limit:          10,
+	})
+	if err != nil {
+		t.Fatalf("ReceiveMessages returned error for empty query: %v", err)
+	}
+	if len(empty.Messages) != 0 {
+		t.Fatalf("empty query returned %+v, want none", empty.Messages)
+	}
+}
+
 func encryptPacketLineForTest(t *testing.T, plaintext string) string {
 	t.Helper()
 	block, err := des.NewCipher([]byte("hyhccdtm"))
--- a/internal/tools/isphere_read.go
+++ b/internal/tools/isphere_read.go
@@ -18,7 +18,9 @@ type ReceiveMessagesSource interface {
 }

 type ReceiveMessagesArgs struct {
-	Limit int `json:"limit,omitempty" jsonschema:"maximum number of messages to return; zero or negative returns all available messages"`
+	ConversationID string `json:"conversation_id,omitempty" jsonschema:"exact conversation id filter for local readonly message logs"`
+	Query          string `json:"query,omitempty" jsonschema:"case-insensitive keyword filter over message text, subject, ids, and participants"`
+	Limit          int    `json:"limit,omitempty" jsonschema:"maximum number of messages to return; zero or negative returns all available messages"`
 }

 func RegisterISphereReadTools(server *mcp.Server, source ReceiveMessagesSource) {
@@ -31,7 +33,11 @@ func RegisterISphereReadTools(server *mcp.Server, source ReceiveMessagesSource)
 		Description: "Return iSphere messages from the configured read-only log-backed source.",
 	}, func(ctx context.Context, _ *mcp.CallToolRequest, input ReceiveMessagesArgs) (*mcp.CallToolResult, map[string]any, error) {
 		started := time.Now().UTC()
-		result, err := source.ReceiveMessages(ctx, isphere.ReceiveMessagesQuery{Limit: input.Limit})
+		result, err := source.ReceiveMessages(ctx, isphere.ReceiveMessagesQuery{
+			ConversationID: input.ConversationID,
+			Query:          input.Query,
+			Limit:          input.Limit,
+		})
 		if err != nil {
 			return nil, nil, err
 		}
--- a/internal/tools/isphere_read_test.go
+++ b/internal/tools/isphere_read_test.go
@@ -42,7 +42,7 @@ func TestISphereReceiveMessagesToolReturnsSourceMessages(t *testing.T) {

 	callResult, err := session.CallTool(context.Background(), &mcp.CallToolParams{
 		Name:      ToolNameReceiveMessages,
-		Arguments: map[string]any{"limit": 5},
+		Arguments: map[string]any{"conversation_id": "sender@imopenfire1-lanzhou|receiver@imopenfire1-lanzhou", "query": "REPORT", "limit": 5},
 	})
 	if err != nil {
 		t.Fatalf("call %s: %v", ToolNameReceiveMessages, err)
@@ -50,8 +50,8 @@ func TestISphereReceiveMessagesToolReturnsSourceMessages(t *testing.T) {
 	if callResult.IsError {
 		t.Fatalf("call result is error: %+v", callResult)
 	}
-	if len(fake.queries) != 1 || fake.queries[0].Limit != 5 {
-		t.Fatalf("queries = %+v, want one limit=5", fake.queries)
+	if len(fake.queries) != 1 || fake.queries[0].Limit != 5 || fake.queries[0].ConversationID != "sender@imopenfire1-lanzhou|receiver@imopenfire1-lanzhou" || fake.queries[0].Query != "REPORT" {
+		t.Fatalf("queries = %+v, want conversation_id/query/limit", fake.queries)
 	}

 	var decoded struct {