upsert WIP

Author: pdphilip

src/Eloquent/Builder.php

@@ -75,6 +75,7 @@ class Builder extends BaseEloquentBuilder
         'openpit',
         'bulkinsert',
         'createonly',
+        'upsert',
     ];
 
     /**

src/Eloquent/Docs/ModelDocs.php

@@ -8,8 +8,6 @@
 use PDPhilip\Elasticsearch\Data\ModelMeta;
 
 /**
- * @method bool|int push(string $column = null, mixed $values = null, bool $unique = false)
- *
  * @property object $searchHighlights
  * @property array $searchHighlightsAsArray
  * @property object $withHighlights

src/Eloquent/Model.php

@@ -748,6 +748,99 @@ public function bulkInsert(array $values): array
         return $this->processor->processBulkInsert($this, $this->connection->insert($this->grammar->compileInsert($this, $values), [], true));
     }
 
+    /**
+     * Insert or update records matching the unique key.
+     *
+     * ES has no unique column constraints, so this queries for existing
+     * documents first, then issues a single bulk request mixing index
+     * (for new docs) and update (for existing docs) actions.
+     */
+    public function upsert(array $values, $uniqueBy, $update = null): int
+    {
+        // Normalize to batch format
+        if (! array_is_list($values)) {
+            $values = [$values];
+        }
+
+        $uniqueBy = (array) $uniqueBy;
+
+        // Empty update means plain insert
+        if ($update === []) {
+            $this->insert($values);
+
+            return count($values);
+        }
+
+        // Collect the unique field values from the input
+        $lookupValues = [];
+        foreach ($values as &$doc) {
+            $key = $this->buildUpsertKey($doc, $uniqueBy);
+            $doc['_upsert_key'] = $key;
+            $lookupValues[] = $key;
+        }
+        unset($doc);
+
+        // Query ES for existing documents matching the unique fields
+        $existingIds = $this->findExistingIds($uniqueBy, $lookupValues);
+
+        // Compile and execute the bulk request
+        $dsl = $this->grammar->compileUpsert($this, $values, $existingIds, $update);
+        $result = $this->connection->insert($dsl);
+
+        return $this->processor->processUpsert($this, $result);
+    }
+
+    /**
+     * Build a lookup key from the document's unique field values.
+     */
+    private function buildUpsertKey(array $doc, array $uniqueBy): string
+    {
+        $parts = [];
+        foreach ($uniqueBy as $field) {
+            $parts[] = (string) ($doc[$field] ?? '');
+        }
+
+        return implode('|', $parts);
+    }
+
+    /**
+     * Find existing document IDs by unique field values.
+     *
+     * Returns a map of lookup_key => _id.
+     */
+    private function findExistingIds(array $uniqueBy, array $lookupValues): array
+    {
+        // Build a fresh query against the same index
+        $query = $this->newQuery();
+
+        if (count($uniqueBy) === 1) {
+            // Single field: simple whereIn
+            $field = $uniqueBy[0];
+            $fieldValues = array_unique(array_map(fn ($key) => explode('|', $key)[0], $lookupValues));
+            $results = $query->whereIn($field, $fieldValues)->get();
+        } else {
+            // Multi-field: use bool should with exact match per combination
+            foreach (array_unique($lookupValues) as $key) {
+                $parts = explode('|', $key);
+                $query->orWhere(function ($q) use ($uniqueBy, $parts) {
+                    foreach ($uniqueBy as $i => $field) {
+                        $q->where($field, $parts[$i] ?? '');
+                    }
+                });
+            }
+            $results = $query->get();
+        }
+
+        // Map lookup_key => _id
+        $map = [];
+        foreach ($results as $result) {
+            $key = $this->buildUpsertKey((array) $result, $uniqueBy);
+            $map[$key] = $result['_id'];
+        }
+
+        return $map;
+    }
+
     /**
      * {@inheritdoc}
      */

src/Query/Builder.php

@@ -21,6 +21,11 @@ public static function indexOperation(string $index, mixed $id = null, array $op
         return ['index' => $operation];
     }
 
+    public static function updateOperation(string $index, string $id, array $options = []): array
+    {
+        return ['update' => array_merge(['_index' => $index, '_id' => $id], $options)];
+    }
+
     // ----------------------------------------------------------------------
     // Query
     // ----------------------------------------------------------------------

src/Query/DSL/DslFactory.php

@@ -112,6 +112,56 @@ public function compileInsert($query, array $values): array
         return $dsl->getDsl();
     }
 
+    /**
+     * Compile an upsert bulk request.
+     *
+     * Existing docs (with _id) use the update action with doc_as_upsert.
+     * New docs use the index action.
+     */
+    public function compileUpsert($query, array $values, array $existingIds, ?array $updateColumns): array
+    {
+        $dsl = new DslBuilder;
+        $index = $query->getFrom();
+
+        foreach ($values as $doc) {
+            // Convert DateTime values
+            foreach ($doc as &$property) {
+                $property = $this->getStringValue($property);
+            }
+            unset($property);
+
+            // Clean internal fields from the doc body
+            $cleanDoc = $doc;
+            unset($cleanDoc['id'], $cleanDoc['_id']);
+
+            $lookupKey = $doc['_upsert_key'] ?? null;
+            unset($cleanDoc['_upsert_key']);
+
+            $existingId = $lookupKey !== null ? ($existingIds[$lookupKey] ?? null) : null;
+
+            if ($existingId) {
+                // Existing doc: update action with partial doc
+                $updateDoc = $updateColumns
+                    ? array_intersect_key($cleanDoc, array_flip($updateColumns))
+                    : $cleanDoc;
+
+                $dsl->appendBody(DslFactory::updateOperation($index, $existingId));
+                $dsl->appendBody([
+                    'doc' => $updateDoc,
+                    'upsert' => $cleanDoc,
+                ]);
+            } else {
+                // New doc: index action
+                $dsl->appendBody(DslFactory::indexOperation($index));
+                $dsl->appendBody($cleanDoc);
+            }
+        }
+
+        $dsl->setRefresh($query->getOption('refresh', true));
+
+        return $dsl->getDsl();
+    }
+
     /**
      * @param  Builder  $query
      *

src/Query/Grammar/Grammar.php

@@ -289,6 +289,26 @@ public function processBulkInsert(Builder $query, Elasticsearch $result): array
         return $outcome;
     }
 
+    public function processUpsert(Builder $query, Elasticsearch $result): int
+    {
+        $this->rawResponse = $result;
+        $this->query = $query;
+
+        $process = $result->asArray();
+        $count = 0;
+
+        foreach ($process['items'] ?? [] as $item) {
+            // Bulk response items are keyed by action type (index or update)
+            $action = $item['index'] ?? $item['update'] ?? null;
+
+            if ($action && empty($action['error'])) {
+                $count++;
+            }
+        }
+
+        return $count;
+    }
+
     public function processRaw($query, $response)
     {
         $this->rawResponse = $response;

src/Query/Processor/Processor.php

@@ -107,7 +107,7 @@
     expect($result)->toBe(1)
         ->and(User::count())->toBe(2)
         ->and(User::where('email', 'foo')->first()->name)->toBe('bar3');
-})->todo();
+});
 
 it('tests manual string id', function () {
     $user = new User;