admin/isphere-ai-bridge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: filter receive messages by since

Browse Source
This commit is contained in:
zhaoyilun
2026-07-10 01:27:27 +08:00
parent e7324d8aa6
commit aa636f5a33
9 changed files with 164 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
internal/isphere/source.go
View File

@@ -3,7 +3,9 @@ package isphere
import (
"context"
"fmt"
"strconv"
"strings"
"time"
"isphere-ai-bridge/internal/isphere/logcodec"
"isphere-ai-bridge/internal/isphere/packetlog"
@@ -25,6 +27,7 @@ type Message struct {
type ReceiveMessagesQuery struct {
ConversationID string
Query string
Since string
Limit int
}
@@ -37,6 +40,10 @@ type EncryptedPacketLogSource struct {
}
func (s EncryptedPacketLogSource) ReceiveMessages(ctx context.Context, query ReceiveMessagesQuery) (ReceiveMessagesResult, error) {
since, hasSince, err := parseReceiveMessagesSince(query.Since)
if err != nil {
return ReceiveMessagesResult{}, err
}
limit := query.Limit
if limit <= 0 || limit > len(s.Lines) {
limit = len(s.Lines)
@@ -57,7 +64,7 @@ func (s EncryptedPacketLogSource) ReceiveMessages(ctx context.Context, query Rec
return ReceiveMessagesResult{}, fmt.Errorf("parse packet log line %d: %w", index, err)
}
message := normalizePacketMessage(parsed)
if !receiveMessageMatchesQuery(message, query) {
if !receiveMessageMatchesQuery(message, query) || !receiveMessageMatchesSince(message, since, hasSince) {
continue
}
messages = append(messages, message)
@@ -68,6 +75,18 @@ func (s EncryptedPacketLogSource) ReceiveMessages(ctx context.Context, query Rec
return ReceiveMessagesResult{Messages: messages}, nil
}
func parseReceiveMessagesSince(value string) (time.Time, bool, error) {
trimmed := strings.TrimSpace(value)
if trimmed == "" {
return time.Time{}, false, nil
}
parsed, err := time.Parse(time.RFC3339Nano, trimmed)
if err != nil {
return time.Time{}, false, fmt.Errorf("parse receive messages since %q: %w", value, err)
}
return parsed.UTC(), true, nil
}
func receiveMessageMatchesQuery(message Message, query ReceiveMessagesQuery) bool {
if strings.TrimSpace(query.ConversationID) != "" && message.ConversationID != strings.TrimSpace(query.ConversationID) {
return false
@@ -93,6 +112,32 @@ func receiveMessageMatchesQuery(message Message, query ReceiveMessagesQuery) boo
return false
}
func receiveMessageMatchesSince(message Message, since time.Time, hasSince bool) bool {
if !hasSince {
return true
}
messageTime, ok := packetMessageTimestamp(message.Timestamp)
if !ok {
return false
}
return !messageTime.Before(since)
}
func packetMessageTimestamp(value string) (time.Time, bool) {
trimmed := strings.TrimSpace(value)
if trimmed == "" {
return time.Time{}, false
}
epoch, err := strconv.ParseInt(trimmed, 10, 64)
if err != nil {
return time.Time{}, false
}
if epoch > 1_000_000_000_000 {
return time.UnixMilli(epoch).UTC(), true
}
return time.Unix(epoch, 0).UTC(), true
}
func normalizePacketMessage(msg packetlog.Message) Message {
sender := bareJID(msg.From)
receiver := bareJID(msg.To)

41
internal/isphere/source_test.go
View File

@@ -94,6 +94,47 @@ func TestEncryptedPacketLogSourceFiltersByConversationAndQuery(t *testing.T) {
}
}
func TestEncryptedPacketLogSourceFiltersBySince(t *testing.T) {
first := `--------------------------------------------------------------------------------------------------------------------------------------------
2026/7/7 15:30:07
<message id="msg-before" from="sender@imopenfire1-lanzhou/imp_pc_4.1.2.6842" to="receiver@imopenfire1-lanzhou" type="chat">
<body>before threshold</body>
<received xmlns="urn:xmpp:receipts" id="receipt-before" 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-after" from="sender@imopenfire1-lanzhou/imp_pc_4.1.2.6842" to="receiver@imopenfire1-lanzhou" type="chat">
<body>after threshold</body>
<received xmlns="urn:xmpp:receipts" id="receipt-after" type="1" stamp="" />
<subject>TASK_UPDATE</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{
Since: "2026-07-07T11:31:00Z",
Limit: 10,
})
if err != nil {
t.Fatalf("ReceiveMessages returned error: %v", err)
}
if len(got.Messages) != 1 || got.Messages[0].ID != "msg-after" {
t.Fatalf("messages = %+v, want only msg-after", got.Messages)
}
_, err = source.ReceiveMessages(context.Background(), ReceiveMessagesQuery{Since: "not-a-time"})
if err == nil {
t.Fatalf("ReceiveMessages accepted invalid since")
}
}
func encryptPacketLineForTest(t *testing.T, plaintext string) string {
t.Helper()
block, err := des.NewCipher([]byte("hyhccdtm"))

2
internal/tools/isphere_read.go
View File

@@ -20,6 +20,7 @@ type ReceiveMessagesSource interface {
type ReceiveMessagesArgs struct {
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"`
Since string `json:"since,omitempty" jsonschema:"RFC3339 timestamp; return messages at or after this time when packet timestamps are available"`
Limit int `json:"limit,omitempty" jsonschema:"maximum number of messages to return; zero or negative returns all available messages"`
}
@@ -36,6 +37,7 @@ func RegisterISphereReadTools(server *mcp.Server, source ReceiveMessagesSource)
result, err := source.ReceiveMessages(ctx, isphere.ReceiveMessagesQuery{
ConversationID: input.ConversationID,
Query: input.Query,
Since: input.Since,
Limit: input.Limit,
})
if err != nil {

6
internal/tools/isphere_read_test.go
View File

@@ -42,7 +42,7 @@ func TestISphereReceiveMessagesToolReturnsSourceMessages(t *testing.T) {
callResult, err := session.CallTool(context.Background(), &mcp.CallToolParams{
Name: ToolNameReceiveMessages,
Arguments: map[string]any{"conversation_id": "sender@imopenfire1-lanzhou|receiver@imopenfire1-lanzhou", "query": "REPORT", "limit": 5},
Arguments: map[string]any{"conversation_id": "sender@imopenfire1-lanzhou|receiver@imopenfire1-lanzhou", "query": "REPORT", "since": "2026-07-07T11:30:00Z", "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 || 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)
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" || fake.queries[0].Since != "2026-07-07T11:30:00Z" {
t.Fatalf("queries = %+v, want conversation_id/query/since/limit", fake.queries)
}
var decoded struct {