fix(csv): report rows that cannot be imported instead of changing or dropping them

src/Migration/Destinations/Appwrite.php

@@ -160,6 +160,13 @@ class Appwrite extends Destination
      */
     private array $rowBuffer = [];
 
+    /**
+     * The rows whose documents are in $rowBuffer.
+     *
+     * @var array<Row>
+     */
+    private array $rowBufferResources = [];
+
     /**
      * Overwrite-mode orphan tracking, keyed by (database, table). Orphans are
      * destination keys not in `attributeKeys` / `indexKeys`. Entries removed
@@ -300,6 +307,7 @@ public function run(
     private function resetRunState(): void
     {
         $this->rowBuffer = [];
+        $this->rowBufferResources = [];
         $this->orphansByTable = [];
         $this->processedTwoWayPairs = [];
         $this->provisioningDatabases = [];
@@ -549,6 +557,7 @@ protected function import(array $resources, callable $callback): void
         }
 
         $total = \count($resources);
+        $records = [];
 
         foreach ($resources as $index => $resource) {
             $resource->setStatus(Resource::STATUS_PROCESSING);
@@ -590,9 +599,17 @@ protected function import(array $resources, callable $callback): void
                 $this->dbForPlatform->setPreserveDates(false);
             }
 
+            // Rows are counted once their batch is written, since writing it can still fail them.
+            if ($responseResource instanceof Row) {
+                $records[] = $responseResource;
+                continue;
+            }
+
             $this->cache->update($responseResource);
         }
 
+        $this->cache->updateAll($records);
+
         $callback($resources);
     }
 
@@ -1552,13 +1569,7 @@ protected function createIndex(Index $resource): bool
         return true;
     }
 
-    /**
-     * @throws AuthorizationException
-     * @throws DatabaseException
-     * @throws StructureException
-     * @throws Exception
-     */
-    protected function createRecord(Row $resource, bool $isLast): bool
+    private function bufferRecord(Row $resource): bool
     {
         if ($resource->getId() == 'unique()') {
             $resource->setId(ID::unique());
@@ -1622,8 +1633,25 @@ protected function createRecord(Row $resource, bool $isLast): bool
             '$id' => $resource->getId(),
             '$permissions' => $resource->getPermissions(),
         ], $data));
+        $this->rowBufferResources[] = $resource;
+
+        return true;
+    }
+
+    /**
+     * Rows are written in one batch, so a failed write fails every row in it.
+     *
+     * @throws AuthorizationException
+     * @throws DatabaseException
+     * @throws StructureException
+     * @throws Exception
+     */
+    protected function createRecord(Row $resource, bool $isLast): bool
+    {
+        $buffered = $this->bufferRecord($resource);
 
-        if ($isLast) {
+        // Write the batch even when its last row was rejected, or the rows before it are lost.
+        if ($isLast && !empty($this->rowBuffer)) {
             try {
                 $database = $this->dbForProject->getDocument(
                     self::META_DATABASES,
@@ -1669,48 +1697,45 @@ protected function createRecord(Row $resource, bool $isLast): bool
                 }
                 $collectionId = $this->tableCollectionId($database, $table);
 
-                try {
-                    match ($this->onDuplicate) {
-                        OnDuplicate::Overwrite => $dbForDatabases->skipRelationshipsExistCheck(
-                            fn () => $dbForDatabases->upsertDocuments($collectionId, $this->rowBuffer)
-                        ),
-                        OnDuplicate::Skip => $dbForDatabases->skipDuplicates(
-                            fn () => $dbForDatabases->skipRelationshipsExistCheck(
-                                fn () => $dbForDatabases->createDocuments($collectionId, $this->rowBuffer)
-                            )
-                        ),
-                        OnDuplicate::Fail => $dbForDatabases->skipRelationshipsExistCheck(
+                match ($this->onDuplicate) {
+                    OnDuplicate::Overwrite => $dbForDatabases->skipRelationshipsExistCheck(
+                        fn () => $dbForDatabases->upsertDocuments($collectionId, $this->rowBuffer)
+                    ),
+                    OnDuplicate::Skip => $dbForDatabases->skipDuplicates(
+                        fn () => $dbForDatabases->skipRelationshipsExistCheck(
                             fn () => $dbForDatabases->createDocuments($collectionId, $this->rowBuffer)
-                        ),
-                    };
-                } catch (DuplicateException $e) {
-                    $resource->setStatus(Resource::STATUS_ERROR, 'Document already exists');
-                    $this->addError(new Exception(
-                        resourceName: $resource->getName(),
-                        resourceGroup: $resource->getGroup(),
-                        resourceId: $resource->getId(),
-                        message: 'Document already exists',
-                        previous: $e,
-                    ));
-                    return false;
-                } catch (StructureException $e) {
-                    $resource->setStatus(Resource::STATUS_ERROR, $e->getMessage());
-                    $this->addError(new Exception(
-                        resourceName: $resource->getName(),
-                        resourceGroup: $resource->getGroup(),
-                        resourceId: $resource->getId(),
-                        message: $e->getMessage(),
-                        previous: $e,
-                    ));
-                    return false;
+                        )
+                    ),
+                    OnDuplicate::Fail => $dbForDatabases->skipRelationshipsExistCheck(
+                        fn () => $dbForDatabases->createDocuments($collectionId, $this->rowBuffer)
+                    ),
+                };
+            } catch (\Throwable $e) {
+                $message = $e instanceof DuplicateException ? 'Document already exists' : $e->getMessage();
+
+                foreach ($this->rowBufferResources as $row) {
+                    $row->setStatus(Resource::STATUS_ERROR, $message);
                 }
+
+                if (!$e instanceof DuplicateException && !$e instanceof StructureException) {
+                    throw $e;
+                }
+
+                $this->addError(new Exception(
+                    resourceName: $resource->getName(),
+                    resourceGroup: $resource->getGroup(),
+                    resourceId: $resource->getId(),
+                    message: $message,
+                    previous: $e,
+                ));
+                return false;
             } finally {
                 $this->rowBuffer = [];
+                $this->rowBufferResources = [];
             }
         }
 
-
-        return true;
+        return $buffered;
     }
 
     /** Relationships route through deleteRelationship since deleteAttribute throws for VAR_RELATIONSHIP. */

src/Migration/Destinations/CSV.php

@@ -240,8 +240,14 @@ protected function writeCSVLine($handle, array $fields): bool
         $parts = [];
 
         foreach ($fields as $field) {
+            if ($field === null) {
+                $parts[] = 'null';
+                continue;
+            }
+
             $field = (string)$field;
-            if (\strpbrk($field, $this->delimiter . "\n\r" . $this->enclosure) !== false) {
+            // Quoted, the text "null" is not read back as NULL.
+            if ($field === 'null' || \strpbrk($field, $this->delimiter . "\n\r" . $this->enclosure) !== false) {
                 $parts[] = $this->enclosure . \str_replace($this->enclosure, $this->enclosure . $this->enclosure, $field) . $this->enclosure;
             } else {
                 $parts[] = $field;
@@ -316,10 +322,10 @@ protected function resourceToCSVData(Row $resource): array
     /**
      * Convert a single value to CSV-compatible format
      */
-    protected function convertValueToCSV(mixed $value): string
+    protected function convertValueToCSV(mixed $value): ?string
     {
         if (\is_null($value)) {
-            return 'null';
+            return null;
         }
         if (\is_bool($value)) {
             return $value ? 'true' : 'false';