fix: improve loadSources error handling (#288)

companion PR: nextcloud/context_chat#228

---------

Signed-off-by: Anupam Kumar <kyteinsky@gmail.com>

Author: kyteinsky

context_chat_backend/controller.py

@@ -343,9 +343,15 @@ def _(sources: list[UploadFile]):
 	if len(sources) == 0:
 		return JSONResponse('No sources provided', 400)
 
+	filtered_sources = []
+
 	for source in sources:
 		if not value_of(source.filename):
-			return JSONResponse(f'Invalid source filename for: {source.headers.get("title")}', 400)
+			logger.warning('Skipping source with invalid source_id', extra={
+				'source_id': source.filename,
+				'title': source.headers.get('title'),
+			})
+			continue
 
 		with index_lock:
 			if source.filename in _indexing:
@@ -364,12 +370,14 @@ def _(sources: list[UploadFile]):
 			and source.headers['modified'].isdigit()
 			and value_of(source.headers.get('provider'))
 		):
-			logger.error('Invalid/missing headers received', extra={
+			logger.warning('Skipping source with invalid/missing headers', extra={
 				'source_id': source.filename,
 				'title': source.headers.get('title'),
 				'headers': source.headers,
 			})
-			return JSONResponse(f'Invaild/missing headers for: {source.filename}', 400)
+			continue
+
+		filtered_sources.append(source)
 
 	# wait for 10 minutes before failing the request
 	semres = doc_parse_semaphore.acquire(block=True, timeout=10*60)
@@ -381,27 +389,27 @@ def _(sources: list[UploadFile]):
 		)
 
 	with index_lock:
-		for source in sources:
+		for source in filtered_sources:
 			_indexing[source.filename] = source.size
 
 	try:
 		loaded_sources, not_added_sources = exec_in_proc(
 			target=embed_sources,
-			args=(vectordb_loader, app.extra['CONFIG'], sources)
+			args=(vectordb_loader, app.extra['CONFIG'], filtered_sources)
 		)
 	except (DbException, EmbeddingException):
 		raise
 	except Exception as e:
 		raise DbException('Error: failed to load sources') from e
 	finally:
 		with index_lock:
-			for source in sources:
+			for source in filtered_sources:
 				_indexing.pop(source.filename, None)
 		doc_parse_semaphore.release()
 
-	if len(loaded_sources) != len(sources):
+	if len(loaded_sources) != len(filtered_sources):
 		logger.debug('Some sources were not loaded', extra={
-			'Count of loaded sources': f'{len(loaded_sources)}/{len(sources)}',
+			'Count of loaded sources': f'{len(loaded_sources)}/{len(filtered_sources)}',
 			'source_ids': loaded_sources,
 		})
 

context_chat_backend/vectordb/pgvector.py

@@ -6,6 +6,7 @@
 import os
 from datetime import datetime
 
+import psycopg
 import sqlalchemy as sa
 import sqlalchemy.dialects.postgresql as postgresql_dialects
 import sqlalchemy.orm as orm
@@ -155,7 +156,18 @@ def add_indocuments(self, indocuments: list[InDocument]) -> tuple[list[str], lis
 						chunks=chunk_ids,
 					)
 					session.add(doc)
-					session.commit()
+					try:
+						session.commit()
+					except sa.exc.IntegrityError as ie:  # pyright: ignore[reportAttributeAccessIssue]
+						if not isinstance(ie.orig, psycopg.errors.UniqueViolation):
+							raise
+
+						# it's already in the db, continue updating the access
+						logger.debug(
+							'Unique violation: document already exists in the database',
+							exc_info=ie,
+							extra={ 'source_id': indoc.source_id },
+						)
 
 					self.decl_update_access(indoc.userIds, indoc.source_id, session)
 					added_sources.append(indoc.source_id)