123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184
import { Collection } from "../collection/collection.ts";import { FindCursor } from "../collection/commands/find.ts";import { Database } from "../database.ts";import { Filter } from "../types.ts";import {  Chunk,  File,  FileId,  GridFSBucketOptions,  GridFSFindOptions,  GridFSUploadOptions,} from "../types/gridfs.ts";import { checkIndexes } from "./indexes.ts";import { createUploadStream } from "./upload.ts";import { MongoRuntimeError } from "../error.ts";import { ObjectId } from "../../deps.ts";
export class GridFSBucket {  #chunksCollection: Collection<Chunk>;  #filesCollection: Collection<File>;  #chunkSizeBytes: number;  #checkedIndexes = false;
  private readonly getBucketData = () => ({    filesCollection: this.#filesCollection,    chunksCollection: this.#chunksCollection,    chunkSizeBytes: this.#chunkSizeBytes,  });
  /**   * Create a new GridFSBucket object on @db with the given @options.   */  constructor(db: Database, options: GridFSBucketOptions = {}) {    const newLocal = options.bucketName ?? "fs";    this.#chunksCollection = db.collection(`${newLocal}.chunks`);    this.#filesCollection = db.collection(`${newLocal}.files`);    this.#chunkSizeBytes = options.chunkSizeBytes ?? 255 * 1024;  }
  /**   * Opens a Stream that the application can write the contents of the file to.   * The driver generates the file id.   *   * Returns a Stream to which the application will write the contents.   *   * Note: this method is provided for backward compatibility. In languages   * that use generic type parameters, this method may be omitted since   * the TFileId type might not be an ObjectId.   */  openUploadStream(filename: string, options?: GridFSUploadOptions) {    return this.openUploadStreamWithId(      new ObjectId(),      filename,      options,    );  }
  /**   * Opens a Stream that the application can write the contents of the file to.   * The application provides a custom file id.   *   * Returns a Stream to which the application will write the contents.   */  async openUploadStreamWithId(    id: FileId,    filename: string,    options?: GridFSUploadOptions,  ) {    if (!this.#checkedIndexes) await this.#checkIndexes();    return createUploadStream(this.getBucketData(), filename, id, options);  }
  /**   * Uploads a user file to a GridFS bucket. The driver generates the file id.   *   * Reads the contents of the user file from the @source Stream and uploads it   * as chunks in the chunks collection. After all the chunks have been uploaded,   * it creates a files collection document for @filename in the files collection.   *   * Returns the id of the uploaded file.   *   * Note: this method is provided for backward compatibility. In languages   * that use generic type parameters, this method may be omitted since   * the TFileId type might not be an ObjectId.   */  async uploadFromStream(    filename: string,    source: ReadableStream,    options?: GridFSUploadOptions,  ): Promise<ObjectId> {    const objectid = new ObjectId();    await source.pipeTo(      await this.openUploadStreamWithId(objectid, filename, options),    );    return objectid;  }
  /**   * Uploads a user file to a GridFS bucket. The application supplies a custom file id.   *   * Reads the contents of the user file from the @source Stream and uploads it   * as chunks in the chunks collection. After all the chunks have been uploaded,   * it creates a files collection document for @filename in the files collection.   *   * Note: there is no need to return the id of the uploaded file because the application   * already supplied it as a parameter.   */  async uploadFromStreamWithId(    id: FileId,    filename: string,    source: ReadableStream,    options: GridFSUploadOptions,  ): Promise<void> {    await source.pipeTo(      await this.openUploadStreamWithId(id, filename, options),    );  }
  /** Opens a Stream from which the application can read the contents of the stored file   * specified by @id.   *   * Returns a Stream.   */  async openDownloadStream(id: FileId) {    if (!this.#checkedIndexes) await this.#checkIndexes();
    return new ReadableStream<Uint8Array>({      start: async (controller) => {        const collection = this.#chunksCollection.find({ files_id: id });        await collection.forEach((value) =>          controller.enqueue(value?.data.buffer)        );        controller.close();      },    });  }
  /**   * Downloads the contents of the stored file specified by @id and writes   * the contents to the @destination Stream.   */  async downloadToStream(id: FileId, destination: WritableStream<Uint8Array>) {    await (await this.openDownloadStream(id)).pipeTo(destination);  }
  /**   * Given a @id, delete this stored file’s files collection document and   * associated chunks from a GridFS bucket.   */  async delete(id: FileId) {    await this.#filesCollection.deleteOne({ _id: id });    const response = await this.#chunksCollection.deleteMany({ files_id: id });    if (!response) {      throw new MongoRuntimeError(`File not found for id ${id}`);    }  }
  /**   * Find and return the files collection documents that match @filter.   */  find(    filter: Filter<File>,    options: GridFSFindOptions = {},  ): FindCursor<File> {    return this.#filesCollection.find(filter ?? {}, options);  }
  /**   * Drops the files and chunks collections associated with   * this bucket.   */  async drop() {    await this.#filesCollection.drop();    await this.#chunksCollection.drop();  }
  #checkIndexes = () =>    checkIndexes(      this.#filesCollection,      this.#chunksCollection,      (value) => (this.#checkedIndexes = value),    );}