feat: add pooled Docker Compose instances

.changeset/fuzzy-pools-compose.md

@@ -0,0 +1,5 @@
+---
+'prool': patch
+---
+
+Added pooled Docker Compose instances with named service endpoints.

src/testcontainers/Instance.ts

@@ -17,6 +17,167 @@ import * as ContainerOptions from './containerOptions.js'
 
 export type { Endpoint, Instance, InstanceOptions } from '../Instance.js'
 
+/**
+ * Defines an instance backed by a Docker Compose environment.
+ *
+ * @example
+ * ```ts
+ * const instance = Instance.compose({
+ *   name: 'services',
+ *   environment: () => new DockerComposeEnvironment('.', 'compose.yml'),
+ *   services: ['api'],
+ *   endpoints: {
+ *     default: { container: 'api-1', protocol: 'http', port: 8080 },
+ *   },
+ * })
+ * const pool = Pool.define({ instance })
+ * await pool.start(Number(process.env.VITEST_POOL_ID ?? 1))
+ * ```
+ */
+export function compose<
+  const endpointDefinitions extends compose.EndpointDefinitions,
+>(
+  parameters: compose.Parameters<endpointDefinitions>,
+  options?: Instance.InstanceOptions,
+): Instance.Instance<undefined, compose.Endpoints<endpointDefinitions>> {
+  const initialEndpoints = Object.fromEntries(
+    Object.entries(parameters.endpoints).flatMap(([name, endpoint]) =>
+      endpoint
+        ? [
+            [
+              name,
+              {
+                host: 'localhost',
+                port: endpoint.port,
+                protocol: endpoint.protocol,
+              },
+            ],
+          ]
+        : [],
+    ),
+  ) as compose.Endpoints<endpointDefinitions>
+
+  const definition = Instance.define<
+    undefined,
+    undefined,
+    compose.Endpoints<endpointDefinitions>
+  >(() => {
+    let environment: compose.StartedEnvironment | undefined
+
+    async function stopEnvironment() {
+      if (!environment) return
+      const started = environment
+      await started.down(parameters.down)
+      if (environment === started) environment = undefined
+    }
+
+    return {
+      endpoints: initialEndpoints,
+      host: initialEndpoints.default.host,
+      name: parameters.name,
+      port: initialEndpoints.default.port,
+      async start(_, { setEndpoint }) {
+        await stopEnvironment()
+        const started = await parameters
+          .environment()
+          .up(parameters.services ? [...parameters.services] : undefined)
+        environment = started
+
+        try {
+          const endpoints = Object.entries(parameters.endpoints).flatMap(
+            ([name, definition]) => {
+              if (!definition) return []
+              const container = started.getContainer(definition.container)
+              return [
+                [
+                  name,
+                  {
+                    host: container.getHost(),
+                    port: container.getMappedPort(definition.port),
+                    protocol: definition.protocol,
+                  },
+                ] as const,
+              ]
+            },
+          )
+          const applyEndpoint = setEndpoint as
+            | ((name: string, endpoint: Instance.Endpoint) => void)
+            | undefined
+          for (const [name, endpoint] of endpoints)
+            applyEndpoint?.(name, endpoint)
+        } catch (error) {
+          const [result] = await Promise.allSettled([
+            started.down(parameters.down),
+          ])
+          if (result?.status === 'fulfilled') environment = undefined
+          throw error
+        }
+      },
+      async stop() {
+        await stopEnvironment()
+      },
+    }
+  })
+
+  return definition(options)
+}
+
+export declare namespace compose {
+  export type EndpointDefinition<
+    protocol extends Instance.Endpoint.Protocol = Instance.Endpoint.Protocol,
+  > = {
+    container: string
+    port: number
+    protocol: protocol
+  }
+
+  export type EndpointDefinitions = {
+    default: EndpointDefinition
+    [name: string]: EndpointDefinition | undefined
+  }
+
+  export type Endpoints<definitions extends EndpointDefinitions> = {
+    default: Instance.Endpoint<definitions['default']['protocol']>
+  } & {
+    [name in Exclude<keyof definitions, 'default'>]: Endpoint<definitions[name]>
+  }
+
+  export type Endpoint<definition extends EndpointDefinition | undefined> =
+    definition extends EndpointDefinition
+      ? Instance.Endpoint<definition['protocol']>
+      : undefined
+
+  export type DownOptions = {
+    removeVolumes?: boolean | undefined
+    timeout?: number | undefined
+  }
+
+  export type Environment = {
+    up(services?: string[] | undefined): Promise<StartedEnvironment>
+  }
+
+  export type Parameters<definitions extends EndpointDefinitions> = {
+    /** Options passed to `docker compose down`. */
+    down?: DownOptions | undefined
+    /** Container ports exposed as named instance endpoints. */
+    endpoints: definitions
+    /** Creates a fresh Compose environment for each pooled instance. */
+    environment: () => Environment
+    /** Instance name reported by Prool. */
+    name: string
+    /** Compose services to start. Dependencies start automatically. */
+    services?: readonly string[] | undefined
+  }
+
+  export type StartedEnvironment = {
+    down(options?: DownOptions | undefined): Promise<unknown>
+    getContainer(name: string): {
+      getHost(): string
+      getMappedPort(port: number): number
+    }
+  }
+}
+
 /**
  * Defines an instance backed by a Testcontainers container.
  *

src/testcontainers/compose.test.ts

@@ -0,0 +1,197 @@
+import { Pool } from 'prool'
+import { Instance } from 'prool/testcontainers'
+import { expect, expectTypeOf, test, vi } from 'vitest'
+
+function environment(parameters: {
+  down: () => Promise<unknown>
+  host: string
+  port: number
+  services: (services: string[] | undefined) => void
+}) {
+  return {
+    async up(services: string[] | undefined) {
+      parameters.services(services)
+      return {
+        down: parameters.down,
+        getContainer() {
+          return {
+            getHost: () => parameters.host,
+            getMappedPort: () => parameters.port,
+          }
+        },
+      }
+    },
+  }
+}
+
+test('pools compose environments with mapped endpoints', async () => {
+  const downs: ReturnType<typeof vi.fn>[] = []
+  const metrics = true as boolean
+  const services: (string[] | undefined)[] = []
+  let environments = 0
+  const instance = Instance.compose({
+    down: { removeVolumes: true },
+    endpoints: {
+      default: { container: 'api-1', port: 8080, protocol: 'http' },
+      database: { container: 'database-1', port: 5432, protocol: 'tcp' },
+      logs: undefined,
+      metrics: metrics
+        ? { container: 'metrics-1', port: 9090, protocol: 'ws' }
+        : undefined,
+    },
+    environment() {
+      environments++
+      const down = vi.fn(async () => {})
+      downs.push(down)
+      return environment({
+        down,
+        host: '127.0.0.1',
+        port: environments * 10_000,
+        services: (value) => services.push(value),
+      })
+    },
+    name: 'services',
+    services: ['api', 'database'],
+  })
+  const pool = Pool.define({ instance })
+
+  const [first, second] = await Promise.all([pool.start(1), pool.start(2)])
+
+  expect(first.endpoints.default).toEqual({
+    host: '127.0.0.1',
+    port: expect.any(Number),
+    protocol: 'http',
+  })
+  expect(second.endpoints.database).toEqual({
+    host: '127.0.0.1',
+    port: expect.any(Number),
+    protocol: 'tcp',
+  })
+  expect(
+    [first.endpoints.default.port, second.endpoints.database.port].sort(
+      (a, b) => a - b,
+    ),
+  ).toEqual([10_000, 20_000])
+  expectTypeOf(first.endpoints.default).toEqualTypeOf<
+    Instance.Endpoint<'http'>
+  >()
+  expectTypeOf(first.endpoints.database).toEqualTypeOf<
+    Instance.Endpoint<'tcp'>
+  >()
+  expectTypeOf(first.endpoints.logs).toEqualTypeOf<undefined>()
+  expectTypeOf(first.endpoints.metrics).toEqualTypeOf<
+    Instance.Endpoint<'ws'> | undefined
+  >()
+  expect(services).toEqual([
+    ['api', 'database'],
+    ['api', 'database'],
+  ])
+
+  await pool.restart(1)
+
+  expect(first.endpoints.default.port).toBe(30_000)
+
+  await pool.destroyAll()
+
+  expect(downs).toHaveLength(3)
+  for (const down of downs) {
+    expect(down).toHaveBeenCalledOnce()
+    expect(down).toHaveBeenCalledWith({ removeVolumes: true })
+  }
+})
+
+test('tears down an environment when endpoint mapping fails', async () => {
+  const down = vi.fn(async () => {})
+  const instance = Instance.compose({
+    endpoints: {
+      default: { container: 'missing-1', port: 8080, protocol: 'http' },
+    },
+    environment: () => ({
+      async up() {
+        return {
+          down,
+          getContainer() {
+            throw new Error('missing container')
+          },
+        }
+      },
+    }),
+    name: 'services',
+  })
+
+  await expect(instance.start()).rejects.toThrow('missing container')
+  expect(down).toHaveBeenCalledOnce()
+  expect(instance.status).toBe('idle')
+})
+
+test('cleans a retained environment before retrying start', async () => {
+  const retainedDown = vi
+    .fn<() => Promise<void>>()
+    .mockRejectedValueOnce(new Error('down failed'))
+    .mockResolvedValueOnce(undefined)
+  const replacementDown = vi.fn(async () => {})
+  let starts = 0
+  const instance = Instance.compose({
+    endpoints: {
+      default: { container: 'api-1', port: 8080, protocol: 'http' },
+    },
+    environment: () => ({
+      async up() {
+        starts++
+        if (starts === 1)
+          return {
+            down: retainedDown,
+            getContainer() {
+              throw new Error('missing container')
+            },
+          }
+        return {
+          down: replacementDown,
+          getContainer() {
+            return {
+              getHost: () => '127.0.0.1',
+              getMappedPort: (port: number) => port,
+            }
+          },
+        }
+      },
+    }),
+    name: 'services',
+  })
+
+  await expect(instance.start()).rejects.toThrow('missing container')
+  await instance.start()
+
+  expect(retainedDown).toHaveBeenCalledTimes(2)
+  expect(instance.status).toBe('started')
+
+  await instance.stop()
+  expect(replacementDown).toHaveBeenCalledOnce()
+})
+
+test('retains the environment when stopping fails', async () => {
+  const down = vi
+    .fn<() => Promise<void>>()
+    .mockRejectedValueOnce(new Error('down failed'))
+    .mockResolvedValueOnce(undefined)
+  const instance = Instance.compose({
+    endpoints: {
+      default: { container: 'api-1', port: 8080, protocol: 'http' },
+    },
+    environment: () =>
+      environment({
+        down,
+        host: '127.0.0.1',
+        port: 8080,
+        services: () => {},
+      }),
+    name: 'services',
+  })
+
+  await instance.start()
+  await expect(instance.stop()).rejects.toThrow('down failed')
+  expect(instance.status).toBe('started')
+
+  await instance.stop()
+  expect(down).toHaveBeenCalledTimes(2)
+})