refactor(canal): remove event buffer

trim21 · trim21 · commit 49670fd55cca · 2025-03-18T05:23:48.000+08:00
diff --git a/canal/canal.go b/canal/canal.go
@@ -114,5 +114,5 @@ func Main() error {
 		}
 	})
 
-	wg.Wait()
+	return wg.Wait()
 }
diff --git a/go.mod b/go.mod
@@ -35,7 +35,6 @@ require (
 	github.com/trim21/errgo v0.0.3
 	github.com/trim21/go-phpserialize v0.1.0
 	github.com/trim21/htest v0.0.4
-	github.com/trim21/pkg v0.0.4
 	go.uber.org/fx v1.23.0
 	go.uber.org/zap v1.27.0
 	golang.org/x/sync v0.12.0
diff --git a/go.sum b/go.sum
@@ -239,8 +239,6 @@ github.com/trim21/go-phpserialize v0.1.0 h1:4wLvVYkP6Ya07M+2YweFdx4WOrydZzVyxKRD
 github.com/trim21/go-phpserialize v0.1.0/go.mod h1:99ibWRqdNj3DTX+dbHvo+L47TsoEXwArpvBPRLrvz20=
 github.com/trim21/htest v0.0.4 h1:dDIzKNdIClgtB158DlO+Xf0sfwNycmx3kfo/FJuY+eE=
 github.com/trim21/htest v0.0.4/go.mod h1:W+zaYAGCBqx38eMrMGvXrALnbcXR6OBtZiRiHahgo+E=
-github.com/trim21/pkg v0.0.4 h1:0nYODKdqNUzmUaPFvqSiR420u2uXQgIYyVyiNfH7olc=
-github.com/trim21/pkg v0.0.4/go.mod h1:edl6xdqBOJrhMuIGvcY2lg5L9cqp/hVuwHRM/kdzbMg=
 github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=
 github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
 github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
diff --git a/internal/search/character/client.go b/internal/search/character/client.go
@@ -8,7 +8,6 @@ import (
 
 	"github.com/meilisearch/meilisearch-go"
 	"github.com/trim21/errgo"
-	"github.com/trim21/pkg/queue"
 	"go.uber.org/zap"
 
 	"github.com/bangumi/server/config"
@@ -54,22 +53,12 @@ type client struct {
 	meili meilisearch.ServiceManager
 	log   *zap.Logger
 	q     *query.Query
-
-	queue *queue.Batched[searcher.Document]
-}
-
-func (c *client) Close() {
-	if c.queue != nil {
-		c.queue.Close()
-	}
 }
 
 func (c *client) canalInit(cfg config.AppConfig) error {
 	if err := searcher.ValidateConfigs(cfg); err != nil {
 		return errgo.Wrap(err, "validate search config")
 	}
-	c.queue = searcher.NewBatchQueue(cfg, c.log, c.index)
-	searcher.RegisterQueueMetrics(idx, c.queue)
 	shouldCreateIndex, err := searcher.NeedFirstRun(c.meili, idx)
 	if err != nil {
 		return err
diff --git a/internal/search/character/event.go b/internal/search/character/event.go
@@ -45,9 +45,9 @@ func (c *client) OnUpdate(ctx context.Context, id model.CharacterID) error {
 
 	extracted := extract(&s)
 
-	c.queue.Push(extracted)
+	_, err = c.index.UpdateDocumentsWithContext(ctx, extracted, "id")
 
-	return nil
+	return err
 }
 
 func (c *client) OnDelete(ctx context.Context, id model.CharacterID) error {
diff --git a/internal/search/person/client.go b/internal/search/person/client.go
@@ -8,7 +8,6 @@ import (
 
 	"github.com/meilisearch/meilisearch-go"
 	"github.com/trim21/errgo"
-	"github.com/trim21/pkg/queue"
 	"go.uber.org/zap"
 
 	"github.com/bangumi/server/config"
@@ -54,22 +53,12 @@ type client struct {
 	meili meilisearch.ServiceManager
 	log   *zap.Logger
 	q     *query.Query
-
-	queue *queue.Batched[searcher.Document]
-}
-
-func (c *client) Close() {
-	if c.queue != nil {
-		c.queue.Close()
-	}
 }
 
 func (c *client) canalInit(cfg config.AppConfig) error {
 	if err := searcher.ValidateConfigs(cfg); err != nil {
 		return errgo.Wrap(err, "validate search config")
 	}
-	c.queue = searcher.NewBatchQueue(cfg, c.log, c.index)
-	searcher.RegisterQueueMetrics(idx, c.queue)
 	shouldCreateIndex, err := searcher.NeedFirstRun(c.meili, idx)
 	if err != nil {
 		return err
diff --git a/internal/search/person/event.go b/internal/search/person/event.go
@@ -45,9 +45,9 @@ func (c *client) OnUpdate(ctx context.Context, id model.PersonID) error {
 
 	extracted := extract(&s)
 
-	c.queue.Push(extracted)
+	_, err = c.index.UpdateDocumentsWithContext(ctx, extracted, "id")
 
-	return nil
+	return err
 }
 
 func (c *client) OnDelete(ctx context.Context, id model.PersonID) error {
diff --git a/internal/search/search.go b/internal/search/search.go
@@ -143,8 +143,4 @@ func (s *Search) EventDelete(ctx context.Context, id uint32, target SearchTarget
 	return searcher.OnDelete(ctx, id)
 }
 
-func (s *Search) Close() {
-	for _, searcher := range s.searchers {
-		searcher.Close()
-	}
-}
+func (s *Search) Close() {}
diff --git a/internal/search/searcher/client.go b/internal/search/searcher/client.go
@@ -12,10 +12,8 @@ import (
 	"github.com/avast/retry-go/v4"
 	"github.com/labstack/echo/v4"
 	"github.com/meilisearch/meilisearch-go"
-	"github.com/prometheus/client_golang/prometheus"
 	"github.com/samber/lo"
 	"github.com/trim21/errgo"
-	"github.com/trim21/pkg/queue"
 	"go.uber.org/zap"
 
 	"github.com/bangumi/server/config"
@@ -25,8 +23,6 @@ import (
 type Searcher interface {
 	Handle(c echo.Context) error
 
-	Close()
-
 	OnAdded(ctx context.Context, id uint32) error
 	OnUpdate(ctx context.Context, id uint32) error
 	OnDelete(ctx context.Context, id uint32) error
@@ -140,32 +136,6 @@ func NewDedupeFunc() func([]Document) []Document {
 	}
 }
 
-func NewBatchQueue(cfg config.AppConfig, log *zap.Logger, index meilisearch.IndexManager) *queue.Batched[Document] {
-	return queue.NewBatchedDedupe(
-		NewSendBatch(log, index),
-		cfg.Search.SearchBatchSize,
-		cfg.Search.SearchBatchInterval,
-		NewDedupeFunc(),
-	)
-}
-
-func RegisterQueueMetrics(idx string, queue *queue.Batched[Document]) {
-	prometheus.DefaultRegisterer.MustRegister(
-		prometheus.NewGaugeFunc(
-			prometheus.GaugeOpts{
-				Namespace: "chii",
-				Name:      "meilisearch_queue_batch",
-				Help:      "meilisearch update queue batch size",
-				ConstLabels: prometheus.Labels{
-					"index": idx,
-				},
-			},
-			func() float64 {
-				return float64(queue.Len())
-			},
-		))
-}
-
 func GetAttributes(rt reflect.Type, tag string) *[]string {
 	var s []string
 	for i := 0; i < rt.NumField(); i++ {
diff --git a/internal/search/subject/client.go b/internal/search/subject/client.go
@@ -8,7 +8,6 @@ import (
 
 	"github.com/meilisearch/meilisearch-go"
 	"github.com/trim21/errgo"
-	"github.com/trim21/pkg/queue"
 	"go.uber.org/zap"
 
 	"github.com/bangumi/server/config"
@@ -54,23 +53,12 @@ type client struct {
 	meili meilisearch.ServiceManager
 	log   *zap.Logger
 	q     *query.Query
-
-	queue *queue.Batched[searcher.Document]
-}
-
-func (c *client) Close() {
-	if c.queue != nil {
-		c.queue.Close()
-	}
 }
 
 func (c *client) canalInit(cfg config.AppConfig) error {
 	if err := searcher.ValidateConfigs(cfg); err != nil {
 		return errgo.Wrap(err, "validate search config")
 	}
-	c.queue = searcher.NewBatchQueue(cfg, c.log, c.index)
-	searcher.RegisterQueueMetrics(idx, c.queue)
-
 	shouldCreateIndex, err := searcher.NeedFirstRun(c.meili, idx)
 	if err != nil {
 		return err
diff --git a/internal/search/subject/event.go b/internal/search/subject/event.go
@@ -46,9 +46,9 @@ func (c *client) OnUpdate(ctx context.Context, id model.SubjectID) error {
 
 	extracted := extract(&s)
 
-	c.queue.Push(extracted)
+	_, err = c.index.UpdateDocumentsWithContext(ctx, extracted, "id")
 
-	return nil
+	return err
 }
 
 func (c *client) OnDelete(ctx context.Context, id model.SubjectID) error {

Original file line number	Diff line number	Diff line change
`@@ -114,5 +114,5 @@ func Main() error {`
`114`	`114`	`}`
`115`	`115`	`})`
`116`	`116`
`117`		`- wg.Wait()`
	`117`	`+ return wg.Wait()`
`118`	`118`	`}`
Original file line number	Diff line number	Diff line change
`@@ -45,9 +45,9 @@ func (c *client) OnUpdate(ctx context.Context, id model.CharacterID) error {`
`45`	`45`
`46`	`46`	`extracted := extract(&s)`
`47`	`47`
`48`		`- c.queue.Push(extracted)`
	`48`	`+ _, err = c.index.UpdateDocumentsWithContext(ctx, extracted, "id")`
`49`	`49`
`50`		`- return nil`
	`50`	`+ return err`
`51`	`51`	`}`
`52`	`52`
`53`	`53`	`func (c *client) OnDelete(ctx context.Context, id model.CharacterID) error {`
Original file line number	Diff line number	Diff line change
`@@ -143,8 +143,4 @@ func (s *Search) EventDelete(ctx context.Context, id uint32, target SearchTarget`
`143`	`143`	`return searcher.OnDelete(ctx, id)`
`144`	`144`	`}`
`145`	`145`
`146`		`-func (s *Search) Close() {`
`147`		`- for _, searcher := range s.searchers {`
`148`		`- searcher.Close()`
`149`		`- }`
`150`		`-}`
	`146`	`+func (s *Search) Close() {}`
Original file line number	Diff line number	Diff line change
`@@ -46,9 +46,9 @@ func (c *client) OnUpdate(ctx context.Context, id model.SubjectID) error {`
`46`	`46`
`47`	`47`	`extracted := extract(&s)`
`48`	`48`
`49`		`- c.queue.Push(extracted)`
	`49`	`+ _, err = c.index.UpdateDocumentsWithContext(ctx, extracted, "id")`
`50`	`50`
`51`		`- return nil`
	`51`	`+ return err`
`52`	`52`	`}`
`53`	`53`
`54`	`54`	`func (c *client) OnDelete(ctx context.Context, id model.SubjectID) error {`