diff --git a/src/govoplan_calendar/backend/outbox.py b/src/govoplan_calendar/backend/outbox.py index c31ebce..d59a9c2 100644 --- a/src/govoplan_calendar/backend/outbox.py +++ b/src/govoplan_calendar/backend/outbox.py @@ -858,15 +858,18 @@ def _fail_operation( session.flush() -def execute_calendar_outbox_operation( +def _lock_leased_outbox_operation( session: Session, *, operation_id: str, lease_token: str, - client_factory: Callable[[Session, CalendarSyncSource], object] | None = None, -) -> CalendarOutboxOperation: +) -> tuple[CalendarOutboxOperation, CalendarSyncSource | None]: operation = session.get(CalendarOutboxOperation, operation_id) - if operation is None or operation.status != "in_progress" or operation.lease_token != lease_token: + if ( + operation is None + or operation.status != "in_progress" + or operation.lease_token != lease_token + ): raise ValueError("Calendar outbox lease is no longer valid") source = ( session.query(CalendarSyncSource) @@ -882,151 +885,282 @@ def execute_calendar_outbox_operation( ) if operation.status != "in_progress" or operation.lease_token != lease_token: raise ValueError("Calendar outbox lease is no longer valid") - operation_metadata = operation.metadata_ or {} - unavailable_reason = _operation_source_unavailable_reason(operation, source) - if unavailable_reason: - if source is not None and source.tenant_id == operation.tenant_id: - _fail_operation( - session, - operation=operation, - source=source, - error=unavailable_reason, - terminal_status="cancelled", - ) - else: - operation.status = "cancelled" - operation.completed_at = utcnow() - operation.last_error = unavailable_reason - operation.lease_token = None - operation.lease_expires_at = None - session.flush() - return operation + return operation, source - overwrite = bool(operation_metadata.get("overwrite")) if isinstance(operation_metadata, dict) else False - client: object | None = None - try: - if client_factory is None: - from govoplan_calendar.backend.service import caldav_client_for_source - client = caldav_client_for_source(session, source) - else: - client = client_factory(session, source) - if operation.operation_kind == "put": - result = client.put_object( - operation.resource_href, - operation.payload_ics or "", - etag=operation.expected_etag, - create=not bool(operation.expected_etag), - overwrite=overwrite, - ) - remote_etag = result.etag - reconciled = False - if not remote_etag: - try: - matched, remote_etag = _remote_matches_operation(client, operation) - except Exception: - matched, remote_etag = False, None - if not matched or not remote_etag: - _fail_operation( - session, - operation=operation, - source=source, - error="CalDAV PUT succeeded without an ETag and could not be reconciled safely", - ) - return operation - reconciled = True - _complete_operation( - session, - operation=operation, - source=source, - remote_etag=remote_etag, - reconciled=reconciled, - ) - return operation - - if not operation.expected_etag and not overwrite: - matched, remote_etag = _remote_matches_operation(client, operation) - if matched: - _complete_operation( - session, - operation=operation, - source=source, - remote_etag=remote_etag, - reconciled=True, - ) - return operation - _fail_operation( - session, - operation=operation, - source=source, - error=( - "CalDAV DELETE has no known ETag and the remote resource exists; " - "refusing to delete an unknown remote version" - ), - terminal_status="conflict", - ) - return operation - result = client.delete_object( - operation.resource_href, - etag=operation.expected_etag, - overwrite=overwrite, +def _cancel_undeliverable_operation( + session: Session, + *, + operation: CalendarOutboxOperation, + source: CalendarSyncSource | None, + reason: str, +) -> None: + if source is not None and source.tenant_id == operation.tenant_id: + _fail_operation( + session, + operation=operation, + source=source, + error=reason, + terminal_status="cancelled", ) + return + operation.status = "cancelled" + operation.completed_at = utcnow() + operation.last_error = reason + operation.lease_token = None + operation.lease_expires_at = None + session.flush() + + +def _calendar_outbox_client( + session: Session, + *, + source: CalendarSyncSource, + client_factory: Callable[[Session, CalendarSyncSource], object] | None, +) -> object: + if client_factory is not None: + return client_factory(session, source) + from govoplan_calendar.backend.service import caldav_client_for_source + + return caldav_client_for_source(session, source) + + +def _try_remote_match( + client: object, + operation: CalendarOutboxOperation, +) -> tuple[bool, str | None] | None: + try: + return _remote_matches_operation(client, operation) + except Exception: + return None + + +def _complete_if_remote_match_is_usable( + session: Session, + *, + operation: CalendarOutboxOperation, + source: CalendarSyncSource, + remote_match: tuple[bool, str | None] | None, +) -> bool: + if remote_match is None: + return False + matched, remote_etag = remote_match + if not matched or (operation.operation_kind != "delete" and not remote_etag): + return False + _complete_operation( + session, + operation=operation, + source=source, + remote_etag=remote_etag, + reconciled=True, + ) + return True + + +def _execute_calendar_outbox_put( + session: Session, + *, + operation: CalendarOutboxOperation, + source: CalendarSyncSource, + client: object, + overwrite: bool, +) -> None: + result = client.put_object( + operation.resource_href, + operation.payload_ics or "", + etag=operation.expected_etag, + create=not bool(operation.expected_etag), + overwrite=overwrite, + ) + remote_etag = result.etag + if remote_etag: _complete_operation( session, operation=operation, source=source, - remote_etag=result.etag, - reconciled=result.status == 404, + remote_etag=remote_etag, + reconciled=False, + ) + return + remote_match = _try_remote_match(client, operation) + if _complete_if_remote_match_is_usable( + session, + operation=operation, + source=source, + remote_match=remote_match, + ): + return + _fail_operation( + session, + operation=operation, + source=source, + error="CalDAV PUT succeeded without an ETag and could not be reconciled safely", + ) + + +def _execute_calendar_outbox_delete( + session: Session, + *, + operation: CalendarOutboxOperation, + source: CalendarSyncSource, + client: object, + overwrite: bool, +) -> None: + if not operation.expected_etag and not overwrite: + remote_match = _remote_matches_operation(client, operation) + if _complete_if_remote_match_is_usable( + session, + operation=operation, + source=source, + remote_match=remote_match, + ): + return + _fail_operation( + session, + operation=operation, + source=source, + error=( + "CalDAV DELETE has no known ETag and the remote resource exists; " + "refusing to delete an unknown remote version" + ), + terminal_status="conflict", + ) + return + result = client.delete_object( + operation.resource_href, + etag=operation.expected_etag, + overwrite=overwrite, + ) + _complete_operation( + session, + operation=operation, + source=source, + remote_etag=result.etag, + reconciled=result.status == 404, + ) + + +def _handle_outbox_precondition_failure( + session: Session, + *, + operation: CalendarOutboxOperation, + source: CalendarSyncSource, + client: object | None, + error: CalDAVPreconditionFailed, +) -> None: + remote_match = _try_remote_match(client, operation) if client is not None else None + if _complete_if_remote_match_is_usable( + session, + operation=operation, + source=source, + remote_match=remote_match, + ): + return + if remote_match is None: + _fail_operation(session, operation=operation, source=source, error=error) + return + matched, _remote_etag = remote_match + _fail_operation( + session, + operation=operation, + source=source, + error=( + "Matching CalDAV resource has no usable ETag; retrying reconciliation" + if matched + else "CalDAV resource changed remotely; reconcile or sync before retrying" + ), + terminal_status=None if matched else "conflict", + ) + + +def _handle_outbox_delivery_failure( + session: Session, + *, + operation: CalendarOutboxOperation, + source: CalendarSyncSource, + client: object | None, + error: Exception, +) -> None: + remote_match = _try_remote_match(client, operation) if client is not None else None + if _complete_if_remote_match_is_usable( + session, + operation=operation, + source=source, + remote_match=remote_match, + ): + return + _fail_operation(session, operation=operation, source=source, error=error) + + +def execute_calendar_outbox_operation( + session: Session, + *, + operation_id: str, + lease_token: str, + client_factory: Callable[[Session, CalendarSyncSource], object] | None = None, +) -> CalendarOutboxOperation: + operation, source = _lock_leased_outbox_operation( + session, + operation_id=operation_id, + lease_token=lease_token, + ) + unavailable_reason = _operation_source_unavailable_reason(operation, source) + if unavailable_reason: + _cancel_undeliverable_operation( + session, + operation=operation, + source=source, + reason=unavailable_reason, ) return operation - except CalDAVPreconditionFailed as exc: - if client is None: - _fail_operation(session, operation=operation, source=source, error=exc) - return operation - try: - matched, remote_etag = _remote_matches_operation(client, operation) - except Exception: - _fail_operation(session, operation=operation, source=source, error=exc) - else: - if matched and (operation.operation_kind == "delete" or remote_etag): - _complete_operation( - session, - operation=operation, - source=source, - remote_etag=remote_etag, - reconciled=True, - ) - else: - _fail_operation( - session, - operation=operation, - source=source, - error=( - "Matching CalDAV resource has no usable ETag; retrying reconciliation" - if matched - else "CalDAV resource changed remotely; reconcile or sync before retrying" - ), - terminal_status=None if matched else "conflict", - ) - return operation - except Exception as exc: - matched, remote_etag = False, None - if client is not None: - try: - matched, remote_etag = _remote_matches_operation(client, operation) - except Exception: - matched, remote_etag = False, None - if matched and (operation.operation_kind == "delete" or remote_etag): - _complete_operation( + + assert source is not None + operation_metadata = operation.metadata_ or {} + overwrite = ( + bool(operation_metadata.get("overwrite")) + if isinstance(operation_metadata, dict) + else False + ) + client: object | None = None + try: + client = _calendar_outbox_client( + session, + source=source, + client_factory=client_factory, + ) + if operation.operation_kind == "put": + _execute_calendar_outbox_put( session, operation=operation, source=source, - remote_etag=remote_etag, - reconciled=True, + client=client, + overwrite=overwrite, ) else: - _fail_operation(session, operation=operation, source=source, error=exc) - return operation + _execute_calendar_outbox_delete( + session, + operation=operation, + source=source, + client=client, + overwrite=overwrite, + ) + except CalDAVPreconditionFailed as exc: + _handle_outbox_precondition_failure( + session, + operation=operation, + source=source, + client=client, + error=exc, + ) + except Exception as exc: + _handle_outbox_delivery_failure( + session, + operation=operation, + source=source, + client=client, + error=exc, + ) + return operation def dispatch_calendar_outbox(