From 92c6c7d99ac20cd93a838b672d1c992d8e69c1cc Mon Sep 17 00:00:00 2001
From: Mazyar <m.nazemi@ravanertebat.com>
Date: Tue, 30 Jun 2026 13:03:39 +0330
Subject: [PATCH] Refactor Elasticsearch client error handling and introduce
 bulk index action marshaling

- Updated error handling in the Elasticsearch client to utilize a new `readErrorBody` function for better diagnostics when pinging and searching.
- Refactored the `flushBatch` method to marshal bulk index actions using a dedicated `marshalBulkIndexAction` function, improving code clarity and error handling.
- Enhanced the overall structure of the Elasticsearch client for improved maintainability and readability.

This update enhances the robustness of the Elasticsearch client, ensuring more informative error messages and cleaner code organization.
---
 pkg/elasticsearch/client.go | 35 +++++++++++++++++++++++++++--------
 1 file changed, 27 insertions(+), 8 deletions(-)

diff --git a/pkg/elasticsearch/client.go b/pkg/elasticsearch/client.go
index 16caf02..e55990e 100644
--- a/pkg/elasticsearch/client.go
+++ b/pkg/elasticsearch/client.go
@@ -86,8 +86,7 @@ func (c *client) ping(ctx context.Context) error {
 	defer res.Body.Close()
 
 	if res.IsError() {
-		body, _ := io.ReadAll(res.Body)
-		return fmt.Errorf("elasticsearch ping failed: %s", string(body))
+		return fmt.Errorf("elasticsearch ping failed: %s", readErrorBody(res.Body))
 	}
 
 	return nil
@@ -126,8 +125,7 @@ func (c *client) Search(ctx context.Context, filter SearchFilter) (*SearchResult
 	defer res.Body.Close()
 
 	if res.IsError() {
-		body, _ := io.ReadAll(res.Body)
-		return nil, fmt.Errorf("elasticsearch search error: %s", string(body))
+		return nil, fmt.Errorf("elasticsearch search error: %s", readErrorBody(res.Body))
 	}
 
 	return decodeSearchResponse(res.Body)
@@ -191,8 +189,11 @@ func (c *client) flushBatch(batch []AuditDocument) error {
 	var buf bytes.Buffer
 	for _, doc := range batch {
 		indexName := indexNameFor(c.config.IndexPrefix, doc.At)
-		meta := fmt.Sprintf(`{"index":{"_index":"%s"}}`, indexName)
-		buf.WriteString(meta)
+		meta, err := marshalBulkIndexAction(indexName)
+		if err != nil {
+			return fmt.Errorf("marshal bulk index action: %w", err)
+		}
+		buf.Write(meta)
 		buf.WriteByte('\n')
 
 		body, err := json.Marshal(doc)
@@ -213,13 +214,31 @@ func (c *client) flushBatch(batch []AuditDocument) error {
 	defer res.Body.Close()
 
 	if res.IsError() {
-		responseBody, _ := io.ReadAll(res.Body)
-		return fmt.Errorf("elasticsearch bulk error: %s", string(responseBody))
+		return fmt.Errorf("elasticsearch bulk error: %s", readErrorBody(res.Body))
 	}
 
 	return nil
 }
 
+type bulkIndexAction struct {
+	Index struct {
+		Index string `json:"_index"`
+	} `json:"index"`
+}
+
+func marshalBulkIndexAction(indexName string) ([]byte, error) {
+	action := bulkIndexAction{}
+	action.Index.Index = indexName
+	return json.Marshal(action)
+}
+
+// readErrorBody best-effort reads an Elasticsearch error response for diagnostics.
+// Read errors are intentionally ignored because the HTTP call already failed.
+func readErrorBody(body io.Reader) string {
+	responseBody, _ := io.ReadAll(body)
+	return string(responseBody)
+}
+
 func indexNameFor(prefix string, at int64) string {
 	t := time.Unix(at, 0).UTC()
 	return fmt.Sprintf("%s-%04d.%02d", prefix, t.Year(), int(t.Month()))