Fix Minio for Export

packages/api/src/jobs/export.ts

@@ -14,20 +14,19 @@ import { logger } from '../utils/logger'
 import { highlightToMarkdown } from '../utils/parser'
 import { contentFilePath } from '../utils/uploads'
 import { env } from '../env'
-import { File, Storage } from '@google-cloud/storage'
+import { storage } from '../repository/storage/storage'
+import { File } from '../repository/storage/StorageClient'
+import { Readable } from 'stream'
 
 export interface ExportJobData {
   userId: string
   exportId: string
 }
 
-export const storage = env.fileUpload?.gcsUploadSAKeyFilePath
-  ? new Storage({ keyFilename: env.fileUpload.gcsUploadSAKeyFilePath })
-  : new Storage()
 const bucketName = env.fileUpload.gcsUploadBucket
 
 const createGCSFile = (filename: string): File => {
-  return storage.bucket(bucketName).file(filename)
+  return storage.createFile(bucketName, filename)
 }
 
 export const EXPORT_JOB_NAME = 'export'
@@ -60,7 +59,7 @@ const uploadContent = async (
   const file = createGCSFile(filePath)
 
   // check if file is already uploaded
-  const [exists] = await file.exists()
+  const exists = await file.exists()
   if (!exists) {
     logger.info(`File not found: ${filePath}`)
 
@@ -80,10 +79,14 @@ const uploadContent = async (
       contentType: 'text/html',
       private: true,
     })
+    archive.append(Readable.from(item.readableContent), {
+      name: `content/${libraryItem.slug}.html`,
+    })
   }
 
   // append the existing file to the archive
-  archive.append(file.createReadStream(), {
+  const content = await file.download()
+  archive.append(Readable.from(content.toString()), {
     name: `content/${libraryItem.slug}.html`,
   })
 }
@@ -171,7 +174,7 @@ export const exportJob = async (jobData: ExportJobData) => {
       logger.error('Failed to send export job email', {
         userId,
       })
-      return
+      // return
     }
 
     logger.info('exporting all items...', {
@@ -193,9 +196,18 @@ export const exportJob = async (jobData: ExportJobData) => {
 
     // Create a write stream
     const writeStream = file.createWriteStream({
-      metadata: {
-        contentType: 'application/zip',
-      },
+      contentType: 'application/zip',
+    })
+
+    const finishedPromise = new Promise<void>((resolve, reject) => {
+      if (writeStream.closed) {
+        resolve()
+      }
+      writeStream.on('finish', () => {
+        logger.info('File successfully written to GCS')
+        resolve()
+      })
+      writeStream.on('error', reject)
     })
 
     // Handle any errors in the streams
@@ -203,10 +215,6 @@ export const exportJob = async (jobData: ExportJobData) => {
       logger.error('Error writing to GCS:', err)
     })
 
-    writeStream.on('finish', () => {
-      logger.info('File successfully written to GCS')
-    })
-
     // Initialize archiver for zipping files
     const archive = archiver('zip', {
       zlib: { level: 9 }, // Compression level
@@ -252,17 +260,14 @@ export const exportJob = async (jobData: ExportJobData) => {
     }
 
     // Ensure that the writeStream has finished
-    await new Promise((resolve, reject) => {
-      writeStream.on('finish', resolve)
-      writeStream.on('error', reject)
-    })
+    await finishedPromise
 
     logger.info('export completed', {
       userId,
     })
 
     // generate a temporary signed url for the zip file
-    const [signedUrl] = await file.getSignedUrl({
+    const signedUrl = await storage.signedUrl(file.bucket, file.key, {
       action: 'read',
       expires: Date.now() + 86400 * 1000, // 15 minutes
     })
@@ -274,6 +279,7 @@ export const exportJob = async (jobData: ExportJobData) => {
 
     await saveExport(userId, {
       id: exportId,
+      signedUrl,
       state: TaskState.Succeeded,
     })
 

packages/api/src/repository/storage/GcsStorageClient.ts

@@ -1,4 +1,10 @@
-import { SignedUrlParameters, StorageClient, File } from './StorageClient'
+import {
+  SignedUrlParameters,
+  StorageClient,
+  File,
+  SaveOptions,
+  SaveData,
+} from './StorageClient'
 import { Storage, File as GCSFile } from '@google-cloud/storage'
 
 export class GcsStorageClient implements StorageClient {
@@ -24,6 +30,13 @@ export class GcsStorageClient implements StorageClient {
         const [metadata] = await gcsFile.getMetadata()
         return metadata.md5Hash
       },
+      createWriteStream: (saveOptions: SaveOptions) =>
+        gcsFile.createWriteStream({
+          metadata: { contentType: saveOptions.contentType },
+        }),
+      save: (saveData: SaveData, saveOptions: SaveOptions) =>
+        gcsFile.save(saveData, saveOptions),
+      key: gcsFile.name,
     }
   }
 
@@ -32,6 +45,10 @@ export class GcsStorageClient implements StorageClient {
     return Promise.resolve(this.convertFileToGeneric(file))
   }
 
+  createFile(bucket: string, filePath: string): File {
+    return this.convertFileToGeneric(this.storage.bucket(bucket).file(filePath))
+  }
+
   async getFilesFromPrefix(bucket: string, prefix: string): Promise<File[]> {
     const [filesWithPrefix] = await this.storage
       .bucket(bucket)

packages/api/src/repository/storage/S3StorageClient.ts

@@ -1,17 +1,94 @@
-import { SignedUrlParameters, StorageClient, File } from './StorageClient'
+import {
+  SignedUrlParameters,
+  StorageClient,
+  File,
+  SaveOptions,
+  SaveData,
+} from './StorageClient'
+import { Upload } from '@aws-sdk/lib-storage'
 import {
   GetObjectCommand,
   GetObjectCommandOutput,
   S3Client,
   ListObjectsV2Command,
   PutObjectCommand,
+  HeadObjectCommand,
+  S3ServiceException,
 } from '@aws-sdk/client-s3'
 import { getSignedUrl } from '@aws-sdk/s3-request-presigner'
-import type { Readable } from 'stream'
+import { Readable } from 'stream'
+import * as stream from 'node:stream'
 
 // While this is listed as S3, for self hosting we will use MinIO, which is
 // S3 Compatible.
 export class S3StorageClient implements StorageClient {
+  BlankFile = class implements File {
+    bucket: string
+    key: string
+    s3Client: S3StorageClient
+    downloadedFile: File | undefined
+
+    constructor(s3StorageClass: S3StorageClient, bucket: string, file: string) {
+      this.bucket = bucket
+      this.key = file
+      this.s3Client = s3StorageClass
+    }
+
+    isPublic() {
+      return Promise.resolve(true)
+    }
+
+    publicUrl() {
+      return `${this.s3Client.urlOverride ?? ''}/${this.bucket}/${this.key}`
+    }
+
+    async download(): Promise<Buffer> {
+      this.downloadedFile = await this.s3Client.downloadFile(
+        this.bucket,
+        this.key
+      )
+      return this.downloadedFile.download()
+    }
+
+    async exists() {
+      try {
+        await this.s3Client.s3Client.send(
+          new HeadObjectCommand({
+            Bucket: this.bucket,
+            Key: this.key,
+          })
+        )
+
+        return true
+      } catch (e) {
+        if (
+          e instanceof S3ServiceException &&
+          e.$metadata.httpStatusCode == 404
+        ) {
+          return false
+        }
+
+        throw e
+      }
+    }
+
+    save(saveData: SaveData, saveOptions: SaveOptions): Promise<void> {
+      return this.s3Client.upload(this.bucket, this.key, saveData, saveOptions)
+    }
+
+    createWriteStream(saveOptions: SaveOptions) {
+      return this.s3Client.createS3UploadStream(
+        this.bucket,
+        this.key,
+        saveOptions
+      )
+    }
+
+    getMetadataMd5() {
+      return this.downloadedFile?.getMetadataMd5() || Promise.resolve('')
+    }
+  }
+
   private s3Client: S3Client
   private urlOverride: string | undefined
 
@@ -23,16 +100,48 @@ export class S3StorageClient implements StorageClient {
     })
   }
 
+  private createS3UploadStream = (
+    bucket: string,
+    key: string,
+    saveOptions: SaveOptions
+  ) => {
+    const passThroughStream = new stream.PassThrough()
+
+    const upload = new Upload({
+      client: this.s3Client,
+      params: {
+        Bucket: bucket,
+        Key: key,
+        Body: passThroughStream,
+        ContentType: saveOptions.contentType,
+      },
+    })
+
+    void upload.done().then((res) => {
+      console.log(`Successfully Uploaded File ${res.Key ?? ''}`)
+    })
+
+    return passThroughStream
+  }
+
   private convertFileToGeneric(
-    s3File: GetObjectCommandOutput
-  ): Omit<File, 'bucket' | 'publicUrl'> {
+    s3File: GetObjectCommandOutput,
+    bucket: string,
+    key: string
+  ): File {
     return {
       exists: () => {
         return Promise.resolve(s3File.$metadata.httpStatusCode == 200)
       },
+      save: async () => Promise.resolve(),
       isPublic: async () => Promise.resolve(true),
       download: async () => this.getFileFromReadable(s3File.Body as Readable),
       getMetadataMd5: () => Promise.resolve(s3File.ETag),
+      createWriteStream: (saveOptions: SaveOptions) =>
+        this.createS3UploadStream(bucket, key, saveOptions),
+      publicUrl: () => `${this.urlOverride ?? ''}/${bucket}/${key}`,
+      bucket,
+      key,
     }
   }
 
@@ -53,11 +162,11 @@ export class S3StorageClient implements StorageClient {
       })
     )
 
-    return {
-      ...this.convertFileToGeneric(s3File),
-      bucket: bucket,
-      publicUrl: () => `${this.urlOverride ?? ''}/${bucket}/${filePath}`,
-    }
+    return this.convertFileToGeneric(s3File, bucket, filePath)
+  }
+
+  createFile(bucket: string, filePath: string): File {
+    return new this.BlankFile(this, bucket, filePath) as unknown as File
   }
 
   async getFilesFromPrefix(bucket: string, prefix: string): Promise<File[]> {
@@ -72,8 +181,9 @@ export class S3StorageClient implements StorageClient {
 
     return prefixKeys
       .map(({ Prefix }) => Prefix)
-      .map((key) => {
+      .map((key: string | undefined) => {
         return {
+          key: key || '',
           exists: () => Promise.resolve(true),
           isPublic: async () => Promise.resolve(true),
           download: async () => {
@@ -86,9 +196,12 @@ export class S3StorageClient implements StorageClient {
 
             return this.getFileFromReadable(s3File.Body as Readable)
           },
+          save: () => Promise.resolve(),
+          createWriteStream: (saveOptions: SaveOptions) =>
+            new stream.PassThrough(),
           getMetadataMd5: () => Promise.resolve(key),
           bucket: bucket,
-          publicUrl: () => `${this.urlOverride ?? ''}/${bucket}/${key}`,
+          publicUrl: () => `${this.urlOverride ?? ''}/${bucket}/${key ?? ''}`,
         }
       })
   }
@@ -117,24 +230,23 @@ export class S3StorageClient implements StorageClient {
     return url
   }
 
-  upload(
+  async upload(
     bucket: string,
     filePath: string,
-    data: Buffer,
+    data: SaveData,
     options: {
       contentType?: string
       public?: boolean
       timeout?: number
     }
   ): Promise<void> {
-    return this.s3Client
-      .send(
-        new PutObjectCommand({
-          Bucket: bucket,
-          Key: filePath,
-          Body: data,
-        })
-      )
-      .then(() => {})
+    await this.s3Client.send(
+      new PutObjectCommand({
+        Bucket: bucket,
+        Key: filePath,
+        Body: data.toString(),
+        ContentType: options.contentType,
+      })
+    )
   }
 }

packages/api/src/repository/storage/StorageClient.ts

@@ -1,20 +1,37 @@
+import { PipelineSource, Writable } from 'stream'
+
 export type SignedUrlParameters = {
   action: 'read' | 'write' | 'delete' | 'resumable'
   expires: number
 }
 
+export type SaveData = string | Buffer | PipelineSource<string | Buffer>
+export type SaveOptions = {
+  contentType?: string
+  gzip?: string | boolean
+  resumable?: boolean
+  timeout?: number
+  validation?: string | boolean
+  private?: boolean | undefined
+}
+
 export type File = {
   isPublic: () => Promise<boolean>
   publicUrl: () => string
   download: () => Promise<Buffer>
   exists: () => Promise<boolean>
+  save: (saveData: SaveData, saveOptions: SaveOptions) => Promise<void>,
+  createWriteStream: (saveOptions: SaveOptions) => Writable
   getMetadataMd5: () => Promise<string | undefined>
   bucket: string
+  key: string
 }
 
 export interface StorageClient {
   downloadFile(bucket: string, filePath: string): Promise<File>
 
+  createFile(bucket: string, filePath: string): File
+
   getFilesFromPrefix(bucket: string, filePrefix: string): Promise<File[]>
 
   upload(

packages/discover/src/lib/inputSources/articles/rss/rssConverters/atom.ts

@@ -3,9 +3,7 @@
 /* eslint-disable @typescript-eslint/no-unsafe-return */
 /* eslint-disable @typescript-eslint/restrict-template-expressions */
 
-import { OmnivoreArticle } from '../../../../../types/OmnivoreArticle'
 import { slugify } from 'voca'
-import { Observable, tap } from 'rxjs'
 import { fromArrayLike } from 'rxjs/internal/observable/innerFrom'
 import { mapOrNull } from '../../../../utils/reactive'
 import {