-
Notifications
You must be signed in to change notification settings - Fork 2
feat: 🎸 Implement message prefetching and link previews #22
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.
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
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
|
|
@@ -18,6 +18,7 @@ import ( | |||||||||||||||||
| "io" | ||||||||||||||||||
| "net/http" | ||||||||||||||||||
| "os" | ||||||||||||||||||
| "regexp" | ||||||||||||||||||
| "strconv" | ||||||||||||||||||
| "strings" | ||||||||||||||||||
| "sync" | ||||||||||||||||||
|
|
@@ -473,9 +474,57 @@ func (lc *LineClient) Connect(ctx context.Context) { | |||||||||||||||||
| } | ||||||||||||||||||
|
|
||||||||||||||||||
| go lc.syncChats(ctx) | ||||||||||||||||||
| go lc.prefetchMessages(ctx) | ||||||||||||||||||
| go lc.pollLoop(ctx) | ||||||||||||||||||
| } | ||||||||||||||||||
|
|
||||||||||||||||||
| func (lc *LineClient) prefetchMessages(ctx context.Context) { | ||||||||||||||||||
| client := line.NewClient(lc.AccessToken) | ||||||||||||||||||
| opts := line.MessageBoxesOptions{ | ||||||||||||||||||
| ActiveOnly: true, | ||||||||||||||||||
| MessageBoxCountLimit: 100, | ||||||||||||||||||
| WithUnreadCount: true, | ||||||||||||||||||
| LastMessagesPerMessageBoxCount: 0, | ||||||||||||||||||
| } | ||||||||||||||||||
|
|
||||||||||||||||||
| res, err := client.GetMessageBoxes(opts) | ||||||||||||||||||
| if err != nil && lc.isRefreshRequired(err) { | ||||||||||||||||||
| if errRefresh := lc.refreshAndSave(ctx); errRefresh == nil { | ||||||||||||||||||
| client = line.NewClient(lc.AccessToken) | ||||||||||||||||||
| res, err = client.GetMessageBoxes(opts) | ||||||||||||||||||
| } | ||||||||||||||||||
| } | ||||||||||||||||||
| if err != nil { | ||||||||||||||||||
| lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to prefetch message boxes") | ||||||||||||||||||
| return | ||||||||||||||||||
| } | ||||||||||||||||||
|
|
||||||||||||||||||
| for _, box := range res.MessageBoxes { | ||||||||||||||||||
| // Fetch recent messages for all active chats to ensure history is populated | ||||||||||||||||||
| msgs, err := client.GetRecentMessagesV2(box.ID, 50) | ||||||||||||||||||
| if err != nil { | ||||||||||||||||||
| lc.UserLogin.Bridge.Log.Warn().Err(err).Str("chat_mid", box.ID).Msg("Failed to fetch recent messages") | ||||||||||||||||||
| continue | ||||||||||||||||||
| } | ||||||||||||||||||
|
|
||||||||||||||||||
| // Reverse messages to process oldest first | ||||||||||||||||||
| for i := len(msgs) - 1; i >= 0; i-- { | ||||||||||||||||||
| msg := msgs[i] | ||||||||||||||||||
|
|
||||||||||||||||||
| existing, err := lc.UserLogin.Bridge.DB.Message.GetPartByID(ctx, lc.UserLogin.ID, networkid.MessageID(msg.ID), "") | ||||||||||||||||||
| if err == nil && existing != nil { | ||||||||||||||||||
| continue | ||||||||||||||||||
| } | ||||||||||||||||||
|
|
||||||||||||||||||
| opType := 26 | ||||||||||||||||||
| if msg.From == lc.Mid { | ||||||||||||||||||
| opType = 25 | ||||||||||||||||||
| } | ||||||||||||||||||
| lc.queueIncomingMessage(msg, opType) | ||||||||||||||||||
| } | ||||||||||||||||||
| } | ||||||||||||||||||
|
Comment on lines
+481
to
+525
|
||||||||||||||||||
| } | ||||||||||||||||||
|
|
||||||||||||||||||
| func (lc *LineClient) syncChats(ctx context.Context) { | ||||||||||||||||||
| client := line.NewClient(lc.AccessToken) | ||||||||||||||||||
| midsResp, err := client.GetAllChatMids(true, true) | ||||||||||||||||||
|
|
@@ -948,25 +997,39 @@ func (lc *LineClient) queueIncomingMessage(msg *line.Message, opType int) { | |||||||||||||||||
| } | ||||||||||||||||||
| decryptedBody := bodyText | ||||||||||||||||||
|
|
||||||||||||||||||
| var ts time.Time | ||||||||||||||||||
| if tsInt, err := msg.CreatedTime.Int64(); err != nil { | ||||||||||||||||||
| lc.UserLogin.Bridge.Log.Warn(). | ||||||||||||||||||
| Err(err). | ||||||||||||||||||
| Str("msg_id", msg.ID). | ||||||||||||||||||
| Msg("Failed to convert message CreatedTime to int64, using current time") | ||||||||||||||||||
| ts = time.Now() | ||||||||||||||||||
| } else { | ||||||||||||||||||
| ts = time.UnixMilli(tsInt) | ||||||||||||||||||
| if ts.IsZero() { | ||||||||||||||||||
| ts = time.Now() | ||||||||||||||||||
| } | ||||||||||||||||||
| } | ||||||||||||||||||
|
|
||||||||||||||||||
| lc.UserLogin.Bridge.QueueRemoteEvent(lc.UserLogin, &simplevent.Message[line.Message]{ | ||||||||||||||||||
| EventMeta: simplevent.EventMeta{ | ||||||||||||||||||
| Type: bridgev2.RemoteEventMessage, | ||||||||||||||||||
| LogContext: func(c zerolog.Context) zerolog.Context { return c.Str("msg_id", msg.ID) }, | ||||||||||||||||||
| PortalKey: portalKey, | ||||||||||||||||||
| CreatePortal: true, | ||||||||||||||||||
| Sender: bridgev2.EventSender{Sender: senderID, IsFromMe: opType == 25}, | ||||||||||||||||||
| Timestamp: time.Now(), | ||||||||||||||||||
| Timestamp: ts, | ||||||||||||||||||
| }, | ||||||||||||||||||
| Data: *msg, | ||||||||||||||||||
| ID: networkid.MessageID(msg.ID), | ||||||||||||||||||
| ConvertMessageFunc: func(ctx context.Context, portal *bridgev2.Portal, intent bridgev2.MatrixAPI, data line.Message) (*bridgev2.ConvertedMessage, error) { | ||||||||||||||||||
| replyRelatesTo := lc.resolveReplyRelatesTo(ctx, &data) | ||||||||||||||||||
| // Handle Images | ||||||||||||||||||
| client := line.NewClient(lc.AccessToken) | ||||||||||||||||||
| if data.ContentType == 1 { // Image | ||||||||||||||||||
| oid := data.ContentMetadata["OID"] | ||||||||||||||||||
|
|
||||||||||||||||||
| if oid != "" { | ||||||||||||||||||
| client := line.NewClient(lc.AccessToken) | ||||||||||||||||||
| imgData, err := client.DownloadOBS(oid, data.ID) | ||||||||||||||||||
|
|
||||||||||||||||||
| // Refresh token if we get a 401 | ||||||||||||||||||
|
|
@@ -1048,7 +1111,6 @@ func (lc *LineClient) queueIncomingMessage(msg *line.Message, opType int) { | |||||||||||||||||
| } | ||||||||||||||||||
|
|
||||||||||||||||||
| if oid != "" { | ||||||||||||||||||
| client := line.NewClient(lc.AccessToken) | ||||||||||||||||||
| videoData, err := client.DownloadOBSWithSID(oid, data.ID, "emv") | ||||||||||||||||||
|
|
||||||||||||||||||
| if err != nil && (strings.Contains(err.Error(), "401") || lc.isRefreshRequired(err)) { | ||||||||||||||||||
|
|
@@ -1188,7 +1250,6 @@ func (lc *LineClient) queueIncomingMessage(msg *line.Message, opType int) { | |||||||||||||||||
| } | ||||||||||||||||||
|
|
||||||||||||||||||
| if oid != "" { | ||||||||||||||||||
| client := line.NewClient(lc.AccessToken) | ||||||||||||||||||
| fileData, err := client.DownloadOBSWithSID(oid, data.ID, "emf") | ||||||||||||||||||
| if err != nil { | ||||||||||||||||||
| lc.UserLogin.Bridge.Log.Error(). | ||||||||||||||||||
|
|
@@ -1370,15 +1431,40 @@ func (lc *LineClient) queueIncomingMessage(msg *line.Message, opType int) { | |||||||||||||||||
| } | ||||||||||||||||||
|
|
||||||||||||||||||
| // Default to Text | ||||||||||||||||||
| content := &event.MessageEventContent{ | ||||||||||||||||||
| MsgType: event.MsgText, | ||||||||||||||||||
| Body: unwrappedText, | ||||||||||||||||||
| RelatesTo: replyRelatesTo, | ||||||||||||||||||
| } | ||||||||||||||||||
|
|
||||||||||||||||||
| urlRegex := regexp.MustCompile(`(https?://)?([a-zA-Z0-9.-]+\.[a-zA-Z]{2,})(/[^\s]*)?`) | ||||||||||||||||||
| if match := urlRegex.FindString(unwrappedText); match != "" { | ||||||||||||||||||
| match = strings.TrimRight(match, ".,;:!?") | ||||||||||||||||||
| requestURL := match | ||||||||||||||||||
| if !strings.HasPrefix(match, "http") { | ||||||||||||||||||
| requestURL = "https://" + match | ||||||||||||||||||
| } | ||||||||||||||||||
| if info, err := client.GetPageInfo(requestURL); err == nil { | ||||||||||||||||||
| preview := &event.BeeperLinkPreview{ | ||||||||||||||||||
| MatchedURL: match, | ||||||||||||||||||
| LinkPreview: event.LinkPreview{ | ||||||||||||||||||
| Title: info.Title, | ||||||||||||||||||
| Description: info.Summary, | ||||||||||||||||||
| CanonicalURL: info.Domain, | ||||||||||||||||||
| }, | ||||||||||||||||||
| } | ||||||||||||||||||
|
||||||||||||||||||
| } | |
| } | |
| // Ensure a bounded timeout for any HTTP requests made by GetPageInfo. | |
| if http.DefaultClient != nil { | |
| clientCopy := *http.DefaultClient | |
| clientCopy.Timeout = 5 * time.Second | |
| http.DefaultClient = &clientCopy | |
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -538,3 +538,43 @@ func (c *Client) constructTalkMeta(messageID string) string { | |
|
|
||
| return base64.StdEncoding.EncodeToString(metaBytes) | ||
| } | ||
|
|
||
| func (c *Client) GetPageInfo(url string) (*PageInfoResult, error) { | ||
| apiURL := "https://legy-jp.line-apps.com/sc/api/v2/pageinfo/get" | ||
| req, err := http.NewRequest("GET", apiURL, nil) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| q := req.URL.Query() | ||
| q.Add("url", url) | ||
|
||
| q.Add("caller", "LINE_CHROME") | ||
| req.URL.RawQuery = q.Encode() | ||
|
|
||
| req.Header.Set("User-Agent", UserAgent) | ||
| if c.AccessToken != "" { | ||
| req.Header.Set("x-line-access", c.AccessToken) | ||
| req.Header.Set("Cookie", fmt.Sprintf("lct=%s", c.AccessToken)) | ||
| } | ||
|
|
||
| resp, err := c.HTTPClient.Do(req) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| defer resp.Body.Close() | ||
|
|
||
| if resp.StatusCode != 200 { | ||
| return nil, fmt.Errorf("pageinfo request failed: %d", resp.StatusCode) | ||
| } | ||
|
|
||
| var wrapper PageInfoResponse | ||
| if err := json.NewDecoder(resp.Body).Decode(&wrapper); err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| if wrapper.Code != 0 { | ||
| return nil, fmt.Errorf("pageinfo API error: %s", wrapper.Message) | ||
| } | ||
|
|
||
| return &wrapper.Result, nil | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
During prefetch, a database query is executed for every message (up to 5000 messages) to check if it already exists. This could create a performance bottleneck, especially with a large number of messages. Consider batching the existence checks or using a bloom filter to reduce database queries.