feat: implement smr queue #119

LemonNekoGH · 2023-05-26T14:50:30Z

No description provided.

pkg/bots/slackbot/services/services_test.go

nekomeowww · 2023-05-31T06:25:10Z

pkg/bots/slackbot/slackbot.go

+	if !errors.Is(err, context.Canceled) {
+		s.logger.WithField("error", err.Error()).Error("slack bot server shutdown failed")
+		return err
+	}


不论如何都打印。cancelled 说明 graceful 耗时太长了，如果是正常情况，是不是要改 stop timeout 阈值？

nekomeowww · 2023-05-31T06:25:30Z

pkg/bots/slackbot/slackbot.go

+
+func (s *BotService) Stop(ctx context.Context) error {
+	err := s.server.Shutdown(ctx)
+


没必要多空一行？

nekomeowww · 2023-05-31T06:25:38Z

pkg/bots/slackbot/slackbot.go

+
+func (s *BotService) Stop(ctx context.Context) error {
+	err := s.server.Shutdown(ctx)
+


Suggested change

nekomeowww · 2023-05-31T06:31:02Z

pkg/bots/discordbot/discordbot.go

+	b.logger.Info("discord: shutting down...")
+	b.Close(ctx)
+	b.webhookStarted = false


给 health check 用的没必要设回 false，要不然会触发告警的

nekomeowww · 2023-05-31T06:31:58Z

internal/services/smr/utils.go

+
+	parsedURL, err2 := url.Parse(urlString)
+	if err2 != nil {
+		return ErrParse


Suggested change

return ErrParse

return fmt.Errorf("%w: %w", err2, ErrParse)

nekomeowww · 2023-05-31T06:32:12Z

internal/services/smr/utils.go

+		return ErrParse
+	}
+	if parsedURL.Scheme == "" || !lo.Contains([]string{"http", "https"}, parsedURL.Scheme) {
+		return ErrScheme


Suggested change

return ErrScheme

return fmt.Errorf("%w: %w", err2, ErrScheme)

要正确传递 err 链，要不然上层日志打印的时候没办法排错了

nekomeowww · 2023-05-31T06:34:25Z

internal/services/smr/taskmgr.go

+		return info, err
+	}
+
+	err = json.Unmarshal([]byte(res[1]), &info)


好像没保证 res 有 len() == 2？

以及为什么是 index 1？

nekomeowww · 2023-05-31T06:36:10Z

internal/services/smr/taskmgr.go

+	close(s.closeChan)
+}
+
+func (s *Service) getTask() (types.TaskInfo, error) {


用 pointer pattern 省去调用方解 err

nekomeowww · 2023-05-31T06:36:31Z

internal/services/smr/taskmgr.go

+	s.closeChan <- struct{}{}
+	close(s.closeChan)


用 context.CancelFunc 关？

nekomeowww · 2023-05-31T06:38:03Z

internal/services/smr/taskmgr.go

+		if needToClose {
+			break
+		}


可以放 select 上面方便读

nekomeowww · 2023-05-31T06:42:30Z

internal/services/services.go

@@ -12,5 +13,6 @@ func NewModules() fx.Option {
 		fx.Provide(health.NewHealth()),
 		fx.Provide(pprof.NewPprof()),
 		fx.Provide(autorecap.NewAutoRecapService()),
+		fx.Provide(smr.NewService()),


其实我没有太理解为什么要单独加一个 service，是为了避免并发造成的 OOM 吗？

nekomeowww · 2023-05-31T07:01:32Z

internal/bots/telegram/handlers/summarize/smr_command.go

-	defer cancel()
+	err = h.smrService.AddTask(types.TaskInfo{
+		Platform:  smr.FromPlatformTelegram,
+		Url:       urlString,


nekomeowww · 2023-05-31T15:59:09Z

internal/bots/slack/handlers/handlers.go

+	var body receivedCommandInfo
+	if err := ctx.Bind(&body); err != nil {
+		ctx.AbortWithStatus(http.StatusBadRequest)
+		h.logger.WithField("error", err.Error()).Warn("failed to bind request body, maybe slack request definition changed")


Suggested change

h.logger.WithField("error", err.Error()).Warn("failed to bind request body, maybe slack request definition changed")

h.logger.WithField("error", err.Error()).Warn("failed to bind request body, type definition of slack request body may have changed")

nekomeowww · 2023-05-31T16:00:08Z

internal/bots/slack/handlers/handlers.go

+	h.logger.WithFields(logrus.Fields{
+		"user_id":    body.UserID,
+		"channel_id": body.ChannelID,
+	}).Infof("slack: command received: /smr %s", body.Text)


Suggested change

}).Infof("slack: command received: /smr %s", body.Text)

}).Trace("slack: command received: /smr %s", body.Text)

nekomeowww · 2023-05-31T16:01:21Z

internal/bots/slack/handlers/handlers.go

+	err := smrutils.CheckUrl(urlString)
+	if err != nil {
+		if smrutils.IsUrlCheckError(err) {
+			ctx.JSON(http.StatusOK, slackbot.NewSlackWebhookMessage(err.Error()))


这种也可以用 AbortWithStatus 么？

不可以，Slack 会提示 smr 因为 dispatch 错误而失败

nekomeowww · 2023-05-31T16:04:19Z

internal/services/smr/error.go

+	ErrNoLink = errors.New("没有找到链接，可以发送一个有效的链接吗？用法：/smr <链接>")
+	ErrParse  = errors.New("你发来的链接无法被理解，可以重新发一个试试。用法：/smr <链接>")
+	ErrScheme = errors.New("你发来的链接无法被理解，可以重新发一个试试。用法：/smr <链接>")


最好不要把这种有程序错误作为上游的错误文案放到 error 里面，文案应该是业务 specific 的。比如 Telegram 的 /smr <链接> 应该用 <code></code> 扩起来，其他平台不行。

nekomeowww · 2023-05-31T16:04:45Z

internal/services/smr/processor.go

+		return result.FormatSummarizationAsDiscordMarkdown()
+	}
+
+	return ""


直接写 default 吧

nekomeowww · 2023-05-31T16:05:47Z

internal/services/smr/processor.go

+
+		_, err := s.tgBot.Send(msgEdit)
+		if err != nil {
+			s.logger.WithError(err).WithField("platform", info.Platform).Warn("smr service: failed to send result message")


太长了，链式调用换个行吧

nekomeowww · 2023-05-31T16:06:12Z

internal/services/smr/processor.go

+
+		_, err := s.tgBot.Send(msgEdit)
+		if err != nil {
+			s.logger.WithError(err).WithField("platform", info.Platform).Warn("smr service: failed to send result message")


Suggested change

s.logger.WithError(err).WithField("platform", info.Platform).Warn("smr service: failed to send result message")

s.logger.WithError(err).

WithField("platform", info.Platform).

Warn("smr service: failed to send result message")

nekomeowww · 2023-05-31T16:06:41Z

internal/services/smr/processor.go

+			First(context.Background())
+
+		if err != nil {
+			s.logger.WithError(err).WithField("platform", info.Platform).Warn("smr service: failed to get team's access token")


Suggested change

s.logger.WithError(err).WithField("platform", info.Platform).Warn("smr service: failed to get team's access token")

s.logger.WithError(err).

WithField("platform", info.Platform).

Warn("smr service: failed to get team's access token")

nekomeowww · 2023-05-31T16:06:51Z

internal/services/smr/processor.go

+		)
+
+		if err != nil {
+			s.logger.WithError(err).WithField("platform", info.Platform).Warn("smr service: failed to send result message")


Suggested change

s.logger.WithError(err).WithField("platform", info.Platform).Warn("smr service: failed to send result message")

s.logger.WithError(err).

WithField("platform", info.Platform).

Warn("smr service: failed to send result message")

nekomeowww · 2023-05-31T16:07:02Z

internal/services/smr/processor.go

+			)
+
+		if err != nil {
+			s.logger.WithError(err).WithField("platform", info.Platform).Warn("smr service: failed to send result message")


Suggested change

s.logger.WithError(err).WithField("platform", info.Platform).Warn("smr service: failed to send result message")

s.logger.WithError(err).

WithField("platform", info.Platform).

Warn("smr service: failed to send result message")

nekomeowww · 2023-05-31T16:07:39Z

internal/services/smr/processor.go

+	}
+}
+
+func (s *Service) botExists(platform smr.FromPlatform) bool {


返回 bool 的函数用动词 verb 开头，isBotExists 或者 hasBot

nekomeowww · 2023-05-31T16:08:20Z

internal/services/smr/smr.go

+	closeChan     chan struct{}
+	alreadyClosed bool


用 lazy init 的 cancelFunc context.CancelFunc 来解决呗？

nekomeowww · 2023-05-31T16:08:43Z

internal/services/smr/smr.go

+		case <-s.closeChan:
+			s.logger.WithField("last tasks count", s.queue.Count()).Info("smr service: received stop signal, waiting for all tasks done")
+
+			needToClose = true


用 context.Context.Done() 可以吗

确实，这样优雅一些

internal/services/smr/smr.go

nekomeowww · 2023-05-31T16:32:32Z

internal/services/smr/types/types.go

+type OngoingTaskPool struct {
+	tasks []TaskInfo
+	mu    *sync.RWMutex
+}
+
+func NewOngoingTaskPool() *OngoingTaskPool {
+	return &OngoingTaskPool{
+		mu: &sync.RWMutex{},
+	}
+}
+
+func (t *OngoingTaskPool) Add(info TaskInfo) {
+	t.mu.Lock()
+	t.tasks = append(t.tasks, info)
+	t.mu.Unlock()
+}
+
+func (t *OngoingTaskPool) Remove() {
+	t.mu.Lock()
+	if len(t.tasks) == 0 {
+		t.mu.Unlock()
+		return
+	}
+
+	t.tasks = t.tasks[1:]
+	t.mu.Unlock()
+}
+
+func (t *OngoingTaskPool) Len() int {
+	t.mu.RLock()
+	l := len(t.tasks)
+	t.mu.RUnlock()
+
+	return l
+}


internal/services/smr/smr.go L100 又定义了一个 pool，和这个有啥区别？

nekomeowww · 2023-05-31T16:32:47Z

internal/services/smr/types/types.go

+	Platform smr.FromPlatform `json:"platform"`
+	Url      string           `json:"url"` // url to summarize
+
+	ChatID    int64 `json:"chatID"`    // only for telegram
+	MessageID int   `json:"messageID"` // used to edit the reply message of request, not work in slack or discordbot currently
+
+	ChannelID string `json:"channelID"` // for slack and discordbot
+
+	TeamID string `json:"teamID"` // only for slack, used to query access token and refresh token


能拆成三个附属结构体的吗？

那样的话创建这个结构体实例时没那么方便了，会多出好几行

pkg/bots/discordbot/discordbot.go

nekomeowww · 2023-05-31T16:34:25Z

internal/services/smr/smrutils/utils.go

+	if urlString == "" {
+		return smr.ErrNoLink
+	}
+
+	parsedURL, err2 := url.Parse(urlString)
+	if err2 != nil {
+		return smr.ErrParse
+	}
+	if parsedURL.Scheme == "" || !lo.Contains([]string{"http", "https"}, parsedURL.Scheme) {
+		return smr.ErrScheme
+	}
+
+	return nil


能正确传递 err 链吗？写成这样的话外面怎么打日志呢？

internal/services/smr/smr.go

nekomeowww · 2023-06-01T05:46:52Z

internal/bots/slack/handlers/handlers.go

+			ctx.JSON(http.StatusOK, slackbot.NewSlackWebhookMessage(smrutils.FormatUrlCheckError(err, smr.FromPlatformSlack)))
+			return
+		}
+
+		ctx.JSON(http.StatusOK, slackbot.NewSlackWebhookMessage("出现了一些问题，可以再试试？"))


能用 ctx.AbortWithStatusJSON() 么？

nekomeowww · 2023-06-01T05:47:35Z

internal/bots/slack/handlers/handlers.go

+			ctx.JSON(http.StatusOK, slackbot.NewSlackWebhookMessage("本应用没有权限向这个频道发送消息，尝试重新安装一下？"))
+			return
+		}
+
+		ctx.JSON(http.StatusOK, slackbot.NewSlackWebhookMessage("出现了一些问题，可以再试试？"))


能用 ctx.AbortWithStatusJSON() 么？

nekomeowww · 2023-06-01T05:47:46Z

internal/bots/slack/handlers/handlers.go

+	})
+	if err != nil {
+		h.logger.WithError(err).Warn("slack: failed to add task")
+		ctx.JSON(http.StatusOK, slackbot.NewSlackWebhookMessage("量子速读请求发送失败了，可以再试试？"))


能用 ctx.AbortWithStatusJSON() 么？

nekomeowww · 2023-06-01T05:48:25Z

internal/bots/slack/slack.go

+				ctx, cancel := context.WithTimeout(ctx, time.Second*5)
+				defer cancel()


Lifecycle 本身有 timeout 信号的，不用自己写

* fix: add line break because invoke chain too long * fix: rename function botExists to isBotExists * fix: use cancel func to stop service * fix: return origin error when check url * fix: unit test * fix: lint issue * fix: remove conc * fix: require pop result length equals to 2 * fix: format error text instead * fix: change some error handling method in telegram smr command * fix: change slack command receive log level * fix: grammar issue in log text * fix: rename fields in TaskInfo * fix: remove code will cause health check alert * fix: remove redundant blank line * fix: add timeout for slack webhook server shutdown, do log for any error when shutdown * fix: rename some identifier Co-authored-by: Neko Ayaka <neko@ayaka.moe>

LemonNekoGH marked this pull request as draft May 26, 2023 14:50

LemonNekoGH linked an issue May 26, 2023 that may be closed by this pull request

improve: use a queue to manage request for smr command #110

Closed

LemonNekoGH self-assigned this May 26, 2023

LemonNekoGH force-pushed the feat/smr-queue branch 2 times, most recently from b78e8bd to 2c6dd84 Compare May 26, 2023 17:30

LemonNekoGH added 4 commits May 31, 2023 00:53

feat: implement smr queue

3440df3

fix: lint issues

74b84f8

fix: run ent generation

b77ea8d

fix: add some test, fix panic when send smr result for slack

a709ac1

LemonNekoGH force-pushed the feat/smr-queue branch from 0c803e5 to a709ac1 Compare May 30, 2023 16:54

fix: change back the log prefix

468cd52

LemonNekoGH requested a review from nekomeowww May 30, 2023 17:28

LemonNekoGH marked this pull request as ready for review May 30, 2023 17:28