Add recursive watch

packages/js-sdk/src/sandbox/filesystem/index.ts

@@ -1,27 +1,27 @@
 import {
+  Code,
+  ConnectError,
   createPromiseClient,
-  Transport,
   PromiseClient,
-  ConnectError,
-  Code,
+  Transport,
 } from '@connectrpc/connect'
 import {
   ConnectionConfig,
+  ConnectionOpts,
   defaultUsername,
   Username,
-  ConnectionOpts,
 } from '../../connectionConfig'
+import { handleEnvdApiError } from '../../envd/api'
+import { authenticationHeader, handleRpcError } from '../../envd/rpc'
 import {
   SandboxError
 } from '../../errors'
-import { handleEnvdApiError } from '../../envd/api'
-import { authenticationHeader, handleRpcError } from '../../envd/rpc'
 
 import { EnvdApiClient } from '../../envd/api'
 import { Filesystem as FilesystemService } from '../../envd/filesystem/filesystem_connect'
 import { FileType as FsFileType, WatchDirResponse } from '../../envd/filesystem/filesystem_pb'
 
-import { WatchHandle, FilesystemEvent } from './watchHandle'
+import { FilesystemEvent, FilesystemEventType, WatchHandle } from './watchHandle'
 
 export interface EntryInfo {
   name: string
@@ -233,7 +233,7 @@ export class Filesystem {
   async watch(
     path: string,
     onEvent: (event: FilesystemEvent) => void | Promise<void>,
-    opts?: FilesystemRequestOpts & { timeout?: number, onExit?: (err?: Error) => void | Promise<void> },
+    opts?: FilesystemRequestOpts & { timeout?: number, onExit?: (err?: Error) => void | Promise<void>, eventTypes?: Set<FilesystemEventType>},
   ): Promise<WatchHandle> {
     const requestTimeoutMs = opts?.requestTimeoutMs ?? this.connectionConfig.requestTimeoutMs
 
@@ -251,6 +251,15 @@ export class Filesystem {
       timeoutMs: opts?.timeout ?? this.defaultWatchTimeout,
     })
 
+    let onEvent_ = onEvent
+
+    if (opts?.eventTypes) {
+      // wrap handler with event type filtering
+      onEvent_ = (event: FilesystemEvent) => {
+        if (opts.eventTypes!.has(event.type)) return onEvent(event)
+      }
+    }
+
     try {
       const startEvent: WatchDirResponse = (await events[Symbol.asyncIterator]().next()).value
 
@@ -263,7 +272,7 @@ export class Filesystem {
       return new WatchHandle(
         () => controller.abort(),
         events,
-        onEvent,
+        onEvent_,
         opts?.onExit,
       )
     } catch (err) {

packages/js-sdk/src/sandbox/index.ts

@@ -1,17 +1,19 @@
 import { createConnectTransport } from '@connectrpc/connect-web'
 
-import { ConnectionOpts, ConnectionConfig, defaultUsername } from '../connectionConfig'
+import { ConnectionConfig, ConnectionOpts, defaultUsername } from '../connectionConfig'
+import { EnvdApiClient, handleEnvdApiError } from '../envd/api'
 import { createRpcLogger } from '../logs'
 import { Filesystem } from './filesystem'
+import { FilesystemEvent, FilesystemEventType } from './filesystem/watchHandle'
 import { Process } from './process'
 import { Pty } from './pty'
 import { SandboxApi } from './sandboxApi'
-import { EnvdApiClient, handleEnvdApiError } from '../envd/api'
 
 export interface SandboxOpts extends ConnectionOpts {
   metadata?: Record<string, string>
   envs?: Record<string, string>
   timeoutMs?: number
+  onFileCreation?: (event: FilesystemEvent) => void | Promise<void>
 }
 
 export class Sandbox extends SandboxApi {
@@ -45,6 +47,7 @@ export class Sandbox extends SandboxApi {
 
     this.envdApi = new EnvdApiClient({ apiUrl: this.envdApiUrl, logger: opts?.logger })
     this.files = new Filesystem(rpcTransport, this.envdApi, this.connectionConfig)
+
     this.commands = new Process(rpcTransport, this.connectionConfig)
     this.pty = new Pty(rpcTransport, this.connectionConfig)
   }
@@ -63,13 +66,21 @@ export class Sandbox extends SandboxApi {
       : await this.createSandbox(template, sandboxOpts?.timeoutMs ?? this.defaultSandboxTimeoutMs, sandboxOpts)
 
     const sbx = new this({ sandboxId, ...config }) as InstanceType<S>
+
+    if (sandboxOpts?.onFileCreation)
+      await sbx.files.watch('/', sandboxOpts.onFileCreation, { eventTypes: new Set([FilesystemEventType.CREATE])})
+
     return sbx
   }
 
   static async connect<S extends typeof Sandbox>(this: S, sandboxId: string, opts?: Omit<SandboxOpts, 'metadata' | 'envs' | 'timeoutMs'>): Promise<InstanceType<S>> {
     const config = new ConnectionConfig(opts)
 
     const sbx = new this({ sandboxId, ...config }) as InstanceType<S>
+
+    if (opts?.onFileCreation)
+      await sbx.files.watch('/', opts.onFileCreation, { eventTypes: new Set([FilesystemEventType.CREATE])})
+
     return sbx
   }
 

packages/js-sdk/tests/sandbox/connect.test.ts

@@ -1,6 +1,6 @@
-import { test, assert } from 'vitest'
+import { assert, onTestFinished, test } from 'vitest'
 
-import { Sandbox } from '../../src'
+import { FilesystemEvent, FilesystemEventType, Sandbox } from '../../src'
 import { isDebug, template } from '../setup.js'
 
 test('connect', async () => {
@@ -19,3 +19,33 @@ test('connect', async () => {
     }
   }
 })
+
+test.skipIf(isDebug)('connect with file creation handler', async () => {
+  const sbx = await Sandbox.create(template, { timeoutMs: 10_000 })
+  const dirPath = '/new-dir2'
+  onTestFinished(async () =>  await sbx.files.remove(dirPath))
+
+  let trigger: () => void
+
+  const eventPromise = new Promise<void>((resolve) => {
+    trigger = resolve
+  })
+
+  const onFileCreation = async (e: FilesystemEvent) => {
+    if (e.type === FilesystemEventType.CREATE && e.name === dirPath) {
+      trigger()
+    }
+  }
+
+  const isRunning = await sbx.isRunning()
+  assert.isTrue(isRunning)
+
+  const sbxConnection = await Sandbox.create(sbx.sandboxId, { onFileCreation})
+  const isRunning2 = await sbxConnection.isRunning()
+  assert.isTrue(isRunning2)
+
+  const ok = await sbx.files.makeDir(dirPath)
+  assert.isTrue(ok)
+
+  await eventPromise
+})

packages/js-sdk/tests/sandbox/create.test.ts

@@ -1,7 +1,7 @@
-import { test, assert } from 'vitest'
+import { assert, onTestFinished, test } from 'vitest'
 
-import { Sandbox } from '../../src'
-import { template, isDebug } from '../setup.js'
+import { FilesystemEvent, FilesystemEventType, Sandbox } from '../../src'
+import { isDebug, template } from '../setup.js'
 
 test.skipIf(isDebug)('create', async () => {
   const sbx = await Sandbox.create(template, { timeoutMs: 5_000 })
@@ -13,7 +13,7 @@ test.skipIf(isDebug)('create', async () => {
   }
 })
 
-test.skipIf(isDebug)('metadata', async () => {
+test.skipIf(isDebug)('create with metadata', async () => {
   const metadata = {
     'test-key': 'test-value',
   }
@@ -29,3 +29,26 @@ test.skipIf(isDebug)('metadata', async () => {
     await sbx.kill()
   }
 })
+
+test.skipIf(isDebug)('create with file creation handler', async () => {
+  const dirPath = '/new-dir'
+
+  let trigger: () => void
+
+  const eventPromise = new Promise<void>((resolve) => {
+    trigger = resolve
+  })
+
+  const onFileCreation = async (e: FilesystemEvent) => {
+    if (e.type === FilesystemEventType.CREATE && e.name === dirPath) {
+      trigger()
+    }
+  }
+  const sbx = await Sandbox.create(template, { timeoutMs: 5_000, onFileCreation })
+  onTestFinished(async () =>  await sbx.files.remove(dirPath))
+
+  const ok = await sbx.files.makeDir(dirPath)
+  assert.isTrue(ok)
+
+  await eventPromise
+})

packages/js-sdk/tests/sandbox/files/list.test.ts

@@ -1,9 +1,10 @@
-import { assert } from 'vitest'
+import { assert, onTestFinished } from 'vitest'
 
 import { sandboxTest } from '../../setup.js'
 
 sandboxTest('list directory', async ({ sandbox }) => {
   const dirName = 'test_directory4'
+  onTestFinished(async () =>  await sandbox.files.remove(dirName))
 
   await sandbox.files.makeDir(dirName)
 

packages/js-sdk/tests/sandbox/files/watch.test.ts

@@ -1,7 +1,7 @@
-import { expect } from 'vitest'
+import { assert, expect, onTestFinished } from 'vitest'
 
-import { sandboxTest } from '../../setup.js'
 import { FilesystemEventType, NotFoundError } from '../../../src'
+import { sandboxTest } from '../../setup.js'
 
 sandboxTest('watch directory changes', async ({ sandbox }) => {
   const dirname = 'test_watch_dir'
@@ -20,7 +20,7 @@ sandboxTest('watch directory changes', async ({ sandbox }) => {
 
 
   const handle = await sandbox.files.watch(dirname, async (event) => {
-    if (event.type === FilesystemEventType.WRITE && event.name === filename) {
+    if (event.type === FilesystemEventType.WRITE && event.name === `/home/user/${dirname}/${filename}`) {
       trigger()
     }
   })
@@ -37,3 +37,49 @@ sandboxTest('watch non-existing directory', async ({ sandbox }) => {
 
   await expect(sandbox.files.watch(dirname, () => { })).rejects.toThrowError(NotFoundError)
 })
+
+sandboxTest('watch recursive directory changes', async ({ sandbox }) => {
+  const parentDir = 'a'
+  const childDir = 'b'
+  const fileName = 'test_watch.txt'
+  const filePath = `/${parentDir}/${childDir}/${fileName}`
+  const content = 'This file will be watched.'
+
+  onTestFinished(async () =>  {
+    await sandbox.files.remove('/'+parentDir)
+  })
+
+  let trigger1: () => void
+  let trigger2: () => void
+  let trigger3: () => void
+
+  const eventPromise1 = new Promise<void>((resolve) => {
+    trigger1 = resolve
+  })
+
+  const eventPromise2 = new Promise<void>((resolve) => {
+    trigger2 = resolve
+  })
+
+  const eventPromise3 = new Promise<void>((resolve) => {
+    trigger3 = resolve
+  })
+
+  const handle = await sandbox.files.watch('/', async (event) => {
+    assert.strictEqual(event.type, FilesystemEventType.CREATE)
+    // watch parent dir creation
+    if (event.name === `/${parentDir}`) trigger1()
+    // watch child dir creation
+    if (event.name === `/${parentDir}/${childDir}`) trigger2()
+    // watch nested file creation
+    if (event.name === filePath) trigger3()
+  }, { eventTypes: new Set([FilesystemEventType.CREATE])})
+
+  await sandbox.files.makeDir('/' + parentDir)
+  await sandbox.files.makeDir(`/${parentDir}/${childDir}`)
+  await sandbox.files.write(filePath, content)
+
+  await Promise.all([eventPromise1, eventPromise2, eventPromise3])
+
+  await handle.close()
+})