feat(flow-php/filesystem): exclusive create, local permissions, non-recursive rm

documentation/components/bridges/filesystem-async-aws-bridge.md

@@ -58,3 +58,7 @@ A source or sink given a filesystem that does not serve its path throws at const
 `FileStatus` values returned from `list()` and `status()` carry `size` (from S3 `Size` / `ContentLength`)
 and `lastModifiedAt` (from `LastModified`) populated directly from the S3 response - no extra HEAD call
 is issued when the CLI `flow:filesystem:ls --long` or `flow:filesystem:stat` prints them.
+
+`create()` commits the multipart upload with `If-None-Match: *`, so it throws `FileExistsException` when the key
+exists. When a conflicting write aborts the upload, it throws `RuntimeException`; call `create()` again.
+S3-compatible stores that ignore conditional writes do not enforce it.

documentation/components/bridges/filesystem-azure-bridge.md

@@ -99,4 +99,7 @@ A source or sink given a filesystem that does not serve its path throws at const
 `FileStatus` values returned from `list()` and `status()` carry `size` (from the `Content-Length`
 header / listing property) and `lastModifiedAt` (from the `Last-Modified` header, parsed as RFC 7231)
 populated directly from the Azure response - no extra stream is opened when the CLI
-`flow:filesystem:ls --long` or `flow:filesystem:stat` prints them.
+`flow:filesystem:ls --long` or `flow:filesystem:stat` prints them.
+
+`create()` commits the blob with `If-None-Match: *`, so it throws `FileExistsException` when the blob exists. A
+blob written in blocks is invisible until `close()`.

documentation/components/bridges/filesystem-sftp-bridge.md

@@ -130,6 +130,8 @@ $filesystem = sftp_filesystem(
 | `withBlockSize()` | 4 MB | size of a single block buffered locally before it is uploaded |
 | `withReadChunkSize()` | 1 MB | bytes `readLines()` pulls in one request when no length is given |
 
+`create()` is not supported and throws `RuntimeException`.
+
 ## Lost connections
 
 This bridge does not reconnect on your behalf. phpseclib raises `InvalidStateException` once a

documentation/components/libs/filesystem.md

@@ -76,12 +76,13 @@ final readonly class Mount
 <?php
 
 Filesystem::appendTo(Path $path) : DestinationStream;
+Filesystem::create(Path $path) : DestinationStream; // visible whole at close(), FileExistsException when the path exists
 Filesystem::getSystemTmpDir() : Path;
 Filesystem::list(Path $path, Filter $pathFilter = new KeepAll()) : \Generator; // yields FileStatus
 Filesystem::mount() : Mount;
 Filesystem::mv(Path $from, Path $to) : bool;
 Filesystem::readFrom(Path $path) : SourceStream;
-Filesystem::rm(Path $path) : bool;
+Filesystem::rm(Path $path, bool $recursive = true) : bool; // $recursive = false refuses a non-empty directory
 Filesystem::status(Path $path) : ?FileStatus;
 Filesystem::writeTo(Path $path) : DestinationStream;
 ```
@@ -183,6 +184,43 @@ native_local_filesystem()->list(path(__DIR__ . '/data/**.parquet'));   // data/f
 Local listing follows symlinks a pattern names; `**` never descends into a symlinked directory. There is no escape
 character.
 
+## Exclusive create
+
+`create()` writes a file that becomes visible whole at `close()`. If the path exists by then, `close()` throws
+`FileExistsException` and the existing file stays as it was. Local, memory, S3 and Azure filesystems support it; SFTP
+and StdOut throw.
+
+```php
+<?php
+
+use Flow\Filesystem\Exception\FileExistsException;
+
+use function Flow\Filesystem\DSL\{native_local_filesystem, path};
+
+try {
+    native_local_filesystem()->create(path(__DIR__ . '/claims/42'))->append((string) getmypid())->close();
+} catch (FileExistsException) {
+    // another process created it first
+}
+```
+
+## Local permissions
+
+`NativeLocalFilesystem` applies `Permissions` to every file it writes and every directory it creates. Existing
+directories are not changed. Without `Permissions`, the umask decides.
+
+```php
+<?php
+
+use Flow\Filesystem\Local\Permissions;
+
+use function Flow\Filesystem\DSL\{native_local_filesystem, path};
+
+$filesystem = native_local_filesystem(permissions: Permissions::private()); // files 0600, directories 0700
+
+$filesystem->writeTo(path(__DIR__ . '/secrets/token.txt'))->append($token)->close();
+```
+
 ## Cross-filesystem copy & move
 
 `FilesystemTable` coupled with the `Copy` / `Move` operations lets you copy or move files between any

documentation/upgrading.md

@@ -1866,6 +1866,29 @@ by Spark's hidden-path rule.
 | `Calculator::modulus(7, -3)` on `brick/math` 0.15+: `Brick\Math\Exception\InvalidArgumentException` | `Flow\ETL\Foundation\Exception\InvalidArgumentException` `Modulus must not be negative` |
 | `ref('a')->mod(lit(-3))`: a result or an `EvaluationException`, by `brick/math` version             | always an `EvaluationException` caused by `InvalidArgumentException`                    |
 
+### 123) `flow-php/filesystem` and the filesystem bridges - `Filesystem::create()` and non-recursive `rm()`
+
+| Before                                                   | After                                                                                    |
+|----------------------------------------------------------|------------------------------------------------------------------------------------------|
+| `interface Filesystem`                                   | `+ create(Path $path): DestinationStream` - custom implementations add it                |
+| `Filesystem::rm(Path $path): bool`                       | `rm(Path $path, bool $recursive = true): bool` - custom implementations add the argument |
+| `final class Flow\Filesystem\Exception\RuntimeException` | not final; `FileExistsException` extends it                                              |
+| `SFTPFilesystem`, `StdOutFilesystem`                     | `create()` throws `RuntimeException`                                                     |
+
+### 124) `flow-php/filesystem` - `NativeLocalFilesystem` takes `Permissions`
+
+| Before                                                                                                                                                 | After                                                                                  |
+|--------------------------------------------------------------------------------------------------------------------------------------------------------|----------------------------------------------------------------------------------------|
+| `NativeLocalFilesystem::normalize()`: `{protocol}`                                                                                                     | `{protocol, permissions}`; `fromArray()` rejects a configuration without `permissions` |
+| `AsyncAWS\Options::normalize()`, `Azure\Options::normalize()`, `SFTP\Options::normalize()`                                                             | `+ block_permissions`; `fromArray()` rejects a configuration without it                |
+| `NativeLocalFilesystem::mv()` from a missing source raised an `E_WARNING`, created the target's parent directory, and removed an existing target first | returns `false` and changes nothing; an existing file target is replaced by `rename()` |
+
+### 125) `flow-php/filesystem-azure-bridge` - a blob written in blocks is invisible until `close()`
+
+| Before                                               | After                                                                 |
+|------------------------------------------------------|-----------------------------------------------------------------------|
+| the first filled block put an empty blob at the path | nothing is visible at the path until `close()` commits the block list |
+
 ## Upgrading from 0.45.x to 0.46.x
 
 ### 1) `flow-php/etl-adapter-postgresql`, `flow-php/etl-adapter-doctrine` - keyset reads reject NULL keys up front

src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Double/LengthCapturingFilesystem.php

@@ -44,6 +44,11 @@ public function appendTo(Path $path): DestinationStream
         return $this->wrapped->appendTo($path);
     }
 
+    public function create(Path $path): DestinationStream
+    {
+        return $this->wrapped->create($path);
+    }
+
     public function normalize(): ?array
     {
         return null;
@@ -85,9 +90,9 @@ public function readFrom(Path $path): SourceStream
         );
     }
 
-    public function rm(Path $path): bool
+    public function rm(Path $path, bool $recursive = true): bool
     {
-        return $this->wrapped->rm($path);
+        return $this->wrapped->rm($path, $recursive);
     }
 
     public function status(Path $path): ?FileStatus

src/bridge/filesystem/async-aws/src/Flow/Filesystem/Bridge/AsyncAWS/AsyncAWSS3DestinationStream.php

@@ -4,12 +4,15 @@
 
 namespace Flow\Filesystem\Bridge\AsyncAWS;
 
+use AsyncAws\Core\Exception\Http\ClientException;
 use AsyncAws\S3\Exception\NoSuchKeyException;
 use AsyncAws\S3\Input\CreateMultipartUploadRequest;
 use AsyncAws\S3\S3Client;
 use Flow\Filesystem\Bridge\AsyncAWS\AsyncAWSS3DesintationStream\AsyncAWSS3BlockLifecycle;
 use Flow\Filesystem\DestinationStream;
+use Flow\Filesystem\Exception\FileExistsException;
 use Flow\Filesystem\Exception\InvalidArgumentException;
+use Flow\Filesystem\Exception\RuntimeException;
 use Flow\Filesystem\Path;
 use Flow\Filesystem\SizeUnits;
 use Flow\Filesystem\Stream\Block\NativeLocalFileBlocksFactory;
@@ -30,6 +33,8 @@ final class AsyncAWSS3DestinationStream implements DestinationStream
 {
     private bool $closed;
 
+    private ?string $ifNoneMatch = null;
+
     public function __construct(
         private readonly S3Client $s3Client,
         private readonly string $uploadId,
@@ -162,6 +167,22 @@ public static function openBlank(
         );
     }
 
+    /**
+     * @param int<1, max> $blockSize
+     */
+    public static function openExclusive(
+        S3Client $s3Client,
+        string $bucket,
+        Path $path,
+        BlockFactory $blockFactory = new NativeLocalFileBlocksFactory(),
+        int $blockSize = 1024 * 1024 * 4,
+    ): self {
+        $stream = self::openBlank($s3Client, $bucket, $path, $blockFactory, $blockSize);
+        $stream->ifNoneMatch = '*';
+
+        return $stream;
+    }
+
     public function append(string $data): DestinationStream
     {
         $this->blocks->append($data);
@@ -171,6 +192,10 @@ public function append(string $data): DestinationStream
 
     public function close(): void
     {
+        if ($this->closed) {
+            return;
+        }
+
         if ($this->blocks->size() === 0) {
             $handle = fopen($this->blocks->block()->path()->path(), 'rb');
 
@@ -189,27 +214,18 @@ public function close(): void
             fclose($handle);
             unlink($this->blocks->block()->path()->path());
 
-            $this->s3Client->completeMultipartUpload([
-                'Bucket' => $this->bucket,
-                'Key' => ltrim($this->path->path(), '/'),
-                'UploadId' => $this->uploadId,
-                'MultipartUpload' => [
-                    'Parts' => [
-                        ['PartNumber' => 1, 'ETag' => $partResponse->getEtag()],
-                    ],
-                ],
-            ]);
+            $this->complete([['PartNumber' => 1, 'ETag' => $partResponse->getEtag()]]);
 
             if ($this->path->endsWith('_flow_php_append_')) {
                 $this->s3Client->deleteObject([
                     'Bucket' => $this->bucket,
-                    'Key' => ltrim(str_replace('_flow_php_append_', '', $this->path->path()), '/'),
+                    'Key' => ltrim(str_replace('/_flow_php_append_', '', $this->path->path()), '/'),
                 ]);
 
                 $this->s3Client->copyObject([
                     'Bucket' => $this->bucket,
-                    'Key' => ltrim(str_replace('_flow_php_append_', '', $this->path->path()), '/'),
-                    'CopySource' => $this->bucket . '/' . $this->path->path(),
+                    'Key' => ltrim(str_replace('/_flow_php_append_', '', $this->path->path()), '/'),
+                    'CopySource' => $this->bucket . '/' . ltrim($this->path->path(), '/'),
                 ]);
 
                 $this->s3Client->deleteObject([
@@ -225,14 +241,7 @@ public function close(): void
 
         $this->blocks->done();
 
-        $this->s3Client->completeMultipartUpload([
-            'Bucket' => $this->bucket,
-            'Key' => ltrim($this->path->path(), '/'),
-            'UploadId' => $this->uploadId,
-            'MultipartUpload' => [
-                'Parts' => $this->blockList->toArray(),
-            ],
-        ]);
+        $this->complete($this->blockList->toArray());
 
         if ($this->path->endsWith('_flow_php_append_')) {
             $this->s3Client->deleteObject([
@@ -283,4 +292,44 @@ public function path(): Path
     {
         return $this->path;
     }
+
+    /**
+     * @param array<array{PartNumber: int, ETag: null|string}> $parts
+     *
+     * @throws FileExistsException when the stream is exclusive and the key exists
+     * @throws RuntimeException when the stream is exclusive and a conflicting write aborted the upload
+     */
+    private function complete(array $parts): void
+    {
+        try {
+            $this->s3Client->completeMultipartUpload([
+                'Bucket' => $this->bucket,
+                'Key' => ltrim($this->path->path(), '/'),
+                'UploadId' => $this->uploadId,
+                'IfNoneMatch' => $this->ifNoneMatch,
+                'MultipartUpload' => ['Parts' => $parts],
+            ])->resolve();
+        } catch (ClientException $e) {
+            $status = $e->getResponse()->getStatusCode();
+
+            if ($this->ifNoneMatch === null || $status !== 409 && $status !== 412) {
+                throw $e;
+            }
+
+            $this->s3Client->abortMultipartUpload([
+                'Bucket' => $this->bucket,
+                'Key' => ltrim($this->path->path(), '/'),
+                'UploadId' => $this->uploadId,
+            ])->resolve();
+            $this->closed = true;
+
+            if ($status === 409) {
+                throw new RuntimeException(
+                    "A conflicting write to {$this->path->uri()} aborted the upload, call create() again",
+                );
+            }
+
+            throw new FileExistsException('File already exists: ' . $this->path->uri());
+        }
+    }
 }

src/bridge/filesystem/async-aws/src/Flow/Filesystem/Bridge/AsyncAWS/AsyncAWSS3Filesystem.php

@@ -95,6 +95,23 @@ public function normalize(): ?array
         ];
     }
 
+    public function create(Path $path): DestinationStream
+    {
+        if ($path->isEqual($this->getSystemTmpDir())) {
+            throw new RuntimeException('Cannot write to system tmp directory');
+        }
+
+        $this->mount->supports($path) || throw new InvalidSchemeException($path->protocol(), $this->mount->protocol);
+
+        return AsyncAWSS3DestinationStream::openExclusive(
+            $this->client->client(),
+            $this->bucket,
+            $path,
+            $this->options->blockFactory(),
+            $this->options->partSize(),
+        );
+    }
+
     public function getSystemTmpDir(): Path
     {
         return $this->options->tmpDir();
@@ -217,7 +234,7 @@ public function readFrom(Path $path): SourceStream
         return new AsyncAWSS3SourceStream($path, $this->bucket, $this->client->client());
     }
 
-    public function rm(Path $path): bool
+    public function rm(Path $path, bool $recursive = true): bool
     {
         if ($path->isEqual($this->getSystemTmpDir())) {
             return false;
@@ -258,6 +275,14 @@ public function rm(Path $path): bool
              */
             $folderPath = path(trim($path->uri(), DIRECTORY_SEPARATOR) . DIRECTORY_SEPARATOR, $path->options());
 
+            if (!$recursive) {
+                foreach ($this->list($folderPath) as $fileStatus) {
+                    if ($fileStatus->path->path() !== $folderPath->path()) {
+                        throw new RuntimeException("Directory {$path->uri()} was not removed: it is not empty");
+                    }
+                }
+            }
+
             $deletedCount = 0;
 
             foreach ($this->list($folderPath) as $fileStatus) {

src/bridge/filesystem/async-aws/src/Flow/Filesystem/Bridge/AsyncAWS/Options.php

@@ -5,17 +5,21 @@
 namespace Flow\Filesystem\Bridge\AsyncAWS;
 
 use Flow\Filesystem\Exception\InvalidArgumentException;
+use Flow\Filesystem\Local\Permissions;
 use Flow\Filesystem\Path;
 use Flow\Filesystem\Stream\Block\NativeLocalFileBlocksFactory;
 use Flow\Filesystem\Stream\BlockFactory;
 
+use function array_key_exists;
 use function Flow\Filesystem\DSL\path;
+use function is_array;
 use function is_bool;
 use function is_int;
 use function is_string;
 
 /**
- * @type S3Options = array{block_location: string, file_fast_path: bool, part_size: int}
+ * @type S3Options = array{block_location: string, block_permissions: null|PermissionsShape, file_fast_path: bool, part_size: int}
+ * @import-type PermissionsShape from Permissions
  */
 final class Options
 {
@@ -61,16 +65,23 @@ public static function fromArray(array $data): self
     {
         if (
             !is_string($data['block_location'] ?? null)
+            || !array_key_exists('block_permissions', $data)
+            || $data['block_permissions'] !== null && !is_array($data['block_permissions'])
             || !is_bool($data['file_fast_path'] ?? null)
             || !is_int($data['part_size'] ?? null)
         ) {
             throw new InvalidArgumentException(
-                'AsyncAWS options must hold block_location (string), file_fast_path (bool) and part_size (int)',
+                'AsyncAWS options must hold block_location (string), block_permissions (null or array), file_fast_path (bool) and part_size (int)',
             );
         }
 
         return (new self())
-            ->withBlockFactory(new NativeLocalFileBlocksFactory($data['block_location']))
+            ->withBlockFactory(
+                new NativeLocalFileBlocksFactory(
+                    $data['block_location'],
+                    $data['block_permissions'] === null ? null : Permissions::fromArray($data['block_permissions']),
+                ),
+            )
             ->withFileFastPath($data['file_fast_path'])
             ->withBlockSize($data['part_size']);
     }
@@ -86,6 +97,7 @@ public function normalize(): ?array
 
         return [
             'block_location' => $this->blockFactory->blockLocation(),
+            'block_permissions' => $this->blockFactory->permissions()?->normalize(),
             'file_fast_path' => $this->fileFastPath,
             'part_size' => $this->partSize,
         ];