diff --git a/application/single_app/background_tasks.py b/application/single_app/background_tasks.py index 6d3f7c530..54564f0d7 100644 --- a/application/single_app/background_tasks.py +++ b/application/single_app/background_tasks.py @@ -34,6 +34,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, @@ -881,6 +882,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_app_maintenance_loop(): """Run idempotent app maintenance tasks under a distributed lock.""" while True: @@ -956,6 +987,7 @@ def start_background_task_threads(app=None): ('File Sync scheduler background task started.', run_file_sync_scheduler_loop), ('Tabular generated-output scheduler background task started.', run_tabular_generated_output_scheduler_loop), ('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), ('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 632b5360e..d5f127d2a 100644 --- a/application/single_app/config.py +++ b/application/single_app/config.py @@ -98,7 +98,7 @@ EXECUTOR_TYPE = 'thread' EXECUTOR_MAX_WORKERS = 30 SESSION_TYPE = 'filesystem' -VERSION = "0.261.051" +VERSION = "0.261.052" 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 d690a6fbb..e4adf2677 100644 --- a/application/single_app/functions_documents.py +++ b/application/single_app/functions_documents.py @@ -7,8 +7,10 @@ import traceback import zipfile import hashlib +from concurrent.futures import ThreadPoolExecutor from io import BytesIO from flask import make_response +from azure.core import MatchConditions from azure.core.exceptions import ResourceExistsError from config import * from functions_appinsights import log_event @@ -73,6 +75,14 @@ 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 XsdIngestionCapabilityError(RuntimeError): """Raised when an XSD cannot be accepted without exact-source storage.""" @@ -3303,24 +3313,17 @@ 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 is projected by a durable background sync, so the save does + no per-chunk work. Access-control changes are projected to the chunks before they are saved. + Returns {'updated': bool, 'search_sync': {...}}. + """ document_id = kwargs.get('document_id') user_id = kwargs.get('user_id') group_id = kwargs.get('group_id') public_workspace_id = kwargs.get('public_workspace_id') num_chunks_increment = kwargs.pop('num_chunks_increment', 0) - chunk_sync_limit = kwargs.pop('chunk_sync_limit', None) - chunk_sync_offset = max(int(kwargs.pop('chunk_sync_offset', 0) or 0), 0) - requested_chunk_sync_fields = set(kwargs.pop('chunk_sync_fields', []) or []) - supported_chunk_sync_fields = {'title', 'authors', 'file_name', 'document_classification', 'tags', 'shared_group_ids'} - requested_chunk_sync_fields = requested_chunk_sync_fields.intersection(supported_chunk_sync_fields) - chunk_sync_result = { - 'required': False, - 'complete': True, - 'fields': [], - 'total': 0, - 'processed': 0, - 'next_offset': None, - } if not document_id or not user_id: # Cannot proceed without these identifiers @@ -3418,7 +3421,9 @@ def update_document(**kwargs): # 2. Apply updates from kwargs update_occurred = False - updated_fields_requiring_chunk_sync = set() # Track fields needing propagation + metadata_fields_requiring_sync = set() + acl_fields_requiring_projection = {} + 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) @@ -3431,17 +3436,22 @@ def update_document(**kwargs): ) for key, value in kwargs.items(): - 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_fields_requiring_projection[key] = value # 3. If any update happened, handle timestamps and percentage if update_occurred: @@ -3473,83 +3483,41 @@ def update_document(**kwargs): else: existing_document['percentage_complete'] = new_percentage - updated_fields_requiring_chunk_sync.update(requested_chunk_sync_fields) - - # 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. - if updated_fields_requiring_chunk_sync: - try: - chunks_to_update = get_all_chunks( - document_id, - user_id, - group_id=group_id, - public_workspace_id=public_workspace_id - ) - total_chunks = len(chunks_to_update) - chunk_sync_result['required'] = total_chunks > 0 - chunk_sync_result['fields'] = sorted(updated_fields_requiring_chunk_sync) - chunk_sync_result['total'] = total_chunks - - if chunk_sync_limit is None: - chunk_sync_end = total_chunks - else: - chunk_sync_limit = max(int(chunk_sync_limit), 1) - chunk_sync_end = min(total_chunks, chunk_sync_offset + chunk_sync_limit) - - for chunk in chunks_to_update[chunk_sync_offset:chunk_sync_end]: - 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 - } - - # 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) - chunk_sync_result['processed'] += 1 - - if chunk_sync_end < total_chunks: - chunk_sync_result['complete'] = False - chunk_sync_result['next_offset'] = chunk_sync_end - 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 {chunk_sync_result['processed']} of {total_chunks} search chunks." - ) - except Exception as 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 to the search chunks before saving them, so the index + # never grants access that Cosmos does not record and a revocation is enforced first. + if acl_fields_requiring_projection: + projected_chunk_count = project_document_acl_to_chunks( + document_id, + acl_fields_requiring_projection, + user_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"Projected access changes for fields {sorted(acl_fields_requiring_projection)} to {projected_chunk_count} search chunks." + ) + # 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': 'not_required'} + if metadata_fields_requiring_sync: + 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 + # 6. Upsert the document if changes were made if update_occurred: _upsert_document_and_sync_access_index( cosmos_container, @@ -3557,7 +3525,19 @@ def update_document(**kwargs): operation='document_updated', ) - return {'updated': update_occurred, 'chunk_sync': chunk_sync_result} + if metadata_fields_requiring_sync: + 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']}." + ) + + return {'updated': update_occurred, 'search_sync': search_sync} except CosmosResourceNotFoundError as e: # Error already logged where it was first detected @@ -4230,6 +4210,758 @@ def update_chunk_metadata(chunk_id, user_id, group_id=None, public_workspace_id= raise +# Search chunk metadata 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 fields are different: they are projected synchronously, before the change is +# saved, 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_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', +} +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_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 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, +): + """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. + 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 + + 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], + ) + 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): + """Project access-control fields to every active chunk before the access change is saved. + + Raises DocumentSearchAclProjectionDeferredError while target Search writes are frozen and + DocumentSearchAclProjectionError for any other failure, so callers never save 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, + ) + except DataManagementSearchWritesFrozenError as exc: + raise DocumentSearchAclProjectionDeferredError(DOCUMENT_SEARCH_ACL_DEFERRED_MESSAGE) from exc + except Exception as exc: + log_event( + "[DOCUMENT_SEARCH_SYNC] Search ACL projection failed; the access change was not saved.", + 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 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 _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 _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 _read_document_search_sync_record(record_id): + try: + return cosmos_settings_container.read_item(item=record_id, partition_key=record_id) + except CosmosResourceNotFoundError: + return None + + +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. 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): + 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 revision 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 CosmosResourceNotFoundError: + return 'complete', None + except Exception as exc: + 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 CosmosResourceNotFoundError: + return + + +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 CosmosResourceNotFoundError: + return None + 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 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. Failures are + recorded on the record with a retry backoff. 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} + + # Personal chunks are scoped by the owner; group and public chunks by their workspace. + chunks_updated = project_fields_to_document_chunks( + document_id, + build_document_search_metadata_fields(document_item, pending_fields), + document_item.get('user_id') or record.get('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: + propagate_tags_to_blob_metadata( + document_id, + document_item.get('tags') or [], + document_item.get('user_id') or record.get('user_id'), + group_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 its 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. @@ -11461,27 +12193,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 @@ -11522,24 +12247,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') @@ -12306,50 +13018,3 @@ def propagate_tags_to_blob_metadata(document_id, tags, user_id, group_id=None, p except Exception as 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) - - except Exception as e: - print(f"Error propagating tags to chunks for document {document_id}: {e}") - raise \ No newline at end of file diff --git a/application/single_app/route_backend_documents.py b/application/single_app/route_backend_documents.py index d6d7ea773..8939e56b2 100644 --- a/application/single_app/route_backend_documents.py +++ b/application/single_app/route_backend_documents.py @@ -1045,94 +1045,78 @@ def api_patch_user_document(document_id): return jsonify({'error': 'User not authenticated'}), 401 data = request.get_json() or {} # new metadata values from the client - chunk_sync_state = data.get('_metadata_chunk_sync') if isinstance(data.get('_metadata_chunk_sync'), dict) else {} - try: - chunk_sync_offset = max(int(chunk_sync_state.get('next_offset', 0) or 0), 0) - except (TypeError, ValueError): - return jsonify({'error': 'Invalid metadata chunk sync state'}), 400 - chunk_sync_fields = chunk_sync_state.get('fields') if isinstance(chunk_sync_state.get('fields'), list) else [] - is_chunk_sync_continuation = chunk_sync_offset > 0 and bool(chunk_sync_fields) # 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: - 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 + 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 and chunk propagation - 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 + # 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 - # Save updates back to Cosmos - try: - update_result = {'chunk_sync': {'complete': True}} + # 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, - chunk_sync_limit=75, - chunk_sync_offset=chunk_sync_offset, - chunk_sync_fields=chunk_sync_fields, **document_updates, - ) or update_result - - if 'tags' in updated_fields and not is_chunk_sync_continuation: - try: - propagate_tags_to_blob_metadata(document_id, updated_fields['tags'], user_id) - except Exception as blob_metadata_error: - debug_print(f"Warning: Failed to propagate tags to blob metadata: {blob_metadata_error}") + ) or {} # Log the metadata update transaction if any fields were updated - if updated_fields and not is_chunk_sync_continuation: + if updated_fields: # Get document details for logging doc_response = get_document(user_id, document_id) doc = None @@ -1155,10 +1139,9 @@ def api_patch_user_document(document_id): file_type=doc.get('file_type') ) - chunk_sync = update_result.get('chunk_sync', {}) if isinstance(update_result, dict) else {} response_payload = {'message': 'Document metadata updated successfully'} - if chunk_sync.get('required'): - response_payload['metadata_chunk_sync'] = chunk_sync + 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: @@ -1630,7 +1613,7 @@ def api_bulk_tag_documents(): from functions_documents import ( validate_tags, get_document, update_document, - propagate_tags_to_chunks, get_or_create_tag_definition + get_or_create_tag_definition ) # Validate and normalize tags @@ -1703,19 +1686,13 @@ def api_bulk_tag_documents(): debug_print(f"[BULK_TAG] New tags for doc {doc_id}: {new_tags}") - # Update document + # Update document; search chunks and blob tags sync in the background 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}") - results['success'].append({ 'document_id': doc_id, 'tags': new_tags @@ -1773,7 +1750,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 ) from functions_settings import get_user_settings, update_user_settings from utils_cache import invalidate_personal_search_cache @@ -1833,18 +1810,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] + # Search chunks and blob tags sync in the background 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}") - updated_count += 1 debug_print(f"[UPDATE_TAG] Updated {updated_count} documents") @@ -1940,8 +1912,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 ) from functions_settings import get_user_settings, update_user_settings @@ -1985,18 +1956,13 @@ def api_delete_tag(tag_name): # Remove tag 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'], 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}") - updated_count += 1 # Remove tag definition @@ -2270,6 +2236,13 @@ 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. + 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) @@ -2278,25 +2251,6 @@ 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}") # Invalidate cache for user who approved (their search results changed) if updated: @@ -2305,7 +2259,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 as exc: + response = jsonify({'error': str(exc)}) + response.headers['Retry-After'] = '150' + return response, 503 + except DocumentSearchAclProjectionError as exc: + return jsonify({'error': str(exc)}), 500 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 448ddd315..e6b301d71 100644 --- a/application/single_app/route_backend_group_documents.py +++ b/application/single_app/route_backend_group_documents.py @@ -127,6 +127,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 enforce.""" + response = jsonify({'error': str(error)}) + if isinstance(error, DocumentSearchAclProjectionDeferredError): + response.headers['Retry-After'] = '150' + return response, 503 + return response, 500 + + def _get_group_name(group_doc, fallback='Unknown Group'): return str((group_doc or {}).get('name') or fallback).strip() @@ -942,79 +951,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: @@ -1025,13 +995,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: @@ -1060,9 +1035,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()) @@ -1527,6 +1511,8 @@ def api_approve_shared_group_document(document_id): invalidate_group_search_cache(active_group_id) return jsonify({'message': 'Share approved' if updated else 'Already approved'}), 200 + except (DocumentSearchAclProjectionDeferredError, DocumentSearchAclProjectionError) 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 @@ -1895,6 +1881,8 @@ def api_share_document_with_group(document_id): 'document_id': document_id, 'shared_with_group': target_group_id }), 200 + except (DocumentSearchAclProjectionDeferredError, DocumentSearchAclProjectionError) as exc: + return _group_share_search_acl_error_response(exc) except Exception as e: return jsonify({'error': f'Error sharing document: {str(e)}'}), 500 @@ -1964,6 +1952,8 @@ def api_unshare_document_with_group(document_id): 'document_id': document_id, 'unshared_with_group': target_group_id }), 200 + except (DocumentSearchAclProjectionDeferredError, DocumentSearchAclProjectionError) as exc: + return _group_share_search_acl_error_response(exc) except Exception as e: return jsonify({'error': f'Error unsharing document: {str(e)}'}), 500 @@ -2041,6 +2031,8 @@ 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) 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 @@ -2218,7 +2210,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) @@ -2268,6 +2260,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, @@ -2275,11 +2268,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 @@ -2331,7 +2319,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) @@ -2363,6 +2351,7 @@ def api_update_group_tag(tag_name): 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, @@ -2370,11 +2359,6 @@ def api_update_group_tag(tag_name): 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 tag_defs = group_doc.get('tag_definitions', {}) @@ -2442,7 +2426,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) @@ -2466,6 +2450,7 @@ def api_delete_group_tag(tag_name): 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, @@ -2473,11 +2458,6 @@ def api_delete_group_tag(tag_name): 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 tag_defs = group_doc.get('tag_definitions', {}) diff --git a/application/single_app/route_backend_public_documents.py b/application/single_app/route_backend_public_documents.py index 149bd43d0..1db97a0ae 100644 --- a/application/single_app/route_backend_public_documents.py +++ b/application/single_app/route_backend_public_documents.py @@ -660,27 +660,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 @@ -690,9 +691,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 @@ -709,9 +720,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()) @@ -1370,7 +1389,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) @@ -1420,6 +1439,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, @@ -1427,11 +1447,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 @@ -1484,7 +1499,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) @@ -1516,6 +1531,7 @@ def api_update_public_workspace_tag(tag_name): 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, @@ -1523,11 +1539,6 @@ def api_update_public_workspace_tag(tag_name): 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 tag_defs = ws_doc.get('tag_definitions', {}) @@ -1596,7 +1607,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) @@ -1620,6 +1631,7 @@ def api_delete_public_workspace_tag(tag_name): 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, @@ -1627,11 +1639,6 @@ def api_delete_public_workspace_tag(tag_name): 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 tag_defs = ws_doc.get('tag_definitions', {}) diff --git a/application/single_app/static/js/public/public_workspace.js b/application/single_app/static/js/public/public_workspace.js index 3410be79a..6a66ecacd 100644 --- a/application/single_app/static/js/public/public_workspace.js +++ b/application/single_app/static/js/public/public_workspace.js @@ -3037,6 +3037,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 667b8c60b..5be858b06 100644 --- a/application/single_app/static/js/workspace/workspace-documents.js +++ b/application/single_app/static/js/workspace/workspace-documents.js @@ -802,27 +802,26 @@ function renderWorkspaceDocumentView() { window.renderWorkspaceDocumentView = renderWorkspaceDocumentView; -async function patchDocumentMetadata(documentId, payload) { - let chunkSync = null; - do { - const requestPayload = { ...payload }; - if (chunkSync && !chunkSync.complete) { - requestPayload._metadata_chunk_sync = chunkSync; - } +const SEARCH_SYNC_PENDING_MESSAGE = "Metadata saved. Search and chat results will reflect the change once the search index finishes updating."; - const response = await fetch(`/api/documents/${documentId}`, { - method: "PATCH", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify(requestPayload), - }); +async function patchDocumentMetadata(documentId, payload) { + const response = await fetch(`/api/documents/${documentId}`, { + method: "PATCH", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(payload), + }); - const responseData = await response.json().catch(() => ({})); - if (!response.ok) { - throw responseData.error ? responseData : { error: `Server responded with status ${response.status}` }; - } + const responseData = await response.json().catch(() => ({})); + if (!response.ok) { + throw responseData.error ? responseData : { error: `Server responded with status ${response.status}` }; + } + return responseData; +} - chunkSync = responseData.metadata_chunk_sync || null; - } while (chunkSync && !chunkSync.complete); +function notifySearchSyncPending(responseData) { + if (responseData?.search_sync?.status === "pending" && typeof window.showToast === "function") { + window.showToast(SEARCH_SYNC_PENDING_MESSAGE, "info"); + } } function getDocumentDeleteModalContent(documentCount) { @@ -1452,8 +1451,9 @@ if (docMetadataForm && docMetadataModalEl) { // Check both exist } patchDocumentMetadata(docId, payload) - .then(() => { + .then((responseData) => { if (docMetadataModalEl) docMetadataModalEl.hide(); + notifySearchSyncPending(responseData); 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 b715d2486..2c7327a99 100644 --- a/application/single_app/templates/group_workspaces.html +++ b/application/single_app/templates/group_workspaces.html @@ -6725,6 +6725,9 @@