diff --git a/services/search/pkg/bleve/index.go b/services/search/pkg/bleve/index.go
index 097fea9560..ed093e8078 100644
--- 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
}
diff --git a/services/search/pkg/bleve/migrate.go b/services/search/pkg/bleve/migrate.go
new file mode 100644
index 0000000000..760844f8dd
--- /dev/null
+++ 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))
+}
diff --git a/services/search/pkg/bleve/migrate_fields_test.go b/services/search/pkg/bleve/migrate_fields_test.go
new file mode 100644
index 0000000000..b570fa0424
--- /dev/null
+++ 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)
+}
diff --git a/services/search/pkg/bleve/migrate_test.go b/services/search/pkg/bleve/migrate_test.go
new file mode 100644
index 0000000000..0c1ab3489a
--- /dev/null
+++ 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)
+}
diff --git a/services/search/pkg/command/index.go b/services/search/pkg/command/index.go
index 3951ee6fd3..e459e6e0f7 100644
--- 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
}
diff --git a/services/search/pkg/command/migrate.go b/services/search/pkg/command/migrate.go
new file mode 100644
index 0000000000..c389e56257
--- /dev/null
+++ 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
+ },
+ }
+}
diff --git a/services/search/pkg/command/migrate_test.go b/services/search/pkg/command/migrate_test.go
new file mode 100644
index 0000000000..d42c3b9d46
--- /dev/null
+++ 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())
+}
diff --git a/services/search/pkg/command/server.go b/services/search/pkg/command/server.go
index 5e9d8fee05..9723ae371b 100644
--- 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)
diff --git a/services/search/pkg/config/defaults/defaultconfig.go b/services/search/pkg/config/defaults/defaultconfig.go
index 3e504350fa..137d91c824 100644
--- 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{
diff --git a/services/search/pkg/config/engine.go b/services/search/pkg/config/engine.go
index a22d2a92d1..e7e5109069 100644
--- 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
diff --git a/services/search/pkg/opensearch/client.go b/services/search/pkg/opensearch/client.go
new file mode 100644
index 0000000000..ec2e6d9b96
--- /dev/null
+++ 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
+}
diff --git a/services/search/pkg/opensearch/index_test.go b/services/search/pkg/opensearch/index_test.go
index d6b7204880..11b866d277 100644
--- 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"
diff --git a/services/search/pkg/opensearch/migrate.go b/services/search/pkg/opensearch/migrate.go
new file mode 100644
index 0000000000..cd047f5167
--- /dev/null
+++ 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 (-v) 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 -v with k < SchemaRevision plus the legacy unversioned
+// . 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 -v below the current SchemaRevision, or, if none
+// exists yet, the legacy unversioned 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 -v 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)
+}
diff --git a/services/search/pkg/opensearch/migrate_fields_test.go b/services/search/pkg/opensearch/migrate_fields_test.go
new file mode 100644
index 0000000000..d3ab8e9a06
--- /dev/null
+++ 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)
+}
diff --git a/services/search/pkg/opensearch/migrate_test.go b/services/search/pkg/opensearch/migrate_test.go
new file mode 100644
index 0000000000..a05c68b6be
--- /dev/null
+++ 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")
+}
diff --git a/services/search/pkg/search/search.go b/services/search/pkg/search/search.go
index c0dac8fe99..047d7fc45b 100644
--- 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