Add skipDuplicates() scope guard to createDocuments

src/Database/Adapter.php

@@ -33,6 +33,8 @@ abstract class Adapter
 
     protected bool $alterLocks = false;
 
+    protected bool $skipDuplicates = false;
+
     /**
      * @var array<string, mixed>
      */
@@ -392,6 +394,27 @@ public function inTransaction(): bool
         return $this->inTransaction > 0;
     }
 
+    /**
+     * Run a callback with skipDuplicates enabled.
+     * Duplicate key errors during createDocuments() will be silently skipped
+     * instead of thrown. Nestable — saves and restores previous state.
+     *
+     * @template T
+     * @param callable(): T $callback
+     * @return T
+     */
+    public function skipDuplicates(callable $callback): mixed
+    {
+        $previous = $this->skipDuplicates;
+        $this->skipDuplicates = true;
+
+        try {
+            return $callback();
+        } finally {
+            $this->skipDuplicates = $previous;
+        }
+    }
+
     /**
      * @template T
      * @param callable(): T $callback

src/Database/Adapter/Mongo.php

@@ -122,6 +122,11 @@ public function withTransaction(callable $callback): mixed
             return $callback();
         }
 
+        // upsert + $setOnInsert hits WriteConflict (E112) under txn snapshot isolation.
+        if ($this->skipDuplicates) {
+            return $callback();
+        }
+
         try {
             $this->startTransaction();
             $result = $callback();
@@ -1492,6 +1497,42 @@ public function createDocuments(Document $collection, array $documents): array
             $records[] = $record;
         }
 
+        // insertMany aborts the txn on any duplicate; upsert + $setOnInsert no-ops instead.
+        if ($this->skipDuplicates) {
+            if (empty($records)) {
+                return [];
+            }
+
+            $operations = [];
+            foreach ($records as $record) {
+                $filter = ['_uid' => $record['_uid'] ?? ''];
+                if ($this->sharedTables) {
+                    $filter['_tenant'] = $record['_tenant'] ?? $this->getTenant();
+                }
+
+                // Filter fields can't reappear in $setOnInsert (mongo path-conflict error).
+                $setOnInsert = $record;
+                unset($setOnInsert['_uid'], $setOnInsert['_tenant']);
+
+                if (empty($setOnInsert)) {
+                    continue;
+                }
+
+                $operations[] = [
+                    'filter' => $filter,
+                    'update' => ['$setOnInsert' => $setOnInsert],
+                ];
+            }
+
+            try {
+                $this->client->upsert($name, $operations, $options);
+            } catch (MongoException $e) {
+                throw $this->processException($e);
+            }
+
+            return $documents;
+        }
+
         try {
             $documents = $this->client->insertMany($name, $records, $options);
         } catch (MongoException $e) {

src/Database/Adapter/Pool.php

@@ -43,6 +43,11 @@ public function __construct(UtopiaPool $pool)
     public function delegate(string $method, array $args): mixed
     {
         if ($this->pinnedAdapter !== null) {
+            if ($this->skipDuplicates) {
+                return $this->pinnedAdapter->skipDuplicates(
+                    fn () => $this->pinnedAdapter->{$method}(...$args)
+                );
+            }
             return $this->pinnedAdapter->{$method}(...$args);
         }
 
@@ -66,6 +71,11 @@ public function delegate(string $method, array $args): mixed
                 $adapter->setMetadata($key, $value);
             }
 
+            if ($this->skipDuplicates) {
+                return $adapter->skipDuplicates(
+                    fn () => $adapter->{$method}(...$args)
+                );
+            }
             return $adapter->{$method}(...$args);
         });
     }
@@ -146,6 +156,11 @@ public function withTransaction(callable $callback): mixed
 
             $this->pinnedAdapter = $adapter;
             try {
+                if ($this->skipDuplicates) {
+                    return $adapter->skipDuplicates(
+                        fn () => $adapter->withTransaction($callback)
+                    );
+                }
                 return $adapter->withTransaction($callback);
             } finally {
                 $this->pinnedAdapter = null;

src/Database/Adapter/Postgres.php

@@ -2350,6 +2350,35 @@ public function getSupportForOptionalSpatialAttributeWithExistingRows(): bool
         return false;
     }
 
+    protected function getInsertKeyword(): string
+    {
+        return 'INSERT INTO';
+    }
+
+    protected function getInsertSuffix(string $table): string
+    {
+        if (!$this->skipDuplicates) {
+            return '';
+        }
+
+        $conflictTarget = $this->sharedTables ? '("_uid", "_tenant")' : '("_uid")';
+
+        return "ON CONFLICT {$conflictTarget} DO NOTHING";
+    }
+
+    protected function getInsertPermissionsSuffix(): string
+    {
+        if (!$this->skipDuplicates) {
+            return '';
+        }
+
+        $conflictTarget = $this->sharedTables
+            ? '("_type", "_permission", "_document", "_tenant")'
+            : '("_type", "_permission", "_document")';
+
+        return "ON CONFLICT {$conflictTarget} DO NOTHING";
+    }
+
     public function decodePoint(string $wkb): array
     {
         if (str_starts_with(strtoupper($wkb), 'POINT(')) {

src/Database/Adapter/SQL.php

@@ -1029,6 +1029,33 @@ public function getSupportForHostname(): bool
         return true;
     }
 
+    /**
+     * Returns the INSERT keyword, optionally with IGNORE for duplicate handling.
+     * Override in adapter subclasses for DB-specific syntax.
+     */
+    protected function getInsertKeyword(): string
+    {
+        return $this->skipDuplicates ? 'INSERT IGNORE INTO' : 'INSERT INTO';
+    }
+
+    /**
+     * Returns a suffix appended after VALUES clause for duplicate handling.
+     * Override in adapter subclasses (e.g., Postgres uses ON CONFLICT DO NOTHING).
+     */
+    protected function getInsertSuffix(string $table): string
+    {
+        return '';
+    }
+
+    /**
+     * Returns a suffix for the permissions INSERT statement when ignoring duplicates.
+     * Override in adapter subclasses for DB-specific syntax.
+     */
+    protected function getInsertPermissionsSuffix(): string
+    {
+        return '';
+    }
+
     /**
      * Get current attribute count from collection document
      *
@@ -2476,6 +2503,7 @@ public function createDocuments(Document $collection, array $documents): array
         if (empty($documents)) {
             return $documents;
         }
+
         $spatialAttributes = $this->getSpatialAttributes($collection);
         $collection = $collection->getId();
         try {
@@ -2573,8 +2601,9 @@ public function createDocuments(Document $collection, array $documents): array
             $batchKeys = \implode(', ', $batchKeys);
 
             $stmt = $this->getPDO()->prepare("
-                INSERT INTO {$this->getSQLTable($name)} {$columns}
+                {$this->getInsertKeyword()} {$this->getSQLTable($name)} {$columns}
                 VALUES {$batchKeys}
+                {$this->getInsertSuffix($name)}
             ");
 
             foreach ($bindValues as $key => $value) {
@@ -2588,8 +2617,9 @@ public function createDocuments(Document $collection, array $documents): array
                 $permissions = \implode(', ', $permissions);
 
                 $sqlPermissions = "
-                    INSERT INTO {$this->getSQLTable($name . '_perms')} (_type, _permission, _document {$tenantColumn})
-                    VALUES {$permissions};
+                    {$this->getInsertKeyword()} {$this->getSQLTable($name . '_perms')} (_type, _permission, _document {$tenantColumn})
+                    VALUES {$permissions}
+                    {$this->getInsertPermissionsSuffix()}
                 ";
 
                 $stmtPermissions = $this->getPDO()->prepare($sqlPermissions);

src/Database/Adapter/SQLite.php

@@ -1936,4 +1936,9 @@ public function getSupportForTTLIndexes(): bool
     {
         return false;
     }
+
+    protected function getInsertKeyword(): string
+    {
+        return $this->skipDuplicates ? 'INSERT OR IGNORE INTO' : 'INSERT INTO';
+    }
 }

src/Database/Database.php

@@ -417,6 +417,8 @@ class Database
 
     protected bool $preserveDates = false;
 
+    protected bool $skipDuplicates = false;
+
     protected bool $preserveSequence = false;
 
     protected int $maxQueryValues = 5000;
@@ -842,6 +844,29 @@ public function skipRelationshipsExistCheck(callable $callback): mixed
         }
     }
 
+    public function skipDuplicates(callable $callback): mixed
+    {
+        $previous = $this->skipDuplicates;
+        $this->skipDuplicates = true;
+
+        try {
+            return $callback();
+        } finally {
+            $this->skipDuplicates = $previous;
+        }
+    }
+
+    /**
+     * Build a tenant-aware identity key for a document.
+     * Returns "<tenant>:<id>" in tenant-per-document shared-table mode, otherwise just the id.
+     */
+    private function tenantKey(Document $document): string
+    {
+        return ($this->adapter->getSharedTables() && $this->adapter->getTenantPerDocument())
+            ? $document->getTenant() . ':' . $document->getId()
+            : $document->getId();
+    }
+
     /**
      * Trigger callback for events
      *
@@ -5700,9 +5725,11 @@ public function createDocuments(
         }
 
         foreach (\array_chunk($documents, $batchSize) as $chunk) {
-            $batch = $this->withTransaction(function () use ($collection, $chunk) {
-                return $this->adapter->createDocuments($collection, $chunk);
-            });
+            $insert = fn () => $this->withTransaction(fn () => $this->adapter->createDocuments($collection, $chunk));
+            // Set adapter flag before withTransaction so Mongo can opt out of a real txn.
+            $batch = $this->skipDuplicates
+                ? $this->adapter->skipDuplicates($insert)
+                : $insert();
 
             $batch = $this->adapter->getSequences($collection->getId(), $batch);
 
@@ -7116,18 +7143,57 @@ public function upsertDocumentsWithIncrease(
         $created = 0;
         $updated = 0;
         $seenIds = [];
-        foreach ($documents as $key => $document) {
-            if ($this->getSharedTables() && $this->getTenantPerDocument()) {
-                $old = $this->authorization->skip(fn () => $this->withTenant($document->getTenant(), fn () => $this->silent(fn () => $this->getDocument(
-                    $collection->getId(),
-                    $document->getId(),
-                ))));
-            } else {
-                $old = $this->authorization->skip(fn () => $this->silent(fn () => $this->getDocument(
-                    $collection->getId(),
-                    $document->getId(),
-                )));
+
+        // Batch-fetch existing documents in one query instead of N individual getDocument() calls.
+        // tenantPerDocument: group ids by tenant and run one find() per tenant under withTenant,
+        // so cross-tenant batches (e.g. StatsUsage worker) don't get silently scoped to the
+        // session tenant and miss rows belonging to other tenants.
+        $existingDocs = [];
+
+        if ($this->getSharedTables() && $this->getTenantPerDocument()) {
+            $idsByTenant = [];
+            foreach ($documents as $doc) {
+                if ($doc->getId() !== '') {
+                    $idsByTenant[$doc->getTenant()][] = $doc->getId();
+                }
+            }
+            foreach ($idsByTenant as $tenant => $tenantIds) {
+                $tenantIds = \array_values(\array_unique($tenantIds));
+                foreach (\array_chunk($tenantIds, \max(1, $this->maxQueryValues)) as $chunk) {
+                    $found = $this->authorization->skip(fn () => $this->withTenant($tenant, fn () => $this->silent(
+                        fn () => $this->find($collection->getId(), [
+                            Query::equal('$id', $chunk),
+                            Query::limit($this->maxQueryValues),
+                        ])
+                    )));
+                    foreach ($found as $doc) {
+                        $existingDocs[$this->tenantKey($doc)] = $doc;
+                    }
+                }
             }
+        } else {
+            $docIds = \array_values(\array_unique(\array_filter(
+                \array_map(fn (Document $doc) => $doc->getId(), $documents),
+                fn ($id) => $id !== ''
+            )));
+
+            if (!empty($docIds)) {
+                foreach (\array_chunk($docIds, \max(1, $this->maxQueryValues)) as $chunk) {
+                    $existing = $this->authorization->skip(fn () => $this->silent(
+                        fn () => $this->find($collection->getId(), [
+                            Query::equal('$id', $chunk),
+                            Query::limit($this->maxQueryValues),
+                        ])
+                    ));
+                    foreach ($existing as $doc) {
+                        $existingDocs[$this->tenantKey($doc)] = $doc;
+                    }
+                }
+            }
+        }
+
+        foreach ($documents as $key => $document) {
+            $old = $existingDocs[$this->tenantKey($document)] ?? new Document();
 
             // Extract operators early to avoid comparison issues
             $documentArray = $document->getArrayCopy();
@@ -7294,7 +7360,7 @@ public function upsertDocumentsWithIncrease(
                 $document = $this->silent(fn () => $this->createDocumentRelationships($collection, $document));
             }
 
-            $seenIds[] = $document->getId();
+            $seenIds[] = $this->tenantKey($document);
             $old = $this->adapter->castingBefore($collection, $old);
             $document = $this->adapter->castingBefore($collection, $document);