{"version":3,"file":"http_helpers.cjs","names":[],"sources":["../../../src/pframe/internal_api/http_helpers.ts"],"sourcesContent":["import type { Readable } from \"node:stream\";\nimport type { RequestListener } from \"node:http\";\nimport type { Branded, Base64Encoded } from \"@milaboratories/pl-model-common\";\nimport type { Logger } from \"./common\";\n\n/** Parquet file extension */\nexport const ParquetExtension = \".parquet\" as const;\n\n/** Parquet file name */\nexport type ParquetFileName = Branded<`${string}.parquet`, \"PFrameInternal.ParquetFileName\">;\n\nexport type FileRange = {\n  /** Start byte position (inclusive) */\n  start: number;\n  /** End byte position (inclusive) */\n  end: number;\n};\n\n/** HTTP range as of RFC 9110 <https://datatracker.ietf.org/doc/html/rfc9110#name-range> */\nexport type HttpRange =\n  | {\n      /**\n       * Get file content in the specified byte range\n       *\n       * @example\n       * ```\n       * GET /file.parquet HTTP/1.1\n       * Range: bytes=0-1023\n       * ```\n       */\n      type: \"bounded\";\n      /** Start byte position (inclusive) */\n      start: number;\n      /** End byte position (inclusive) */\n      end: number;\n    }\n  | {\n      /**\n       * Get byte range starting from the specified offset\n       *\n       * @example\n       * ```\n       * GET /file.parquet HTTP/1.1\n       * Range: bytes=1024-\n       * ```\n       */\n      type: \"offset\";\n      /** Start byte position (inclusive) */\n      offset: number;\n    }\n  | {\n      /**\n       * Get byte range starting from the specified suffix\n       *\n       * @example\n       * ```\n       * GET /file.parquet HTTP/1.1\n       * Range: bytes=-1024\n       * ```\n       */\n      type: \"suffix\";\n      /** End byte position (inclusive) */\n      suffix: number;\n    };\n\n/** HTTP method passed to object store */\nexport type HttpMethod = \"GET\" | \"HEAD\";\n\n/** HTTP response from object store */\nexport type ObjectStoreResponse =\n  | {\n      /**\n       * Will be translated to 500 Internal Server Error by the handler\n       * or 408 Request Timeout if the request was aborted\n       */\n      type: \"InternalError\";\n    }\n  | {\n      /** Will be translated to 404 Not Found by the handler */\n      type: \"NotFound\";\n    }\n  | {\n      /** Will be translated to 416 Range Not Satisfiable by the handler */\n      type: \"RangeNotSatisfiable\";\n      /** Total file size in bytes */\n      size: number;\n    }\n  | {\n      /** Will be translated to 200 OK or 206 Partial Content by the handler */\n      type: \"Ok\";\n      /** Total file size in bytes */\n      size: number;\n      /** File range translated from HTTP range */\n      range: FileRange;\n      /** Stream of file content, undefined for HEAD requests */\n      data?: Readable;\n    };\n\n/** Common options for object store creation */\nexport interface ObjectStoreOptions {\n  /** Logger instance, no logging is performed when not provided */\n  logger?: Logger;\n}\n\n/** Options for file system object store creation */\nexport interface FsStoreOptions extends ObjectStoreOptions {\n  /** Local directory to serve files from */\n  rootDir: string;\n}\n\nexport interface ObjectStore {\n  /**\n   * Proxy HTTP(S) request for parquet file to object store.\n   * Callback promise resolves when stream is closed by handler @see HttpHelpers.createRequestHandler\n   * Callback API is used so that ObjectStore can limit the number of concurrent requests.\n   */\n  request(\n    filename: ParquetFileName,\n    params: {\n      method: HttpMethod;\n      range?: HttpRange;\n      signal: AbortSignal;\n      callback: (response: ObjectStoreResponse) => Promise<void>;\n    },\n  ): void;\n}\n\n/** File system abstraction for request handler factory, @see HttpHelpers.createRequestHandler */\nexport abstract class BaseObjectStore implements ObjectStore {\n  protected readonly logger: Logger;\n\n  constructor(options: ObjectStoreOptions) {\n    this.logger = options.logger ?? (() => {});\n  }\n\n  /** Translate HTTP range to file range, @returns null if the range is not satisfiable */\n  protected translate(fileSize: number, range?: HttpRange): FileRange | null {\n    if (!range) return { start: 0, end: fileSize - 1 };\n    switch (range.type) {\n      case \"bounded\":\n        if (range.end >= fileSize) return null;\n        return { start: range.start, end: range.end };\n      case \"offset\":\n        if (range.offset >= fileSize) return null;\n        return { start: range.offset, end: fileSize - 1 };\n      case \"suffix\":\n        if (range.suffix > fileSize) return null;\n        return { start: fileSize - range.suffix, end: fileSize - 1 };\n    }\n  }\n\n  /**\n   * Proxy HTTP(S) request for parquet file to object store.\n   * Callback promise resolves when stream is closed by handler @see HttpHelpers.createRequestHandler\n   * Callback API is used so that ObjectStore can limit the number of concurrent requests.\n   */\n  abstract request(\n    filename: ParquetFileName,\n    params: {\n      method: HttpMethod;\n      range?: HttpRange;\n      signal: AbortSignal;\n      callback: (response: ObjectStoreResponse) => Promise<void>;\n    },\n  ): Promise<void>;\n}\n\n/** Configuration for {@link HttpHelpers.createCachingObjectStore} */\nexport type CacheConfig = {\n  /** Filesystem path where the cache persists its data */\n  cachePath: string;\n  /** Hard total size budget in bytes; the cache self-stabilizes near this size */\n  maxSizeBytes: number;\n  /**\n   * Max share of the budget a single file may occupy (0..1); when a file exceeds it, that\n   * file's oldest ranges are evicted first. Stops one large scan from flushing the working set.\n   */\n  admissionFraction: number;\n  /** Upper bound on previously-evicted (ghost) files remembered to guide readmission; resident files are already byte-bounded */\n  maxFilesTracked: number;\n};\n\n/**\n * Cumulative cache event counters. Surfaced both as lifetime totals (persisted across restarts)\n * and as a per-process session view, @see CacheMetrics.\n */\nexport type CacheCounters = {\n  /** Requests fully served from cache */\n  hits: number;\n  /** Requests that fell through to upstream */\n  misses: number;\n  /** Bytes served from cache */\n  bytesServed: number;\n  /**\n   * Bytes requested by clients that were not cached; compare with {@link bytesFetched} for read-ahead\n   * amplification.\n   */\n  bytesMissed: number;\n  /** Bytes downloaded from upstream on misses, including read-ahead and read-behind */\n  bytesFetched: number;\n  /** Bytes evicted to keep the cache within its total size budget, @see CacheConfig.maxSizeBytes */\n  bytesEvictedByBudget: number;\n  /** Bytes evicted because a single file exceeded its allowed share of the budget, @see CacheConfig.admissionFraction */\n  bytesEvictedByFileCap: number;\n  /** Files promoted on observed reuse, so they are retained longer than first-time entries */\n  promotions: number;\n  /** Previously evicted files that were re-requested and readmitted */\n  ghostReadmissions: number;\n};\n\n/** Snapshot of cache state and counters, @see CachingObjectStore.getMetrics */\nexport type CacheMetrics = {\n  /** Current resident bytes */\n  sizeBytes: number;\n  /** Number of files with resident (cached) bytes */\n  residentFiles: number;\n  /** Number of previously-evicted (ghost) files remembered, <= @see CacheConfig.maxFilesTracked */\n  ghostFiles: number;\n  /** Counters cumulative across all sessions (persisted) */\n  lifetime: CacheCounters;\n  /** The same counters scoped to the current process */\n  session: CacheCounters;\n};\n\n/**\n * Options for caching object store creation, @see HttpHelpers.createCachingObjectStore.\n * Standalone (not extending {@link ObjectStoreOptions}): a caching store is a decorator over\n * `upstream`, so source-store options do not apply to it.\n */\nexport interface CachingObjectStoreOptions {\n  /** Upstream store consulted on cache misses */\n  upstream: ObjectStore;\n  /** Cache configuration */\n  config: CacheConfig;\n  /** Logger instance, no logging is performed when not provided */\n  logger?: Logger;\n}\n\n/**\n * An {@link ObjectStore} that serves byte ranges from a persistent local cache, fetching misses\n * from an upstream store. This is the single handle for the cache: pass it as\n * {@link RequestHandlerOptions.store}, read {@link CachingObjectStore.getMetrics}, and dispose it\n * to release the cache, @see HttpHelpers.createCachingObjectStore.\n */\nexport interface CachingObjectStore extends ObjectStore, AsyncDisposable {\n  /** Instantaneous cache state plus lifetime/session counters */\n  getMetrics(): CacheMetrics;\n  /** Drop all cached data and zero the counters (test/benchmark use) */\n  reset(): Promise<void>;\n}\n\n/** Object store base URL in format accepted by Apache DataFusion and DuckDB */\nexport type ObjectStoreUrl = Branded<string, \"PFrameInternal.ObjectStoreUrl\">;\n\n/** HTTP(S) request handler creation options */\nexport type RequestHandlerOptions = {\n  /**\n   * Object store to serve files from. Compose caching by wrapping the upstream store with\n   * {@link HttpHelpers.createCachingObjectStore} and passing the result here,\n   * @see HttpHelpers.createFsStore\n   */\n  store: ObjectStore;\n};\n\n/** Server configuration options */\nexport type HttpServerOptions = {\n  /** HTTP(S) request handler function, @see HttpHelpers.createRequestHandler */\n  handler: RequestListener;\n  /** Port to bind to, @default 0 for auto-assignment */\n  port?: number;\n  /** Do not apply authorization middleware to @param handler */\n  noAuth?: true;\n  /** Downgrade default HTTPS server to plain HTTP, @warning use only for testing */\n  noHttps?: true;\n};\n\n/**\n * Long unique opaque string for use in Bearer authorization header\n *\n * @example\n * ```ts\n * request.setHeader('Authorization', `Bearer ${authToken}`);\n * ```\n */\nexport type HttpAuthorizationToken = Branded<string, \"PFrameInternal.HttpAuthorizationToken\">;\n\n/**\n * TLS certificate in PEM format\n *\n * @example\n * ```txt\n * -----BEGIN CERTIFICATE-----\n * MIIC2zCCAcOgAwIBAgIJaVW7...\n * ...\n * ...Yf9CRK8fgnukKM7TJ\n * -----END CERTIFICATE-----\n * ```\n */\nexport type PemCertificate = Branded<string, \"PFrameInternal.PemCertificate\">;\n\n/** HTTP(S) server connection settings, {@link HttpHelpers.createHttpServer} */\nexport type HttpServerInfo = {\n  /** URL of the HTTP(S) server formatted as `http{s}://<host>:<port>/` */\n  url: ObjectStoreUrl;\n  /** Authorization token for Bearer scheme, undefined when @see HttpServerOptions.noAuth flag is set */\n  authToken?: HttpAuthorizationToken;\n  /** Encoded CA certificate of HTTPS server, undefined when @see HttpServerOptions.noHttps flag is set */\n  encodedCaCert?: Base64Encoded<PemCertificate>;\n};\n\n/** HTTP(S) server information and controls, @see HttpHelpers.createHttpServer */\nexport interface HttpServer {\n  /** Server configuration information for initiating connections from clients */\n  get info(): HttpServerInfo;\n  /** Promise that resolves when the server is stopped */\n  get stopped(): Promise<void>;\n  /** Request server stop, returns the same promise as @see HttpServer.stopped */\n  stop(): Promise<void>;\n}\n\n/** List of HTTP(S) related helper functions exposed by PFrame module */\nexport interface HttpHelpers {\n  /**\n   * Create an object store for serving files from a local directory.\n   * Rejects if the provided path does not exist or is not a directory.\n   */\n  createFsStore(options: FsStoreOptions): Promise<ObjectStore>;\n\n  /**\n   * Wrap an upstream object store in a persistent byte-range cache backed by SQLite.\n   * The returned store is itself an {@link ObjectStore} (pass it to {@link createRequestHandler})\n   * and additionally exposes metrics and reset. Dispose it to close the cache database.\n   */\n  createCachingObjectStore(options: CachingObjectStoreOptions): Promise<CachingObjectStore>;\n\n  /**\n   * Create an HTTP request handler for serving files from an object store.\n   * Accepts only paths of the form `/<filename>.parquet`, returns 410 otherwise.\n   * Assumes that files are immutable (and sets cache headers accordingly).\n   */\n  createRequestHandler(options: RequestHandlerOptions): RequestListener;\n\n  /**\n   * Serve HTTP(S) requests using the provided handler on localhost port.\n   * @returns promise that resolves when the server has stopped.\n   *\n   * @example\n   * ```ts\n   * const rootDir = '/path/to/directory/with/parquet/files';\n   *\n   * let store = await HttpHelpers.createFsStore({ rootDir }).catch((err: unknown) => {\n   *   throw new Error(`Failed to create file store for ${rootDir} - ${ensureError(err)}`);\n   * });\n   *\n   * const server = await HttpHelpers.createHttpServer({\n   *   handler: HttpHelpers.createRequestHandler({ store }),\n   * }).catch((err: unknown) => {\n   *   throw new Error(`Failed to start HTTPS server - ${ensureError(err)}`);\n   * });\n   *\n   * const { url, authToken, encodedCaCert } = server.info;\n   *\n   * await server.stop();\n   * ```\n   */\n  createHttpServer(options: HttpServerOptions): Promise<HttpServer>;\n}\n"],"mappings":";;AAMA,MAAa,mBAAmB;;AA0HhC,IAAsB,kBAAtB,MAA6D;CAC3D;CAEA,YAAY,SAA6B;EACvC,KAAK,SAAS,QAAQ,iBAAiB,CAAC;CAC1C;;CAGA,UAAoB,UAAkB,OAAqC;EACzE,IAAI,CAAC,OAAO,OAAO;GAAE,OAAO;GAAG,KAAK,WAAW;EAAE;EACjD,QAAQ,MAAM,MAAd;GACE,KAAK;IACH,IAAI,MAAM,OAAO,UAAU,OAAO;IAClC,OAAO;KAAE,OAAO,MAAM;KAAO,KAAK,MAAM;IAAI;GAC9C,KAAK;IACH,IAAI,MAAM,UAAU,UAAU,OAAO;IACrC,OAAO;KAAE,OAAO,MAAM;KAAQ,KAAK,WAAW;IAAE;GAClD,KAAK;IACH,IAAI,MAAM,SAAS,UAAU,OAAO;IACpC,OAAO;KAAE,OAAO,WAAW,MAAM;KAAQ,KAAK,WAAW;IAAE;EAC/D;CACF;AAgBF"}