Function layer

packages/core/integrations-registry/src/registry.ts:116–266

(
  config: IntegrationsRegistryConfig,
)

};

export const layer = (
  config: IntegrationsRegistryConfig,
): Layer.Layer<IntegrationsRegistry, never, HttpClient.HttpClient | FileSystem.FileSystem> =>
  Layer.effect(
    IntegrationsRegistry,
    Effect.gen(function* () {
      const fs = yield* FileSystem.FileSystem;
      const http = yield* HttpClient.HttpClient;

      const source =
        config.url ?? process.env.EXECUTOR_INTEGRATIONS_URL ?? DEFAULT_INTEGRATIONS_URL;
      const cacheDir = resolveCacheDir(config.cacheDir);
      const cacheFile = cacheFileFor(cacheDir, source);
      const lockFile = `${cacheFile}.lock`;
      const ttl = Duration.fromInputUnsafe(config.cacheTtl ?? Duration.hours(12));
      const refreshEvery = Duration.fromInputUnsafe(config.refreshInterval ?? Duration.hours(12));
      const disabled = config.disabled ?? isFetchDisabled();

      const isFresh = Effect.gen(function* () {
        const stat = yield* fs.stat(cacheFile).pipe(Effect.catch(() => Effect.succeed(undefined)));
        if (!stat) return false;
        const mtime = Option.getOrElse(stat.mtime, () => new Date(0)).getTime();
        return Date.now() - mtime < Duration.toMillis(ttl);
      });

      const fetchText = Effect.gen(function* () {
        const request = HttpClientRequest.get(source).pipe(
          HttpClientRequest.setHeader("user-agent", config.userAgent),
          HttpClientRequest.setHeader("accept", "application/json"),
        );
        const response = yield* http.execute(request);
        return yield* response.text;
      }).pipe(Effect.timeout(Duration.seconds(10)), Effect.withSpan("IntegrationsRegistry.fetch"));

      const writeCache = (text: string) =>
        Effect.gen(function* () {
          yield* fs.makeDirectory(cacheDir, { recursive: true }).pipe(Effect.ignore);
          yield* fs.writeFileString(cacheFile, text);
        });

      // Cross-process advisory lock so concurrent CLI invocations don't all
      // race to refresh the same file. Best-effort: if we can't acquire,
      // another process is already refreshing — skip and read what they wrote.
      const withWriteLock = <A, E>(use: Effect.Effect<A, E>) =>
        Effect.acquireUseRelease(
          Effect.gen(function* () {
            yield* fs.makeDirectory(cacheDir, { recursive: true }).pipe(Effect.ignore);
            return yield* fs.writeFileString(lockFile, `${process.pid}\n`, { flag: "wx" }).pipe(
              Effect.as(true),
              Effect.catch(() => Effect.succeed(false)),
            );
          }),
          (acquired): Effect.Effect<Option.Option<A>, E> =>
            acquired ? Effect.map(use, Option.some) : Effect.succeed(Option.none<A>()),
          (acquired) =>
            acquired ? fs.remove(lockFile, { force: true }).pipe(Effect.ignore) : Effect.void,
        );