diff --git a/addons/slack/manifest.json b/addons/slack/manifest.json index 3154a498f9..7e765a7d85 100644 --- a/addons/slack/manifest.json +++ b/addons/slack/manifest.json @@ -6,7 +6,7 @@ "os": "linux", "arch": "amd64", "host": { "min": 1, "max": 1 }, - "api": { "min": 1, "max": 1 }, + "api": { "min": 2, "max": 2 }, "protocol": { "min": 1, "max": 1 }, "dataFormat": { "min": 1, "max": 1 }, "capabilities": { @@ -18,7 +18,8 @@ "description": "Talk to one of your saved agents from Slack, in direct messages and in channel threads.", "senderHint": "A user ID (U...), a channel ID (C...) or an @name", "delivery": "pull", - "page": "apps" + "page": "apps", + "history": { "thread": true, "maxMessages": 30 } } ] }, diff --git a/addons/slack/slacktest/fake.go b/addons/slack/slacktest/fake.go index 26ff6a994f..0076d0c4c8 100644 --- a/addons/slack/slacktest/fake.go +++ b/addons/slack/slacktest/fake.go @@ -58,6 +58,8 @@ type Server struct { refuse map[string]string limit map[string]int calls map[string]int + thread map[string][]map[string]any // by channel and thread: the messages conversations.replies returns + failing map[string]string // method -> error code, once seq int links int upgrader websocket.Upgrader @@ -93,7 +95,7 @@ func (c *conn) send(v any) error { // New starts a fake Slack. Close it when the test ends. func New() *Server { s := &Server{apps: map[string]*App{}, tickets: map[string]string{}, conns: map[string][]*conn{}, pending: map[string][]any{}, - users: map[string]map[string]any{}, files: map[string]file{}, uploads: map[string]*Upload{}, refuse: map[string]string{}, limit: map[string]int{}, calls: map[string]int{}} + users: map[string]map[string]any{}, files: map[string]file{}, uploads: map[string]*Upload{}, refuse: map[string]string{}, limit: map[string]int{}, calls: map[string]int{}, thread: map[string][]map[string]any{}, failing: map[string]string{}} mux := http.NewServeMux() mux.HandleFunc("/api/", s.api) mux.HandleFunc("/link", s.link) @@ -145,6 +147,27 @@ func (s *Server) AddUser(id, name, real string) { s.users[id] = map[string]any{"id": id, "name": name, "profile": map[string]any{"real_name": real, "display_name": ""}} } +// AddThreadMessage appends a message to a thread, as conversations.replies +// returns it. The first message of a thread is its parent: ts equals thread. +// fields are the keys of a Slack message, such as user, text, files or bot_id; +// ts and thread_ts are set here. +func (s *Server) AddThreadMessage(channel, thread, ts string, fields map[string]any) { + m := map[string]any{"type": "message", "ts": ts, "thread_ts": thread} + for k, v := range fields { + m[k] = v + } + s.mu.Lock() + defer s.mu.Unlock() + s.thread[channel+"\x00"+thread] = append(s.thread[channel+"\x00"+thread], m) +} + +// FailOnce makes the next call of a method answer ok:false with code. +func (s *Server) FailOnce(method, code string) { + s.mu.Lock() + defer s.mu.Unlock() + s.failing[method] = code +} + // AddFile registers a file that a message can carry. func (s *Server) AddFile(id, name, mime string, data []byte) { s.mu.Lock() @@ -313,6 +336,10 @@ func (s *Server) api(w http.ResponseWriter, r *http.Request) { return } code := s.refuse[token] + if once, ok := s.failing[method]; ok { + delete(s.failing, method) + code = once + } app := s.apps[token] s.mu.Unlock() switch { @@ -365,6 +392,8 @@ func (s *Server) api(w http.ResponseWriter, r *http.Request) { s.posts = append(s.posts, Post{Token: token, Channel: channel, Thread: thread, Text: text, TS: ts, Edited: method == "chat.update"}) s.mu.Unlock() s.reply(w, map[string]any{"ok": true, "channel": channel, "ts": ts}) + case "conversations.replies": + s.replies(w, r) case "users.info": s.mu.Lock() u, ok := s.users[r.Form.Get("user")] @@ -420,6 +449,41 @@ func (s *Server) api(w http.ResponseWriter, r *http.Request) { } } +// replies answers conversations.replies: the messages of a thread, oldest +// first, those before latest when it is given, a page of limit messages from +// the cursor on. +func (s *Server) replies(w http.ResponseWriter, r *http.Request) { + channel, thread := r.Form.Get("channel"), r.Form.Get("ts") + s.mu.Lock() + all := append([]map[string]any(nil), s.thread[channel+"\x00"+thread]...) + s.mu.Unlock() + if len(all) == 0 { + s.reply(w, map[string]any{"ok": false, "error": "thread_not_found"}) + return + } + if latest, err := strconv.ParseFloat(r.Form.Get("latest"), 64); err == nil { + kept := all[:0:0] + for _, m := range all { + if t, _ := strconv.ParseFloat(m["ts"].(string), 64); t < latest { + kept = append(kept, m) + } + } + all = kept + } + limit, _ := strconv.Atoi(r.Form.Get("limit")) + if limit < 1 || limit > 1000 { + limit = 100 + } + from, _ := strconv.Atoi(r.Form.Get("cursor")) + from = min(max(from, 0), len(all)) + to := min(from+limit, len(all)) + out := map[string]any{"ok": true, "messages": all[from:to], "has_more": to < len(all)} + if to < len(all) { + out["response_metadata"] = map[string]any{"next_cursor": strconv.Itoa(to)} + } + s.reply(w, out) +} + func (s *Server) download(w http.ResponseWriter, r *http.Request) { s.mu.Lock() app := s.apps[bearer(r)] diff --git a/addons/slack/transport/api.go b/addons/slack/transport/api.go index 144e65509d..864ba1c77c 100644 --- a/addons/slack/transport/api.go +++ b/addons/slack/transport/api.go @@ -205,6 +205,43 @@ func (a *API) UserInfo(ctx context.Context, bot, id string) (User, error) { return r.User, err } +// Reply is a message of a thread as conversations.replies returns it. +type Reply struct { + User string `json:"user"` + BotID string `json:"bot_id"` + Username string `json:"username"` + Subtype string `json:"subtype"` + Text string `json:"text"` + TS string `json:"ts"` + Files []struct { + Name string `json:"name"` + ID string `json:"id"` + } `json:"files"` +} + +// Replies returns one page of the messages of a thread, oldest first, the +// parent message included. Only messages before latest are returned, when it +// is set. It needs the scopes channels:history, groups:history, im:history and +// mpim:history, by the kind of the conversation. The cursor is empty for the +// first page; the next one is empty on the last. +func (a *API) Replies(ctx context.Context, bot, channel, thread, latest, cursor string, limit int) ([]Reply, string, error) { + form := url.Values{"channel": {channel}, "ts": {thread}, "limit": {strconv.Itoa(limit)}} + if latest != "" { + form.Set("latest", latest) + } + if cursor != "" { + form.Set("cursor", cursor) + } + var r struct { + Messages []Reply `json:"messages"` + Meta struct { + Next string `json:"next_cursor"` + } `json:"response_metadata"` + } + err := a.call(ctx, bot, "conversations.replies", form, &r) + return r.Messages, r.Meta.Next, err +} + // File is a file that a message carries. type File struct { ID string `json:"id"` diff --git a/addons/slack/transport/history_test.go b/addons/slack/transport/history_test.go new file mode 100644 index 0000000000..a6061ddff3 --- /dev/null +++ b/addons/slack/transport/history_test.go @@ -0,0 +1,104 @@ +// SPDX-License-Identifier: MIT +package transport_test + +import ( + "context" + "fmt" + "strings" + + "github.com/mudler/LocalAI/addons/slack/slacktest" + "github.com/mudler/LocalAI/pkg/addons/contract" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("the history of a thread", func() { + ctx := context.Background() + history := func(f *slacktest.Server, limit int, before string) contract.ConnectorResponse { + r := runtimeOf(f) + out, err := r.Connector(ctx, contract.ConnectorRequest{Op: "history", Kind: "slack", Token: secret, Chat: "C1", Thread: "100.000001", Limit: limit, Before: before}) + Expect(err).NotTo(HaveOccurred()) + return out + } + add := func(f *slacktest.Server, ts string, fields map[string]any) { + f.AddThreadMessage("C1", "100.000001", ts, fields) + } + + It("returns the messages oldest first with names, the bot's own marked, and files by name", func() { + f := fake() + f.AddUser("U2", "bo", "Bo") + add(f, "100.000001", map[string]any{"user": "U1", "text": "the deploy failed"}) + add(f, "100.000002", map[string]any{"user": "U2", "text": "retrying", "files": []any{map[string]any{"id": "F1", "name": "out.txt"}}}) + add(f, "100.000003", map[string]any{"user": "UBOT", "text": "Done."}) + add(f, "100.000004", map[string]any{"user": "U1", "text": "<@UBOT> and now?"}) + out := history(f, 30, "100.000004") + Expect(out.Error).To(BeNil()) + Expect(out.Truncated).To(BeFalse()) + Expect(out.Messages).To(HaveLen(3), "the message named by before is not returned") + m := out.Messages + Expect([]string{m[0].SenderName, m[0].SenderUsername, m[0].Text}).To(Equal([]string{"Ada Lovelace", "ada", "the deploy log (https://x.example) failed"})) + Expect(m[1].SenderName).To(Equal("Bo")) + Expect([]any{m[1].Media, m[1].MediaNames}).To(Equal([]any{1, []string{"out.txt"}})) + Expect(m[2].Own).To(BeTrue()) + Expect(m[0].Own).To(BeFalse()) + Expect(m[0].TS.Unix()).To(Equal(int64(100))) + }) + + It("keeps the newest messages within the limit and says that older ones exist", func() { + f := fake() + for i := 1; i <= 10; i++ { + add(f, fmt.Sprintf("100.%06d", i), map[string]any{"user": "U1", "text": fmt.Sprintf("m%d", i)}) + } + out := history(f, 4, "") + Expect(out.Truncated).To(BeTrue()) + Expect(out.Messages).To(HaveLen(4)) + Expect(out.Messages[0].Text).To(Equal("m7")) + Expect(out.Messages[3].Text).To(Equal("m10")) + }) + + It("reads a long thread page by page and looks each person up once", func() { + f := fake() + for i := 1; i <= 450; i++ { + add(f, fmt.Sprintf("100.%06d", i), map[string]any{"user": "U1", "text": fmt.Sprintf("m%d", i)}) + } + out := history(f, 3, "") + Expect(out.Messages[2].Text).To(Equal("m450")) + Expect(f.Calls("conversations.replies")).To(Equal(3)) + Expect(f.Calls("users.info")).To(Equal(1)) + }) + + It("refuses a thread that is longer than one history reads, rather than show its middle", func() { + f := fake() + for i := 1; i <= 1100; i++ { + add(f, fmt.Sprintf("100.%06d", i), map[string]any{"user": "U1", "text": "x"}) + } + out := history(f, 3, "") + Expect(out.Error).NotTo(BeNil()) + Expect(out.Error.Code).To(Equal(contract.ConnectorFailed)) + Expect(f.Calls("conversations.replies")).To(Equal(5)) + }) + + It("falls back to the user ID when the name cannot be read, and skips joins", func() { + f := fake() + add(f, "100.000001", map[string]any{"user": "U9", "text": "hello"}) + add(f, "100.000002", map[string]any{"user": "U1", "subtype": "channel_join", "text": "joined"}) + out := history(f, 30, "") + Expect(out.Messages).To(HaveLen(1)) + Expect([]string{out.Messages[0].SenderID, out.Messages[0].SenderName}).To(Equal([]string{"U9", ""})) + }) + + It("types a missing scope, a missing thread and a rate limit", func() { + f := fake() + add(f, "100.000001", map[string]any{"user": "U1", "text": "hello"}) + f.FailOnce("conversations.replies", "missing_scope") + out := history(f, 30, "") + Expect(out.Error.Code).To(Equal(contract.ConnectorFailed)) + Expect(out.Error.Message).To(ContainSubstring("missing_scope")) + f.RateLimitOnce("conversations.replies", 7) + out = history(f, 30, "") + Expect(out.Error.Code).To(Equal(contract.ConnectorRateLimited)) + Expect(out.Error.RetryAfter).To(Equal(7)) + Expect(f.Calls("conversations.replies")).To(Equal(2), "a rate limit is not waited out") + Expect(strings.Count(out.Error.Message, "xoxb")).To(BeZero()) + }) +}) diff --git a/addons/slack/transport/runtime.go b/addons/slack/transport/runtime.go index f3467d597d..4bc365edbc 100644 --- a/addons/slack/transport/runtime.go +++ b/addons/slack/transport/runtime.go @@ -8,6 +8,8 @@ import ( "errors" "fmt" "regexp" + "slices" + "strconv" "sync" "time" @@ -68,10 +70,11 @@ func (r *Runtime) Close() { r.Socket.Close() } type platform struct{ r *Runtime } var ( - _ connector.Formatter = platform{} - _ connector.Splitter = platform{} - _ connector.MediaReader = platform{} - _ connector.FileSender = platform{} + _ connector.Formatter = platform{} + _ connector.Splitter = platform{} + _ connector.MediaReader = platform{} + _ connector.FileSender = platform{} + _ connector.HistoryReader = platform{} ) func (platform) Kind() string { return Kind } @@ -223,3 +226,77 @@ func (r *Runtime) take(req contract.ConnectorRequest) (string, bool) { delete(r.working, key) return ph.ts, ok && time.Since(ph.at) < placeholderLife } + +// Limits of the history of a thread. +const ( + // repliesPage is the size of one page of conversations.replies. + repliesPage = 200 + // repliesPages is how many pages one history reads. Slack returns a thread + // oldest first, so the newest messages are at its end: a thread longer + // than repliesPages pages is refused, not shown from its middle. + repliesPages = 5 + // historyNames bounds the people whose names one history looks up. + historyNames = 30 +) + +// errThreadTooLong means the thread has more messages than one history reads. +var errThreadTooLong = errors.New("slack: the thread is too long to read") + +// History returns the earlier messages of a thread: the newest req.Limit +// before req.Before, oldest first, with the names of their authors. A message +// of the bot is marked Own. Messages that no person or app wrote (joins, +// deletions) are left out. The file bytes are never read, only their names. +func (p platform) History(ctx context.Context, req contract.ConnectorRequest) ([]contract.ConnectorHistoryMessage, bool, error) { + b, err := bot(req.Token) + if err != nil { + return nil, false, err + } + var all []Reply + cursor := "" + for page := 0; ; page++ { + if page == repliesPages { + return nil, false, errThreadTooLong + } + got, next, err := p.r.API.Replies(ctx, b, req.Chat, req.Thread, req.Before, cursor, repliesPage) + if err != nil { + return nil, false, err + } + all = append(all, got...) + if next == "" { + break + } + cursor = next + } + all = slices.DeleteFunc(all, func(m Reply) bool { + return m.TS == req.Before || (m.Subtype != "" && !personal[m.Subtype] && m.Subtype != "bot_message") + }) + more := len(all) > req.Limit + if more { + all = all[len(all)-req.Limit:] + } + self := p.r.Socket.botUser(ctx, req.Token) + users := p.r.Socket.namer(ParseTokens(req.Token)) + out := make([]contract.ConnectorHistoryMessage, 0, len(all)) + for _, m := range all { + h := contract.ConnectorHistoryMessage{MessageID: m.TS, SenderID: m.User, Text: Plain(m.Text, ""), Own: self != "" && m.User == self} + if sec, err := strconv.ParseFloat(m.TS, 64); err == nil { + h.TS = time.UnixMicro(int64(sec * 1e6)).UTC() + } + switch { + case m.User != "": + if u, ok := users(ctx, m.User); ok { + h.SenderName, h.SenderUsername = u.Display(), u.Name + } + case m.BotID != "": + h.SenderID, h.SenderName = m.BotID, m.Username + } + for _, f := range m.Files { + h.Media++ + if len(h.MediaNames) < contract.MaxHistoryMedia && f.Name != "" { + h.MediaNames = append(h.MediaNames, f.Name) + } + } + out = append(out, h) + } + return out, more, nil +} diff --git a/addons/slack/transport/socket.go b/addons/slack/transport/socket.go index 97d6e0aa86..65c0f6f4a7 100644 --- a/addons/slack/transport/socket.go +++ b/addons/slack/transport/socket.go @@ -414,3 +414,49 @@ func (s *Socket) user(ctx context.Context, ss *session, id string) (User, bool) ss.mu.Unlock() return u, true } + +// botUser is the Slack user ID of the bot of a secret: from the open link when +// there is one, else from the Web API. It is empty when it cannot be told. +func (s *Socket) botUser(ctx context.Context, secret string) string { + t := ParseTokens(secret) + s.mu.Lock() + ss := s.sessions[keyOf(t)] + s.mu.Unlock() + if ss != nil { + ss.mu.Lock() + id := ss.bot.UserID + ss.mu.Unlock() + if id != "" { + return id + } + } + id, err := s.API.AuthTest(ctx, t.Bot) + if err != nil { + return "" + } + return id.UserID +} + +// namer returns a function that looks up the name of a person. It shares the +// cache of the link of the tokens, when one is open; else it keeps a cache +// for the one call. It looks up at most historyNames people, then gives IDs. +func (s *Socket) namer(t Tokens) func(ctx context.Context, id string) (User, bool) { + s.mu.Lock() + ss := s.sessions[keyOf(t)] + s.mu.Unlock() + if ss == nil { + ss = &session{tokens: t, users: map[string]cachedUser{}} + } + asked := 0 + return func(ctx context.Context, id string) (User, bool) { + ss.mu.Lock() + c, ok := ss.users[id] + ss.mu.Unlock() + if !ok || time.Since(c.at) >= userLife { + if asked++; asked > historyNames { + return User{}, false + } + } + return s.user(ctx, ss, id) + } +} diff --git a/tests/studiobot/transport.go b/tests/studiobot/transport.go index 9ab99dbc23..35ddf8b0b3 100644 --- a/tests/studiobot/transport.go +++ b/tests/studiobot/transport.go @@ -38,7 +38,7 @@ func TelegramPaced(api *transport.API) Addon { // Slack is the Slack add-on over api. The caller closes the runtime. func Slack(rt *slack.Runtime) Addon { - return Addon{ID: "slack", Kind: contract.ConnectorKind{ID: slack.Kind, Title: "Slack", Delivery: contract.DeliveryPull}, Runtime: rt} + return Addon{ID: "slack", Kind: contract.ConnectorKind{ID: slack.Kind, Title: "Slack", Delivery: contract.DeliveryPull, History: &contract.ConnectorHistory{Thread: true, MaxMessages: 30}}, Runtime: rt} } // BotTransport is a studio.ConnectorTransport that serves the real connector