feat(search): schema-revision index migration for bleve and OpenSearch

Migrate the search index in place instead of a full Tika re-crawl when the shared search.SchemaRevision is bumped. bleve (single instance): rebuild from the existing documents into a temp index and atomically swap, with crash recovery; auto-migrate on startup (opt-out via SEARCH_ENGINE_BLEVE_AUTO_MIGRATE). OpenSearch (scaled): versioned indices <base>-v<rev> (no alias), filled from the newest older index via a create-only scroll reindex - idempotent, never overwrites, safe to run live. 'opencloud search index migrate' fills the current-revision index; 'opencloud search index prune' drops older ones. Per-document fixups are gated on the source schema version. Trash preserved, geopoint siblings synthesized, all fields round-tripped.
2026-08-01 02:11:15 -04:00 · 2026-07-16 21:52:59 +02:00
parent cbd196218b
commit 4f71ff8552
16 changed files with 1400 additions and 41 deletions
--- a/services/search/pkg/bleve/index.go
+++ b/services/search/pkg/bleve/index.go
@@ -46,6 +46,12 @@ func NewIndex(root string) (bleve.Index, searchmapping.Classification, error) {
 		if err != nil {
 			return nil, searchmapping.Classification{}, err
 		}
+		// stamp the current revision so a fresh index is not seen as outdated
+		// and needlessly migrated on the next start
+		if err := writeRevision(index); err != nil {
+			_ = index.Close()
+			return nil, searchmapping.Classification{}, err
+		}

 		return index, searchmapping.Classification{Verdict: searchmapping.VerdictEqual}, nil
 	}
--- a/services/search/pkg/bleve/migrate.go
+++ b/services/search/pkg/bleve/migrate.go
@@ -0,0 +1,262 @@
+package bleve
+
+import (
+	"errors"
+	"fmt"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"strconv"
+
+	"github.com/blevesearch/bleve/v2"
+
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+)
+
+const (
+	revisionKey        = "_schema_revision"
+	migratingSuffix    = ".migrating"
+	backupSuffix       = ".bak"
+	migrateBatch       = 100
+	migrateLogInterval = 50000 // how often buildReplacement logs progress, in documents
+)
+
+// OpenOrMigrate opens the index and, when autoMigrate is set and the stored
+// schema is incompatible, rebuilds it in place first. The migration triggers on
+// a schema revision bump, not on a mapping diff, so a breaking change without a
+// matching bump still refuses to start. Returns the number of documents migrated
+// (0 for none). Opening first means a fresh install and the common up-to-date
+// case open the index only once; the migration path is taken only on a mismatch.
+func OpenOrMigrate(root string, autoMigrate bool, logger log.Logger) (bleve.Index, searchmapping.Classification, int, error) {
+	idx, classification, openErr := NewIndex(root)
+	if openErr == nil || !autoMigrate || !errors.Is(openErr, searchmapping.ErrManualActionRequired) {
+		// opened cleanly, or auto-migrate is off, or the failure is not a
+		// schema mismatch we could migrate away
+		return idx, classification, 0, openErr
+	}
+
+	// the stored schema is incompatible; rebuild only if the revision was bumped,
+	// otherwise surface the original error (a schema change without a bump)
+	migrated, err := MigrateIndex(root, logger)
+	if err != nil {
+		return nil, classification, 0, err
+	}
+	if migrated == 0 {
+		return nil, classification, 0, openErr
+	}
+	idx, classification, err = NewIndex(root)
+	return idx, classification, migrated, err
+}
+
+// MigrateIndex rebuilds the index from the documents it already holds when the
+// stored schema revision is older than search.SchemaRevision, else a no-op. It
+// takes the exclusive bolt lock, so the search service must be stopped.
+func MigrateIndex(root string, logger log.Logger) (int, error) {
+	dest := filepath.Join(root, "bleve")
+	tmp := dest + migratingSuffix
+	bak := dest + backupSuffix
+
+	if err := recoverMigration(root); err != nil {
+		return 0, err
+	}
+
+	old, err := bleve.OpenUsing(dest, openRuntimeConfig)
+	if errors.Is(err, bleve.ErrorIndexPathDoesNotExist) {
+		return 0, nil // nothing to migrate yet; NewIndex creates a fresh index
+	}
+	if err != nil {
+		return 0, fmt.Errorf("open index (is the search service still running?): %w", err)
+	}
+
+	stored, err := readRevision(old)
+	if err != nil {
+		_ = old.Close()
+		return 0, err
+	}
+	if stored >= search.SchemaRevision {
+		_ = old.Close()
+		return 0, nil
+	}
+
+	total, err := old.DocCount()
+	if err != nil {
+		_ = old.Close()
+		return 0, err
+	}
+	logger.Info().Uint64("documents", total).Msg("starting search index migration")
+
+	copied, err := buildReplacement(old, tmp, total, logger)
+	if err != nil {
+		_ = old.Close()
+		_ = os.RemoveAll(tmp)
+		return 0, err
+	}
+
+	// verify before the swap, so a failure leaves the original untouched
+	if err := verifyMigrated(tmp, total); err != nil {
+		_ = old.Close()
+		_ = os.RemoveAll(tmp)
+		return 0, err
+	}
+
+	if err := swap(old, dest, tmp, bak); err != nil {
+		return 0, err
+	}
+	logger.Info().Int("documents", copied).Msg("search index migration complete")
+	return copied, nil
+}
+
+// buildReplacement copies every document from src into a fresh index at tmp via
+// the same Deserialize -> PrepareForIndex round-trip the Move/Delete/Restore
+// paths use, and stamps the current revision.
+func buildReplacement(src bleve.Index, tmp string, total uint64, logger log.Logger) (int, error) {
+	if err := os.RemoveAll(tmp); err != nil {
+		return 0, err
+	}
+	m, err := NewMapping()
+	if err != nil {
+		return 0, err
+	}
+	dst, err := bleve.New(tmp, m)
+	if err != nil {
+		return 0, err
+	}
+
+	batch, err := NewBatch(dst, migrateBatch)
+	if err != nil {
+		_ = dst.Close()
+		return 0, err
+	}
+
+	var copied int
+	nextLog := migrateLogInterval
+	var after []string
+	for {
+		req := bleve.NewSearchRequest(bleve.NewMatchAllQuery())
+		req.Size = migrateBatch
+		req.Fields = []string{"*"}
+		req.SortBy([]string{"_id"}) // total order, no skips or dupes across pages
+		req.SearchAfter = after
+		res, err := src.Search(req)
+		if err != nil {
+			_ = dst.Close()
+			return 0, err
+		}
+		if len(res.Hits) == 0 {
+			break
+		}
+		for _, hit := range res.Hits {
+			r := legacyHitToResource(hit.Fields)
+			if err := batch.Upsert(hit.ID, r); err != nil { // hit.ID: the bleve key is authoritative
+				_ = dst.Close()
+				return 0, err
+			}
+			copied++
+		}
+		if copied >= nextLog {
+			logger.Info().Msgf("migrating search index: %d of %d documents", copied, total)
+			nextLog += migrateLogInterval
+		}
+		after = []string{res.Hits[len(res.Hits)-1].ID}
+	}
+	if err := batch.Push(); err != nil {
+		_ = dst.Close()
+		return 0, err
+	}
+	if err := writeRevision(dst); err != nil {
+		_ = dst.Close()
+		return 0, err
+	}
+	return copied, dst.Close()
+}
+
+// legacyHitToResource is the fixup hook before re-indexing, symmetric to the
+// opensearch side. bleve Deserialize is fail-soft, so nothing is stripped here.
+func legacyHitToResource(fields map[string]any) search.Resource {
+	return *searchmapping.Deserialize[search.Resource](fields)
+}
+
+// verifyMigrated checks the rebuilt index holds want documents before the swap
+// replaces the original. The count catches a pagination drop or batch-push loss;
+// the mapping needs no check, it is built deterministically from NewMapping().
+func verifyMigrated(path string, want uint64) error {
+	idx, err := bleve.OpenUsing(path, openRuntimeConfig)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = idx.Close() }()
+
+	got, err := idx.DocCount()
+	if err != nil {
+		return err
+	}
+	if got != want {
+		return fmt.Errorf("migrated index holds %d documents, expected %d", got, want)
+	}
+	return nil
+}
+
+// swap replaces the live index with the rebuilt one. old still holds the lock
+// and is closed here right before the rename.
+func swap(old bleve.Index, dest, tmp, bak string) error {
+	if err := old.Close(); err != nil {
+		return err
+	}
+	if err := os.Rename(dest, bak); err != nil {
+		return err
+	}
+	if err := os.Rename(tmp, dest); err != nil {
+		_ = os.Rename(bak, dest) // best effort roll back
+		return err
+	}
+	return os.RemoveAll(bak)
+}
+
+// recoverMigration cleans up a crashed MigrateIndex and must run before the
+// index is opened, else NewIndex would recreate an empty index mid-swap. Rolls
+// back, never forward: a truncated tmp is a valid index, completeness is not
+// observable.
+func recoverMigration(root string) error {
+	dest := filepath.Join(root, "bleve")
+	tmp := dest + migratingSuffix
+	bak := dest + backupSuffix
+
+	if _, err := os.Stat(dest); err == nil {
+		return errors.Join(os.RemoveAll(tmp), os.RemoveAll(bak)) // index present: tmp/bak are leftovers
+	} else if !errors.Is(err, fs.ErrNotExist) {
+		return err
+	}
+
+	if _, err := os.Stat(bak); err == nil {
+		// crashed mid-swap: roll back
+		if err := os.RemoveAll(tmp); err != nil {
+			return err
+		}
+		return os.Rename(bak, dest)
+	} else if !errors.Is(err, fs.ErrNotExist) {
+		return err
+	}
+
+	return os.RemoveAll(tmp) // fresh install, or index removed by hand
+}
+
+// writeRevision stamps the current revision so a fresh or migrated index is not
+// seen as outdated.
+func writeRevision(index bleve.Index) error {
+	return index.SetInternal([]byte(revisionKey), []byte(strconv.Itoa(search.SchemaRevision)))
+}
+
+// readRevision returns the stored schema revision, or 0 for an index that
+// predates the marker.
+func readRevision(index bleve.Index) (int, error) {
+	raw, err := index.GetInternal([]byte(revisionKey))
+	if err != nil {
+		return 0, err
+	}
+	if len(raw) == 0 {
+		return 0, nil
+	}
+	return strconv.Atoi(string(raw))
+}
--- a/services/search/pkg/bleve/migrate_fields_test.go
+++ b/services/search/pkg/bleve/migrate_fields_test.go
@@ -0,0 +1,89 @@
+package bleve_test
+
+import (
+	"path/filepath"
+	"sort"
+	"testing"
+	"time"
+
+	bleveSearch "github.com/blevesearch/bleve/v2"
+	libregraph "github.com/opencloud-eu/libre-graph-api-go"
+	"github.com/stretchr/testify/require"
+
+	"github.com/opencloud-eu/opencloud/pkg/conversions"
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/bleve"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/content"
+	searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+)
+
+// fullyPopulated exercises every field, including the facets and slices the
+// count-only test skipped.
+func fullyPopulated() search.Resource {
+	return search.Resource{
+		ID:       "1$2!3",
+		RootID:   "1$2!2",
+		ParentID: "1$2!2",
+		Path:     "./a/b/song.mp3",
+		Type:     2,
+		Deleted:  true,
+		Hidden:   true,
+		Document: content.Document{
+			Title:     "the title",
+			Name:      "song.mp3",
+			Content:   "some extracted content",
+			Size:      123456,
+			Mtime:     conversions.ToPointer(time.Date(2026, 3, 15, 12, 0, 0, 0, time.UTC)),
+			MimeType:  "audio/mpeg",
+			Tags:      []string{"alpha", "beta", "gamma"},
+			Favorites: []string{"user-a", "user-b"},
+			Audio: &libregraph.Audio{
+				Album:  libregraph.PtrString("the album"),
+				Artist: libregraph.PtrString("the artist"),
+				Track:  libregraph.PtrInt32(7),
+				Year:   libregraph.PtrInt32(1998),
+				HasDrm: libregraph.PtrBool(false),
+			},
+			Image: &libregraph.Image{Width: libregraph.PtrInt32(1920), Height: libregraph.PtrInt32(1080)},
+			Photo: &libregraph.Photo{CameraMake: libregraph.PtrString("Canon"), Iso: libregraph.PtrInt32(400), FNumber: libregraph.PtrFloat64(2.8)},
+			Location: &libregraph.GeoCoordinates{
+				Longitude: libregraph.PtrFloat64(11.103870357204285),
+				Latitude:  libregraph.PtrFloat64(49.48675890884328),
+				Altitude:  libregraph.PtrFloat64(300.0),
+			},
+		},
+	}
+}
+
+func TestMigratePreservesAllFields(t *testing.T) {
+	root := t.TempDir()
+	want := fullyPopulated()
+
+	old, err := bleveSearch.New(filepath.Join(root, "bleve"), oldMainMapping(t))
+	require.NoError(t, err)
+	require.NoError(t, old.Index(want.ID, want))
+	require.NoError(t, old.Close())
+
+	_, err = bleve.MigrateIndex(root, log.NopLogger())
+	require.NoError(t, err)
+
+	idx, _, err := bleve.NewIndex(root)
+	require.NoError(t, err)
+	defer func() { _ = idx.Close() }()
+
+	// read the migrated document back through the production reconstruction path
+	req := bleveSearch.NewSearchRequest(bleveSearch.NewMatchAllQuery())
+	req.Fields = []string{"*"}
+	res, err := idx.Search(req)
+	require.NoError(t, err)
+	require.Len(t, res.Hits, 1)
+	got := searchmapping.Deserialize[search.Resource](res.Hits[0].Fields)
+
+	require.NotNil(t, got.Mtime, "Mtime")
+	require.True(t, want.Mtime.Equal(*got.Mtime), "Mtime: want %v got %v", want.Mtime, got.Mtime)
+	got.Mtime = want.Mtime            // normalize time.Time location/monotonic
+	sort.Strings(got.Tags)            // bleve may return multi-value fields reordered
+	sort.Strings(got.Favorites)       //
+	require.Equal(t, want, *got)
+}
--- a/services/search/pkg/bleve/migrate_test.go
+++ b/services/search/pkg/bleve/migrate_test.go
@@ -0,0 +1,164 @@
+package bleve_test
+
+import (
+	"path/filepath"
+	"testing"
+	"time"
+
+	bleveSearch "github.com/blevesearch/bleve/v2"
+	bleveMapping "github.com/blevesearch/bleve/v2/mapping"
+	"github.com/blevesearch/bleve/v2/search/query"
+	libregraph "github.com/opencloud-eu/libre-graph-api-go"
+	"github.com/stretchr/testify/require"
+
+	"github.com/opencloud-eu/opencloud/pkg/conversions"
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/bleve"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/content"
+	searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+)
+
+// oldMainMapping reconstructs the pre-refactor bleve mapping: only Name, Tags,
+// Favorites and Content explicit, everything else dynamic. Reuses NewMapping's
+// registered analyzers so indexing works.
+func oldMainMapping(t *testing.T) bleveMapping.IndexMapping {
+	m, err := bleve.NewMapping()
+	require.NoError(t, err)
+	impl := m.(*bleveMapping.IndexMappingImpl)
+
+	doc := bleveSearch.NewDocumentMapping()
+	name := bleveSearch.NewTextFieldMapping()
+	name.Analyzer = "lowercaseKeyword"
+	lc := bleveSearch.NewTextFieldMapping()
+	lc.Analyzer = "lowercaseKeyword"
+	lc.IncludeInAll = false
+	ft := bleveSearch.NewTextFieldMapping()
+	ft.Analyzer = "fulltext"
+	ft.IncludeInAll = false
+	doc.AddFieldMappingsAt("Name", name)
+	doc.AddFieldMappingsAt("Tags", lc)
+	doc.AddFieldMappingsAt("Favorites", lc)
+	doc.AddFieldMappingsAt("Content", ft)
+	impl.DefaultMapping = doc
+	return impl
+}
+
+func fullDoc(id, name string, deleted bool) search.Resource {
+	lon, lat, alt := 11.103870357204285, 49.48675890884328, 300.0
+	return search.Resource{
+		ID:       id,
+		RootID:   "1$1!1",
+		ParentID: "1$1!1",
+		Path:     "./" + name,
+		Type:     2,
+		Deleted:  deleted,
+		Document: content.Document{
+			Name:     name,
+			Title:    "title",
+			Content:  "hello world",
+			Size:     42,
+			Mtime:    conversions.ToPointer(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)),
+			MimeType: "image/jpeg",
+			Tags:     []string{"tag1"},
+			Location: &libregraph.GeoCoordinates{Longitude: &lon, Latitude: &lat, Altitude: &alt},
+		},
+	}
+}
+
+func TestOpenOrMigrate(t *testing.T) {
+	buildOld := func(t *testing.T) string {
+		root := t.TempDir()
+		old, err := bleveSearch.New(filepath.Join(root, "bleve"), oldMainMapping(t))
+		require.NoError(t, err)
+		require.NoError(t, old.Index("1$1!1", fullDoc("1$1!1", "a.jpg", false)))
+		require.NoError(t, old.Close())
+		return root
+	}
+
+	t.Run("migrates an outdated revision when auto-migrate is on", func(t *testing.T) {
+		// the old index carries no revision (0), the code revision is higher
+		idx, classification, migrated, err := bleve.OpenOrMigrate(buildOld(t), true, log.NopLogger())
+		require.NoError(t, err)
+		require.NotNil(t, idx)
+		defer func() { _ = idx.Close() }()
+		require.Equal(t, searchmapping.VerdictEqual, classification.Verdict)
+		require.Equal(t, 1, migrated)
+	})
+
+	t.Run("refuses to start when auto-migrate is off", func(t *testing.T) {
+		idx, _, migrated, err := bleve.OpenOrMigrate(buildOld(t), false, log.NopLogger())
+		require.ErrorIs(t, err, searchmapping.ErrManualActionRequired)
+		require.Nil(t, idx)
+		require.Equal(t, 0, migrated)
+	})
+
+	t.Run("does not migrate a fresh index", func(t *testing.T) {
+		root := t.TempDir()
+		// NewIndex creates a fresh index and stamps the current revision
+		idx, _, err := bleve.NewIndex(root)
+		require.NoError(t, err)
+		require.NoError(t, idx.Close())
+
+		// a fresh index is already at the current revision, so no migration runs
+		idx2, classification, migrated, err := bleve.OpenOrMigrate(root, true, log.NopLogger())
+		require.NoError(t, err)
+		defer func() { _ = idx2.Close() }()
+		require.Equal(t, searchmapping.VerdictEqual, classification.Verdict)
+		require.Equal(t, 0, migrated)
+	})
+}
+
+func TestMigrateIndex(t *testing.T) {
+	root := t.TempDir()
+
+	// 1) an index left behind by the old release: main-style mapping, no revision
+	old, err := bleveSearch.New(filepath.Join(root, "bleve"), oldMainMapping(t))
+	require.NoError(t, err)
+	docs := []search.Resource{
+		fullDoc("1$1!1", "photo.jpg", false),
+		fullDoc("1$1!2", "song.mp3", false),
+		fullDoc("1$1!3", "trashed.txt", true), // trash must survive the migration
+	}
+	for _, r := range docs {
+		require.NoError(t, old.Index(r.ID, r))
+	}
+	require.NoError(t, old.Close())
+
+	// 2) sanity: the old index is incompatible, the service would refuse to start
+	_, _, err = bleve.NewIndex(root)
+	require.ErrorIs(t, err, searchmapping.ErrManualActionRequired)
+
+	// 3) migrate
+	n, err := bleve.MigrateIndex(root, log.NopLogger())
+	require.NoError(t, err)
+	require.Equal(t, 3, n)
+
+	// 4) the migrated index classifies as equal and holds every document
+	migrated, classification, err := bleve.NewIndex(root)
+	require.NoError(t, err)
+	require.Equal(t, searchmapping.VerdictEqual, classification.Verdict)
+
+	count, err := migrated.DocCount()
+	require.NoError(t, err)
+	require.Equal(t, uint64(3), count)
+
+	// 4a) the trashed document survived (a rescan would have dropped it)
+	trashed, err := migrated.Document("1$1!3")
+	require.NoError(t, err)
+	require.NotNil(t, trashed)
+
+	// 5) location_geopoint was synthesized during migration (absent in the old index)
+	near := query.NewGeoDistanceQuery(11.103870357204285, 49.48675890884328, "1km")
+	near.SetField("location" + searchmapping.GeopointSuffix)
+	res, err := migrated.Search(bleveSearch.NewSearchRequest(near))
+	require.NoError(t, err)
+	require.Equal(t, uint64(3), res.Total, "all three docs match a geo-distance query on the new sibling field")
+
+	require.NoError(t, migrated.Close()) // release the lock before the second run
+
+	// 6) idempotent: a second run is a no-op because the revision is now current
+	n, err = bleve.MigrateIndex(root, log.NopLogger())
+	require.NoError(t, err)
+	require.Equal(t, 0, n)
+}
--- a/services/search/pkg/command/index.go
+++ b/services/search/pkg/command/index.go
@@ -95,5 +95,7 @@ func Index(cfg *config.Config) *cobra.Command {
 		"disable TLS for the gRPC connection.",
 	)

+	indexCmd.AddCommand(Migrate(cfg), Prune(cfg))
+
 	return indexCmd
 }
--- a/services/search/pkg/command/migrate.go
+++ b/services/search/pkg/command/migrate.go
@@ -0,0 +1,103 @@
+package command
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	"github.com/opencloud-eu/opencloud/pkg/config/configlog"
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/bleve"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/config"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/config/parser"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch"
+)
+
+// Migrate rebuilds the index for the current schema revision from the documents
+// it holds, no re-crawl. Stop the service first. Idempotent.
+func Migrate(cfg *config.Config) *cobra.Command {
+	return &cobra.Command{
+		Use:   "migrate",
+		Short: "rebuild the search index for the current schema revision",
+		Long:  "Rebuild the search index from the documents it already holds when the schema revision changed, instead of re-crawling storage. Stop the search service before running it. Idempotent: a no-op when the index is already current.",
+		PreRunE: func(cmd *cobra.Command, args []string) error {
+			return configlog.ReturnFatal(parser.ParseConfig(cfg))
+		},
+		RunE: func(cmd *cobra.Command, args []string) error {
+			logger := log.Configure(cfg.Service.Name, cfg.Commons, cfg.LogLevel)
+
+			switch cfg.Engine.Type {
+			case "bleve":
+				n, err := bleve.MigrateIndex(cfg.Engine.Bleve.Datapath, logger)
+				if err != nil {
+					return err
+				}
+				reportMigration(logger, n)
+			case "open-search":
+				client, err := opensearch.NewClient(cfg.Engine.OpenSearch.Client)
+				if err != nil {
+					return err
+				}
+				n, err := opensearch.MigrateIndex(cmd.Context(), cfg.Engine.OpenSearch.ResourceIndex.Name, client, logger)
+				if err != nil {
+					return err
+				}
+				reportMigration(logger, n)
+			default:
+				return fmt.Errorf("unknown search engine: %s", cfg.Engine.Type)
+			}
+			return nil
+		},
+	}
+}
+
+// rescanRecommendation is appended after a rebuild/migration: the index holds
+// the migrated documents, but a full crawl is the only guaranteed-clean state.
+const rescanRecommendation = "re-indexing with 'opencloud search index --all-spaces --force-rescan' is still recommended for a guaranteed-clean index"
+
+func reportMigration(logger log.Logger, migrated int) {
+	if migrated == 0 {
+		logger.Info().Msg("the search index is already at the current schema revision, nothing to migrate")
+		return
+	}
+	logger.Info().Int("documents", migrated).Msg("the search index was migrated; " + rescanRecommendation)
+}
+
+// Prune deletes the OpenSearch indices left by older schema revisions. Run it
+// after a rollout has fully drained the old instances. bleve keeps one index in
+// place, so there is nothing to prune there.
+func Prune(cfg *config.Config) *cobra.Command {
+	return &cobra.Command{
+		Use:   "prune",
+		Short: "delete search indices older than the current schema revision",
+		Long:  "Delete search indices left by older schema revisions, after a rollout has fully drained the old instances. OpenSearch only; refuses if the current-revision index does not exist yet.",
+		PreRunE: func(cmd *cobra.Command, args []string) error {
+			return configlog.ReturnFatal(parser.ParseConfig(cfg))
+		},
+		RunE: func(cmd *cobra.Command, args []string) error {
+			logger := log.Configure(cfg.Service.Name, cfg.Commons, cfg.LogLevel)
+
+			switch cfg.Engine.Type {
+			case "bleve":
+				logger.Info().Msg("bleve keeps a single index in place, nothing to prune")
+			case "open-search":
+				client, err := opensearch.NewClient(cfg.Engine.OpenSearch.Client)
+				if err != nil {
+					return err
+				}
+				n, err := opensearch.PruneOldIndices(cmd.Context(), client, cfg.Engine.OpenSearch.ResourceIndex.Name, logger)
+				if err != nil {
+					return err
+				}
+				if n == 0 {
+					logger.Info().Msg("no old search indices to prune")
+				} else {
+					logger.Info().Int("indices", n).Msg("pruned old search indices")
+				}
+			default:
+				return fmt.Errorf("unknown search engine: %s", cfg.Engine.Type)
+			}
+			return nil
+		},
+	}
+}
--- a/services/search/pkg/command/migrate_test.go
+++ b/services/search/pkg/command/migrate_test.go
@@ -0,0 +1,49 @@
+package command_test
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+
+	bleveSearch "github.com/blevesearch/bleve/v2"
+	"github.com/stretchr/testify/require"
+
+	"github.com/opencloud-eu/opencloud/pkg/shared"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/bleve"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/command"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/config/defaults"
+	searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping"
+)
+
+func TestMigrateCommandBleve(t *testing.T) {
+	root := t.TempDir()
+
+	// an index left behind by an older release: a plain dynamic mapping, which
+	// classifies as breaking against the code schema
+	old, err := bleveSearch.New(filepath.Join(root, "bleve"), bleveSearch.NewIndexMapping())
+	require.NoError(t, err)
+	require.NoError(t, old.Index("doc1", map[string]any{"Name": "file.txt", "Mtime": "2026-01-01T00:00:00Z"}))
+	require.NoError(t, old.Close())
+
+	cfg := defaults.DefaultConfig()
+	defaults.EnsureDefaults(cfg)
+	cfg.Commons = &shared.Commons{Log: &shared.Log{}} // normally filled by the config pipeline
+	cfg.Engine.Type = "bleve"
+	cfg.Engine.Bleve.Datapath = root
+
+	run := func() error {
+		cmd := command.Migrate(cfg)
+		cmd.SetContext(context.Background())
+		return cmd.RunE(cmd, nil) // bypass PreRunE (parser requires service credentials)
+	}
+
+	// first run migrates, the index then classifies equal
+	require.NoError(t, run())
+	idx, classification, err := bleve.NewIndex(root)
+	require.NoError(t, err)
+	require.NoError(t, idx.Close())
+	require.Equal(t, searchmapping.VerdictEqual, classification.Verdict)
+
+	// second run is a no-op (idempotent), still succeeds
+	require.NoError(t, run())
+}
--- a/services/search/pkg/command/server.go
+++ b/services/search/pkg/command/server.go
@@ -2,10 +2,7 @@ package command

 import (
 	"context"
-	"crypto/tls"
 	"fmt"
-	"net/http"
-	"os"
 	"os/signal"
 	"time"

@@ -32,8 +29,6 @@ import (

 	"github.com/opencloud-eu/reva/v2/pkg/events/raw"
 	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
-	opensearchgo "github.com/opensearch-project/opensearch-go/v4"
-	opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi"
 	"github.com/spf13/cobra"
 )

@@ -73,7 +68,13 @@ func Server(cfg *config.Config) *cobra.Command {
 			var eng search.Engine
 			switch cfg.Engine.Type {
 			case "bleve":
-				idx, classification, err := bleve.NewIndex(cfg.Engine.Bleve.Datapath)
+				// with AutoMigrate a schema revision bump rebuilds the index in
+				// place instead of refusing to start. Safe for bleve because a
+				// single instance holds the datapath lock.
+				idx, classification, migrated, err := bleve.OpenOrMigrate(cfg.Engine.Bleve.Datapath, cfg.Engine.Bleve.AutoMigrate, logger)
+				if migrated > 0 {
+					logger.Info().Int("documents", migrated).Msgf("the bleve index at %s was rebuilt for the new schema revision; %s", cfg.Engine.Bleve.Datapath, rescanRecommendation)
+				}
 				// warn before the error check: the new mapping may already be
 				// persisted, then later startups classify equal and stay silent
 				if classification.Verdict == searchmapping.VerdictAdditive {
@@ -93,44 +94,18 @@ func Server(cfg *config.Config) *cobra.Command {

 				eng = bleve.NewBackend(idx, bleveQuery.DefaultCreator, logger)
 			case "open-search":
-				clientConfig := opensearchgo.Config{
-					Addresses:             cfg.Engine.OpenSearch.Client.Addresses,
-					Username:              cfg.Engine.OpenSearch.Client.Username,
-					Password:              cfg.Engine.OpenSearch.Client.Password,
-					Header:                cfg.Engine.OpenSearch.Client.Header,
-					RetryOnStatus:         cfg.Engine.OpenSearch.Client.RetryOnStatus,
-					DisableRetry:          cfg.Engine.OpenSearch.Client.DisableRetry,
-					EnableRetryOnTimeout:  cfg.Engine.OpenSearch.Client.EnableRetryOnTimeout,
-					MaxRetries:            cfg.Engine.OpenSearch.Client.MaxRetries,
-					CompressRequestBody:   cfg.Engine.OpenSearch.Client.CompressRequestBody,
-					DiscoverNodesOnStart:  cfg.Engine.OpenSearch.Client.DiscoverNodesOnStart,
-					DiscoverNodesInterval: cfg.Engine.OpenSearch.Client.DiscoverNodesInterval,
-					EnableMetrics:         cfg.Engine.OpenSearch.Client.EnableMetrics,
-					EnableDebugLogger:     cfg.Engine.OpenSearch.Client.EnableDebugLogger,
-					Transport: &http.Transport{
-						TLSClientConfig: &tls.Config{
-							MinVersion:         tls.VersionTLS12,
-							InsecureSkipVerify: cfg.Engine.OpenSearch.Client.Insecure,
-						},
-					},
-				}
-
-				if cfg.Engine.OpenSearch.Client.CACert != "" {
-					certBytes, err := os.ReadFile(cfg.Engine.OpenSearch.Client.CACert)
-					if err != nil {
-						return fmt.Errorf("failed to read CA cert: %w", err)
-					}
-					clientConfig.CACert = certBytes
-				}
-
-				client, err := opensearchgoAPI.NewClient(opensearchgoAPI.Config{Client: clientConfig})
+				client, err := opensearch.NewClient(cfg.Engine.OpenSearch.Client)
 				if err != nil {
-					return fmt.Errorf("failed to create OpenSearch client: %w", err)
+					return err
 				}

+				// the revision is in the index name, so each instance uses its own
+				// index and rollouts never share one; a pre-rollout migrate builds it
+				indexName := opensearch.TargetIndex(cfg.Engine.OpenSearch.ResourceIndex.Name)
+
 				// a hung cluster must fail the start, not block it forever
 				startupCtx, cancelStartup := context.WithTimeout(ctx, time.Minute)
-				openSearchBackend, err := opensearch.NewBackend(startupCtx, cfg.Engine.OpenSearch.ResourceIndex.Name, client, logger)
+				openSearchBackend, err := opensearch.NewBackend(startupCtx, indexName, client, logger)
 				cancelStartup()
 				if err != nil {
 					return fmt.Errorf("failed to create OpenSearch backend: %w", err)
--- a/services/search/pkg/config/defaults/defaultconfig.go
+++ b/services/search/pkg/config/defaults/defaultconfig.go
@@ -37,7 +37,8 @@ func DefaultConfig() *config.Config {
 		Engine: config.Engine{
 			Type: "bleve",
 			Bleve: config.EngineBleve{
-				Datapath: filepath.Join(defaults.BaseDataPath(), "search"),
+				Datapath:    filepath.Join(defaults.BaseDataPath(), "search"),
+				AutoMigrate: true,
 			},
 			OpenSearch: config.EngineOpenSearch{
 				ResourceIndex: config.EngineOpenSearchResourceIndex{
--- a/services/search/pkg/config/engine.go
+++ b/services/search/pkg/config/engine.go
@@ -14,7 +14,8 @@ type Engine struct {

 // EngineBleve configures the bleve engine
 type EngineBleve struct {
-	Datapath string `yaml:"data_path" env:"SEARCH_ENGINE_BLEVE_DATA_PATH" desc:"The directory where the filesystem will store search data. If not defined, the root directory derives from $OC_BASE_DATA_PATH/search." introductionVersion:"1.0.0"`
+	Datapath    string `yaml:"data_path" env:"SEARCH_ENGINE_BLEVE_DATA_PATH" desc:"The directory where the filesystem will store search data. If not defined, the root directory derives from $OC_BASE_DATA_PATH/search." introductionVersion:"1.0.0"`
+	AutoMigrate bool   `yaml:"auto_migrate" env:"SEARCH_ENGINE_BLEVE_AUTO_MIGRATE" desc:"Rebuild the index on startup when the schema revision was bumped, instead of refusing to start. Safe for bleve because a single instance holds the data path lock. Defaults to 'true'; set to 'false' to migrate manually with 'opencloud search migrate'." introductionVersion:"%%NEXT%%"`
 }

 // EngineOpenSearch configures the OpenSearch engine
--- a/services/search/pkg/opensearch/client.go
+++ b/services/search/pkg/opensearch/client.go
@@ -0,0 +1,53 @@
+package opensearch
+
+import (
+	"crypto/tls"
+	"fmt"
+	"net/http"
+	"os"
+
+	opensearchgo "github.com/opensearch-project/opensearch-go/v4"
+	opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi"
+
+	"github.com/opencloud-eu/opencloud/services/search/pkg/config"
+)
+
+// NewClient builds an OpenSearch API client from the engine client config. It is
+// shared by the server startup and the migrate CLI so both connect identically.
+func NewClient(cfg config.EngineOpenSearchClient) (*opensearchgoAPI.Client, error) {
+	clientConfig := opensearchgo.Config{
+		Addresses:             cfg.Addresses,
+		Username:              cfg.Username,
+		Password:              cfg.Password,
+		Header:                cfg.Header,
+		RetryOnStatus:         cfg.RetryOnStatus,
+		DisableRetry:          cfg.DisableRetry,
+		EnableRetryOnTimeout:  cfg.EnableRetryOnTimeout,
+		MaxRetries:            cfg.MaxRetries,
+		CompressRequestBody:   cfg.CompressRequestBody,
+		DiscoverNodesOnStart:  cfg.DiscoverNodesOnStart,
+		DiscoverNodesInterval: cfg.DiscoverNodesInterval,
+		EnableMetrics:         cfg.EnableMetrics,
+		EnableDebugLogger:     cfg.EnableDebugLogger,
+		Transport: &http.Transport{
+			TLSClientConfig: &tls.Config{
+				MinVersion:         tls.VersionTLS12,
+				InsecureSkipVerify: cfg.Insecure,
+			},
+		},
+	}
+
+	if cfg.CACert != "" {
+		certBytes, err := os.ReadFile(cfg.CACert)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read CA cert: %w", err)
+		}
+		clientConfig.CACert = certBytes
+	}
+
+	client, err := opensearchgoAPI.NewClient(opensearchgoAPI.Config{Client: clientConfig})
+	if err != nil {
+		return nil, fmt.Errorf("failed to create OpenSearch client: %w", err)
+	}
+	return client, nil
+}
--- a/services/search/pkg/opensearch/index_test.go
+++ b/services/search/pkg/opensearch/index_test.go
@@ -51,6 +51,7 @@ func TestIndexManager(t *testing.T) {
 		require.NoError(t, indexManager.Apply(t.Context(), indexName, tc.Client(), log.NopLogger()))
 	})

+
 	t.Run("fails to create index if it already exists but is not up to date", func(t *testing.T) {
 		indexManager := opensearch.IndexManagerLatest
 		indexName := "opencloud-test-resource"
--- a/services/search/pkg/opensearch/migrate.go
+++ b/services/search/pkg/opensearch/migrate.go
@@ -0,0 +1,319 @@
+package opensearch
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi"
+
+	"github.com/opencloud-eu/opencloud/pkg/conversions"
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/mapping"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+)
+
+const (
+	// migratePageSize is the scroll page size for reading the source index.
+	migratePageSize = 1000
+	// migrateLogInterval is how often reindex logs progress, in documents.
+	migrateLogInterval = 50000
+	// migrateScrollKeepAlive keeps the source snapshot alive across scroll pages.
+	migrateScrollKeepAlive = 5 * time.Minute
+)
+
+// TargetIndex is the concrete index for the current schema revision. The
+// revision is the index name suffix, so instances of different revisions use
+// different indices and never share one.
+func TargetIndex(base string) string {
+	return fmt.Sprintf("%s-v%d", base, search.SchemaRevision)
+}
+
+// MigrateIndex fills the current-revision index (<base>-v<rev>) from the newest
+// older-revision index, in-process. Because the revision is in the index name,
+// instances of different revisions never share an index and no alias flip is
+// needed: run it as a pre-rollout step so new instances find their index ready.
+// Reindexing uses create-only bulk ops, so a document already present is skipped,
+// never overwritten: the run is idempotent and safe against a target an instance
+// is already writing to. A no-op when there is no older index to migrate from
+// (fresh install, or pruned). Returns the number of documents newly created.
+func MigrateIndex(ctx context.Context, base string, client *opensearchgoAPI.Client, logger log.Logger) (int, error) {
+	target := TargetIndex(base)
+
+	source, oldVersion, err := previousIndex(ctx, client, base)
+	if err != nil {
+		return 0, err
+	}
+	if source == "" {
+		return 0, nil // nothing older to migrate from
+	}
+
+	exists, err := indexExists(ctx, client, target)
+	if err != nil {
+		return 0, err
+	}
+	if !exists {
+		mappingB, err := IndexManagerLatest.MarshalJSON()
+		if err != nil {
+			return 0, err
+		}
+		resp, err := client.Indices.Create(ctx, opensearchgoAPI.IndicesCreateReq{
+			Index: target,
+			Body:  strings.NewReader(string(mappingB)),
+		})
+		switch {
+		case err != nil:
+			return 0, fmt.Errorf("create %s: %w", target, err)
+		case !resp.Acknowledged:
+			return 0, fmt.Errorf("create %s not acknowledged", target)
+		}
+	}
+	logger.Info().Str("source", source).Str("target", target).Msg("starting search index migration")
+
+	created, skipped, err := reindex(ctx, client, source, target, oldVersion, logger)
+	if err != nil {
+		return 0, err
+	}
+	logger.Info().Str("target", target).Int("created", created).Int("skipped", skipped).Msg("search index migration complete")
+	return created, nil
+}
+
+// PruneOldIndices deletes every index older than the current schema revision:
+// the versioned <base>-v<k> with k < SchemaRevision plus the legacy unversioned
+// <base>. Run it after a rollout has fully drained the old instances. It refuses
+// when the current-revision index does not exist yet, so a premature prune can
+// not wipe the only data.
+func PruneOldIndices(ctx context.Context, client *opensearchgoAPI.Client, base string, logger log.Logger) (int, error) {
+	current, err := indexExists(ctx, client, TargetIndex(base))
+	if err != nil {
+		return 0, err
+	}
+	if !current {
+		return 0, fmt.Errorf("current index %s does not exist; migrate before pruning", TargetIndex(base))
+	}
+
+	older, err := olderVersionedIndices(ctx, client, base)
+	if err != nil {
+		return 0, err
+	}
+	old := make([]string, 0, len(older))
+	for _, name := range older {
+		old = append(old, name)
+	}
+	if legacy, err := indexExists(ctx, client, base); err != nil {
+		return 0, err
+	} else if legacy {
+		old = append(old, base)
+	}
+	if len(old) == 0 {
+		return 0, nil
+	}
+
+	delResp, err := client.Indices.Delete(ctx, opensearchgoAPI.IndicesDeleteReq{Indices: old})
+	if err != nil {
+		return 0, fmt.Errorf("delete old indices %v: %w", old, err)
+	}
+	if !delResp.Acknowledged {
+		return 0, fmt.Errorf("prune not acknowledged")
+	}
+	logger.Info().Strs("indices", old).Msg("pruned old search indices")
+	return len(old), nil
+}
+
+// previousIndex returns the newest index to migrate from and its schema version:
+// the highest-revision <base>-v<k> below the current SchemaRevision, or, if none
+// exists yet, the legacy unversioned <base> index (version 0) left by a
+// pre-versioning release. name is "" when there is nothing older to migrate from.
+func previousIndex(ctx context.Context, client *opensearchgoAPI.Client, base string) (name string, version int, err error) {
+	older, err := olderVersionedIndices(ctx, client, base)
+	if err != nil {
+		return "", 0, err
+	}
+	best := -1
+	for k := range older {
+		if k > best {
+			best = k
+		}
+	}
+	if best >= 0 {
+		return older[best], best, nil
+	}
+
+	// no versioned index yet: fall back to the legacy unversioned index (version 0)
+	legacy, err := indexExists(ctx, client, base)
+	if err != nil {
+		return "", 0, err
+	}
+	if legacy {
+		return base, 0, nil
+	}
+	return "", 0, nil
+}
+
+// olderVersionedIndices returns the existing <base>-v<k> indices with k below the
+// current SchemaRevision, keyed by version.
+func olderVersionedIndices(ctx context.Context, client *opensearchgoAPI.Client, base string) (map[int]string, error) {
+	resp, err := client.Indices.Get(ctx, opensearchgoAPI.IndicesGetReq{Indices: []string{base + "-v*"}})
+	if err != nil {
+		return nil, fmt.Errorf("list %s-v* indices: %w", base, err)
+	}
+	prefix := base + "-v"
+	out := map[int]string{}
+	for name := range resp.Indices {
+		suffix, ok := strings.CutPrefix(name, prefix)
+		if !ok {
+			continue
+		}
+		if k, err := strconv.Atoi(suffix); err == nil && k < search.SchemaRevision {
+			out[k] = name
+		}
+	}
+	return out, nil
+}
+
+// indexExists reports whether the concrete index exists. A transient error is
+// returned so callers do not treat "unknown" as "absent".
+func indexExists(ctx context.Context, client *opensearchgoAPI.Client, index string) (bool, error) {
+	resp, err := client.Indices.Exists(ctx, opensearchgoAPI.IndicesExistsReq{Indices: []string{index}})
+	switch {
+	case resp != nil && resp.StatusCode == 404:
+		return false, nil
+	case err != nil:
+		return false, fmt.Errorf("check index %s: %w", index, err)
+	}
+	return true, nil
+}
+
+// reindex copies source into target with create-only bulk ops (same
+// PrepareForIndex transform as the live write path). It uses a scroll: a
+// consistent snapshot of the source even while old instances keep writing to it.
+// Returns the number of documents created and skipped (already present).
+func reindex(ctx context.Context, client *opensearchgoAPI.Client, source, target string, oldVersion int, logger log.Logger) (int, int, error) {
+	first, err := client.Search(ctx, &opensearchgoAPI.SearchReq{
+		Indices: []string{source},
+		Params:  opensearchgoAPI.SearchParams{Scroll: migrateScrollKeepAlive},
+		// track_total_hits so Total.Value is exact, not capped at 10000: the
+		// completeness check below relies on it
+		Body: strings.NewReader(fmt.Sprintf(`{"size":%d,"track_total_hits":true,"query":{"match_all":{}}}`, migratePageSize)),
+	})
+	if err != nil {
+		return 0, 0, err
+	}
+	scrollID := ""
+	if first.ScrollID != nil {
+		scrollID = *first.ScrollID
+	}
+	defer func() {
+		if scrollID != "" {
+			_, _ = client.Scroll.Delete(ctx, opensearchgoAPI.ScrollDeleteReq{ScrollIDs: []string{scrollID}})
+		}
+	}()
+
+	total := first.Hits.Total.Value
+	var created, skipped int
+	nextLog := migrateLogInterval
+	hits := first.Hits.Hits
+	for len(hits) > 0 {
+		c, s, err := bulkCreate(ctx, client, target, hits, oldVersion)
+		if err != nil {
+			return created, skipped, err
+		}
+		created += c
+		skipped += s
+		if created+skipped >= nextLog {
+			logger.Info().Msgf("migrating search index: %d of %d documents", created+skipped, total)
+			nextLog += migrateLogInterval
+		}
+		next, err := client.Scroll.Get(ctx, opensearchgoAPI.ScrollGetReq{
+			ScrollID: scrollID,
+			Params:   opensearchgoAPI.ScrollGetParams{Scroll: migrateScrollKeepAlive},
+		})
+		if err != nil {
+			return created, skipped, err
+		}
+		if next.ScrollID != nil {
+			scrollID = *next.ScrollID
+		}
+		hits = next.Hits.Hits
+	}
+	if created+skipped != total {
+		return created, skipped, fmt.Errorf("processed %d of %d source documents", created+skipped, total)
+	}
+	return created, skipped, nil
+}
+
+// bulkCreate inserts hits into target with create semantics: a document already
+// present is a 409 conflict, counted as skipped and never overwritten.
+func bulkCreate(ctx context.Context, client *opensearchgoAPI.Client, target string, hits []opensearchgoAPI.SearchHit, oldVersion int) (int, int, error) {
+	var body strings.Builder
+	for _, hit := range hits {
+		r, err := legacyHitToResource(hit.Source, oldVersion)
+		if err != nil {
+			return 0, 0, fmt.Errorf("convert document %s: %w", hit.ID, err)
+		}
+		doc, err := mapping.PrepareForIndex(r, r.SearchFieldOverrides())
+		if err != nil {
+			return 0, 0, fmt.Errorf("prepare document %s: %w", hit.ID, err)
+		}
+		action, err := json.Marshal(map[string]any{"create": map[string]any{"_index": target, "_id": hit.ID}})
+		if err != nil {
+			return 0, 0, err
+		}
+		docJSON, err := json.Marshal(doc)
+		if err != nil {
+			return 0, 0, err
+		}
+		body.Write(action)
+		body.WriteByte('\n')
+		body.Write(docJSON)
+		body.WriteByte('\n')
+	}
+
+	resp, err := client.Bulk(ctx, opensearchgoAPI.BulkReq{Body: strings.NewReader(body.String())})
+	if err != nil {
+		return 0, 0, fmt.Errorf("bulk create: %w", err)
+	}
+	var created, skipped int
+	for _, item := range resp.Items {
+		res, ok := item["create"]
+		if !ok {
+			continue
+		}
+		switch {
+		case res.Status == 409:
+			skipped++ // already present, keep the existing (newer) document
+		case res.Error != nil || res.Status >= 300:
+			reason := ""
+			if res.Error != nil {
+				reason = res.Error.Type + ": " + res.Error.Reason
+			}
+			return created, skipped, fmt.Errorf("create document %s failed: %s", res.ID, reason)
+		default:
+			created++
+		}
+	}
+	return created, skipped, nil
+}
+
+// legacyHitToResource is the per-document fixup applied while reindexing.
+// oldVersion is the schema version of the SOURCE index (0 for the legacy
+// unversioned index), so fixups are gated on where the document comes from: the
+// empty-Mtime strip only runs for documents from the pre-date schema. The switch
+// is the hook for future revision-dependent migrations.
+func legacyHitToResource(source json.RawMessage, oldVersion int) (search.Resource, error) {
+	var m map[string]any
+	if err := json.Unmarshal(source, &m); err != nil {
+		return search.Resource{}, err
+	}
+	switch {
+	case oldVersion <= 1:
+		// pre-date schema: an empty Mtime is not a valid date, strip it
+		if v, ok := m["Mtime"]; ok && v == "" {
+			delete(m, "Mtime")
+		}
+	}
+	return conversions.To[search.Resource](m)
+}
--- a/services/search/pkg/opensearch/migrate_fields_test.go
+++ b/services/search/pkg/opensearch/migrate_fields_test.go
@@ -0,0 +1,79 @@
+package opensearch_test
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	libregraph "github.com/opencloud-eu/libre-graph-api-go"
+	"github.com/stretchr/testify/require"
+
+	"github.com/opencloud-eu/opencloud/pkg/conversions"
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/content"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch"
+	opensearchtest "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/test"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+)
+
+func fullyPopulated() search.Resource {
+	return search.Resource{
+		ID:       "1$2!3",
+		RootID:   "1$2!2",
+		ParentID: "1$2!2",
+		Path:     "./a/b/song.mp3",
+		Type:     2,
+		Deleted:  true,
+		Hidden:   true,
+		Document: content.Document{
+			Title:     "the title",
+			Name:      "song.mp3",
+			Content:   "some extracted content",
+			Size:      123456,
+			Mtime:     conversions.ToPointer(time.Date(2026, 3, 15, 12, 0, 0, 0, time.UTC)),
+			MimeType:  "audio/mpeg",
+			Tags:      []string{"alpha", "beta", "gamma"},
+			Favorites: []string{"user-a", "user-b"},
+			Audio: &libregraph.Audio{
+				Album:  libregraph.PtrString("the album"),
+				Artist: libregraph.PtrString("the artist"),
+				Track:  libregraph.PtrInt32(7),
+				Year:   libregraph.PtrInt32(1998),
+				HasDrm: libregraph.PtrBool(false),
+			},
+			Image: &libregraph.Image{Width: libregraph.PtrInt32(1920), Height: libregraph.PtrInt32(1080)},
+			Photo: &libregraph.Photo{CameraMake: libregraph.PtrString("Canon"), Iso: libregraph.PtrInt32(400), FNumber: libregraph.PtrFloat64(2.8)},
+			Location: &libregraph.GeoCoordinates{
+				Longitude: libregraph.PtrFloat64(11.103870357204285),
+				Latitude:  libregraph.PtrFloat64(49.48675890884328),
+				Altitude:  libregraph.PtrFloat64(300.0),
+			},
+		},
+	}
+}
+
+func TestMigratePreservesAllFields(t *testing.T) {
+	ctx := context.Background()
+	base := "opencloud-migrate-fields"
+	tc, target := newMigrateTest(t, base)
+
+	want := fullyPopulated()
+	tc.Require.IndicesCreate(base, strings.NewReader(`{}`))
+	tc.Require.DocumentCreate(base, want.ID, strings.NewReader(opensearchtest.JSONMustMarshal(t, want)))
+	tc.Require.IndicesCount([]string{base}, nil, 1)
+
+	_, err := opensearch.MigrateIndex(ctx, base, tc.Client(), log.NopLogger())
+	require.NoError(t, err)
+
+	// read the migrated document back from the versioned target; every field must round-trip
+	hits := tc.Require.Search(target, strings.NewReader(`{"size":10,"query":{"match_all":{}}}`))
+	resources := opensearchtest.SearchHitsMustBeConverted[search.Resource](t, hits.Hits)
+	require.Len(t, resources, 1)
+	got := resources[0]
+
+	require.NotNil(t, got.Mtime, "Mtime")
+	require.True(t, want.Mtime.Equal(*got.Mtime), "Mtime: want %v got %v", want.Mtime, got.Mtime)
+	got.Mtime = want.Mtime // normalize time.Time location/monotonic before the deep-equal
+	require.Equal(t, want, got)
+}
--- a/services/search/pkg/opensearch/migrate_test.go
+++ b/services/search/pkg/opensearch/migrate_test.go
@@ -0,0 +1,249 @@
+package opensearch_test
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	libregraph "github.com/opencloud-eu/libre-graph-api-go"
+	opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi"
+	"github.com/stretchr/testify/require"
+
+	"github.com/opencloud-eu/opencloud/pkg/conversions"
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/content"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch"
+	opensearchtest "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/test"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+)
+
+// newMigrateTest returns a client and the current-revision target for a fresh
+// migration test, deleting base, its versioned siblings and the target both
+// before the test and via t.Cleanup.
+func newMigrateTest(t *testing.T, base string) (*opensearchtest.TestClient, string) {
+	tc := opensearchtest.NewDefaultTestClient(t, defaultConfig.Engine.OpenSearch.Client)
+	indices := []string{base, base + "-v0", opensearch.TargetIndex(base)}
+	reset := func() {
+		for _, i := range indices {
+			_, _ = tc.Client().Indices.Delete(context.Background(), opensearchgoAPI.IndicesDeleteReq{Indices: []string{i}})
+		}
+	}
+	reset()
+	t.Cleanup(reset)
+	return tc, opensearch.TargetIndex(base)
+}
+
+func migrateDoc(id, name string, deleted bool) search.Resource {
+	lon, lat, alt := 11.103870357204285, 49.48675890884328, 300.0
+	return search.Resource{
+		ID:       id,
+		RootID:   "1$1!1",
+		ParentID: "1$1!1",
+		Path:     "./" + name,
+		Type:     2,
+		Deleted:  deleted,
+		Document: content.Document{
+			Name:     name,
+			Content:  "hello world",
+			Size:     42,
+			Mtime:    conversions.ToPointer(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)),
+			MimeType: "image/jpeg",
+			Tags:     []string{"tag1"},
+			Location: &libregraph.GeoCoordinates{Longitude: &lon, Latitude: &lat, Altitude: &alt},
+		},
+	}
+}
+
+func contentDoc(id, text string) search.Resource {
+	return search.Resource{
+		ID:       id,
+		RootID:   "1$1!1",
+		ParentID: "1$1!1",
+		Path:     "./" + id,
+		Type:     2,
+		Document: content.Document{
+			Name:    id,
+			Content: text,
+			Mtime:   conversions.ToPointer(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)),
+		},
+	}
+}
+
+// bulkIndex writes total migrateDoc documents into base, in chunks.
+func bulkIndex(t *testing.T, tc *opensearchtest.TestClient, base string, total int) {
+	const chunk = 5000
+	for i := 0; i < total; i += chunk {
+		var bulk strings.Builder
+		for j := i; j < i+chunk && j < total; j++ {
+			id := fmt.Sprintf("doc%05d", j)
+			fmt.Fprintf(&bulk, "{\"create\":{\"_index\":%q,\"_id\":%q}}\n", base, id)
+			bulk.WriteString(opensearchtest.JSONMustMarshal(t, migrateDoc(id, id+".txt", false)))
+			bulk.WriteString("\n")
+		}
+		_, err := tc.Client().Bulk(context.Background(), opensearchgoAPI.BulkReq{Body: strings.NewReader(bulk.String())})
+		require.NoError(t, err)
+	}
+}
+
+func TestMigrateIndex(t *testing.T) {
+	ctx := context.Background()
+	base := "opencloud-migrate-basic"
+	tc, target := newMigrateTest(t, base)
+
+	// 1) a legacy unversioned index from a pre-versioning release: dynamic mapping,
+	//    documents including a trashed one, location as a plain object (no geopoint)
+	tc.Require.IndicesCreate(base, strings.NewReader(`{}`))
+	for _, d := range []search.Resource{
+		migrateDoc("doc1", "photo.jpg", false),
+		migrateDoc("doc2", "song.mp3", false),
+		migrateDoc("doc3", "trashed.txt", true),
+	} {
+		tc.Require.DocumentCreate(base, d.ID, strings.NewReader(opensearchtest.JSONMustMarshal(t, d)))
+	}
+	tc.Require.IndicesCount([]string{base}, nil, 3)
+
+	// 2) migrate builds the versioned target from the legacy index
+	n, err := opensearch.MigrateIndex(ctx, base, tc.Client(), log.NopLogger())
+	require.NoError(t, err)
+	require.Equal(t, 3, n)
+
+	// 3) the versioned target exists and holds every document
+	tc.Require.IndicesCount([]string{target}, nil, 3)
+
+	// 3a) the trashed document survived (a rescan would have dropped it)
+	trash := `{"query":{"term":{"Deleted":true}}}`
+	tc.Require.IndicesCount([]string{target}, strings.NewReader(trash), 1)
+
+	// 4) location_geopoint was synthesized during migration (absent in the legacy index)
+	geo := fmt.Sprintf(`{"query":{"geo_distance":{"distance":"1km","location%s":{"lat":%f,"lon":%f}}}}`,
+		"_geopoint", 49.48675890884328, 11.103870357204285)
+	tc.Require.IndicesCount([]string{target}, strings.NewReader(geo), 3)
+
+	// 5) the service starts against the target and Apply classifies it equal
+	require.NoError(t, opensearch.IndexManagerLatest.Apply(ctx, target, tc.Client(), log.NopLogger()))
+
+	// 6) idempotent: the target now exists, so a second run creates nothing
+	n, err = opensearch.MigrateIndex(ctx, base, tc.Client(), log.NopLogger())
+	require.NoError(t, err)
+	require.Equal(t, 0, n)
+}
+
+// TestMigrateLargeIndex covers reindexing past a single scroll page and past
+// OpenSearch's default track_total_hits cap (10000) — the completeness check
+// must use the exact total.
+func TestMigrateLargeIndex(t *testing.T) {
+	for _, row := range []struct {
+		name  string
+		total int
+	}{
+		{"pages beyond one scroll page", 2500},
+		{"exceeds the 10k track_total_hits cap", 10500},
+	} {
+		t.Run(row.name, func(t *testing.T) {
+			ctx := context.Background()
+			base := fmt.Sprintf("opencloud-migrate-large-%d", row.total)
+			tc, target := newMigrateTest(t, base)
+
+			tc.Require.IndicesCreate(base, strings.NewReader(`{}`))
+			bulkIndex(t, tc, base, row.total)
+			tc.Require.IndicesRefresh([]string{base}, nil)
+			tc.Require.IndicesCount([]string{base}, nil, row.total)
+
+			n, err := opensearch.MigrateIndex(ctx, base, tc.Client(), log.NopLogger())
+			require.NoError(t, err)
+			require.Equal(t, row.total, n, "every document across all pages must be migrated")
+			tc.Require.IndicesCount([]string{target}, nil, row.total)
+		})
+	}
+}
+
+// TestMigrateBackfillsWithoutOverwriting simulates the post-rollout / late run:
+// the target already exists (an instance created it and wrote newer documents),
+// and migrate must backfill the missing ones with create-only ops, never
+// overwriting the newer ones.
+func TestMigrateBackfillsWithoutOverwriting(t *testing.T) {
+	ctx := context.Background()
+	base := "opencloud-migrate-backfill"
+	tc, target := newMigrateTest(t, base)
+
+	// legacy index with three documents (old content)
+	tc.Require.IndicesCreate(base, strings.NewReader(`{}`))
+	for _, id := range []string{"a", "b", "c"} {
+		tc.Require.DocumentCreate(base, id, strings.NewReader(opensearchtest.JSONMustMarshal(t, contentDoc(id, "OLD"))))
+	}
+	tc.Require.IndicesCount([]string{base}, nil, 3)
+
+	// a new instance already created the target and wrote a newer "a" plus a new "d"
+	tc.Require.IndicesCreate(target, strings.NewReader(opensearch.IndexManagerLatest.String()))
+	tc.Require.DocumentCreate(target, "a", strings.NewReader(opensearchtest.JSONMustMarshal(t, contentDoc("a", "NEW"))))
+	tc.Require.DocumentCreate(target, "d", strings.NewReader(opensearchtest.JSONMustMarshal(t, contentDoc("d", "NEW"))))
+	tc.Require.IndicesRefresh([]string{target}, nil)
+
+	// migrate backfills b and c, must not overwrite the newer a
+	n, err := opensearch.MigrateIndex(ctx, base, tc.Client(), log.NopLogger())
+	require.NoError(t, err)
+	require.Equal(t, 2, n, "only b and c are newly created (a is skipped, d untouched)")
+
+	tc.Require.IndicesRefresh([]string{target}, nil)
+	tc.Require.IndicesCount([]string{target}, nil, 4) // a, b, c, d
+
+	hits := tc.Require.Search(target, strings.NewReader(`{"query":{"ids":{"values":["a"]}}}`))
+	got := opensearchtest.SearchHitsMustBeConverted[search.Resource](t, hits.Hits)
+	require.Len(t, got, 1)
+	require.Equal(t, "NEW", got[0].Content, "the newer document must not be overwritten by the migration")
+}
+
+func TestPruneOldIndices(t *testing.T) {
+	ctx := context.Background()
+	base := "opencloud-prune"
+	tc, target := newMigrateTest(t, base)
+	old := base + "-v0"
+
+	// refuses while the current-revision index does not exist yet
+	_, err := opensearch.PruneOldIndices(ctx, tc.Client(), base, log.NopLogger())
+	require.Error(t, err, "prune must refuse when the current index is missing")
+
+	// legacy, an older versioned index, and the current index all exist
+	tc.Require.IndicesCreate(base, strings.NewReader(`{}`))
+	tc.Require.IndicesCreate(old, strings.NewReader(`{}`))
+	tc.Require.IndicesCreate(target, strings.NewReader(opensearch.IndexManagerLatest.String()))
+
+	n, err := opensearch.PruneOldIndices(ctx, tc.Client(), base, log.NopLogger())
+	require.NoError(t, err)
+	require.Equal(t, 2, n, "legacy base and base-v0 are pruned, base-v1 stays")
+
+	for _, gone := range []string{base, old} {
+		resp, _ := tc.Client().Indices.Exists(ctx, opensearchgoAPI.IndicesExistsReq{Indices: []string{gone}})
+		require.Equal(t, 404, resp.StatusCode, "%s must be pruned", gone)
+	}
+	resp, err := tc.Client().Indices.Exists(ctx, opensearchgoAPI.IndicesExistsReq{Indices: []string{target}})
+	require.NoError(t, err)
+	require.Equal(t, 200, resp.StatusCode, "the current index must survive prune")
+}
+
+// TestMigrateStripsEmptyMtime covers the one legacy value that would otherwise
+// break the reindex: an empty Mtime string, which is not a valid date. It must
+// be stripped so the document survives instead of being rejected by the new
+// date mapping.
+func TestMigrateStripsEmptyMtime(t *testing.T) {
+	ctx := context.Background()
+	base := "opencloud-migrate-empty-mtime"
+	tc, target := newMigrateTest(t, base)
+
+	// an old document with an empty Mtime, exactly as the pre-fix code wrote it
+	tc.Require.IndicesCreate(base, strings.NewReader(`{}`))
+	tc.Require.DocumentCreate(base, "x", strings.NewReader(`{"ID":"x","Name":"f.txt","Path":"./f.txt","Mtime":""}`))
+	tc.Require.IndicesCount([]string{base}, nil, 1)
+
+	n, err := opensearch.MigrateIndex(ctx, base, tc.Client(), log.NopLogger())
+	require.NoError(t, err)
+	require.Equal(t, 1, n, "the empty-mtime document must survive, not be dropped")
+
+	tc.Require.IndicesCount([]string{target}, nil, 1)
+	hits := tc.Require.Search(target, strings.NewReader(`{"query":{"match_all":{}}}`))
+	resources := opensearchtest.SearchHitsMustBeConverted[search.Resource](t, hits.Hits)
+	require.Len(t, resources, 1)
+	require.Nil(t, resources[0].Mtime, "the empty mtime should be dropped, not carried over")
+}
--- a/services/search/pkg/search/search.go
+++ b/services/search/pkg/search/search.go
@@ -49,6 +49,12 @@ type BatchOperator interface {
 	Push() error
 }

+// SchemaRevision is the version of the index schema defined by Resource and its
+// SearchFieldOverrides. Bump it deliberately when a change needs a rebuild: the
+// bump triggers the migration, not a classifier diff. Both backends share it,
+// bleve stores it in the index, opensearch carries it as the index name suffix.
+const SchemaRevision = 1
+
 // Resource is the entity that is stored in the index.
 type Resource struct {
 	content.Document