fix(typesense): update tasks in Typesense directly when the change happened

Resolves https://community.vikunja.io/t/no-filters-working-assignee-date-task-done-etc/1910

pkg/cmd/index.go

@@ -26,6 +26,7 @@ import (
 
 func init() {
 	rootCmd.AddCommand(indexCmd)
+	rootCmd.AddCommand(partialReindexCmd)
 }
 
 var indexCmd = &cobra.Command{
@@ -35,8 +36,8 @@ var indexCmd = &cobra.Command{
 		initialize.FullInitWithoutAsync()
 	},
 	Run: func(cmd *cobra.Command, args []string) {
-		if !config.TypesenseEnabled.GetBool() {
-			log.Error("Typesense not enabled")
+		if config.TypesenseURL.GetString() == "" {
+			log.Error("Typesense not configured")
 			return
 		}
 
@@ -56,3 +57,32 @@ var indexCmd = &cobra.Command{
 		log.Infof("Done!")
 	},
 }
+
+var partialReindexCmd = &cobra.Command{
+	Use:   "partial-index",
+	Short: "Reindex any tasks which were not indexed yet into Typesense. This will not remove any existing index.",
+	PreRun: func(cmd *cobra.Command, args []string) {
+		initialize.FullInitWithoutAsync()
+	},
+	Run: func(cmd *cobra.Command, args []string) {
+		if config.TypesenseURL.GetString() == "" {
+			log.Error("Typesense not configured")
+			return
+		}
+
+		log.Infof("Indexing… This may take a while.")
+
+		err := models.CreateTypesenseCollections()
+		if err != nil {
+			log.Criticalf("Could not create Typesense collections: %s", err.Error())
+			return
+		}
+		err = models.SyncUpdatedTasksIntoTypesense()
+		if err != nil {
+			log.Criticalf("Could not reindex all changed tasks into Typesense: %s", err.Error())
+			return
+		}
+
+		log.Infof("Done!")
+	},
+}

pkg/initialize/init.go

@@ -95,7 +95,6 @@ func FullInit() {
 	models.RegisterUserDeletionCron()
 	models.RegisterOldExportCleanupCron()
 	openid.CleanupSavedOpenIDProviders()
-	models.RegisterPeriodicTypesenseResyncCron()
 
 	// Start processing events
 	go func() {

pkg/models/listeners.go

@@ -70,6 +70,7 @@ func RegisterListeners() {
 	if config.TypesenseEnabled.GetBool() {
 		events.RegisterListener((&TaskDeletedEvent{}).Name(), &RemoveTaskFromTypesense{})
 		events.RegisterListener((&TaskCreatedEvent{}).Name(), &AddTaskToTypesense{})
+		events.RegisterListener((&TaskUpdatedEvent{}).Name(), &UpdateTaskInTypesense{})
 	}
 	if config.WebhooksEnabled.GetBool() {
 		RegisterEventForWebhook(&TaskCreatedEvent{})
@@ -511,7 +512,7 @@ type RemoveTaskFromTypesense struct {
 
 // Name defines the name for the RemoveTaskFromTypesense listener
 func (s *RemoveTaskFromTypesense) Name() string {
-	return "remove.task.from.typesense"
+	return "typesense.task.remove"
 }
 
 // Handle is executed when the event RemoveTaskFromTypesense listens on is fired
@@ -537,7 +538,7 @@ type AddTaskToTypesense struct {
 
 // Name defines the name for the AddTaskToTypesense listener
 func (l *AddTaskToTypesense) Name() string {
-	return "add.task.to.typesense"
+	return "typesense.task.add"
 }
 
 // Handle is executed when the event AddTaskToTypesense listens on is fired
@@ -563,6 +564,32 @@ func (l *AddTaskToTypesense) Handle(msg *message.Message) (err error) {
 	return
 }
 
+// UpdateTaskInTypesense  represents a listener
+type UpdateTaskInTypesense struct {
+}
+
+// Name defines the name for the UpdateTaskInTypesense listener
+func (l *UpdateTaskInTypesense) Name() string {
+	return "typesense.task.update"
+}
+
+// Handle is executed when the event UpdateTaskInTypesense listens on is fired
+func (l *UpdateTaskInTypesense) Handle(msg *message.Message) (err error) {
+	event := &TaskUpdatedEvent{}
+	err = json.Unmarshal(msg.Payload, event)
+	if err != nil {
+		return err
+	}
+
+	s := db.NewSession()
+	defer s.Close()
+
+	task := make(map[int64]*Task, 1)
+	task[event.Task.ID] = event.Task // Will be filled with all data by the Typesense connector
+
+	return reindexTasksInTypesense(s, task)
+}
+
 // IncreaseAttachmentCounter  represents a listener
 type IncreaseAttachmentCounter struct {
 }

pkg/models/typesense.go

@@ -22,7 +22,6 @@ import (
 	"time"
 
 	"code.vikunja.io/api/pkg/config"
-	"code.vikunja.io/api/pkg/cron"
 	"code.vikunja.io/api/pkg/db"
 	"code.vikunja.io/api/pkg/log"
 	"code.vikunja.io/api/pkg/user"
@@ -232,7 +231,7 @@ func ReindexAllTasks() (err error) {
 		return fmt.Errorf("could not index dummy task: %w", err)
 	}
 
-	err = reindexTasks(s, tasks)
+	err = reindexTasksInTypesense(s, tasks)
 	if err != nil {
 		return fmt.Errorf("could not reindex all tasks: %s", err.Error())
 	}
@@ -278,7 +277,7 @@ func getTypesenseTaskForTask(s *xorm.Session, task *Task, projectsCache map[int6
 	return
 }
 
-func reindexTasks(s *xorm.Session, tasks map[int64]*Task) (err error) {
+func reindexTasksInTypesense(s *xorm.Session, tasks map[int64]*Task) (err error) {
 
 	if len(tasks) == 0 {
 		log.Infof("No tasks to index")
@@ -314,6 +313,8 @@ func reindexTasks(s *xorm.Session, tasks map[int64]*Task) (err error) {
 		return err
 	}
 
+	log.Debugf("Indexed tasks %v into Typesense", tasks)
+
 	return nil
 }
 
@@ -475,6 +476,8 @@ func convertTaskToTypesenseTask(task *Task) *typesenseTask {
 	return tt
 }
 
+// This function is only used to catch up with the Typesense Sync when it didn't index for some reason
+
 func SyncUpdatedTasksIntoTypesense() (err error) {
 	tasks := make(map[int64]*Task)
 
@@ -517,7 +520,7 @@ func SyncUpdatedTasksIntoTypesense() (err error) {
 	if len(tasks) > 0 {
 		log.Debugf("[Typesense Sync] Updating %d tasks", len(tasks))
 
-		err = reindexTasks(s, tasks)
+		err = reindexTasksInTypesense(s, tasks)
 		if err != nil {
 			_ = s.Rollback()
 			return
@@ -539,20 +542,3 @@ func SyncUpdatedTasksIntoTypesense() (err error) {
 
 	return s.Commit()
 }
-
-func RegisterPeriodicTypesenseResyncCron() {
-	if !config.TypesenseEnabled.GetBool() {
-		log.Debugf("[Typesense Sync] Typesense is disabled, not setting up sync cron")
-		return
-	}
-
-	err := cron.Schedule("* * * * *", func() {
-		err := SyncUpdatedTasksIntoTypesense()
-		if err != nil {
-			log.Fatalf("[Typesense Sync] Could not sync updated tasks into typesense: %s", err)
-		}
-	})
-	if err != nil {
-		log.Fatalf("[Typesense Sync] Could not register typesense resync cron: %s", err)
-	}
-}