[Ingest Manager] Lift up registry/{stream,extract} functions (#83239)

## Summary

  * Move stream utility functions from `server/services/epm/registry/streams.ts` to `server/services/epm/streams.ts`
    * They're only used in registry at the moment but aren't specific to registry 
  * Move archive extraction functions from `server/services/epm/registry/extract.ts` to `server/services/epm/archive.ts`
    * The Registry isn't the only service/code which needs to extract packages. Continue consolidating archive-related code under archive vs registry

x-pack/plugins/fleet/server/services/epm/registry/extract.ts → x-pack/plugins/fleet/server/services/epm/archive/extract.ts

@@ -6,12 +6,8 @@
 
 import tar from 'tar';
 import yauzl from 'yauzl';
-import { bufferToStream, streamToBuffer } from './streams';
-
-export interface ArchiveEntry {
-  path: string;
-  buffer?: Buffer;
-}
+import { bufferToStream, streamToBuffer } from '../streams';
+import { ArchiveEntry } from './index';
 
 export async function untarBuffer(
   buffer: Buffer,

x-pack/plugins/fleet/server/services/epm/archive/index.ts

@@ -14,10 +14,16 @@ import {
   setArchiveFilelist,
   deleteArchiveFilelist,
 } from './cache';
-import { ArchiveEntry, getBufferExtractor } from '../registry/extract';
+import { getBufferExtractor } from './extract';
 import { parseAndVerifyArchiveEntries } from './validation';
 
 export * from './cache';
+export { untarBuffer, unzipBuffer, getBufferExtractor } from './extract';
+
+export interface ArchiveEntry {
+  path: string;
+  buffer?: Buffer;
+}
 
 export async function getArchivePackage({
   archiveBuffer,

x-pack/plugins/fleet/server/services/epm/archive/validation.ts

@@ -15,7 +15,8 @@ import {
   RegistryVarsEntry,
 } from '../../../../common/types';
 import { PackageInvalidArchiveError } from '../../../errors';
-import { ArchiveEntry, pkgToPkgKey } from '../registry';
+import { ArchiveEntry } from './index';
+import { pkgToPkgKey } from '../registry';
 
 const MANIFESTS: Record<string, Buffer> = {};
 const MANIFEST_NAME = 'manifest.yml';

x-pack/plugins/fleet/server/services/epm/elasticsearch/ingest_pipeline/install.ts

@@ -11,8 +11,7 @@ import {
   ElasticsearchAssetType,
   InstallablePackage,
 } from '../../../../types';
-import { ArchiveEntry } from '../../registry';
-import { getAsset, getPathParts } from '../../archive';
+import { ArchiveEntry, getAsset, getPathParts } from '../../archive';
 import { CallESAsCurrentUser } from '../../../../types';
 import { saveInstalledEsRefs } from '../../packages/install';
 import { getInstallationObject } from '../../packages';

x-pack/plugins/fleet/server/services/epm/packages/assets.ts

@@ -6,7 +6,7 @@
 
 import { InstallablePackage } from '../../../types';
 import * as Registry from '../registry';
-import { getArchiveFilelist, getAsset } from '../archive';
+import { ArchiveEntry, getArchiveFilelist, getAsset } from '../archive';
 
 // paths from RegistryPackage are routes to the assets on EPR
 // e.g. `/package/nginx/1.2.0/data_stream/access/fields/fields.yml`
@@ -51,14 +51,14 @@ export async function getAssetsData(
   packageInfo: InstallablePackage,
   filter = (path: string): boolean => true,
   datasetName?: string
-): Promise<Registry.ArchiveEntry[]> {
+): Promise<ArchiveEntry[]> {
   // TODO: Needs to be called to fill the cache but should not be required
 
   await Registry.ensureCachedArchiveInfo(packageInfo.name, packageInfo.version, 'registry');
 
   // Gather all asset data
   const assets = getAssets(packageInfo, filter, datasetName);
-  const entries: Registry.ArchiveEntry[] = assets.map((path) => {
+  const entries: ArchiveEntry[] = assets.map((path) => {
     const buffer = getAsset(path);
 
     return { path, buffer };

x-pack/plugins/fleet/server/services/epm/registry/index.test.ts

@@ -5,9 +5,8 @@
  */
 
 import { AssetParts } from '../../../types';
-import { getPathParts } from '../archive';
-import { getBufferExtractor, splitPkgKey } from './index';
-import { untarBuffer, unzipBuffer } from './extract';
+import { getBufferExtractor, getPathParts, untarBuffer, unzipBuffer } from '../archive';
+import { splitPkgKey } from './index';
 
 const testPaths = [
   {

x-pack/plugins/fleet/server/services/epm/registry/index.ts

@@ -24,13 +24,11 @@ import {
   unpackArchiveToCache,
 } from '../archive';
 import { fetchUrl, getResponse, getResponseStream } from './requests';
-import { streamToBuffer } from './streams';
+import { streamToBuffer } from '../streams';
 import { getRegistryUrl } from './registry_url';
 import { appContextService } from '../..';
 import { PackageNotFoundError, PackageCacheError } from '../../../errors';
 
-export { ArchiveEntry, getBufferExtractor } from './extract';
-
 export interface SearchParams {
   category?: CategoryId;
   experimental?: boolean;

x-pack/plugins/fleet/server/services/epm/registry/requests.ts

@@ -6,7 +6,7 @@
 
 import fetch, { FetchError, Response, RequestInit } from 'node-fetch';
 import pRetry from 'p-retry';
-import { streamToString } from './streams';
+import { streamToString } from '../streams';
 import { appContextService } from '../../app_context';
 import { RegistryError, RegistryConnectionError, RegistryResponseError } from '../../../errors';
 import { getProxyAgent, getRegistryProxyUrl } from './proxy';