diff --git a/application/single_app/background_tasks.py b/application/single_app/background_tasks.py index bc4c64c57..09309cd0f 100644 --- a/application/single_app/background_tasks.py +++ b/application/single_app/background_tasks.py @@ -36,6 +36,7 @@ ) from functions_debug import debug_print from functions_data_management import check_due_data_management_jobs_once +from functions_documents import process_due_document_search_metadata_syncs from functions_file_sync import check_due_file_sync_sources_once from functions_keyvault_reminders import ( KEY_VAULT_SECRET_REMINDER_LOCK_NAME, @@ -981,6 +982,36 @@ def run_data_management_scheduler_loop(app=None): time.sleep(60) +def run_document_search_metadata_sync_loop(): + """Retry search chunk metadata syncs that failed or never ran in their original process.""" + while True: + lock_document = None + try: + lock_document = acquire_distributed_task_lock('document_search_metadata_sync_scan', lease_seconds=600) + if lock_document: + summary = process_due_document_search_metadata_syncs() + if summary.get('due'): + log_event( + '[DOCUMENT_SEARCH_SYNC] Reconciled due search metadata syncs.', + extra=summary, + level=logging.INFO, + ) + else: + debug_print('Skipping search metadata sync reconciliation because another worker holds the lease.') + except Exception as exc: + log_event( + '[DOCUMENT_SEARCH_SYNC] Search metadata sync reconciliation failed.', + extra={'error_type': type(exc).__name__}, + level=logging.ERROR, + exceptionTraceback=True, + ) + finally: + if lock_document: + release_distributed_task_lock(lock_document) + + time.sleep(60) + + def run_content_screening_scheduler_loop(app=None): """Recover completion work even when enrollment of new scans is disabled.""" while True: @@ -1086,6 +1117,7 @@ def start_background_task_threads(app=None): ('Orchestration scheduler background task started.', lambda: run_orchestration_scheduler_loop(app=app)), ('Workflow chat delivery background task started.', lambda: run_workflow_chat_delivery_loop(app=app)), ('Data Management scheduler background task started.', lambda: run_data_management_scheduler_loop(app=app)), + ('Document search metadata sync background task started.', run_document_search_metadata_sync_loop), ('Content screening scheduler background task started.', lambda: run_content_screening_scheduler_loop(app=app)), ('App maintenance background task started.', run_app_maintenance_loop), ('Key Vault secret reminder background task started.', run_key_vault_secret_reminder_loop), diff --git a/application/single_app/config.py b/application/single_app/config.py index f5c1fe6f4..175ad4db0 100644 --- a/application/single_app/config.py +++ b/application/single_app/config.py @@ -101,7 +101,7 @@ EXECUTOR_TYPE = 'thread' EXECUTOR_MAX_WORKERS = 30 SESSION_TYPE = 'filesystem' -VERSION = "0.261.267" +VERSION = "0.261.268" IS_DEVELOPMENT = is_development_env_enabled() # Opt-out for deployments where App Service Easy Auth is active but the platform diff --git a/application/single_app/functions_documents.py b/application/single_app/functions_documents.py index 06ea28076..8c7238129 100644 --- a/application/single_app/functions_documents.py +++ b/application/single_app/functions_documents.py @@ -8,6 +8,7 @@ import zipfile import hashlib from functools import partial +from concurrent.futures import ThreadPoolExecutor from contextlib import nullcontext from io import BytesIO from flask import make_response @@ -75,10 +76,16 @@ hold_data_management_search_write_slot, ) from functions_group_document_projection_fence import ( + GROUP_DOCUMENT_PROJECTION_WRITER, GroupDocumentProjectionConflict, assert_group_document_source_writable, hold_group_document_projection, ) +from utils_cache import ( + invalidate_group_search_cache, + invalidate_personal_search_cache, + invalidate_public_workspace_search_cache, +) from functions_visio import build_visio_page_markdown, parse_vsdx_pages from functions_onenote import ONENOTE_MAX_CHUNKS, OneNoteExtractionError, extract_onenote from functions_content import * @@ -127,12 +134,22 @@ class DocumentSearchAclProjectionDeferredError(RuntimeError): """Raised when an authorization-reducing Search ACL update must be retried safely.""" +class DocumentSearchAclProjectionError(RuntimeError): + """Raised when a Search ACL update failed, so the access change was not saved.""" + + +class DocumentSearchSyncLeaseLostError(RuntimeError): + """Raised when another worker took over a document's search metadata sync lease.""" + + class DocumentMutationPropagationError(RuntimeError): """The source changed, but a required downstream projection needs repair.""" - def __init__(self, message, *, deleted_document_ids=()): + def __init__(self, message, *, deleted_document_ids=(), search_writes_frozen=False): super().__init__(message) self.deleted_document_ids = list(deleted_document_ids) + # True when the projection waits only on a Data Management migration's Search write freeze. + self.search_writes_frozen = bool(search_writes_frozen) class DocumentRevisionDeleteError(RuntimeError): @@ -3542,6 +3559,15 @@ def calculate_processing_percentage(doc_metadata): return max(final_pct, current_pct) def update_document(**kwargs): + """Update a document and its derived search state. + + Metadata mirrored on search chunks (title, authors, file name, classification, tags) is projected + by a durable background sync: the request is recorded before the save, the save carries the + request's token, and the save itself does no per-chunk work. Access-control changes are projected + to the chunks synchronously and fail closed. Non-strict calls return + {'updated': bool, 'search_sync': {...}}; strict calls return the saved document, or + (saved document, search_sync) when return_search_sync is set. + """ # Screening state and every screening_* field, including the generated-content # exemption, are server-managed and set only when a version is created. if SCREENING_FIELD in kwargs or any(str(name).startswith("screening_") for name in kwargs): @@ -3557,6 +3583,7 @@ def update_document(**kwargs): expected_etag = kwargs.pop('expected_etag', None) operation_guard = kwargs.pop('operation_guard', None) num_chunks_increment = kwargs.pop('num_chunks_increment', 0) + return_search_sync = kwargs.pop('return_search_sync', False) if operation_guard is not None: operation_guard() @@ -3672,10 +3699,14 @@ def update_document(**kwargs): # start a new hold. Only the inspected source format of the file stays fixed. validate_screened_metadata_update(existing_document, kwargs) original_percentage = existing_document.get('percentage_complete', 0) # Store for comparison + # update_document only replaces top-level fields, so a shallow copy keeps the stored state. + original_document = dict(existing_document) # 2. Apply updates from kwargs update_occurred = False - updated_fields_requiring_chunk_sync = set() # Track fields needing propagation + metadata_fields_requiring_sync = set() + acl_changes = {} + document_search_acl_fields = get_document_search_acl_fields(group_id, public_workspace_id) if num_chunks_increment > 0: current_num_chunks = existing_document.get('num_chunks', 0) @@ -3690,17 +3721,22 @@ def update_document(**kwargs): for key, value in kwargs.items(): if strict and key in {'document_id', 'user_id', 'group_id', 'public_workspace_id'}: continue - if value is not None and existing_document.get(key) != value: + previous_value = existing_document.get(key) + if value is not None and previous_value != value: # Avoid overwriting num_chunks if it was just incremented if key == 'num_chunks' and num_chunks_increment > 0: continue # Skip direct assignment if increment was used existing_document[key] = value update_occurred = True - if key in ['title', 'authors', 'file_name', 'document_classification', 'tags']: - updated_fields_requiring_chunk_sync.add(key) - # Propagate shared_group_ids to group chunks if changed - if is_group and key == 'shared_group_ids': - updated_fields_requiring_chunk_sync.add('shared_group_ids') + if key in DOCUMENT_SEARCH_METADATA_FIELD_MAP: + metadata_fields_requiring_sync.add(key) + # Project only real access changes; a missing list and an empty list grant the same + # access, so legacy normalization never waits on (or fails because of) Search. + if ( + key in document_search_acl_fields + and _normalize_search_acl_entries(previous_value) != _normalize_search_acl_entries(value) + ): + acl_changes[key] = (previous_value, value) # 3. If any update happened, handle timestamps and percentage if update_occurred: @@ -3733,107 +3769,98 @@ def update_document(**kwargs): existing_document['percentage_complete'] = new_percentage if strict: - if operation_guard is not None: - operation_guard() - saved_document = _upsert_document_and_sync_access_index( - cosmos_container, existing_document, operation='document_updated', strict=True, - ) - # Retry the requested projections even when a previous attempt - # committed the source but failed downstream. - updated_fields_requiring_chunk_sync.update( - field for field in ("title", "authors", "file_name", "document_classification", "tags", "shared_group_ids") + # Retry the requested projections even when a previous attempt committed the source + # but did not finish downstream: a strict caller's projected fields are re-requested. + metadata_fields_requiring_sync.update( + field for field in DOCUMENT_SEARCH_METADATA_FIELD_MAP if field in kwargs and kwargs[field] is not None ) - # 4. Propagate relevant changes to search index chunks - # This happens regardless of 'update_occurred' flag because the *intent* from kwargs might trigger it, - # even if the main doc update didn't happen (e.g., only percentage changed). - # However, it's better to only do this if the relevant fields *actually* changed. - # A held screened document has no released chunks; publication rebuilds them. - if ( - (update_occurred or strict) and updated_fields_requiring_chunk_sync - and (marker is None or document_is_available(existing_document)) - ): - try: - chunks_to_update = get_all_chunks( - document_id, - user_id, - group_id=group_id, - public_workspace_id=public_workspace_id - ) - for chunk in chunks_to_update: - if operation_guard is not None: - operation_guard() - if strict: - current = cosmos_container.read_item(item=document_id, partition_key=document_id) - if current.get("_etag") != saved_document.get("_etag"): - raise DocumentMutationPropagationError("A newer document update must be projected before retrying.") - chunk_updates = {} - if 'title' in updated_fields_requiring_chunk_sync: - chunk_updates['title'] = existing_document.get('title') - if 'authors' in updated_fields_requiring_chunk_sync: - # Ensure authors is a list for the chunk metadata if needed - chunk_updates['author'] = ensure_list(existing_document.get('authors')) - if 'file_name' in updated_fields_requiring_chunk_sync: - chunk_updates['file_name'] = existing_document.get('file_name') - if 'document_classification' in updated_fields_requiring_chunk_sync: - chunk_updates['document_classification'] = existing_document.get('document_classification') - if 'tags' in updated_fields_requiring_chunk_sync: - chunk_updates['document_tags'] = existing_document.get('tags', []) - - if chunk_updates: # Only call update if there's something to change - # Build the call parameters - update_params = { - 'chunk_id': chunk['id'], - 'user_id': user_id, - 'document_id': document_id, - 'group_id': group_id, - 'public_workspace_id': public_workspace_id, - **chunk_updates - } + # A held screened document has no released chunks; publication rebuilds them from the document. + releases_chunks = marker is None or document_is_available(existing_document) - # Only include shared_group_ids for group workspaces - if is_group and 'shared_group_ids' in updated_fields_requiring_chunk_sync: - update_params['shared_group_ids'] = existing_document.get('shared_group_ids') - - update_chunk_metadata(**update_params) - add_file_task_to_file_processing_log( - document_id=document_id, - user_id=public_workspace_id if is_public_workspace else (group_id if is_group else user_id), - content=f"Propagated updates for fields {updated_fields_requiring_chunk_sync} to search chunks." - ) - except Exception as chunk_sync_error: - if strict: - raise DocumentMutationPropagationError("Document saved, but chunk metadata propagation failed.") from chunk_sync_error - # Log error but don't necessarily fail the whole document update - error_msg = f"Warning: Failed to sync metadata updates to search chunks for doc {document_id}: {chunk_sync_error}" - print(error_msg) - add_file_task_to_file_processing_log( - document_id=document_id, - user_id=public_workspace_id if is_public_workspace else (group_id if is_group else user_id), - content=error_msg - ) + # 4. Project access-control changes that must reach the search chunks before they are saved, + # so the index never grants access that Cosmos does not record and a revocation is enforced first. + acl_changes_after_save = {} + if acl_changes: + if operation_guard is not None: + operation_guard() + acl_changes_after_save = _project_document_acl_before_save( + cosmos_container, + existing_document, + original_document, + acl_changes, + user_id, + group_id=group_id, + releases_chunks=releases_chunks, + ) + add_file_task_to_file_processing_log( + document_id=document_id, + user_id=public_workspace_id if is_public_workspace else (group_id if is_group else user_id), + content=f"Projected access changes for fields {sorted(acl_changes)} to search chunks before saving." + ) + # 5. Durably request the chunk metadata projection alongside this save. A background worker + # merges it into the search chunks, so the save never waits on the document's chunk count. + search_sync = {'status': DOCUMENT_SEARCH_SYNC_STATUS_NOT_REQUIRED} + if metadata_fields_requiring_sync and releases_chunks: + sync_record = request_document_search_metadata_sync( + document_id, + metadata_fields_requiring_sync, + user_id, + group_id=group_id, + public_workspace_id=public_workspace_id, + ) or {} + existing_document[DOCUMENT_SEARCH_SYNC_TOKEN_FIELD] = sync_record.get('latest_request_token') + search_sync = { + 'status': DOCUMENT_SEARCH_SYNC_STATUS_PENDING, + 'revision': int(sync_record.get('revision') or 0), + 'fields': sorted(metadata_fields_requiring_sync), + } - # 5. Upsert the document if changes were made - if update_occurred and not strict: + # 6. Save the document. The sync request's token rides in this same save. + saved_document = None + if strict: + if operation_guard is not None: + operation_guard() + saved_document = _upsert_document_and_sync_access_index( + cosmos_container, existing_document, operation='document_updated', strict=True, + ) + elif update_occurred: if operation_guard is not None: operation_guard() _upsert_document_and_sync_access_index( cosmos_container, existing_document, operation='document_updated', ) + + if search_sync['status'] == DOCUMENT_SEARCH_SYNC_STATUS_PENDING: + schedule_document_search_metadata_sync( + document_id, + group_id=group_id, + public_workspace_id=public_workspace_id, + ) + add_file_task_to_file_processing_log( + document_id=document_id, + user_id=public_workspace_id if is_public_workspace else (group_id if is_group else user_id), + content=f"Queued search chunk sync for fields {search_sync['fields']} at revision {search_sync['revision']}." + ) + + # 7. The group projection fence admits new access entries only once Cosmos records them. + if acl_changes_after_save: + _project_document_acl_after_save( + document_id, + acl_changes_after_save, + user_id, + group_id=group_id, + document_version=existing_document.get('version'), + releases_chunks=releases_chunks, + ) + if strict: - if 'tags' in kwargs: - if operation_guard is not None: - operation_guard() - current = cosmos_container.read_item(item=document_id, partition_key=document_id) - if current.get("_etag") != saved_document.get("_etag"): - raise DocumentMutationPropagationError("A newer document update must be projected before retrying.") - propagate_tags_to_blob_metadata( - document_id, kwargs['tags'], user_id, group_id, public_workspace_id, - strict=True, expected_etag=saved_document.get("_etag"), - ) + if return_search_sync: + return saved_document, search_sync return saved_document + return {'updated': update_occurred, 'search_sync': search_sync} except CosmosResourceNotFoundError as e: # Error already logged where it was first detected @@ -4573,6 +4600,1062 @@ def update_chunk_metadata(chunk_id, user_id, group_id=None, public_workspace_id= raise +# Search chunk metadata and access-control synchronization. +# +# Document metadata (title, authors, file name, classification, tags) is mirrored onto every AI +# Search chunk. Edits are projected by a background worker so a save never waits on a document's +# chunk count: update_document() records a durable sync request in the settings container before +# it saves the document, and the worker merges only the changed fields into each chunk in batches. +# Access-control lists are different: they are projected synchronously, so the index never grants +# access that Cosmos does not record. +DOCUMENT_SEARCH_SYNC_RECORD_TYPE = 'document_search_metadata_sync' +DOCUMENT_SEARCH_SYNC_SCHEMA_VERSION = 1 +# Each sync request gets a unique token, and the document save that goes with it stores the same +# token. A worker treats the save as landed only when the tokens match, so a value left on the +# document by an earlier request can never be mistaken for a newer one. +DOCUMENT_SEARCH_SYNC_TOKEN_FIELD = 'search_metadata_sync_token' +DOCUMENT_SEARCH_SYNC_STATUS_PENDING = 'pending' +DOCUMENT_SEARCH_SYNC_STATUS_SYNCING = 'syncing' +DOCUMENT_SEARCH_SYNC_STATUS_FAILED = 'failed' +DOCUMENT_SEARCH_SYNC_STATUS_NOT_REQUIRED = 'not_required' +DOCUMENT_SEARCH_SYNC_BATCH_SIZE = 500 +DOCUMENT_SEARCH_SYNC_MAX_BATCH_SIZE = 1000 +DOCUMENT_SEARCH_SYNC_MAX_BATCH_BYTES = 8 * 1024 * 1024 +DOCUMENT_SEARCH_SYNC_LEASE_SECONDS = 600 +DOCUMENT_SEARCH_SYNC_LEASE_WAIT_SECONDS = 20 +DOCUMENT_SEARCH_SYNC_LEASE_POLL_SECONDS = 2 +DOCUMENT_SEARCH_SYNC_PENDING_GRACE_SECONDS = 120 +DOCUMENT_SEARCH_SYNC_COMMIT_RECHECK_SECONDS = 60 +DOCUMENT_SEARCH_SYNC_UNCOMMITTED_INTENT_SECONDS = 600 +DOCUMENT_SEARCH_SYNC_MAX_PASSES = 5 +DOCUMENT_SEARCH_SYNC_RECORD_WRITE_ATTEMPTS = 8 +DOCUMENT_SEARCH_SYNC_RETRY_DELAYS_SECONDS = (60, 300, 900, 3600, 21600) +# Search writes frozen by a Data Management migration are not failures, so they retry on a short +# fixed delay instead of climbing the backoff ladder. +DOCUMENT_SEARCH_SYNC_FROZEN_RETRY_SECONDS = 300 +DOCUMENT_SEARCH_SYNC_WORKER_COUNT = 2 +DOCUMENT_SEARCH_SYNC_RECONCILE_BATCH_SIZE = 25 +DOCUMENT_SEARCH_SYNC_RECONCILE_TIME_BUDGET_SECONDS = 240 +# Cosmos document field -> AI Search chunk field for metadata the background sync projects. +DOCUMENT_SEARCH_METADATA_FIELD_MAP = { + 'title': 'title', + 'authors': 'author', + 'file_name': 'file_name', + 'document_classification': 'document_classification', + 'tags': 'document_tags', +} +# Access lists a projection may carry. A projection never rewrites a key or scope field. +DOCUMENT_SEARCH_ACL_INDEX_FIELDS = frozenset({'shared_user_ids', 'shared_group_ids'}) +# Cosmos system properties, and the claim the projection fence adds and removes around each fenced +# Search write. Neither is a change to the document. +DOCUMENT_SEARCH_ACL_REBASE_IGNORED_FIELDS = frozenset({ + '_etag', '_ts', '_rid', '_self', '_attachments', GROUP_DOCUMENT_PROJECTION_WRITER, +}) +DOCUMENT_SEARCH_ACL_RETRY_AFTER_SECONDS = 150 +DOCUMENT_SEARCH_ACL_DEFERRED_MESSAGE = ( + "Document access was not changed because target Search writes are temporarily frozen. " + "Retry after the migration finishes." +) +DOCUMENT_SEARCH_ACL_FAILED_MESSAGE = ( + "Document access was not changed because the search index could not be updated. Try again." +) +DOCUMENT_SEARCH_ACL_INCOMPLETE_MESSAGE = ( + "The access change was saved, but the search index has not applied it yet. Try again to finish the update." +) +DOCUMENT_SEARCH_ACL_INCOMPLETE_DEFERRED_MESSAGE = ( + "The access change was saved, but target Search writes are temporarily frozen. " + "Retry after the migration finishes to complete it." +) +DOCUMENT_SEARCH_ACL_CONFLICT_MESSAGE = ( + "The document changed while its access was being updated. Refresh and try again." +) + +_document_search_sync_executor = None +_document_search_sync_lock = threading.Lock() +_document_search_sync_queued_records = set() + + +def _escape_search_filter_literal(value): + """Escape a value for an OData single-quoted string literal.""" + return str(value or '').replace("'", "''") + + +def _resolve_document_chunk_scope(user_id, group_id=None, public_workspace_id=None): + """Return the chunk field and value that scope a document's active search chunks.""" + if public_workspace_id is not None: + return 'public_workspace_id', public_workspace_id + if group_id is not None: + return 'group_id', group_id + return 'user_id', user_id + + +def get_document_search_acl_fields(group_id=None, public_workspace_id=None): + """Return the document fields that control search access for a workspace scope.""" + if public_workspace_id is not None: + return () + if group_id is not None: + return ('shared_group_ids',) + return ('shared_user_ids',) + + +def _normalize_search_acl_entries(value): + """Return access entries in a comparable form; a missing list and an empty list are the same.""" + if value is None: + return [] + entries = value if isinstance(value, (list, tuple, set)) else [value] + return sorted({str(entry).strip() for entry in entries if str(entry or '').strip()}) + + +def _search_acl_grant_entries(value): + """Return the access entries that let a principal find the document in search. + + Search matches only approved entries. An entry without a status is treated as approved, so + removing one is never mistaken for a change that revokes nothing. + """ + return { + entry for entry in _normalize_search_acl_entries(value) + if ',' not in entry or entry.endswith(',approved') + } + + +def iter_document_chunk_key_pages( + document_id, + user_id, + group_id=None, + public_workspace_id=None, + page_size=None, +): + """Yield ordered pages of the active search chunk keys for one document. + + Pages by key instead of $skip, so a document's chunk count is never capped and merges that run + between pages cannot move page boundaries. Only the key is selected, so chunk text and + embeddings are never downloaded. Archived revision chunks carry an archived scope value and are + excluded on purpose. + """ + scope_field, scope_value = _resolve_document_chunk_scope(user_id, group_id, public_workspace_id) + if not document_id or not scope_value: + raise ValueError("A document id and workspace scope are required to list document chunks.") + + if page_size is None: + page_size = DOCUMENT_SEARCH_SYNC_BATCH_SIZE + page_size = max(1, min(int(page_size), DOCUMENT_SEARCH_SYNC_MAX_BATCH_SIZE)) + search_client = _get_search_client(group_id=group_id, public_workspace_id=public_workspace_id) + base_filter = ( + f"document_id eq '{_escape_search_filter_literal(document_id)}' " + f"and {scope_field} eq '{_escape_search_filter_literal(scope_value)}'" + ) + last_key = None + while True: + page_filter = base_filter + if last_key is not None: + page_filter = f"{base_filter} and id gt '{_escape_search_filter_literal(last_key)}'" + results = search_client.search( + search_text='*', + filter=page_filter, + select=['id'], + order_by=['id asc'], + top=page_size, + ) + keys = [str(result.get('id')) for result in results if result.get('id')] + if not keys: + return + if last_key is not None and keys[0] <= last_key: + raise RuntimeError("Search chunk key paging did not advance.") + yield keys + if len(keys) < page_size: + return + last_key = keys[-1] + + +def project_fields_to_document_chunks( + document_id, + field_values, + user_id, + group_id=None, + public_workspace_id=None, + batch_size=None, + before_batch=None, + document_version=None, +): + """Merge index fields into every active chunk of one document and return the chunk count. + + Each merge sends only the chunk key and the supplied fields, so embeddings and unrelated chunk + fields are never read or rewritten, and concurrent projections of other fields cannot be undone. + A projection carries either mirrored metadata or the workspace's access list, never both. A group + access list goes through the document's projection fence, which admits only entries the source + document records. Metadata merges carry no access or scope field, so they skip the fence and + never write to the document. Every merge holds the Data Management Search write slot. Raises + when any batch is not fully acknowledged. before_batch runs ahead of each merge, which lets a + caller confirm it still owns the work. + """ + if not field_values: + return 0 + + field_names = set(field_values) + if field_names & DOCUMENT_SEARCH_ACL_INDEX_FIELDS: + if not field_names <= set(get_document_search_acl_fields(group_id, public_workspace_id)): + raise ValueError("An access projection may carry only the workspace's access list.") + elif not field_names <= set(DOCUMENT_SEARCH_METADATA_FIELD_MAP.values()): + raise ValueError("A metadata projection may carry only mirrored document metadata.") + fenced_group_id = group_id if 'shared_group_ids' in field_names else None + + if batch_size is None: + batch_size = DOCUMENT_SEARCH_SYNC_BATCH_SIZE + action_bytes = len(json.dumps(field_values, default=str).encode('utf-8')) + len(str(document_id)) + 64 + effective_batch_size = max( + 1, + min(int(batch_size), DOCUMENT_SEARCH_SYNC_MAX_BATCH_SIZE, DOCUMENT_SEARCH_SYNC_MAX_BATCH_BYTES // action_bytes), + ) + search_client = _get_search_client(group_id=group_id, public_workspace_id=public_workspace_id) + chunks_updated = 0 + for chunk_keys in iter_document_chunk_key_pages( + document_id, + user_id, + group_id=group_id, + public_workspace_id=public_workspace_id, + page_size=effective_batch_size, + ): + if before_batch is not None: + before_batch() + _execute_document_search_write( + search_client, + "merge_documents", + documents=[{'id': chunk_key, **field_values} for chunk_key in chunk_keys], + group_id=fenced_group_id, + document_id=document_id, + document_version=document_version, + ) + chunks_updated += len(chunk_keys) + return chunks_updated + + +def project_document_acl_to_chunks( + document_id, + acl_fields, + user_id, + group_id=None, + public_workspace_id=None, + document_version=None, +): + """Project access-control fields to every active chunk of a document. + + Raises DocumentSearchAclProjectionDeferredError while target Search writes are frozen and + DocumentSearchAclProjectionError for any other failure. Both carry fixed messages that are safe + to show, so a caller never reports an access change that the search index does not enforce. + """ + if not acl_fields: + return 0 + try: + return project_fields_to_document_chunks( + document_id, + acl_fields, + user_id, + group_id=group_id, + public_workspace_id=public_workspace_id, + document_version=document_version, + ) + except DataManagementSearchWritesFrozenError as exc: + raise DocumentSearchAclProjectionDeferredError(DOCUMENT_SEARCH_ACL_DEFERRED_MESSAGE) from exc + except Exception as exc: + log_event( + "[DOCUMENT_SEARCH_SYNC] Search access projection failed.", + extra={ + 'document_id': document_id, + 'fields': sorted(acl_fields), + 'group_id': group_id, + 'public_workspace_id': public_workspace_id, + 'error_type': type(exc).__name__, + }, + level=logging.ERROR, + ) + raise DocumentSearchAclProjectionError(DOCUMENT_SEARCH_ACL_FAILED_MESSAGE) from exc + + +def _document_search_acl_incomplete_error(error): + """Return the error for a saved access change whose search projection has not finished.""" + search_writes_frozen = isinstance(error, DocumentSearchAclProjectionDeferredError) + return DocumentMutationPropagationError( + DOCUMENT_SEARCH_ACL_INCOMPLETE_DEFERRED_MESSAGE if search_writes_frozen else DOCUMENT_SEARCH_ACL_INCOMPLETE_MESSAGE, + search_writes_frozen=search_writes_frozen, + ) + + +def describe_document_search_acl_error(error): + """Return (message, status_code, retry_after_seconds) for an access change Search could not apply. + + Returns None for any other error. Every message is fixed, so no exception text reaches a client. + """ + if isinstance(error, DocumentSearchAclProjectionDeferredError): + return DOCUMENT_SEARCH_ACL_DEFERRED_MESSAGE, 503, DOCUMENT_SEARCH_ACL_RETRY_AFTER_SECONDS + if isinstance(error, DocumentSearchAclProjectionError): + return DOCUMENT_SEARCH_ACL_FAILED_MESSAGE, 500, None + if isinstance(error, DocumentMutationPropagationError): + if error.search_writes_frozen: + return DOCUMENT_SEARCH_ACL_INCOMPLETE_DEFERRED_MESSAGE, 503, DOCUMENT_SEARCH_ACL_RETRY_AFTER_SECONDS + return DOCUMENT_SEARCH_ACL_INCOMPLETE_MESSAGE, 500, None + if isinstance(error, ScreeningConflictError): + return DOCUMENT_SEARCH_ACL_CONFLICT_MESSAGE, 409, None + return None + + +def _document_search_acl_comparable_content(document_item): + """Return a document without Cosmos system properties or the projection fence's claim.""" + return { + key: value for key, value in (document_item or {}).items() + if key not in DOCUMENT_SEARCH_ACL_REBASE_IGNORED_FIELDS + } + + +def _rebase_document_after_fenced_acl_projection( + cosmos_container, + document_item, + original_document, + user_id, + group_id, +): + """Adopt the etag a fenced access projection left on a document, or refuse a concurrent change. + + The group projection fence claims and releases a marker on the source document, which changes + its etag, so the save that follows has to read the document again. When anything other than + Cosmos system properties and the fence's claim changed, another writer won: Search is projected + back to the access list Cosmos now records, and the update is refused with a conflict. + """ + document_id = document_item['id'] + current_document = cosmos_container.read_item(item=document_id, partition_key=document_id) + if ( + _document_search_acl_comparable_content(current_document) + == _document_search_acl_comparable_content(original_document) + ): + document_item['_etag'] = current_document.get('_etag') + return current_document + + try: + project_document_acl_to_chunks( + document_id, + {'shared_group_ids': list(current_document.get('shared_group_ids') or [])}, + user_id, + group_id=group_id, + document_version=current_document.get('version'), + ) + except (DocumentSearchAclProjectionDeferredError, DocumentSearchAclProjectionError): + log_event( + "[DOCUMENT_SEARCH_SYNC] Search access could not be restored after a concurrent document change.", + extra={'document_id': document_id, 'group_id': group_id}, + level=logging.WARNING, + ) + raise ScreeningConflictError() + + +def _project_document_acl_before_save( + cosmos_container, + document_item, + original_document, + acl_changes, + user_id, + group_id=None, + releases_chunks=True, +): + """Project the part of an access change that has to reach Search before Cosmos records it. + + Search may only ever hold a subset of what Cosmos records, so removing search access is enforced + first and fails closed. A personal list has no projection fence, so its new value is projected + now. The group projection fence admits only entries the document already records, so a group + list projects its reduced form (the new list without anything it adds) now and the rest after + the save. Grants reach only released documents; a held document's chunks are rebuilt from the + document when it is published. Returns the access changes to project after the save. + """ + after_save = {} + for field_name, (previous_value, new_value) in acl_changes.items(): + new_entries = list(new_value or []) + previous_entries = set(_normalize_search_acl_entries(previous_value)) + reduced_entries = [entry for entry in new_entries if str(entry).strip() in previous_entries] + revokes_access = bool(_search_acl_grant_entries(previous_value) - _search_acl_grant_entries(new_entries)) + if field_name == 'shared_group_ids': + if revokes_access: + projected_chunks = project_document_acl_to_chunks( + document_item['id'], + {field_name: reduced_entries}, + user_id, + group_id=group_id, + document_version=document_item.get('version'), + ) + if projected_chunks: + _rebase_document_after_fenced_acl_projection( + cosmos_container, document_item, original_document, user_id, group_id, + ) + if not revokes_access or ( + _normalize_search_acl_entries(reduced_entries) != _normalize_search_acl_entries(new_entries) + ): + after_save[field_name] = (previous_value, new_entries) + elif releases_chunks: + project_document_acl_to_chunks(document_item['id'], {field_name: new_entries}, user_id) + elif revokes_access: + project_document_acl_to_chunks(document_item['id'], {field_name: reduced_entries}, user_id) + return after_save + + +def _project_document_acl_after_save( + document_id, + acl_changes, + user_id, + group_id=None, + document_version=None, + releases_chunks=True, +): + """Project the rest of a group access change once Cosmos records it. + + The projection fence now admits the new entries. A grant that does not reach Search raises + DocumentMutationPropagationError so a retry can complete it; entries that grant nothing, such as + pending shares, are projected best effort. + """ + for field_name, (previous_value, new_entries) in acl_changes.items(): + grants_access = bool(_search_acl_grant_entries(new_entries) - _search_acl_grant_entries(previous_value)) + if grants_access and not releases_chunks: + continue + try: + project_document_acl_to_chunks( + document_id, + {field_name: new_entries}, + user_id, + group_id=group_id, + document_version=document_version, + ) + except (DocumentSearchAclProjectionDeferredError, DocumentSearchAclProjectionError) as exc: + if grants_access: + raise _document_search_acl_incomplete_error(exc) from exc + log_event( + "[DOCUMENT_SEARCH_SYNC] An access change that grants no search access was not projected; it is saved.", + extra={'document_id': document_id, 'group_id': group_id, 'error_type': type(exc).__name__}, + level=logging.WARNING, + ) + + +def reproject_document_search_acl(document_item, user_id, group_id=None): + """Project a document's recorded access list to its search chunks again. + + Completes an access change that was saved before its search projection finished, such as an + approval whose projection failed. Grants reach only released documents; a held document's chunks + are rebuilt from the document when it is published. Raises DocumentMutationPropagationError, with + search_writes_frozen set while Search writes are frozen, when the projection fails. + """ + acl_fields = get_document_search_acl_fields(group_id) + if not acl_fields or not document_is_available(document_item): + return 0 + try: + return project_document_acl_to_chunks( + document_item['id'], + {field_name: list(document_item.get(field_name) or []) for field_name in acl_fields}, + document_item.get('user_id') or user_id, + group_id=group_id, + document_version=document_item.get('version'), + ) + except (DocumentSearchAclProjectionDeferredError, DocumentSearchAclProjectionError) as exc: + raise _document_search_acl_incomplete_error(exc) from exc + + +def build_document_search_metadata_fields(document_item, fields=None): + """Return the AI Search chunk values for a document's projected metadata fields.""" + requested_fields = DOCUMENT_SEARCH_METADATA_FIELD_MAP if fields is None else fields + field_values = {} + for document_field in requested_fields: + index_field = DOCUMENT_SEARCH_METADATA_FIELD_MAP.get(document_field) + if not index_field: + continue + value = document_item.get(document_field) + if document_field == 'authors': + value = ensure_list(value) + elif document_field == 'tags': + value = list(value or []) + elif document_field == 'file_name' and not value: + continue + field_values[index_field] = value + return field_values + + +def summarize_document_search_sync(search_syncs): + """Summarize the search sync outcomes of several document saves for one receipt.""" + pending_count = sum( + 1 for search_sync in search_syncs + if isinstance(search_sync, dict) and search_sync.get('status') == DOCUMENT_SEARCH_SYNC_STATUS_PENDING + ) + if pending_count: + return {'status': DOCUMENT_SEARCH_SYNC_STATUS_PENDING, 'document_count': pending_count} + return {'status': DOCUMENT_SEARCH_SYNC_STATUS_NOT_REQUIRED} + + +def _document_search_sync_now(): + return datetime.now(timezone.utc) + + +def _format_document_search_sync_time(value): + """Format a UTC timestamp at fixed width so stored values sort and compare as strings.""" + return value.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%f+00:00') + + +def _parse_document_search_sync_time(value): + if not value: + return None + try: + parsed = datetime.fromisoformat(str(value)) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed + + +def _get_document_search_sync_scope_name(group_id=None, public_workspace_id=None): + if public_workspace_id is not None: + return 'public' + if group_id is not None: + return 'group' + return 'personal' + + +def get_document_search_sync_record_id(document_id, group_id=None, public_workspace_id=None): + """Return the settings-container id of a document's search metadata sync record.""" + scope_name = _get_document_search_sync_scope_name(group_id, public_workspace_id) + return f"{DOCUMENT_SEARCH_SYNC_RECORD_TYPE}:{scope_name}:{document_id}" + + +def _document_search_sync_lease_is_active(record, now): + expires_at = _parse_document_search_sync_time(record.get('lease_expires_at')) + return bool(record.get('lease_token')) and expires_at is not None and expires_at > now + + +def _document_search_sync_intent_is_stale(record, now): + requested_at = _parse_document_search_sync_time(record.get('requested_at')) + if requested_at is None: + return True + return (now - requested_at).total_seconds() >= DOCUMENT_SEARCH_SYNC_UNCOMMITTED_INTENT_SECONDS + + +def _is_cosmos_write_conflict(exc): + return getattr(exc, 'status_code', None) in (409, 412) + + +def _is_cosmos_not_found(exc): + return isinstance(exc, CosmosResourceNotFoundError) or getattr(exc, 'status_code', None) == 404 + + +def _read_document_search_sync_record(record_id): + try: + return cosmos_settings_container.read_item(item=record_id, partition_key=record_id) + except Exception as exc: + if _is_cosmos_not_found(exc): + return None + raise + + +def _replace_document_search_sync_record(record): + return cosmos_settings_container.replace_item( + item=record['id'], + body=record, + etag=record.get('_etag'), + match_condition=MatchConditions.IfNotModified, + ) + + +def request_document_search_metadata_sync(document_id, changed_fields, user_id, group_id=None, public_workspace_id=None): + """Durably record that changed document metadata must be projected to its search chunks. + + Called before the document itself is saved, so a crash between the two writes leaves a pending + request for the reconciler instead of silently dropping the projection. Every call stores a new + request token; the caller saves the same token on the document. Returns the record. + """ + fields = sorted(set(changed_fields or []) & set(DOCUMENT_SEARCH_METADATA_FIELD_MAP)) + if not fields: + return None + + request_token = uuid.uuid4().hex + record_id = get_document_search_sync_record_id(document_id, group_id, public_workspace_id) + for _attempt in range(DOCUMENT_SEARCH_SYNC_RECORD_WRITE_ATTEMPTS): + now = _document_search_sync_now() + now_text = _format_document_search_sync_time(now) + next_attempt_at = _format_document_search_sync_time( + now + timedelta(seconds=DOCUMENT_SEARCH_SYNC_PENDING_GRACE_SECONDS) + ) + record = _read_document_search_sync_record(record_id) + try: + if record is None: + return cosmos_settings_container.create_item(body={ + 'id': record_id, + 'type': DOCUMENT_SEARCH_SYNC_RECORD_TYPE, + 'schema_version': DOCUMENT_SEARCH_SYNC_SCHEMA_VERSION, + 'document_id': document_id, + 'user_id': user_id, + 'group_id': group_id, + 'public_workspace_id': public_workspace_id, + 'revision': 1, + 'latest_request_token': request_token, + 'pending_fields': fields, + 'status': DOCUMENT_SEARCH_SYNC_STATUS_PENDING, + 'requested_at': now_text, + 'updated_at': now_text, + 'next_attempt_at': next_attempt_at, + 'attempts': 0, + 'last_error_type': None, + 'lease_token': None, + 'lease_expires_at': None, + }) + + record['revision'] = int(record.get('revision') or 0) + 1 + record['latest_request_token'] = request_token + record['pending_fields'] = sorted(set(record.get('pending_fields') or []) | set(fields)) + record['requested_at'] = now_text + record['updated_at'] = now_text + record['attempts'] = 0 + record['last_error_type'] = None + # A running worker picks up the new revision itself; otherwise the request waits briefly + # for the immediate in-process sync before the reconciler may claim it. + if not _document_search_sync_lease_is_active(record, now): + record['status'] = DOCUMENT_SEARCH_SYNC_STATUS_PENDING + record['next_attempt_at'] = next_attempt_at + return _replace_document_search_sync_record(record) + except Exception as exc: + # A conflict means another writer changed the record; not-found means a worker finished + # and removed it after this read. Either way, re-read and try again. + if not (_is_cosmos_write_conflict(exc) or _is_cosmos_not_found(exc)): + raise + raise RuntimeError("Unable to record the document search metadata sync request.") + + +def _get_document_search_sync_executor(): + global _document_search_sync_executor + with _document_search_sync_lock: + if _document_search_sync_executor is None: + # A small dedicated pool keeps bulk tag operations from crowding out document uploads on + # the shared Flask executor and bounds this process's Search write pressure. + _document_search_sync_executor = ThreadPoolExecutor( + max_workers=DOCUMENT_SEARCH_SYNC_WORKER_COUNT, + thread_name_prefix='document-search-sync', + ) + return _document_search_sync_executor + + +def schedule_document_search_metadata_sync(document_id, group_id=None, public_workspace_id=None): + """Queue an immediate in-process sync. The reconciler retries anything that does not run here.""" + record_id = get_document_search_sync_record_id(document_id, group_id, public_workspace_id) + with _document_search_sync_lock: + if record_id in _document_search_sync_queued_records: + return True + _document_search_sync_queued_records.add(record_id) + try: + _get_document_search_sync_executor().submit(_run_scheduled_document_search_metadata_sync, record_id) + return True + except Exception as exc: + with _document_search_sync_lock: + _document_search_sync_queued_records.discard(record_id) + log_event( + "[DOCUMENT_SEARCH_SYNC] Unable to queue a search metadata sync; the reconciler will run it.", + extra={'record_id': record_id, 'error_type': type(exc).__name__}, + level=logging.WARNING, + ) + return False + + +def _run_scheduled_document_search_metadata_sync(record_id): + with _document_search_sync_lock: + _document_search_sync_queued_records.discard(record_id) + try: + run_document_search_metadata_sync(record_id, lease_wait_seconds=DOCUMENT_SEARCH_SYNC_LEASE_WAIT_SECONDS) + except Exception as exc: + log_event( + "[DOCUMENT_SEARCH_SYNC] Search metadata sync stopped unexpectedly; the reconciler will retry it.", + extra={'record_id': record_id, 'error_type': type(exc).__name__}, + level=logging.ERROR, + exceptionTraceback=True, + ) + + +def _acquire_document_search_sync_lease(record_id, lease_wait_seconds=0): + """Lease a document's sync record, waiting up to lease_wait_seconds for a running sync.""" + deadline = time.monotonic() + max(0, lease_wait_seconds) + conflicts = 0 + while True: + record = _read_document_search_sync_record(record_id) + if record is None: + return None + now = _document_search_sync_now() + if not _document_search_sync_lease_is_active(record, now): + lease_expires_at = _format_document_search_sync_time( + now + timedelta(seconds=DOCUMENT_SEARCH_SYNC_LEASE_SECONDS) + ) + record.update({ + 'lease_token': uuid.uuid4().hex, + 'lease_expires_at': lease_expires_at, + 'next_attempt_at': lease_expires_at, + 'status': DOCUMENT_SEARCH_SYNC_STATUS_SYNCING, + 'updated_at': _format_document_search_sync_time(now), + }) + try: + return _replace_document_search_sync_record(record) + except Exception as exc: + if not (_is_cosmos_write_conflict(exc) or _is_cosmos_not_found(exc)): + raise + conflicts += 1 + if conflicts >= DOCUMENT_SEARCH_SYNC_RECORD_WRITE_ATTEMPTS: + return None + continue + if time.monotonic() >= deadline: + return None + time.sleep(DOCUMENT_SEARCH_SYNC_LEASE_POLL_SECONDS) + + +def _renew_document_search_sync_lease(record_id, lease_token): + """Confirm this worker still holds the lease, extending it once half of it has elapsed.""" + for _attempt in range(DOCUMENT_SEARCH_SYNC_RECORD_WRITE_ATTEMPTS): + record = _read_document_search_sync_record(record_id) + if record is None or record.get('lease_token') != lease_token: + raise DocumentSearchSyncLeaseLostError("The document search sync lease was lost.") + now = _document_search_sync_now() + expires_at = _parse_document_search_sync_time(record.get('lease_expires_at')) + if expires_at is not None and (expires_at - now).total_seconds() > DOCUMENT_SEARCH_SYNC_LEASE_SECONDS / 2: + return record + lease_expires_at = _format_document_search_sync_time( + now + timedelta(seconds=DOCUMENT_SEARCH_SYNC_LEASE_SECONDS) + ) + record.update({ + 'lease_expires_at': lease_expires_at, + 'next_attempt_at': lease_expires_at, + 'updated_at': _format_document_search_sync_time(now), + }) + try: + return _replace_document_search_sync_record(record) + except Exception as exc: + if _is_cosmos_not_found(exc): + raise DocumentSearchSyncLeaseLostError("The document search sync lease was lost.") from exc + if not _is_cosmos_write_conflict(exc): + raise + raise DocumentSearchSyncLeaseLostError("The document search sync lease could not be renewed.") + + +def _release_document_search_sync_lease(record_id, lease_token): + """Hand a still-pending record back to the reconciler without waiting for a retry delay.""" + for _attempt in range(DOCUMENT_SEARCH_SYNC_RECORD_WRITE_ATTEMPTS): + record = _read_document_search_sync_record(record_id) + if record is None or record.get('lease_token') != lease_token: + return + now_text = _format_document_search_sync_time(_document_search_sync_now()) + record.update({ + 'status': DOCUMENT_SEARCH_SYNC_STATUS_PENDING, + 'lease_token': None, + 'lease_expires_at': None, + 'next_attempt_at': now_text, + 'updated_at': now_text, + }) + try: + _replace_document_search_sync_record(record) + return + except Exception as exc: + if _is_cosmos_not_found(exc): + return + if not _is_cosmos_write_conflict(exc): + raise + + +def _record_document_search_sync_failure(record_id, lease_token, error): + """Release the lease and schedule a retry: a short fixed delay while Search writes are frozen, + otherwise an increasing backoff.""" + search_writes_frozen = isinstance(error, DataManagementSearchWritesFrozenError) + for _attempt in range(DOCUMENT_SEARCH_SYNC_RECORD_WRITE_ATTEMPTS): + record = _read_document_search_sync_record(record_id) + if record is None or record.get('lease_token') != lease_token: + return None + now = _document_search_sync_now() + attempts = int(record.get('attempts') or 0) + if search_writes_frozen: + status = DOCUMENT_SEARCH_SYNC_STATUS_PENDING + retry_delay_seconds = DOCUMENT_SEARCH_SYNC_FROZEN_RETRY_SECONDS + else: + attempts += 1 + status = DOCUMENT_SEARCH_SYNC_STATUS_FAILED + retry_delays = DOCUMENT_SEARCH_SYNC_RETRY_DELAYS_SECONDS + retry_delay_seconds = retry_delays[min(attempts, len(retry_delays)) - 1] + record.update({ + 'status': status, + 'attempts': attempts, + 'last_error_type': type(error).__name__, + 'last_failed_at': _format_document_search_sync_time(now), + 'next_attempt_at': _format_document_search_sync_time(now + timedelta(seconds=retry_delay_seconds)), + 'lease_token': None, + 'lease_expires_at': None, + 'updated_at': _format_document_search_sync_time(now), + }) + try: + return _replace_document_search_sync_record(record) + except Exception as exc: + if _is_cosmos_not_found(exc): + return None + if not _is_cosmos_write_conflict(exc): + raise + return None + + +def _complete_document_search_sync_pass(record_id, lease_token, synced_revision, document_committed): + """Record the outcome of one projection pass. + + Returns ('complete' | 'continue' | 'deferred' | 'lost', record). 'continue' means a newer edit + arrived during the pass and the worker keeps its lease to project it. 'deferred' means the + document write for this request has not landed yet, so the request is re-checked shortly. + """ + for _attempt in range(DOCUMENT_SEARCH_SYNC_RECORD_WRITE_ATTEMPTS): + record = _read_document_search_sync_record(record_id) + if record is None: + return 'complete', None + if record.get('lease_token') != lease_token: + return 'lost', record + now = _document_search_sync_now() + try: + if int(record.get('revision') or 0) > synced_revision: + lease_expires_at = _format_document_search_sync_time( + now + timedelta(seconds=DOCUMENT_SEARCH_SYNC_LEASE_SECONDS) + ) + record.update({ + 'lease_expires_at': lease_expires_at, + 'next_attempt_at': lease_expires_at, + 'updated_at': _format_document_search_sync_time(now), + }) + return 'continue', _replace_document_search_sync_record(record) + if not document_committed: + record.update({ + 'status': DOCUMENT_SEARCH_SYNC_STATUS_PENDING, + 'lease_token': None, + 'lease_expires_at': None, + 'next_attempt_at': _format_document_search_sync_time( + now + timedelta(seconds=DOCUMENT_SEARCH_SYNC_COMMIT_RECHECK_SECONDS) + ), + 'updated_at': _format_document_search_sync_time(now), + }) + return 'deferred', _replace_document_search_sync_record(record) + cosmos_settings_container.delete_item( + item=record_id, + partition_key=record_id, + etag=record.get('_etag'), + match_condition=MatchConditions.IfNotModified, + ) + return 'complete', None + except Exception as exc: + if _is_cosmos_not_found(exc): + return 'complete', None + if not _is_cosmos_write_conflict(exc): + raise + raise RuntimeError("Unable to record the document search metadata sync result.") + + +def _delete_document_search_sync_record(record_id): + try: + cosmos_settings_container.delete_item(item=record_id, partition_key=record_id) + except Exception as exc: + if not _is_cosmos_not_found(exc): + raise + + +def _read_document_for_search_sync(record): + """Read the document a sync record targets, only while it still belongs to that scope.""" + document_id = record.get('document_id') + group_id = record.get('group_id') + public_workspace_id = record.get('public_workspace_id') + container = _get_documents_container(group_id=group_id, public_workspace_id=public_workspace_id) + try: + document_item = container.read_item(item=document_id, partition_key=document_id) + except Exception as exc: + if _is_cosmos_not_found(exc): + return None + raise + if public_workspace_id is not None: + belongs_to_scope = document_item.get('public_workspace_id') == public_workspace_id + elif group_id is not None: + belongs_to_scope = document_item.get('group_id') == group_id + else: + belongs_to_scope = document_item.get('user_id') == record.get('user_id') + return document_item if belongs_to_scope else None + + +def _approved_search_share_principals(entries): + """Return the users or groups whose approved share lets them find a document in search.""" + principals = [] + for entry in _normalize_search_acl_entries(entries): + principal, _separator, status = entry.partition(',') + if principal and status == 'approved': + principals.append(principal) + return principals + + +def _invalidate_document_search_results_cache(document_item, group_id=None, public_workspace_id=None): + """Drop cached search results that can include a document once a sync has changed its chunks. + + Routes clear the cache when a change is saved, but a search that ran while the sync was still + merging could otherwise keep serving the old values until its cache entry expires. Recipients of + approved shares cache their own results, so theirs are cleared too. Best effort. + """ + try: + if public_workspace_id is not None: + invalidate_public_workspace_search_cache(public_workspace_id) + elif group_id is not None: + invalidate_group_search_cache(group_id) + for shared_group_id in _approved_search_share_principals(document_item.get('shared_group_ids')): + invalidate_group_search_cache(shared_group_id) + else: + owner_user_id = document_item.get('user_id') + if owner_user_id: + invalidate_personal_search_cache(owner_user_id) + for shared_user_id in _approved_search_share_principals(document_item.get('shared_user_ids')): + invalidate_personal_search_cache(shared_user_id) + except Exception as exc: + log_event( + "[DOCUMENT_SEARCH_SYNC] Cached search results were not cleared after a search metadata sync.", + extra={'document_id': document_item.get('id'), 'error_type': type(exc).__name__}, + level=logging.WARNING, + ) + + +def run_document_search_metadata_sync(record_id, lease_wait_seconds=0): + """Project a document's pending metadata to all of its search chunks. + + Serialized per document by a lease on the sync record. Every pass projects the latest saved value + of each pending field, so a retry also heals chunks a failed pass left behind. A held screened + document has no released chunks and publication rebuilds them from the document, so it has + nothing to sync. The worker never writes to the document. Failures are recorded on the record + with a retry delay. Returns a status summary. + """ + record = _acquire_document_search_sync_lease(record_id, lease_wait_seconds) + if record is None: + return {'status': 'not_started', 'record_id': record_id} + + lease_token = record.get('lease_token') + document_id = record.get('document_id') + group_id = record.get('group_id') + public_workspace_id = record.get('public_workspace_id') + chunks_updated = 0 + for _pass in range(DOCUMENT_SEARCH_SYNC_MAX_PASSES): + revision = int(record.get('revision') or 0) + pending_fields = list(record.get('pending_fields') or []) + try: + document_item = _read_document_for_search_sync(record) + if document_item is None: + _delete_document_search_sync_record(record_id) + return {'status': 'document_missing', 'record_id': record_id} + + chunks_updated = 0 + if document_is_available(document_item): + # Personal chunks are scoped by the owner; group and public chunks by their workspace. + owner_user_id = document_item.get('user_id') or record.get('user_id') + chunks_updated = project_fields_to_document_chunks( + document_id, + build_document_search_metadata_fields(document_item, pending_fields), + owner_user_id, + group_id=group_id, + public_workspace_id=public_workspace_id, + before_batch=lambda: _renew_document_search_sync_lease(record_id, lease_token), + ) + if 'tags' in pending_fields: + # Blob tag metadata is informational; this refresh is best effort and never fails the pass. + propagate_tags_to_blob_metadata( + document_id, + document_item.get('tags') or [], + owner_user_id, + group_id, + public_workspace_id, + ) + if chunks_updated: + _invalidate_document_search_results_cache( + document_item, + group_id=group_id, + public_workspace_id=public_workspace_id, + ) + except DocumentSearchSyncLeaseLostError: + log_event( + "[DOCUMENT_SEARCH_SYNC] Search metadata sync stopped after another worker took over.", + extra={'record_id': record_id, 'document_id': document_id}, + level=logging.WARNING, + ) + return {'status': 'lost', 'record_id': record_id} + except Exception as exc: + failure_record = _record_document_search_sync_failure(record_id, lease_token, exc) + log_event( + "[DOCUMENT_SEARCH_SYNC] Search metadata sync failed; it will be retried.", + extra={ + 'record_id': record_id, + 'document_id': document_id, + 'fields': pending_fields, + 'attempts': (failure_record or {}).get('attempts'), + 'next_attempt_at': (failure_record or {}).get('next_attempt_at'), + 'error_type': type(exc).__name__, + }, + level=logging.WARNING, + exceptionTraceback=True, + ) + return {'status': 'failed', 'record_id': record_id, 'error_type': type(exc).__name__} + + now = _document_search_sync_now() + # Only the save that carries this request's token proves the projected values are current. + latest_request_token = record.get('latest_request_token') + document_committed = ( + bool(latest_request_token) + and document_item.get(DOCUMENT_SEARCH_SYNC_TOKEN_FIELD) == latest_request_token + ) or _document_search_sync_intent_is_stale(record, now) + outcome, next_record = _complete_document_search_sync_pass( + record_id, + lease_token, + revision, + document_committed, + ) + if outcome != 'continue': + return {'status': outcome, 'record_id': record_id, 'chunks_updated': chunks_updated} + record = next_record + + _release_document_search_sync_lease(record_id, lease_token) + return {'status': DOCUMENT_SEARCH_SYNC_STATUS_PENDING, 'record_id': record_id, 'chunks_updated': chunks_updated} + + +def process_due_document_search_metadata_syncs(max_records=None, time_budget_seconds=None): + """Run search metadata syncs that are due for a retry or were never picked up. + + Used by the background reconciler. Covers syncs whose in-process run failed, never started + because the process restarted, or lost their lease when a worker stopped mid-sync. + """ + if max_records is None: + max_records = DOCUMENT_SEARCH_SYNC_RECONCILE_BATCH_SIZE + if time_budget_seconds is None: + time_budget_seconds = DOCUMENT_SEARCH_SYNC_RECONCILE_TIME_BUDGET_SECONDS + started = time.monotonic() + due_records = list(cosmos_settings_container.query_items( + query=( + 'SELECT TOP @limit c.id FROM c ' + 'WHERE c.type = @type AND c.next_attempt_at <= @now ' + 'ORDER BY c.next_attempt_at ASC' + ), + parameters=[ + {'name': '@limit', 'value': int(max_records)}, + {'name': '@type', 'value': DOCUMENT_SEARCH_SYNC_RECORD_TYPE}, + {'name': '@now', 'value': _format_document_search_sync_time(_document_search_sync_now())}, + ], + enable_cross_partition_query=True, + )) + summary = {'due': len(due_records), 'processed': 0, 'complete': 0, 'failed': 0} + for due_record in due_records: + if time.monotonic() - started >= time_budget_seconds: + break + try: + result = run_document_search_metadata_sync(due_record.get('id')) + except Exception as exc: + log_event( + "[DOCUMENT_SEARCH_SYNC] Search metadata reconciliation failed for a document.", + extra={'record_id': due_record.get('id'), 'error_type': type(exc).__name__}, + level=logging.ERROR, + exceptionTraceback=True, + ) + summary['failed'] += 1 + continue + summary['processed'] += 1 + if result.get('status') == 'complete': + summary['complete'] += 1 + elif result.get('status') == 'failed': + summary['failed'] += 1 + return summary + + def get_pdf_page_count(pdf_path: str) -> int: """ Returns the total number of pages in the given PDF using PyMuPDF. @@ -12355,27 +13438,20 @@ def share_document_with_user(document_id, owner_user_id, target_user_id): operation='document_shared_with_user', ) - # Update all chunks with the new shared_user_ids + # A pending share grants no search access yet, so projecting it is best effort. The + # approval projects the whole list again before it is saved. try: - chunks = get_all_chunks(document_id, owner_user_id) - for chunk in chunks: - chunk_id = chunk.get('id') - if chunk_id: - try: - update_chunk_metadata( - chunk_id=chunk_id, - user_id=owner_user_id, - group_id=None, - public_workspace_id=None, - document_id=document_id, - shared_user_ids=shared_user_ids - ) - except Exception as chunk_e: - print(f"Warning: Failed to update chunk {chunk_id}: {chunk_e}") - # Continue with other chunks + project_fields_to_document_chunks( + document_id, + {'shared_user_ids': shared_user_ids}, + owner_user_id, + ) except Exception as e: - print(f"Warning: Failed to update chunks for document {document_id}: {e}") - # Don't fail the whole operation if chunk update fails + log_event( + "[DOCUMENT_SEARCH_SYNC] Projecting a pending share to search chunks failed; the share is saved.", + extra={'document_id': document_id, 'error_type': type(e).__name__}, + level=logging.WARNING, + ) return True @@ -12416,24 +13492,11 @@ def unshare_document_from_user(document_id, owner_user_id, target_user_id): if len(new_shared_user_ids) != len(shared_user_ids): # A revoked user must never retain search access after the API reports success. # Update every chunk projection before committing the authoritative Cosmos ACL. - chunks = get_all_chunks(document_id, actual_owner_id) - try: - for chunk in chunks: - chunk_id = chunk.get('id') - if not chunk_id: - continue - update_chunk_metadata( - chunk_id=chunk_id, - user_id=actual_owner_id, - group_id=None, - public_workspace_id=None, - document_id=document_id, - shared_user_ids=new_shared_user_ids - ) - except DataManagementSearchWritesFrozenError as exc: - raise DocumentSearchAclProjectionDeferredError( - "Document access was not changed because target Search writes are temporarily frozen. Retry after the migration finishes." - ) from exc + project_document_acl_to_chunks( + document_id, + {'shared_user_ids': new_shared_user_ids}, + actual_owner_id, + ) document_item['shared_user_ids'] = new_shared_user_ids document_item['last_updated'] = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') @@ -12447,7 +13510,7 @@ def unshare_document_from_user(document_id, owner_user_id, target_user_id): except CosmosResourceNotFoundError: return False - except DocumentSearchAclProjectionDeferredError: + except (DocumentSearchAclProjectionDeferredError, DocumentSearchAclProjectionError): raise except Exception as e: print(f"Error unsharing document {document_id}: {e}") @@ -13212,146 +14275,68 @@ def add_definition(fresh): return stored_tag_def -def propagate_tags_to_blob_metadata( - document_id, tags, user_id, group_id=None, public_workspace_id=None, *, strict=False, expected_etag=None, -): - """ - Update blob metadata with document tags when enhanced citations is enabled. - Tags are stored as a comma-separated string in blob metadata. +def propagate_tags_to_blob_metadata(document_id, tags, user_id, group_id=None, public_workspace_id=None): + """Mirror a document's tags into its source blob's metadata when enhanced citations is enabled. - Args: - document_id: Document ID - tags: Array of normalized tag names - user_id: User ID - group_id: Optional group ID - public_workspace_id: Optional public workspace ID + Tags are stored as a comma-separated string. The search metadata sync worker calls this after + it projects tags to the search chunks. Nothing reads these tags back, so the refresh is best + effort: it logs a failure and returns False instead of raising. The write is conditioned on the + blob ETag it just read, so it never overwrites the metadata of a concurrent blob replacement. + Returns True when the blob metadata was written. """ try: settings = get_settings() if not settings.get('enable_enhanced_citations', False): - return - - is_group = group_id is not None - is_public_workspace = public_workspace_id is not None + return False - # Read document from Cosmos DB to get file_name - if is_public_workspace: + if public_workspace_id is not None: cosmos_container = cosmos_public_documents_container - elif is_group: + elif group_id is not None: cosmos_container = cosmos_group_documents_container else: cosmos_container = cosmos_user_documents_container doc_item = cosmos_container.read_item(document_id, partition_key=document_id) - if strict and ( - not expected_etag or doc_item.get("_etag") != expected_etag - or (group_id is not None and doc_item.get("group_id") != group_id) + if ( + (group_id is not None and doc_item.get("group_id") != group_id) or (public_workspace_id is not None and doc_item.get("public_workspace_id") != public_workspace_id) ): - raise ScreeningConflictError() + return False if SCREENING_FIELD in doc_item: # A screened release pins its blob by ETag and content hash. Rewriting the blob's # metadata would make the released file unreadable, so tags stay in Cosmos and Search. - return - if strict and not _has_persisted_blob_reference(doc_item) and not doc_item.get("enhanced_citations"): - return + return False + if not _has_persisted_blob_reference(doc_item) and not doc_item.get("enhanced_citations"): + return False storage_account_container_name, blob_path = get_document_blob_storage_info( doc_item, user_id=user_id, group_id=group_id, public_workspace_id=public_workspace_id, ) - if not blob_path: - if strict: - raise FileNotFoundError("The document source is unavailable.") - print(f"Warning: No blob path found for document {document_id}, skipping blob metadata update") - return - blob_service_client = CLIENTS.get("storage_account_office_docs_client") - if not blob_service_client: - if strict: - raise RuntimeError("Document source storage is unavailable.") - print(f"Warning: Blob service client not available, skipping blob metadata update") - return + if not blob_path or not blob_service_client: + return False blob_client = blob_service_client.get_blob_client( container=storage_account_container_name, blob=blob_path ) - if not blob_client.exists(): - if strict: - raise FileNotFoundError("The document source is unavailable.") - print(f"Warning: Blob not found at {blob_path}, skipping metadata update") - return + return False - # Get existing metadata and update with tags properties = blob_client.get_blob_properties() existing_metadata = dict(properties.metadata) if properties.metadata else {} existing_metadata['document_tags'] = ','.join(tags) if tags else '' - if strict: - blob_client.set_blob_metadata( - metadata=existing_metadata, etag=properties.etag, match_condition=MatchConditions.IfNotModified, - ) - else: - blob_client.set_blob_metadata(metadata=existing_metadata) - - print(f"Successfully updated blob metadata tags for document {document_id} at {blob_path}") - - except Exception as e: - if strict: - log_event( - "[DOCUMENTS] Document tags were saved but blob metadata propagation failed.", - extra={"document_id": document_id, "exception_type": type(e).__name__}, - level=logging.ERROR, - ) - raise DocumentMutationPropagationError("Document tags were saved, but source metadata could not be updated.") from e - print(f"Warning: Failed to update blob metadata tags for document {document_id}: {e}") - # Non-fatal — tag propagation to chunks is the primary operation - - -def propagate_tags_to_chunks(document_id, tags, user_id, group_id=None, public_workspace_id=None): - """ - Update all chunks for a document with new tags. - This is called immediately after tag updates. - - Args: - document_id: Document ID - tags: Array of normalized tag names - user_id: User ID - group_id: Optional group ID - public_workspace_id: Optional public workspace ID - """ - try: - # Get all chunks for this document - chunks = get_all_chunks(document_id, user_id, group_id, public_workspace_id) - - if not chunks: - print(f"No chunks found for document {document_id}") - return - - # Update each chunk with new tags - chunk_count = 0 - for chunk in chunks: - try: - update_chunk_metadata( - chunk_id=chunk['id'], - user_id=user_id, - group_id=group_id, - public_workspace_id=public_workspace_id, - document_id=document_id, - document_tags=tags - ) - chunk_count += 1 - except Exception as chunk_error: - print(f"Error updating chunk {chunk['id']} with tags: {chunk_error}") - # Continue with other chunks - - print(f"Successfully propagated tags to {chunk_count} chunks for document {document_id}") - - # Also update blob metadata with tags if enhanced citations is enabled - propagate_tags_to_blob_metadata(document_id, tags, user_id, group_id, public_workspace_id) + blob_client.set_blob_metadata( + metadata=existing_metadata, etag=properties.etag, match_condition=MatchConditions.IfNotModified, + ) + return True except Exception as e: - print(f"Error propagating tags to chunks for document {document_id}: {e}") - raise + log_event( + "[DOCUMENT_SEARCH_SYNC] Document tags were saved, but the source blob's tag metadata was not refreshed.", + extra={"document_id": document_id, "exception_type": type(e).__name__}, + level=logging.WARNING, + ) + return False diff --git a/application/single_app/functions_group_document_collaboration.py b/application/single_app/functions_group_document_collaboration.py index dff529e2c..8ac0f2f03 100644 --- a/application/single_app/functions_group_document_collaboration.py +++ b/application/single_app/functions_group_document_collaboration.py @@ -437,6 +437,47 @@ def _search_acl(document, *, operation_guard): ) +def _search_acl_is_current(document): + """Whether every search chunk of the document carries the group access list Cosmos records.""" + client = _get_search_client(group_id=document["group_id"]) + escaped_id = document["id"].replace("'", "''") + available = _available(document) and is_current_group_document(document) + expected = sorted(set(document.get("shared_group_ids") or [])) if available else [] + for chunk in client.search(search_text="*", filter=f"document_id eq '{escaped_id}'", select=["id", "shared_group_ids"]): + if sorted(set(chunk.get("shared_group_ids") or [])) != expected: + return False + return True + + +def _repair_approved_share_search_acl(user_id, group_id, document): + """Re-run the search effect when an approved share's chunks miss the recorded access list. + + Approving a share that is already approved completes an approval whose search update did not + finish, such as one made through the classic workspace or before this repair existed. Chunks + that already match are left alone, so a repeat has no effects. Returns receipt errors, if any. + """ + def guard(): + current = read_group_document_record(document["id"]) + _require_share_action(user_id, group_id, current, "approve_share", group_id) + return current + + try: + if _search_acl_is_current(document): + return [] + _search_acl(document, operation_guard=guard) + except Exception as error: + log_event( + "[DOCUMENT_SHARING] An approved group share could not be projected to search again.", + extra={"document_id": document["id"], "group_id": document["group_id"], "exception_type": type(error).__name__}, + level=logging.WARNING, + ) + return [{ + "stage": "search", "code": "search_repair_required", + "message": "The search update could not be confirmed. Refresh before retrying.", + }] + return [] + + def _notification_effect(document, operation, *, operation_guard): target_id = operation["target_group_id"] source_id = document["group_id"] @@ -606,6 +647,10 @@ def change_group_document_share(user_id, group_id, document_id, action, payload, ) unchanged = action == "share" and before in {"approved", "not_approved"} or action == "approve_share" and before == "approved" if unchanged: + if action == "approve_share": + errors = _repair_approved_share_search_acl(user_id, group_id, document) + if errors: + return _operation_receipt(group_id, document, action, target_group_id, "partial", state, errors), 207 return _operation_receipt(group_id, document, action, target_group_id, "unchanged", state, []), 200 if action == "unshare" and before is None and existing and existing.get("target_group_id") == target_group_id and not _unfinished(existing): return _operation_receipt(group_id, document, action, target_group_id, "unchanged", "removed", []), 200 diff --git a/application/single_app/functions_group_document_management.py b/application/single_app/functions_group_document_management.py index df6e26507..767726049 100644 --- a/application/single_app/functions_group_document_management.py +++ b/application/single_app/functions_group_document_management.py @@ -32,6 +32,7 @@ process_document_upload_background, process_metadata_extraction_background, select_current_documents, + summarize_document_search_sync, update_document, validate_tag_color, validate_tags, @@ -272,9 +273,10 @@ def update_group_document_metadata( authorize_group_document_operation, user_id, group_id, document_id, operation, expected_version=document.get("version"), ) - saved = update_document( + saved, search_sync = update_document( document_id=document_id, user_id=user_id, group_id=group_id, - strict=True, expected_etag=document.get("_etag"), operation_guard=guard, **changes, + strict=True, expected_etag=document.get("_etag"), operation_guard=guard, + return_search_sync=True, **changes, ) if not isinstance(saved, dict) or saved.get("id") != document_id or saved.get("group_id") != group_id: raise DocumentMutationPropagationError("The scoped document update could not be confirmed.") @@ -285,11 +287,13 @@ def update_group_document_metadata( updated_fields={name: "[updated]" for name in changes} if SCREENING_FIELD in document else changes, file_type=document.get("file_type"), ) - # Metadata edits apply directly; they never start a new screening hold. + # Metadata edits apply directly; they never start a new screening hold. Search chunks pick the + # change up from a background sync, which search_sync reports. return { "message": "Group document metadata updated.", "document_id": document_id, "group_id": group_id, "updated_fields": list(changes), "status": "updated", + "search_sync": search_sync, } @@ -313,6 +317,7 @@ def tag_group_documents(user_id, group_id, payload): # the whole batch with no document written. _ensure_document_tag_definitions(user_id, group_id, tags) result = {"success": [], "errors": []} + search_syncs = [] for document_id in document_ids: try: document = authorize_group_document_operation(user_id, group_id, document_id, "tag_documents") @@ -323,14 +328,16 @@ def tag_group_documents(user_id, group_id, payload): updated_tags = [tag for tag in current_tags if normalize_tag(tag) not in tags] else: updated_tags = tags - update_group_document_metadata( + receipt = update_group_document_metadata( user_id, group_id, document_id, {"tags": updated_tags}, operation="tag_documents", ensure_definitions=False, ) + search_syncs.append(receipt.get("search_sync")) result["success"].append({"document_id": document_id, "tags": updated_tags}) except Exception as error: failure, _status = group_operation_error(error, "tag_documents", document_id=document_id, group_id=group_id) result["errors"].append(failure) + result["search_sync"] = summarize_document_search_sync(search_syncs) return result, 207 if result["errors"] else 200 @@ -388,6 +395,7 @@ def change_group_document_tag(user_id, group_id, tag_name, payload=None, *, dele definition["color"] = _new_tag_definition(old_name, new_color)["color"] _patch_tag_definitions(user_id, group_id, group, {old_name: definition}, removals=old_keys) result["tag"] = {"name": old_name, "color": definition["color"]} + result["search_sync"] = summarize_document_search_sync([]) return result, 200 changes = {} @@ -403,6 +411,7 @@ def change_group_document_tag(user_id, group_id, tag_name, payload=None, *, dele changes[new_name] = target_definition result["tag"] = {"name": new_name, "color": target_definition.get("color") or get_default_tag_color(new_name)} group = _patch_tag_definitions(user_id, group_id, group, changes) + search_syncs = [] for target in targets: document_id = target["id"] try: @@ -414,14 +423,16 @@ def change_group_document_tag(user_id, group_id, tag_name, payload=None, *, dele if not (delete and normalize_tag(tag) == old_name) ] updated_tags = list(dict.fromkeys(updated_tags)) - update_group_document_metadata( + receipt = update_group_document_metadata( user_id, group_id, document_id, {"tags": updated_tags}, operation="tag_documents", ensure_definitions=False, ) + search_syncs.append(receipt.get("search_sync")) result["success"].append({"document_id": document_id, "tags": updated_tags}) except Exception as error: failure, _status = group_operation_error(error, "manage_tags", document_id=document_id, group_id=group_id) result["errors"].append(failure) result["documents_updated"] = len(result["success"]) + result["search_sync"] = summarize_document_search_sync(search_syncs) if not result["errors"]: try: _patch_tag_definitions(user_id, group_id, group, {}, removals=old_keys) diff --git a/application/single_app/functions_public_document_management.py b/application/single_app/functions_public_document_management.py index e410c9940..eef548260 100644 --- a/application/single_app/functions_public_document_management.py +++ b/application/single_app/functions_public_document_management.py @@ -37,6 +37,7 @@ process_document_upload_background, process_metadata_extraction_background, select_current_documents, + summarize_document_search_sync, update_document, validate_tag_color, validate_tags, @@ -278,9 +279,10 @@ def update_public_document_metadata( authorize_public_document_operation, user_id, workspace_id, document_id, operation, expected_version=document.get("version"), ) - saved = update_document( + saved, search_sync = update_document( document_id=document_id, user_id=user_id, public_workspace_id=workspace_id, - strict=True, expected_etag=document.get("_etag"), operation_guard=guard, **changes, + strict=True, expected_etag=document.get("_etag"), operation_guard=guard, + return_search_sync=True, **changes, ) if not isinstance(saved, dict) or saved.get("id") != document_id or saved.get("public_workspace_id") != workspace_id: raise DocumentMutationPropagationError("The scoped document update could not be confirmed.") @@ -291,11 +293,13 @@ def update_public_document_metadata( updated_fields={name: "[updated]" for name in changes} if SCREENING_FIELD in document else changes, file_type=document.get("file_type"), ) - # Metadata edits apply directly; they never start a new screening hold. + # Metadata edits apply directly; they never start a new screening hold. Search chunks pick the + # change up from a background sync, which search_sync reports. return { "message": "Public document metadata updated.", "document_id": document_id, "public_workspace_id": workspace_id, "updated_fields": list(changes), "status": "updated", + "search_sync": search_sync, } @@ -319,6 +323,7 @@ def tag_public_documents(user_id, workspace_id, payload): # the whole batch with no document written. _ensure_document_tag_definitions(user_id, workspace_id, tags) result = {"success": [], "errors": []} + search_syncs = [] for document_id in document_ids: try: document = authorize_public_document_operation(user_id, workspace_id, document_id, "tag_documents") @@ -329,14 +334,16 @@ def tag_public_documents(user_id, workspace_id, payload): updated_tags = [tag for tag in current_tags if normalize_tag(tag) not in tags] else: updated_tags = tags - update_public_document_metadata( + receipt = update_public_document_metadata( user_id, workspace_id, document_id, {"tags": updated_tags}, operation="tag_documents", ensure_definitions=False, ) + search_syncs.append(receipt.get("search_sync")) result["success"].append({"document_id": document_id, "tags": updated_tags}) except Exception as error: failure, _status = public_operation_error(error, "tag_documents", document_id=document_id, workspace_id=workspace_id) result["errors"].append(failure) + result["search_sync"] = summarize_document_search_sync(search_syncs) return result, 207 if result["errors"] else 200 @@ -394,6 +401,7 @@ def change_public_document_tag(user_id, workspace_id, tag_name, payload=None, *, definition["color"] = _new_tag_definition(old_name, new_color)["color"] _patch_tag_definitions(user_id, workspace_id, workspace, {old_name: definition}, removals=old_keys) result["tag"] = {"name": old_name, "color": definition["color"]} + result["search_sync"] = summarize_document_search_sync([]) return result, 200 changes = {} @@ -409,6 +417,7 @@ def change_public_document_tag(user_id, workspace_id, tag_name, payload=None, *, changes[new_name] = target_definition result["tag"] = {"name": new_name, "color": target_definition.get("color") or get_default_tag_color(new_name)} workspace = _patch_tag_definitions(user_id, workspace_id, workspace, changes) + search_syncs = [] for target in targets: document_id = target["id"] try: @@ -420,14 +429,16 @@ def change_public_document_tag(user_id, workspace_id, tag_name, payload=None, *, if not (delete and normalize_tag(tag) == old_name) ] updated_tags = list(dict.fromkeys(updated_tags)) - update_public_document_metadata( + receipt = update_public_document_metadata( user_id, workspace_id, document_id, {"tags": updated_tags}, operation="tag_documents", ensure_definitions=False, ) + search_syncs.append(receipt.get("search_sync")) result["success"].append({"document_id": document_id, "tags": updated_tags}) except Exception as error: failure, _status = public_operation_error(error, "manage_tags", document_id=document_id, workspace_id=workspace_id) result["errors"].append(failure) result["documents_updated"] = len(result["success"]) + result["search_sync"] = summarize_document_search_sync(search_syncs) if not result["errors"]: try: _patch_tag_definitions(user_id, workspace_id, workspace, {}, removals=old_keys) diff --git a/application/single_app/route_backend_documents.py b/application/single_app/route_backend_documents.py index e2ea306a0..c6012e1e2 100644 --- a/application/single_app/route_backend_documents.py +++ b/application/single_app/route_backend_documents.py @@ -229,6 +229,15 @@ def _create_personal_document_share_decision_notification(document_item, target_ ) +def _document_search_acl_error_response(error): + """Return the safe response for a sharing change the search index could not apply.""" + message, status_code, retry_after_seconds = describe_document_search_acl_error(error) + response = jsonify({'error': message}) + if retry_after_seconds: + response.headers['Retry-After'] = str(retry_after_seconds) + return response, status_code + + def _escape_citation_odata_literal(value): return _normalize_citation_lookup_value(value).replace("'", "''") @@ -1280,111 +1289,77 @@ def api_patch_user_document(document_id): if not user_id: return jsonify({'error': 'User not authenticated'}), 401 - data = request.get_json() # new metadata values from the client + data = request.get_json() or {} # new metadata values from the client # Track which fields were updated updated_fields = {} + document_updates = {} - # Update allowed fields - # You can decide which fields can be updated from the client - if 'title' in data: - update_document( - document_id=document_id, - user_id=user_id, - title=data['title'] - ) - updated_fields['title'] = data['title'] - if 'abstract' in data: - update_document( - document_id=document_id, - user_id=user_id, - abstract=data['abstract'] - ) - updated_fields['abstract'] = data['abstract'] - if 'keywords' in data: - # Expect a list or a comma-delimited string - if isinstance(data['keywords'], list): - update_document( - document_id=document_id, - user_id=user_id, - keywords=data['keywords'] - ) - updated_fields['keywords'] = data['keywords'] - else: - # if client sends a comma-separated string of keywords - keywords_list = [kw.strip() for kw in data['keywords'].split(',')] - update_document( - document_id=document_id, - user_id=user_id, - keywords=keywords_list - ) - updated_fields['keywords'] = keywords_list - if 'publication_date' in data: - update_document( - document_id=document_id, - user_id=user_id, - publication_date=data['publication_date'] - ) - updated_fields['publication_date'] = data['publication_date'] - if 'document_classification' in data: - update_document( - document_id=document_id, - user_id=user_id, - document_classification=data['document_classification'] - ) - updated_fields['document_classification'] = data['document_classification'] - # Add authors if you want to allow editing that - if 'authors' in data: - # if you want a list, or just store a string - # here is one approach: - if isinstance(data['authors'], list): - update_document( - document_id=document_id, - user_id=user_id, - authors=data['authors'] - ) - updated_fields['authors'] = data['authors'] - else: - authors_list = [data['authors']] - update_document( + try: + # Update allowed fields + # You can decide which fields can be updated from the client + if 'title' in data: + document_updates['title'] = data['title'] + updated_fields['title'] = data['title'] + if 'abstract' in data: + document_updates['abstract'] = data['abstract'] + updated_fields['abstract'] = data['abstract'] + if 'keywords' in data: + # Expect a list or a comma-delimited string + if isinstance(data['keywords'], list): + document_updates['keywords'] = data['keywords'] + updated_fields['keywords'] = data['keywords'] + else: + # if client sends a comma-separated string of keywords + keywords_list = [kw.strip() for kw in data['keywords'].split(',')] + document_updates['keywords'] = keywords_list + updated_fields['keywords'] = keywords_list + if 'publication_date' in data: + document_updates['publication_date'] = data['publication_date'] + updated_fields['publication_date'] = data['publication_date'] + if 'document_classification' in data: + document_updates['document_classification'] = data['document_classification'] + updated_fields['document_classification'] = data['document_classification'] + # Add authors if you want to allow editing that + if 'authors' in data: + # if you want a list, or just store a string + # here is one approach: + if isinstance(data['authors'], list): + document_updates['authors'] = data['authors'] + updated_fields['authors'] = data['authors'] + else: + authors_list = [data['authors']] + document_updates['authors'] = authors_list + updated_fields['authors'] = authors_list + + # Handle tags with validation; chunk and blob propagation runs in the background sync + if 'tags' in data: + from functions_documents import validate_tags, get_or_create_tag_definition + + # Validate and normalize tags + tags_input = data['tags'] if isinstance(data['tags'], list) else [] + is_valid, error_msg, normalized_tags = validate_tags(tags_input) + + if not is_valid: + return jsonify({'error': error_msg}), 400 + + # Ensure tag definitions exist for new tags + for tag in normalized_tags: + get_or_create_tag_definition(user_id, tag, workspace_type='personal') + + # Update document with normalized tags + document_updates['tags'] = normalized_tags + updated_fields['tags'] = normalized_tags + + # Save every change in one update; search chunks are synced in the background + update_result = {} + if document_updates: + update_result = update_document( document_id=document_id, user_id=user_id, - authors=authors_list - ) - updated_fields['authors'] = authors_list - - # Handle tags with validation and chunk propagation - if 'tags' in data: - from functions_documents import validate_tags, propagate_tags_to_chunks, get_or_create_tag_definition - - # Validate and normalize tags - tags_input = data['tags'] if isinstance(data['tags'], list) else [] - is_valid, error_msg, normalized_tags = validate_tags(tags_input) - - if not is_valid: - return jsonify({'error': error_msg}), 400 - - # Ensure tag definitions exist for new tags - for tag in normalized_tags: - get_or_create_tag_definition(user_id, tag, workspace_type='personal') - - # Update document with normalized tags - update_document( - document_id=document_id, - user_id=user_id, - tags=normalized_tags - ) - updated_fields['tags'] = normalized_tags - - # Propagate tags to all chunks immediately - try: - propagate_tags_to_chunks(document_id, normalized_tags, user_id) - except Exception as propagate_error: - debug_print(f"Warning: Failed to propagate tags to chunks: {propagate_error}") - # Continue - document tags are updated, chunk sync will be retried later + **document_updates, + ) or {} - # Save updates back to Cosmos - try: # Log the metadata update transaction if any fields were updated if updated_fields: # Get document details for logging @@ -1408,10 +1383,21 @@ def api_patch_user_document(document_id): updated_fields=updated_fields, file_type=doc.get('file_type') ) - - return jsonify({'message': 'Document metadata updated successfully'}), 200 + + response_payload = {'message': 'Document metadata updated successfully'} + if isinstance(update_result, dict) and update_result.get('search_sync'): + response_payload['search_sync'] = update_result['search_sync'] + + return jsonify(response_payload), 200 + except exceptions.CosmosResourceNotFoundError: + return jsonify({'error': 'Document not found or access denied'}), 404 except Exception as e: - return jsonify({'error': str(e)}), 500 + log_event( + '[DOCUMENT_METADATA] Failed to update personal document metadata', + {'document_id': document_id, 'error_type': type(e).__name__}, + level=logging.ERROR, + ) + return jsonify({'error': 'Unable to update document metadata'}), 500 @bp.route('/api/documents/', methods=['DELETE']) @swagger_route(security=get_auth_security()) @@ -1996,8 +1982,8 @@ def api_bulk_tag_documents(): return jsonify({'error': 'action must be add_tags, remove_tags, or set_tags'}), 400 from functions_documents import ( - validate_tags, get_document, update_document, - propagate_tags_to_chunks, get_or_create_tag_definition + validate_tags, get_document, update_document, + get_or_create_tag_definition, summarize_document_search_sync ) # Validate and normalize tags @@ -2013,6 +1999,7 @@ def api_bulk_tag_documents(): 'success': [], 'errors': [] } + search_syncs = [] try: for doc_id in document_ids: @@ -2070,18 +2057,13 @@ def api_bulk_tag_documents(): debug_print(f"[BULK_TAG] New tags for doc {doc_id}: {new_tags}") - # Update document - update_document( + # Update document; search chunks and blob tags sync in the background + update_result = update_document( document_id=doc_id, user_id=user_id, tags=new_tags - ) - - # Propagate to chunks - try: - propagate_tags_to_chunks(doc_id, new_tags, user_id) - except Exception as propagate_error: - debug_print(f"Warning: Failed to propagate tags for doc {doc_id}: {propagate_error}") + ) or {} + search_syncs.append(update_result.get('search_sync')) results['success'].append({ 'document_id': doc_id, @@ -2090,24 +2072,31 @@ def api_bulk_tag_documents(): debug_print(f"[BULK_TAG] Successfully updated doc {doc_id}") except Exception as doc_error: - error_msg = str(doc_error) - debug_print(f"[BULK_TAG] Exception for doc {doc_id}: {error_msg}") - import traceback - traceback.print_exc() + log_event( + '[DOCUMENT_METADATA] Failed to update personal document tags', + {'document_id': doc_id, 'error_type': type(doc_error).__name__}, + level=logging.WARNING, + ) results['errors'].append({ 'document_id': doc_id, - 'error': error_msg + 'error': 'Unable to update tags for this document' }) # Invalidate cache if results['success']: invalidate_personal_search_cache(user_id) + results['search_sync'] = summarize_document_search_sync(search_syncs) status_code = 200 if not results['errors'] else 207 # Multi-Status return jsonify(results), status_code except Exception as e: - return jsonify({'error': str(e)}), 500 + log_event( + '[DOCUMENT_METADATA] Failed to apply personal bulk tag changes', + {'error_type': type(e).__name__}, + level=logging.ERROR, + ) + return jsonify({'error': 'Unable to update document tags'}), 500 @bp.route('/api/documents/tags/', methods=['PATCH']) @swagger_route(security=get_auth_security()) @@ -2140,7 +2129,7 @@ def api_update_tag(tag_name): from functions_documents import ( normalize_tag, validate_tag_color, validate_tags, get_documents, - update_document, propagate_tags_to_chunks + update_document, summarize_document_search_sync ) from functions_settings import get_user_settings, update_user_settings from utils_cache import invalidate_personal_search_cache @@ -2193,6 +2182,7 @@ def api_update_tag(tag_name): debug_print(f"[UPDATE_TAG] Processing {len(all_docs)} unique documents") updated_count = 0 + search_syncs = [] for doc in all_docs: if normalized_old_tag in doc.get('tags', []): @@ -2200,17 +2190,13 @@ def api_update_tag(tag_name): current_tags = doc['tags'] new_tags = [normalized_new_tag if t == normalized_old_tag else t for t in current_tags] - update_document( + # Search chunks and blob tags sync in the background + update_result = update_document( document_id=doc['id'], user_id=user_id, tags=new_tags - ) - - # Propagate to chunks - try: - propagate_tags_to_chunks(doc['id'], new_tags, user_id) - except Exception as propagate_error: - debug_print(f"Warning: Failed to propagate tags for doc {doc['id']}: {propagate_error}") + ) or {} + search_syncs.append(update_result.get('search_sync')) updated_count += 1 @@ -2243,7 +2229,8 @@ def api_update_tag(tag_name): debug_print(f"[UPDATE_TAG] Rename completed successfully") return jsonify({ 'message': f'Tag renamed from "{normalized_old_tag}" to "{normalized_new_tag}"', - 'documents_updated': updated_count + 'documents_updated': updated_count, + 'search_sync': summarize_document_search_sync(search_syncs), }), 200 # Handle color change only @@ -2290,10 +2277,12 @@ def api_update_tag(tag_name): return jsonify({'error': 'No updates specified'}), 400 except Exception as e: - debug_print(f"[UPDATE_TAG] ERROR: {str(e)}") - import traceback - traceback.print_exc() - return jsonify({'error': str(e)}), 500 + log_event( + '[DOCUMENT_METADATA] Failed to update a personal tag', + {'error_type': type(e).__name__}, + level=logging.ERROR, + ) + return jsonify({'error': 'Unable to update the tag'}), 500 @bp.route('/api/documents/tags/', methods=['DELETE']) @swagger_route(security=get_auth_security()) @@ -2307,8 +2296,7 @@ def api_delete_tag(tag_name): return jsonify({'error': 'User not authenticated'}), 401 from functions_documents import ( - normalize_tag, update_document, - propagate_tags_to_chunks + normalize_tag, update_document, summarize_document_search_sync ) from functions_settings import get_user_settings, update_user_settings @@ -2346,23 +2334,20 @@ def api_delete_tag(tag_name): debug_print(f"[DELETE_TAG] Processing {len(all_docs)} unique documents") updated_count = 0 + search_syncs = [] for doc in all_docs: if normalized_tag in doc.get('tags', []): # Remove tag new_tags = [t for t in doc['tags'] if t != normalized_tag] - update_document( + # Search chunks and blob tags sync in the background + update_result = update_document( document_id=doc['id'], user_id=user_id, tags=new_tags - ) - - # Propagate to chunks - try: - propagate_tags_to_chunks(doc['id'], new_tags, user_id) - except Exception as propagate_error: - debug_print(f"Warning: Failed to propagate tags for doc {doc['id']}: {propagate_error}") + ) or {} + search_syncs.append(update_result.get('search_sync')) updated_count += 1 @@ -2382,11 +2367,17 @@ def api_delete_tag(tag_name): invalidate_personal_search_cache(user_id) return jsonify({ - 'message': f'Tag "{normalized_tag}" deleted from {updated_count} document(s)' + 'message': f'Tag "{normalized_tag}" deleted from {updated_count} document(s)', + 'search_sync': summarize_document_search_sync(search_syncs), }), 200 except Exception as e: - return jsonify({'error': str(e)}), 500 + log_event( + '[DOCUMENT_METADATA] Failed to delete a personal tag', + {'error_type': type(e).__name__}, + level=logging.ERROR, + ) + return jsonify({'error': 'Unable to delete the tag'}), 500 # ============= DOCUMENT SHARING API ENDPOINTS ============= @bp.route('/api/documents//share', methods=['POST']) @@ -2478,14 +2469,17 @@ def api_unshare_document(document_id): return jsonify({'message': 'Document unshared successfully'}), 200 else: return jsonify({'error': 'Failed to unshare document'}), 500 - except DocumentSearchAclProjectionDeferredError as exc: - response = jsonify({'error': str(exc)}) - response.headers['Retry-After'] = '150' - return response, 503 + except (DocumentSearchAclProjectionDeferredError, DocumentSearchAclProjectionError) as exc: + return _document_search_acl_error_response(exc) except exceptions.CosmosResourceNotFoundError: return jsonify({'error': 'Document not found or access denied'}), 404 except Exception as e: - return jsonify({'error': f'Error unsharing document: {str(e)}'}), 500 + log_event( + '[DOCUMENT_SHARING] Failed to unshare a personal document', + {'document_id': document_id, 'error_type': type(e).__name__}, + level=logging.ERROR, + ) + return jsonify({'error': 'Unable to unshare the document'}), 500 @bp.route('/api/documents//shared-users', methods=['GET']) @swagger_route(security=get_auth_security()) @@ -2601,11 +2595,17 @@ def api_remove_self_from_document(document_id): return jsonify({'message': 'Successfully removed from shared document'}), 200 else: return jsonify({'error': 'Failed to remove from shared document'}), 500 + except (DocumentSearchAclProjectionDeferredError, DocumentSearchAclProjectionError) as exc: + return _document_search_acl_error_response(exc) except exceptions.CosmosResourceNotFoundError: return jsonify({'error': 'Document not found or access denied'}), 404 except Exception as e: - debug_print(f"[ERROR] /api/documents/{document_id}/remove-self: {e}", flush=True) - return jsonify({'error': f'Error removing from shared document: {str(e)}'}), 500 + log_event( + '[DOCUMENT_SHARING] Failed to remove the user from a shared personal document', + {'document_id': document_id, 'error_type': type(e).__name__}, + level=logging.ERROR, + ) + return jsonify({'error': 'Unable to remove you from the shared document'}), 500 @bp.route('/api/documents//approve-share', methods=['POST']) @swagger_route(security=get_auth_security()) @@ -2637,6 +2637,15 @@ def api_approve_shared_document(document_id): else: new_shared_user_ids.append(entry) if updated: + # Project the approval to every chunk before saving it, so a failed search update + # leaves the share pending and a retry can complete it. A held document has no + # released chunks; publication rebuilds them from the recorded list. + if document_is_available(document_item): + project_document_acl_to_chunks( + document_id, + {'shared_user_ids': new_shared_user_ids}, + document_item.get('user_id'), + ) document_item['shared_user_ids'] = new_shared_user_ids document_item['last_updated'] = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') persisted_document = cosmos_user_documents_container.upsert_item(document_item) @@ -2645,25 +2654,9 @@ def api_approve_shared_document(document_id): document_for_sync, operation='document_share_approved', ) - # Update all chunks with the new shared_user_ids - try: - chunks = get_all_chunks(document_id, document_for_sync.get('user_id')) - for chunk in chunks: - chunk_id = chunk.get('id') - if chunk_id: - try: - update_chunk_metadata( - chunk_id=chunk_id, - user_id=document_for_sync.get('user_id'), - group_id=None, - public_workspace_id=None, - document_id=document_id, - shared_user_ids=new_shared_user_ids - ) - except Exception as chunk_e: - debug_print(f"Warning: Failed to update chunk {chunk_id}: {chunk_e}") - except Exception as e: - debug_print(f"Warning: Failed to update chunks for document {document_id}: {e}") + elif f"{user_id},approved" in shared_user_ids: + # Approving again completes an approval that was saved before its search update finished. + reproject_document_search_acl(document_item, user_id) # Invalidate cache for user who approved (their search results changed) if updated: @@ -2672,7 +2665,18 @@ def api_approve_shared_document(document_id): invalidate_personal_search_cache(user_id) return jsonify({'message': 'Share approved' if updated else 'Already approved'}), 200 + except ( + DocumentSearchAclProjectionDeferredError, + DocumentSearchAclProjectionError, + DocumentMutationPropagationError, + ) as exc: + return _document_search_acl_error_response(exc) except exceptions.CosmosResourceNotFoundError: return jsonify({'error': 'Document not found or access denied'}), 404 except Exception as e: - return jsonify({'error': f'Error approving shared document: {str(e)}'}), 500 + log_event( + '[DOCUMENT_SHARING] Failed to approve a shared personal document', + {'document_id': document_id, 'error_type': type(e).__name__}, + level=logging.ERROR, + ) + return jsonify({'error': 'Unable to approve the shared document'}), 500 diff --git a/application/single_app/route_backend_group_documents.py b/application/single_app/route_backend_group_documents.py index 2cc3f7dfd..289a985e3 100644 --- a/application/single_app/route_backend_group_documents.py +++ b/application/single_app/route_backend_group_documents.py @@ -339,6 +339,15 @@ def _remove_group_share_entries(shared_group_ids, group_id): ] +def _group_share_search_acl_error_response(error): + """Return the safe response for a sharing change the search index could not apply.""" + message, status_code, retry_after_seconds = describe_document_search_acl_error(error) + response = jsonify({'error': message}) + if retry_after_seconds: + response.headers['Retry-After'] = str(retry_after_seconds) + return response, status_code + + def _get_group_name(group_doc, fallback='Unknown Group'): return str((group_doc or {}).get('name') or fallback).strip() @@ -1455,79 +1464,40 @@ def api_patch_group_document(document_id): if error_response: return error_response - data = request.get_json() + data = request.get_json() or {} # Track which fields were updated updated_fields = {} + document_updates = {} try: if 'title' in data: - update_document( - document_id=document_id, - group_id=active_group_id, - user_id=user_id, - title=data['title'] - ) + document_updates['title'] = data['title'] updated_fields['title'] = data['title'] if 'abstract' in data: - update_document( - document_id=document_id, - group_id=active_group_id, - user_id=user_id, - abstract=data['abstract'] - ) + document_updates['abstract'] = data['abstract'] updated_fields['abstract'] = data['abstract'] if 'keywords' in data: if isinstance(data['keywords'], list): - update_document( - document_id=document_id, - group_id=active_group_id, - user_id=user_id, - keywords=data['keywords'] - ) + document_updates['keywords'] = data['keywords'] updated_fields['keywords'] = data['keywords'] else: keywords_list = [kw.strip() for kw in data['keywords'].split(',')] - update_document( - document_id=document_id, - group_id=active_group_id, - user_id=user_id, - keywords=keywords_list - ) + document_updates['keywords'] = keywords_list updated_fields['keywords'] = keywords_list if 'publication_date' in data: - update_document( - document_id=document_id, - group_id=active_group_id, - user_id=user_id, - publication_date=data['publication_date'] - ) + document_updates['publication_date'] = data['publication_date'] updated_fields['publication_date'] = data['publication_date'] if 'document_classification' in data: - update_document( - document_id=document_id, - group_id=active_group_id, - user_id=user_id, - document_classification=data['document_classification'] - ) + document_updates['document_classification'] = data['document_classification'] updated_fields['document_classification'] = data['document_classification'] if 'authors' in data: if isinstance(data['authors'], list): - update_document( - document_id=document_id, - group_id=active_group_id, - user_id=user_id, - authors=data['authors'] - ) + document_updates['authors'] = data['authors'] updated_fields['authors'] = data['authors'] else: authors_list = [data['authors']] - update_document( - document_id=document_id, - group_id=active_group_id, - user_id=user_id, - authors=authors_list - ) + document_updates['authors'] = authors_list updated_fields['authors'] = authors_list if 'tags' in data: @@ -1538,13 +1508,18 @@ def api_patch_group_document(document_id): return jsonify({'error': error_msg}), 400 for tag in normalized_tags: get_or_create_tag_definition(user_id, tag, workspace_type='group', group_id=active_group_id) - update_document( + document_updates['tags'] = normalized_tags + updated_fields['tags'] = normalized_tags + + # Save every change in one update; search chunks are synced in the background + update_result = {} + if document_updates: + update_result = update_document( document_id=document_id, group_id=active_group_id, user_id=user_id, - tags=normalized_tags - ) - updated_fields['tags'] = normalized_tags + **document_updates, + ) or {} # Log the metadata update transaction if any fields were updated if updated_fields: @@ -1573,9 +1548,18 @@ def api_patch_group_document(document_id): group_id=active_group_id ) - return jsonify({'message': 'Group document metadata updated successfully'}), 200 + response_payload = {'message': 'Group document metadata updated successfully'} + if isinstance(update_result, dict) and update_result.get('search_sync'): + response_payload['search_sync'] = update_result['search_sync'] + + return jsonify(response_payload), 200 except Exception as e: - return jsonify({'error': str(e)}), 500 + log_event( + '[DOCUMENT_METADATA] Failed to update group document metadata', + {'document_id': document_id, 'group_id': active_group_id, 'error_type': type(e).__name__}, + level=logging.ERROR, + ) + return jsonify({'error': 'Unable to update document metadata'}), 500 @bp.route('/api/group_documents/', methods=['DELETE']) @swagger_route(security=get_auth_security()) @@ -2038,10 +2022,25 @@ def api_approve_shared_group_document(document_id): ) # Invalidate cache for the group that approved invalidate_group_search_cache(active_group_id) + else: + # Approving again completes an approval that was saved before its search update finished. + reproject_document_search_acl(document_item, user_id, group_id=document_item.get('group_id')) return jsonify({'message': 'Share approved' if updated else 'Already approved'}), 200 + except ( + DocumentSearchAclProjectionDeferredError, + DocumentSearchAclProjectionError, + DocumentMutationPropagationError, + ScreeningConflictError, + ) as exc: + return _group_share_search_acl_error_response(exc) except Exception as e: - return jsonify({'error': f'Error approving shared document: {str(e)}'}), 500 + log_event( + '[DOCUMENT_SHARING] Failed to approve a shared group document', + {'document_id': document_id, 'group_id': active_group_id, 'error_type': type(e).__name__}, + level=logging.ERROR, + ) + return jsonify({'error': 'Unable to approve the shared document'}), 500 @bp.route('/api/group_documents//approve-generated-artifact', methods=['POST']) @swagger_route(security=get_auth_security()) @@ -2426,8 +2425,20 @@ def api_share_document_with_group(document_id): 'document_id': document_id, 'shared_with_group': target_group_id }), 200 + except ( + DocumentSearchAclProjectionDeferredError, + DocumentSearchAclProjectionError, + DocumentMutationPropagationError, + ScreeningConflictError, + ) as exc: + return _group_share_search_acl_error_response(exc) except Exception as e: - return jsonify({'error': f'Error sharing document: {str(e)}'}), 500 + log_event( + '[DOCUMENT_SHARING] Failed to share a group document', + {'document_id': document_id, 'group_id': active_group_id, 'error_type': type(e).__name__}, + level=logging.ERROR, + ) + return jsonify({'error': 'Unable to share the document'}), 500 @bp.route('/api/group_documents//unshare-with-group', methods=['DELETE']) @swagger_route(security=get_auth_security()) @@ -2495,8 +2506,20 @@ def api_unshare_document_with_group(document_id): 'document_id': document_id, 'unshared_with_group': target_group_id }), 200 + except ( + DocumentSearchAclProjectionDeferredError, + DocumentSearchAclProjectionError, + DocumentMutationPropagationError, + ScreeningConflictError, + ) as exc: + return _group_share_search_acl_error_response(exc) except Exception as e: - return jsonify({'error': f'Error unsharing document: {str(e)}'}), 500 + log_event( + '[DOCUMENT_SHARING] Failed to unshare a group document', + {'document_id': document_id, 'group_id': active_group_id, 'error_type': type(e).__name__}, + level=logging.ERROR, + ) + return jsonify({'error': 'Unable to remove document sharing'}), 500 @bp.route('/api/group_documents//remove-self', methods=['DELETE']) @swagger_route(security=get_auth_security()) @@ -2572,8 +2595,20 @@ def api_remove_self_from_group_document(document_id): return jsonify({ 'message': 'Share denied' if was_pending else 'Successfully removed group from shared document' }), 200 + except ( + DocumentSearchAclProjectionDeferredError, + DocumentSearchAclProjectionError, + DocumentMutationPropagationError, + ScreeningConflictError, + ) as exc: + return _group_share_search_acl_error_response(exc) except Exception as e: - return jsonify({'error': f'Error removing group from shared document: {str(e)}'}), 500 + log_event( + '[DOCUMENT_SHARING] Failed to remove a group from a shared document', + {'document_id': document_id, 'group_id': active_group_id, 'error_type': type(e).__name__}, + level=logging.ERROR, + ) + return jsonify({'error': 'Unable to remove the group from the shared document'}), 500 @bp.route('/api/group_documents/tags', methods=['GET']) @swagger_route(security=get_auth_security()) @@ -2757,7 +2792,7 @@ def api_bulk_tag_group_documents(): from functions_documents import ( validate_tags, update_document, - propagate_tags_to_chunks, get_or_create_tag_definition + get_or_create_tag_definition ) is_valid, error_msg, normalized_tags = validate_tags(tags_input) @@ -2807,6 +2842,7 @@ def api_bulk_tag_group_documents(): elif action == 'set_tags': new_tags = normalized_tags + # Search chunks and blob tags sync in the background update_document( document_id=doc_id, group_id=active_group_id, @@ -2814,11 +2850,6 @@ def api_bulk_tag_group_documents(): tags=new_tags ) - try: - propagate_tags_to_chunks(doc_id, new_tags, user_id, group_id=active_group_id) - except Exception: - pass - results['success'].append({ 'document_id': doc_id, 'tags': new_tags @@ -2870,7 +2901,7 @@ def api_update_group_tag(tag_name): new_name = data.get('new_name') new_color = data.get('color') - from functions_documents import normalize_tag, validate_tag_color, validate_tags, update_document, propagate_tags_to_chunks + from functions_documents import normalize_tag, validate_tag_color, validate_tags, update_document try: normalized_old_tag = normalize_tag(tag_name) @@ -2917,6 +2948,7 @@ def rename_definition(definitions): current_tags = doc['tags'] new_tags = [normalized_new_tag if t == normalized_old_tag else t for t in current_tags] + # Search chunks and blob tags sync in the background update_document( document_id=doc['id'], group_id=active_group_id, @@ -2924,11 +2956,6 @@ def rename_definition(definitions): tags=new_tags ) - try: - propagate_tags_to_chunks(doc['id'], new_tags, user_id, group_id=active_group_id) - except Exception: - pass - updated_count += 1 invalidate_group_search_cache(active_group_id) @@ -2993,7 +3020,7 @@ def api_delete_group_tag(tag_name): if error_response: return error_response - from functions_documents import normalize_tag, update_document, propagate_tags_to_chunks + from functions_documents import normalize_tag, update_document try: normalized_tag = normalize_tag(tag_name) @@ -3031,6 +3058,7 @@ def remove_definition(definitions): if normalized_tag in doc.get('tags', []): new_tags = [t for t in doc['tags'] if t != normalized_tag] + # Search chunks and blob tags sync in the background update_document( document_id=doc['id'], group_id=active_group_id, @@ -3038,11 +3066,6 @@ def remove_definition(definitions): tags=new_tags ) - try: - propagate_tags_to_chunks(doc['id'], new_tags, user_id, group_id=active_group_id) - except Exception: - pass - updated_count += 1 if updated_count > 0: diff --git a/application/single_app/route_backend_public_documents.py b/application/single_app/route_backend_public_documents.py index 9801779e7..6dab14164 100644 --- a/application/single_app/route_backend_public_documents.py +++ b/application/single_app/route_backend_public_documents.py @@ -741,27 +741,28 @@ def api_patch_public_document(doc_id): # Track which fields were updated updated_fields = {} + document_updates = {} try: if 'title' in data: - update_document(document_id=doc_id, public_workspace_id=active_ws, user_id=user_id, title=data['title']) + document_updates['title'] = data['title'] updated_fields['title'] = data['title'] if 'abstract' in data: - update_document(document_id=doc_id, public_workspace_id=active_ws, user_id=user_id, abstract=data['abstract']) + document_updates['abstract'] = data['abstract'] updated_fields['abstract'] = data['abstract'] if 'keywords' in data: kws = data['keywords'] if isinstance(data['keywords'],list) else [k.strip() for k in data['keywords'].split(',')] - update_document(document_id=doc_id, public_workspace_id=active_ws, user_id=user_id, keywords=kws) + document_updates['keywords'] = kws updated_fields['keywords'] = kws if 'authors' in data: auths = data['authors'] if isinstance(data['authors'],list) else [data['authors']] - update_document(document_id=doc_id, public_workspace_id=active_ws, user_id=user_id, authors=auths) + document_updates['authors'] = auths updated_fields['authors'] = auths if 'publication_date' in data: - update_document(document_id=doc_id, public_workspace_id=active_ws, user_id=user_id, publication_date=data['publication_date']) + document_updates['publication_date'] = data['publication_date'] updated_fields['publication_date'] = data['publication_date'] if 'document_classification' in data: - update_document(document_id=doc_id, public_workspace_id=active_ws, user_id=user_id, document_classification=data['document_classification']) + document_updates['document_classification'] = data['document_classification'] updated_fields['document_classification'] = data['document_classification'] if 'tags' in data: from functions_documents import validate_tags, get_or_create_tag_definition @@ -771,9 +772,19 @@ def api_patch_public_document(doc_id): return jsonify({'error': error_msg}), 400 for tag in normalized_tags: get_or_create_tag_definition(user_id, tag, workspace_type='public', public_workspace_id=active_ws) - update_document(document_id=doc_id, public_workspace_id=active_ws, user_id=user_id, tags=normalized_tags) + document_updates['tags'] = normalized_tags updated_fields['tags'] = normalized_tags + # Save every change in one update; search chunks are synced in the background + update_result = {} + if document_updates: + update_result = update_document( + document_id=doc_id, + public_workspace_id=active_ws, + user_id=user_id, + **document_updates, + ) or {} + # Log the metadata update transaction if any fields were updated if updated_fields: from functions_documents import get_document @@ -790,9 +801,17 @@ def api_patch_public_document(doc_id): public_workspace_id=active_ws ) - return jsonify({'message':'Metadata updated'}), 200 + response_payload = {'message': 'Metadata updated'} + if isinstance(update_result, dict) and update_result.get('search_sync'): + response_payload['search_sync'] = update_result['search_sync'] + return jsonify(response_payload), 200 except Exception as e: - return jsonify({'error': str(e)}), 500 + log_event( + '[DOCUMENT_METADATA] Failed to update public document metadata', + {'document_id': doc_id, 'public_workspace_id': active_ws, 'error_type': type(e).__name__}, + level=logging.ERROR, + ) + return jsonify({'error': 'Unable to update document metadata'}), 500 @bp.route('/api/public_documents/', methods=['DELETE']) @swagger_route(security=get_auth_security()) @@ -1485,7 +1504,7 @@ def api_bulk_tag_public_documents(): from functions_documents import ( validate_tags, update_document, - propagate_tags_to_chunks, get_or_create_tag_definition + get_or_create_tag_definition ) is_valid, error_msg, normalized_tags = validate_tags(tags_input) @@ -1535,6 +1554,7 @@ def api_bulk_tag_public_documents(): elif action == 'set_tags': new_tags = normalized_tags + # Search chunks and blob tags sync in the background update_document( document_id=doc_id, public_workspace_id=active_ws, @@ -1542,11 +1562,6 @@ def api_bulk_tag_public_documents(): tags=new_tags ) - try: - propagate_tags_to_chunks(doc_id, new_tags, user_id, public_workspace_id=active_ws) - except Exception: - pass - results['success'].append({ 'document_id': doc_id, 'tags': new_tags @@ -1599,7 +1614,7 @@ def api_update_public_workspace_tag(tag_name): new_name = data.get('new_name') new_color = data.get('color') - from functions_documents import normalize_tag, validate_tag_color, validate_tags, update_document, propagate_tags_to_chunks + from functions_documents import normalize_tag, validate_tag_color, validate_tags, update_document try: normalized_old_tag = normalize_tag(tag_name) @@ -1646,6 +1661,7 @@ def rename_definition(definitions): current_tags = doc['tags'] new_tags = [normalized_new_tag if t == normalized_old_tag else t for t in current_tags] + # Search chunks and blob tags sync in the background update_document( document_id=doc['id'], public_workspace_id=active_ws, @@ -1653,11 +1669,6 @@ def rename_definition(definitions): tags=new_tags ) - try: - propagate_tags_to_chunks(doc['id'], new_tags, user_id, public_workspace_id=active_ws) - except Exception: - pass - updated_count += 1 invalidate_public_workspace_search_cache(active_ws) @@ -1723,7 +1734,7 @@ def api_delete_public_workspace_tag(tag_name): return jsonify({'error': 'You do not have permission to manage tags'}), 403 return error_response - from functions_documents import normalize_tag, update_document, propagate_tags_to_chunks + from functions_documents import normalize_tag, update_document try: normalized_tag = normalize_tag(tag_name) @@ -1761,6 +1772,7 @@ def remove_definition(definitions): if normalized_tag in doc.get('tags', []): new_tags = [t for t in doc['tags'] if t != normalized_tag] + # Search chunks and blob tags sync in the background update_document( document_id=doc['id'], public_workspace_id=active_ws, @@ -1768,11 +1780,6 @@ def remove_definition(definitions): tags=new_tags ) - try: - propagate_tags_to_chunks(doc['id'], new_tags, user_id, public_workspace_id=active_ws) - except Exception: - pass - updated_count += 1 if updated_count > 0: diff --git a/application/single_app/route_external_public_documents.py b/application/single_app/route_external_public_documents.py index 4936bc240..85c3a1101 100644 --- a/application/single_app/route_external_public_documents.py +++ b/application/single_app/route_external_public_documents.py @@ -425,77 +425,48 @@ def external_patch_public_document(document_id): # Track which fields were updated updated_fields = {} + document_updates = {} try: if 'title' in data: - update_document( - document_id=document_id, - public_workspace_id=active_workspace_id, - user_id=user_id, - title=data['title'] - ) + document_updates['title'] = data['title'] updated_fields['title'] = data['title'] if 'abstract' in data: - update_document( - document_id=document_id, - public_workspace_id=active_workspace_id, - user_id=user_id, - abstract=data['abstract'] - ) + document_updates['abstract'] = data['abstract'] updated_fields['abstract'] = data['abstract'] if 'keywords' in data: if isinstance(data['keywords'], list): - update_document( - document_id=document_id, - public_workspace_id=active_workspace_id, - user_id=user_id, - keywords=data['keywords'] - ) + document_updates['keywords'] = data['keywords'] updated_fields['keywords'] = data['keywords'] else: keywords_list = [kw.strip() for kw in data['keywords'].split(',')] - update_document( - document_id=document_id, - public_workspace_id=active_workspace_id, - user_id=user_id, - keywords=keywords_list - ) + document_updates['keywords'] = keywords_list updated_fields['keywords'] = keywords_list if 'publication_date' in data: - update_document( - document_id=document_id, - public_workspace_id=active_workspace_id, - user_id=user_id, - publication_date=data['publication_date'] - ) + document_updates['publication_date'] = data['publication_date'] updated_fields['publication_date'] = data['publication_date'] if 'document_classification' in data: - update_document( - document_id=document_id, - public_workspace_id=active_workspace_id, - user_id=user_id, - document_classification=data['document_classification'] - ) + document_updates['document_classification'] = data['document_classification'] updated_fields['document_classification'] = data['document_classification'] if 'authors' in data: if isinstance(data['authors'], list): - update_document( - document_id=document_id, - public_workspace_id=active_workspace_id, - user_id=user_id, - authors=data['authors'] - ) + document_updates['authors'] = data['authors'] updated_fields['authors'] = data['authors'] else: authors_list = [data['authors']] - update_document( - document_id=document_id, - public_workspace_id=active_workspace_id, - user_id=user_id, - authors=authors_list - ) + document_updates['authors'] = authors_list updated_fields['authors'] = authors_list + # Save every change in one update; search chunks are synced in the background + update_result = {} + if document_updates: + update_result = update_document( + document_id=document_id, + public_workspace_id=active_workspace_id, + user_id=user_id, + **document_updates, + ) or {} + # Log the metadata update transaction if any fields were updated if updated_fields: doc_response = get_document(user_id, document_id, public_workspace_id=active_workspace_id) @@ -520,9 +491,17 @@ def external_patch_public_document(document_id): public_workspace_id=active_workspace_id ) - return jsonify({'message': 'Public document metadata updated successfully'}), 200 + response_payload = {'message': 'Public document metadata updated successfully'} + if isinstance(update_result, dict) and update_result.get('search_sync'): + response_payload['search_sync'] = update_result['search_sync'] + return jsonify(response_payload), 200 except Exception as e: - return jsonify({'error': str(e)}), 500 + log_event( + '[DOCUMENT_METADATA] Failed to update public document metadata through the external API', + {'document_id': document_id, 'public_workspace_id': active_workspace_id, 'error_type': type(e).__name__}, + level=logging.ERROR, + ) + return jsonify({'error': 'Unable to update document metadata'}), 500 diff --git a/application/single_app/static/js/public/public_workspace.js b/application/single_app/static/js/public/public_workspace.js index ccf546200..9c1c2d6c3 100644 --- a/application/single_app/static/js/public/public_workspace.js +++ b/application/single_app/static/js/public/public_workspace.js @@ -3059,6 +3059,9 @@ async function onSavePublicDocMetadata(e) { const updatedDoc = await response.json(); publicDocMetadataModal.hide(); + if (updatedDoc?.search_sync?.status === "pending" && typeof window.showToast === "function") { + window.showToast("Metadata saved. Search and chat results will reflect the change once the search index finishes updating.", "info"); + } fetchPublicDocs(); // Refresh the table loadPublicWorkspaceTags(); // Refresh tag counts } catch (err) { diff --git a/application/single_app/static/js/workspace/workspace-documents.js b/application/single_app/static/js/workspace/workspace-documents.js index 7c95d6b08..5abb199e9 100644 --- a/application/single_app/static/js/workspace/workspace-documents.js +++ b/application/single_app/static/js/workspace/workspace-documents.js @@ -1445,6 +1445,9 @@ if (docMetadataForm && docMetadataModalEl) { // Check both exist .then(r => r.ok ? r.json() : r.json().then(err => Promise.reject(err))) .then(updatedDoc => { if (docMetadataModalEl) docMetadataModalEl.hide(); + if (updatedDoc?.search_sync?.status === "pending" && typeof window.showToast === "function") { + window.showToast("Metadata saved. Search and chat results will reflect the change once the search index finishes updating.", "info"); + } fetchUserDocuments(); // Refresh the table loadWorkspaceTags(); // Refresh tag counts and grid view }) diff --git a/application/single_app/templates/group_workspaces.html b/application/single_app/templates/group_workspaces.html index 712f26dc9..b3c13905a 100644 --- a/application/single_app/templates/group_workspaces.html +++ b/application/single_app/templates/group_workspaces.html @@ -6749,6 +6749,9 @@