sm-journal-plugin/console/reindex.go

package console

import (
	"context"
	"errors"
	"fmt"

	"git.golem15.com/golem15/sm-journal-plugin/models"
	"git.golem15.com/golem15/summercms/modules/backpack"
	"git.golem15.com/golem15/summercms/modules/beachcomber"
	"git.golem15.com/golem15/summercms/modules/bonfire"
	"gorm.io/gorm"
)

// reindexBatch is how many posts one database read loads.
const reindexBatch = 200

// ReindexResult counts what a reindex did: posts upserted into the index,
// posts removed from it (unpublished or scheduled) and posts whose sync
// failed.
type ReindexResult struct {
	Synced  int
	Removed int
	Failed  int
}

// Reindex is journal:reindex.
func Reindex(app *backpack.App) bonfire.Command {
	return bonfire.Command{
		Name:        "journal:reindex",
		Description: "Sync every journal post into the search index",
		Long: `Upserts the search document of every published post and removes the
document of every unpublished or scheduled one, through the same path as the
sync after a save. Posts written before search was turned on reach the index
this way.

--fresh first drops the collection and recreates it from the post schema
(models.Post.SearchIndexSchema). Use it once on a collection that was
auto-created without the schema, or after the schema changed; search answers
from SQL while the collection is empty.

The command refuses, and sends nothing, when the search engine is not
configured (search.driver null, or no search.typesense.api_key) or the
Journal setting search_use_typesense is off. It ends with
"Reindexed: N synced, N removed, N failed" and exits non-zero when any post
failed.`,
		Flags: []bonfire.Flag{
			{Name: "fresh", Description: "Drop and recreate the collection from the schema before syncing", Bare: true},
		},
		Examples: []bonfire.Example{
			{Command: "journal:reindex", Description: "Sync every post into the existing collection"},
			{Command: "journal:reindex --fresh", Description: "Recreate the collection from the schema, then sync every post"},
		},
		Run: func(ctx context.Context, in bonfire.Input, out bonfire.Output) error {
			res, err := ReindexPosts(ctx, app, flagOn(in, "fresh"), out)
			if err != nil {
				return err
			}
			msg := fmt.Sprintf("Reindexed: %d synced, %d removed, %d failed", res.Synced, res.Removed, res.Failed)
			if res.Failed > 0 {
				out.Error(msg)
				return fmt.Errorf("journal: %d posts failed to sync", res.Failed)
			}
			out.Success(msg)
			return nil
		},
	}
}

// ReindexPosts syncs every post through beachcomber, in id batches. With
// fresh it drops the collection and recreates it from the schema first. It
// returns an error, before any request, when the engine is not configured
// or search_use_typesense is off. A post that fails to sync is counted and
// reported on out; the rest still run.
func ReindexPosts(ctx context.Context, app *backpack.App, fresh bool, out bonfire.Output) (ReindexResult, error) {
	var res ReindexResult
	gdb := mustDB(app)
	if gdb == nil {
		return res, errors.New("journal: database is not configured")
	}
	svc, err := beachcomber.From(app)
	if err != nil {
		return res, err
	}
	engine := svc.Engine()
	if engine == nil || !engine.Configured() {
		return res, errors.New("journal: the search engine is not configured (set search.driver and search.typesense.api_key)")
	}
	var on bool
	if err := gdb.WithContext(ctx).Raw(`SELECT search_use_typesense FROM golem15_journal_settings WHERE id = 1`).Scan(&on).Error; err != nil {
		return res, fmt.Errorf("journal: read search_use_typesense: %w", err)
	}
	if !on {
		return res, errors.New("journal: search is off; turn on search_use_typesense in the Journal settings first")
	}
	models.SetSearchGate(true)

	index := svc.IndexName(&models.Post{})
	schema := models.Post{}.SearchIndexSchema()
	if fresh {
		existed, err := beachcomber.DropIndex(ctx, engine, index)
		if err != nil {
			return res, fmt.Errorf("journal: drop %s: %w", index, err)
		}
		if existed {
			line(out, "Dropped %s", index)
		}
	}
	if err := beachcomber.EnsureIndex(ctx, engine, index, schema); err != nil {
		return res, fmt.Errorf("journal: create %s: %w", index, err)
	}

	var batch []models.Post
	err = gdb.WithContext(ctx).Model(&models.Post{}).
		Select("id", "published", "published_at").
		Order("id").
		FindInBatches(&batch, reindexBatch, func(tx *gorm.DB, _ int) error {
			for i := range batch {
				post := &batch[i]
				if err := svc.Sync(ctx, gdb, post); err != nil {
					res.Failed++
					if out != nil {
						out.Warning(fmt.Sprintf("post %d: %v", post.ID, err))
					}
					continue
				}
				if post.ShouldBeSearchable() {
					res.Synced++
				} else {
					res.Removed++
				}
			}
			return ctx.Err()
		}).Error
	if err != nil {
		return res, err
	}
	pluginLog.Info("journal: reindexed posts", "index", index, "fresh", fresh,
		"synced", res.Synced, "removed", res.Removed, "failed", res.Failed)
	return res, nil
}