fix: flush new subject to search engine at once

bangumi · Oct 14, 2024 · dd54b7a · dd54b7a
1 parent 09be440
commit dd54b7a
Show file tree

Hide file tree

Showing 7 changed files with 80 additions and 5 deletions.
diff --git a/canal/on_subject.go b/canal/on_subject.go
@@ -44,7 +44,11 @@ func (e *eventHandler) OnSubjectField(ctx context.Context, key json.RawMessage,
 
 func (e *eventHandler) onSubjectChange(ctx context.Context, subjectID model.SubjectID, op string) error {
 	switch op {
-	case opCreate, opUpdate, opSnapshot:
+	case opCreate:
+		if err := e.search.OnSubjectAdded(ctx, subjectID); err != nil {
+			return errgo.Wrap(err, "search.OnSubjectAdded")
+		}
+	case opUpdate, opSnapshot:
 		if err := e.search.OnSubjectUpdate(ctx, subjectID); err != nil {
 			return errgo.Wrap(err, "search.OnSubjectUpdate")
 		}

diff --git a/go.mod b/go.mod
@@ -32,7 +32,7 @@ require (
 	github.com/trim21/errgo v0.0.3
 	github.com/trim21/go-phpserialize v0.1.0-alpha.5
 	github.com/trim21/htest v0.0.4
-	github.com/trim21/pkg v0.0.3
+	github.com/trim21/pkg v0.0.4
 	go.uber.org/fx v1.23.0
 	go.uber.org/zap v1.27.0
 	golang.org/x/crypto v0.28.0

diff --git a/go.sum b/go.sum
@@ -195,8 +195,8 @@ github.com/trim21/go-phpserialize v0.1.0-alpha.5 h1:bMsUpfwAgPggQzDKdafNBvkPWDCM
 github.com/trim21/go-phpserialize v0.1.0-alpha.5/go.mod h1:/3zMYuOzpcKOevwP3ZN0WxdVRaB3CzJh5T2i41QPgRQ=
 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.3 h1:uAqfoFmmYiIMOSretKj8/tvrQs3KG57020Ff0cx8UtE=
-github.com/trim21/pkg v0.0.3/go.mod h1:JrRIFidkCLeuU5j0vBP5ZN0NOp2JavagHZNr4D3AH6Q=
+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/mocks/SearchClient.go b/internal/mocks/SearchClient.go
diff --git a/internal/search/client.go b/internal/search/client.go
@@ -150,6 +150,26 @@ func (c *client) Close() {
 	}
 }
 
+// OnSubjectAdded is the hook called by canal.
+func (c *client) OnSubjectAdded(ctx context.Context, id model.SubjectID) error {
+	s, err := c.subjectRepo.Get(ctx, id, subject.Filter{})
+	if err != nil {
+		if errors.Is(err, gerr.ErrNotFound) {
+			return nil
+		}
+		return errgo.Wrap(err, "subjectRepo.Get")
+	}
+
+	if s.Redirect != 0 || s.Ban != 0 {
+		return c.OnSubjectDelete(ctx, id)
+	}
+
+	extracted := extractSubject(&s)
+
+	_, err = c.subjectIndex.UpdateDocumentsWithContext(ctx, extracted, "id")
+	return err
+}
+
 // OnSubjectUpdate is the hook called by canal.
 func (c *client) OnSubjectUpdate(ctx context.Context, id model.SubjectID) error {
 	s, err := c.subjectRepo.Get(ctx, id, subject.Filter{})

diff --git a/internal/search/handle.go b/internal/search/handle.go
@@ -38,8 +38,10 @@ import (
 
 type Client interface {
 	Handler
-	OnSubjectUpdate(ctx context.Context, id model.SubjectID) error
 	Close()
+
+	OnSubjectAdded(ctx context.Context, id model.SubjectID) error
+	OnSubjectUpdate(ctx context.Context, id model.SubjectID) error
 	OnSubjectDelete(ctx context.Context, id model.SubjectID) error
 }
 

diff --git a/internal/search/noop.go b/internal/search/noop.go
@@ -28,6 +28,8 @@ var _ Client = NoopClient{}
 type NoopClient struct {
 }
 
+func (n NoopClient) OnSubjectAdded(ctx context.Context, id model.SubjectID) error { return nil }
+
 func (n NoopClient) Handle(c echo.Context) error {
 	return c.String(http.StatusOK, "search is not enable")
 }