Delete files upon ingestion deletion

packages/backend/src/services/IngestionService.ts

@@ -106,13 +106,30 @@ export class IngestionService {
     }
 
     public static async delete(id: string): Promise<IngestionSource> {
+        const source = await this.findById(id);
+        if (!source) {
+            throw new Error('Ingestion source not found');
+        }
+
+        // Delete all emails and attachments from storage
+        const storage = new StorageService();
+        const emailPath = `open-archiver/${source.name.replaceAll(' ', '-')}-${source.id}/`;
+        await storage.delete(emailPath);
+
+
+        // Delete all emails from the database
+        // NOTE: This is done by database CASADE, change when CASADE relation no longer exists.
+        // await db.delete(archivedEmails).where(eq(archivedEmails.ingestionSourceId, id));
+
+        // Delete all documents from Meilisearch
+        const searchService = new SearchService();
+        await searchService.deleteDocumentsByFilter('emails', `ingestionSourceId = ${id}`);
+
         const [deletedSource] = await db
             .delete(ingestionSources)
             .where(eq(ingestionSources.id, id))
             .returning();
-        if (!deletedSource) {
-            throw new Error('Ingestion source not found');
-        }
+
         return this.decryptSource(deletedSource);
     }
 
@@ -188,7 +205,7 @@ export class IngestionService {
             console.log('processing email, ', email.id, email.subject);
             const emlBuffer = email.eml ?? Buffer.from(email.body, 'utf-8');
             const emailHash = createHash('sha256').update(emlBuffer).digest('hex');
-            const emailPath = `email-archive/${source.name.replaceAll(' ', '-')}-${source.id}/emails/${email.id}.eml`;
+            const emailPath = `open-archiver/${source.name.replaceAll(' ', '-')}-${source.id}/emails/${email.id}.eml`;
             await storage.put(emailPath, emlBuffer);
 
             const [archivedEmail] = await db
@@ -218,7 +235,7 @@ export class IngestionService {
                 for (const attachment of email.attachments) {
                     const attachmentBuffer = attachment.content;
                     const attachmentHash = createHash('sha256').update(attachmentBuffer).digest('hex');
-                    const attachmentPath = `email-archive/${source.name.replaceAll(' ', '-')}-${source.id}/attachments/${attachment.filename}`;
+                    const attachmentPath = `open-archiver/${source.name.replaceAll(' ', '-')}-${source.id}/attachments/${attachment.filename}`;
                     await storage.put(attachmentPath, attachmentBuffer);
 
                     const [newAttachment] = await db

packages/backend/src/services/SearchService.ts

@@ -33,6 +33,11 @@ export class SearchService {
         return index.search(query, options);
     }
 
+    public async deleteDocumentsByFilter(indexName: string, filter: string | string[]) {
+        const index = await this.getIndex(indexName);
+        return index.deleteDocuments({ filter });
+    }
+
     public async searchEmails(dto: SearchQuery): Promise<SearchResult> {
         const { query, filters, page = 1, limit = 10, matchingStrategy = 'last' } = dto;
         const index = await this.getIndex<EmailDocument>('emails');

packages/backend/src/services/storage/LocalFileSystemProvider.ts

@@ -37,8 +37,9 @@ export class LocalFileSystemProvider implements IStorageProvider {
     async delete(filePath: string): Promise<void> {
         const fullPath = path.join(this.rootPath, filePath);
         try {
-            await fs.unlink(fullPath);
+            await fs.rm(fullPath, { recursive: true, force: true });
         } catch (error: any) {
+            // Even with force: true, other errors might occur (e.g., permissions)
             if (error.code !== 'ENOENT') {
                 throw error;
             }

packages/backend/src/services/storage/S3StorageProvider.ts

@@ -5,6 +5,8 @@ import {
     DeleteObjectCommand,
     HeadObjectCommand,
     NotFound,
+    ListObjectsV2Command,
+    DeleteObjectsCommand,
 } from '@aws-sdk/client-s3';
 import { Upload } from '@aws-sdk/lib-storage';
 import { Readable } from 'stream';
@@ -60,11 +62,28 @@ export class S3StorageProvider implements IStorageProvider {
     }
 
     async delete(path: string): Promise<void> {
-        const command = new DeleteObjectCommand({
+        // List all objects with the given prefix
+        const listCommand = new ListObjectsV2Command({
             Bucket: this.bucket,
-            Key: path,
+            Prefix: path,
         });
-        await this.client.send(command);
+        const listedObjects = await this.client.send(listCommand);
+
+        if (!listedObjects.Contents || listedObjects.Contents.length === 0) {
+            return;
+        }
+
+        // Create a list of objects to delete
+        const deleteParams = {
+            Bucket: this.bucket,
+            Delete: {
+                Objects: listedObjects.Contents.map(({ Key }) => ({ Key })),
+            },
+        };
+
+        // Delete the objects
+        const deleteCommand = new DeleteObjectsCommand(deleteParams);
+        await this.client.send(deleteCommand);
     }
 
     async exists(path: string): Promise<boolean> {