fix(search): report rejected documents when indexing to OpenSearch

services/search/pkg/opensearch/batch.go

@@ -210,11 +210,20 @@ func (b *Batch) Push() error {
 			body.WriteString("\n")
 		}
 
-		if _, err := b.client.Bulk(context.Background(), opensearchgoAPI.BulkReq{
+		resp, err := b.client.Bulk(context.Background(), opensearchgoAPI.BulkReq{
 			Body:   strings.NewReader(body.String()),
 			Params: opensearchgoAPI.BulkParams{Refresh: "wait_for"},
-		}); err != nil {
+		})
+		switch {
+		case err != nil:
 			return fmt.Errorf("failed to execute bulk operations: %w", err)
+		case resp.Errors:
+			items, err := json.Marshal(resp.Items)
+			if err != nil {
+				return fmt.Errorf("failed to marshal bulk response: %w", err)
+			}
+
+			return fmt.Errorf("failed to execute bulk operations, response: %s", items)
 		}
 
 		bulkOperations = nil

services/search/pkg/opensearch/batch_test.go

@@ -0,0 +1,53 @@
+package opensearch_test
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/test"
+)
+
+func TestBatch_Push(t *testing.T) {
+	tc := opensearchtest.NewDefaultTestClient(t, defaultConfig.Engine.OpenSearch.Client)
+
+	t.Run("reports the documents the bulk API rejected", func(t *testing.T) {
+		indexName := "opencloud-test-batch-push-rejected"
+		tc.Require.IndicesReset([]string{indexName})
+		defer tc.Require.IndicesDelete([]string{indexName})
+
+		// Name is a string, mapping it as a long makes every document fail to parse.
+		tc.Require.IndicesCreate(indexName, strings.NewReader(`{"mappings":{"properties":{"Name":{"type":"long"}}}}`))
+
+		batch, err := opensearch.NewBatch(tc.Client(), indexName, 10)
+		require.NoError(t, err)
+
+		document := opensearchtest.Testdata.Resources.File
+		require.NoError(t, batch.Upsert(document.ID, document))
+
+		err = batch.Push()
+		require.Error(t, err)
+		require.ErrorContains(t, err, document.ID)
+		require.ErrorContains(t, err, "mapper_parsing_exception")
+		tc.Require.IndicesCount([]string{indexName}, nil, 0)
+	})
+
+	t.Run("pushes the documents the bulk API accepted", func(t *testing.T) {
+		indexName := "opencloud-test-batch-push-accepted"
+		tc.Require.IndicesReset([]string{indexName})
+		defer tc.Require.IndicesDelete([]string{indexName})
+
+		tc.Require.IndicesCreate(indexName, strings.NewReader(opensearch.IndexManagerLatest.String()))
+
+		batch, err := opensearch.NewBatch(tc.Client(), indexName, 10)
+		require.NoError(t, err)
+
+		document := opensearchtest.Testdata.Resources.File
+		require.NoError(t, batch.Upsert(document.ID, document))
+		require.NoError(t, batch.Push())
+
+		tc.Require.IndicesCount([]string{indexName}, nil, 1)
+	})
+}