V0.4.0 fix (#204)

* Jobs page responsive fix

* feat(ingestion): Refactor email indexing into a dedicated background job

This commit refactors the email indexing process to improve the performance and reliability of the ingestion pipeline.

Previously, email indexing was performed synchronously within the mailbox processing job. This could lead to timeouts and failed ingestion cycles if the indexing step was slow or encountered errors.

To address this, the indexing logic has been moved into a separate, dedicated background job queue (`indexingQueue`). Now, the mailbox processor simply adds a batch of emails to this queue. A separate worker then processes the indexing job asynchronously.

This decoupling makes the ingestion process more robust:
- It prevents slow indexing from blocking or failing the entire mailbox sync.
- It allows for better resource management and scalability by handling indexing in a dedicated process.
- It improves error handling, as a failed indexing job can be retried independently without affecting the main ingestion flow.

Additionally, this commit includes minor documentation updates and removes a premature timeout in the PDF text extraction helper that was causing issues.

---------

Co-authored-by: Wayne <5291640+ringoinca@users.noreply.github.com>

Author: wayneshn

README.md

@@ -7,7 +7,7 @@
 [![Redis](https://img.shields.io/badge/Redis-DC382D?style=for-the-badge&logo=redis&logoColor=white)](https://redis.io)
 [![SvelteKit](https://img.shields.io/badge/SvelteKit-FF3E00?style=for-the-badge&logo=svelte&logoColor=white)](https://svelte.dev/)
 
-**A secure, sovereign, and open-source platform for email archiving and eDiscovery.**
+**A secure, sovereign, and open-source platform for email archiving.**
 
 Open Archiver provides a robust, self-hosted solution for archiving, storing, indexing, and searching emails from major platforms, including Google Workspace (Gmail), Microsoft 365, PST files, as well as generic IMAP-enabled email inboxes. Use Open Archiver to keep a permanent, tamper-proof record of your communication history, free from vendor lock-in.
 
@@ -48,13 +48,13 @@ Password: openarchiver_demo
     - Zipped .eml files
     - Mbox files
 
-- **Secure & Efficient Storage**: Emails are stored in the standard `.eml` format. The system uses deduplication and compression to minimize storage costs. All data is encrypted at rest.
+- **Secure & Efficient Storage**: Emails are stored in the standard `.eml` format. The system uses deduplication and compression to minimize storage costs. All files are encrypted at rest.
 - **Pluggable Storage Backends**: Support both local filesystem storage and S3-compatible object storage (like AWS S3 or MinIO).
 - **Powerful Search & eDiscovery**: A high-performance search engine indexes the full text of emails and attachments (PDF, DOCX, etc.).
 - **Thread discovery**: The ability to discover if an email belongs to a thread/conversation and present the context.
 - **Compliance & Retention**: Define granular retention policies to automatically manage the lifecycle of your data. Place legal holds on communications to prevent deletion during litigation (TBD).
 - **File Hash and Encryption**: Email and attachment file hash values are stored in the meta database upon ingestion, meaning any attempt to alter the file content will be identified, ensuring legal and regulatory compliance.
-- **Comprehensive Auditing**: An immutable audit trail logs all system activities, ensuring you have a clear record of who accessed what and when (TBD).
+- **Comprehensive Auditing**: An immutable audit trail logs all system activities, ensuring you have a clear record of who accessed what and when.
 
 ## 🛠️ Tech Stack
 

assets/screenshots/job-queue.png

@@ -1,4 +1,4 @@
-# Email Integrity Check
+# Integrity Check
 
 Open Archiver allows you to verify the integrity of your archived emails and their attachments. This guide explains how the integrity check works and what the results mean.
 

docs/user-guides/integrity-check.md

@@ -47,10 +47,10 @@ function extractTextFromPdf(buffer: Buffer): Promise<string> {
 		}
 
 		// reduced Timeout for better performance
-		setTimeout(() => {
-			logger.warn('PDF parsing timed out');
-			finish('');
-		}, 5000);
+		// setTimeout(() => {
+		// 	logger.warn('PDF parsing timed out');
+		// 	finish('');
+		// }, 5000);
 	});
 }
 

packages/backend/src/helpers/textExtractor.ts

@@ -33,7 +33,6 @@ export const processMailboxProcessor = async (job: Job<IProcessMailboxJob, SyncS
 	const searchService = new SearchService();
 	const storageService = new StorageService();
 	const databaseService = new DatabaseService();
-	const indexingService = new IndexingService(databaseService, searchService, storageService);
 
 	try {
 		const source = await IngestionService.findById(ingestionSourceId);
@@ -72,7 +71,8 @@ export const processMailboxProcessor = async (job: Job<IProcessMailboxJob, SyncS
 		return newSyncState;
 	} catch (error) {
 		if (emailBatch.length > 0) {
-			await indexingService.indexEmailBatch(emailBatch);
+			await indexingQueue.add('index-email-batch', { emails: emailBatch });
+			emailBatch = [];
 		}
 
 		logger.error({ err: error, ingestionSourceId, userEmail }, 'Error processing mailbox');

packages/backend/src/jobs/processors/process-mailbox.processor.ts

@@ -49,9 +49,10 @@ export default async (job: Job<ISyncCycleFinishedJob, any, string>) => {
 		// if data doesn't have error property, it is a successful job with SyncState
 		const successfulJobs = allChildJobs.filter((v) => !v || !(v as any).error) as SyncState[];
 
-		const finalSyncState = deepmerge(
-			...successfulJobs.filter((s) => s && Object.keys(s).length > 0)
-		);
+		const finalSyncState =
+			successfulJobs.length > 0
+				? deepmerge(...successfulJobs.filter((s) => s && Object.keys(s).length > 0))
+				: {};
 
 		const source = await IngestionService.findById(ingestionSourceId);
 		let status: IngestionStatus = 'active';
@@ -63,7 +64,9 @@ export default async (job: Job<ISyncCycleFinishedJob, any, string>) => {
 		let message: string;
 
 		// Check for a specific rate-limit message from the successful jobs
-		const rateLimitMessage = successfulJobs.find((j) => j.statusMessage)?.statusMessage;
+		const rateLimitMessage = successfulJobs.find(
+			(j) => j.statusMessage && j.statusMessage.includes('rate limit')
+		)?.statusMessage;
 
 		if (failedJobs.length > 0) {
 			status = 'error';

packages/backend/src/jobs/processors/sync-cycle-finished.processor.ts

@@ -93,21 +93,17 @@ export class IndexingService {
 				const batch = emails.slice(i, i + CONCURRENCY_LIMIT);
 
 				const batchDocuments = await Promise.allSettled(
-					batch.map(async ({ email, sourceId, archivedId }) => {
+					batch.map(async (pendingEmail) => {
 						try {
-							return await this.createEmailDocumentFromRawForBatch(
-								email,
-								sourceId,
-								archivedId,
-								email.userEmail || ''
-							);
+							const document = await this.indexEmailById(pendingEmail.archivedEmailId);
+							if (document) {
+								return document;
+							}
+							return null;
 						} catch (error) {
 							logger.error(
 								{
-									emailId: archivedId,
-									sourceId,
-									userEmail: email.userEmail || '',
-									rawEmailData: JSON.stringify(email, null, 2),
+									emailId: pendingEmail.archivedEmailId,
 									error: error instanceof Error ? error.message : String(error),
 								},
 								'Failed to create document for email in batch'
@@ -118,10 +114,12 @@ export class IndexingService {
 				);
 
 				for (const result of batchDocuments) {
-					if (result.status === 'fulfilled') {
+					if (result.status === 'fulfilled' && result.value) {
 						rawDocuments.push(result.value);
-					} else {
+					} else if (result.status === 'rejected') {
 						logger.error({ error: result.reason }, 'Failed to process email in batch');
+					} else {
+						logger.error({ result: result }, 'Failed to process email in batch, reason unknown.');
 					}
 				}
 			}
@@ -195,10 +193,7 @@ export class IndexingService {
 		}
 	}
 
-	/**
-	 * @deprecated
-	 */
-	private async indexEmailById(emailId: string): Promise<void> {
+	private async indexEmailById(emailId: string): Promise<EmailDocument | null> {
 		const email = await this.dbService.db.query.archivedEmails.findFirst({
 			where: eq(archivedEmails.id, emailId),
 		});
@@ -228,13 +223,13 @@ export class IndexingService {
 			emailAttachmentsResult,
 			email.userEmail
 		);
-		await this.searchService.addDocuments('emails', [document], 'id');
+		return document;
 	}
 
 	/**
 	 * @deprecated
 	 */
-	private async indexByEmail(pendingEmail: PendingEmail): Promise<void> {
+	/* 	private async indexByEmail(pendingEmail: PendingEmail): Promise<void> {
 		const attachments: AttachmentsType = [];
 		if (pendingEmail.email.attachments && pendingEmail.email.attachments.length > 0) {
 			for (const attachment of pendingEmail.email.attachments) {
@@ -254,12 +249,12 @@ export class IndexingService {
 		);
 		// console.log(document);
 		await this.searchService.addDocuments('emails', [document], 'id');
-	}
+	} */
 
 	/**
 	 * Creates a search document from a raw email object and its attachments.
 	 */
-	private async createEmailDocumentFromRawForBatch(
+	/* private async createEmailDocumentFromRawForBatch(
 		email: EmailObject,
 		ingestionSourceId: string,
 		archivedEmailId: string,
@@ -333,7 +328,7 @@ export class IndexingService {
 			timestamp: new Date(email.receivedAt).getTime(),
 			ingestionSourceId: ingestionSourceId,
 		};
-	}
+	} */
 
 	private async createEmailDocumentFromRaw(
 		email: EmailObject,

packages/backend/src/services/IndexingService.ts

@@ -186,7 +186,7 @@ export class IngestionService {
 				(key) =>
 					key !== 'providerConfig' &&
 					originalSource[key as keyof IngestionSource] !==
-						decryptedSource[key as keyof IngestionSource]
+					decryptedSource[key as keyof IngestionSource]
 			);
 			if (changedFields.length > 0) {
 				await this.auditService.createAuditLog({
@@ -518,12 +518,8 @@ export class IngestionService {
 				}
 			}
 
-			email.userEmail = userEmail;
-
 			return {
-				email,
-				sourceId: source.id,
-				archivedId: archivedEmail.id,
+				archivedEmailId: archivedEmail.id,
 			};
 		} catch (error) {
 			logger.error({

packages/backend/src/services/IngestionService.ts

@@ -81,6 +81,79 @@ export class StorageService implements IStorageProvider {
 		return Readable.from(decryptedContent);
 	}
 
+	public async getStream(path: string): Promise<NodeJS.ReadableStream> {
+		const stream = await this.provider.get(path);
+		if (!this.encryptionKey) {
+			return stream;
+		}
+
+		// For encrypted files, we need to read the prefix and IV first.
+		// This part still buffers a small, fixed amount of data, which is acceptable.
+		const prefixAndIvBuffer = await new Promise<Buffer>((resolve, reject) => {
+			const chunks: Buffer[] = [];
+			let totalLength = 0;
+			const targetLength = ENCRYPTION_PREFIX.length + 16;
+
+			const onData = (chunk: Buffer) => {
+				chunks.push(chunk);
+				totalLength += chunk.length;
+				if (totalLength >= targetLength) {
+					stream.removeListener('data', onData);
+					resolve(Buffer.concat(chunks));
+				}
+			};
+
+			stream.on('data', onData);
+			stream.on('error', reject);
+			stream.on('end', () => {
+				// Handle cases where the file is smaller than the prefix + IV
+				if (totalLength < targetLength) {
+					resolve(Buffer.concat(chunks));
+				}
+			});
+		});
+
+		const prefix = prefixAndIvBuffer.subarray(0, ENCRYPTION_PREFIX.length);
+		if (!prefix.equals(ENCRYPTION_PREFIX)) {
+			// File is not encrypted, return a new stream containing the buffered prefix and the rest of the original stream
+			const combinedStream = new Readable({
+				read() { },
+			});
+			combinedStream.push(prefixAndIvBuffer);
+			stream.on('data', (chunk) => {
+				combinedStream.push(chunk);
+			});
+			stream.on('end', () => {
+				combinedStream.push(null); // No more data
+			});
+			stream.on('error', (err) => {
+				combinedStream.emit('error', err);
+			});
+			return combinedStream;
+		}
+
+		try {
+			const iv = prefixAndIvBuffer.subarray(
+				ENCRYPTION_PREFIX.length,
+				ENCRYPTION_PREFIX.length + 16
+			);
+			const decipher = createDecipheriv(this.algorithm, this.encryptionKey, iv);
+
+			// Push the remaining part of the initial buffer to the decipher
+			const remainingBuffer = prefixAndIvBuffer.subarray(ENCRYPTION_PREFIX.length + 16);
+			if (remainingBuffer.length > 0) {
+				decipher.write(remainingBuffer);
+			}
+
+			// Pipe the rest of the stream
+			stream.pipe(decipher);
+
+			return decipher;
+		} catch (error) {
+			throw new Error('Failed to decrypt file. It may be corrupted or the key is incorrect.');
+		}
+	}
+
 	delete(path: string): Promise<void> {
 		return this.provider.delete(path);
 	}

packages/backend/src/services/StorageService.ts

@@ -131,6 +131,7 @@ export class ImapConnector implements IEmailConnector {
 			} catch (err: any) {
 				logger.error({ err, attempt }, `IMAP operation failed on attempt ${attempt}`);
 				this.isConnected = false; // Force reconnect on next attempt
+				this.client = this.createClient(); // Create a new client instance for the next retry
 				if (attempt === maxRetries) {
 					logger.error({ err }, 'IMAP operation failed after all retries.');
 					throw err;