From bfa4dc502a1e92fafc515664576164324529d6be Mon Sep 17 00:00:00 2001 From: curo1305 Date: Mon, 22 Jun 2026 08:29:41 +0200 Subject: [PATCH] =?UTF-8?q?feat(12.1-02):=20centralize=20listing=20freshne?= =?UTF-8?q?ss=20gate=20=E2=80=94=20apply=5Flisting=5Fand=5Ffinalize?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add ListingResult dataclass to cloud_items.py - Add apply_listing_and_finalize() — single shared gate for complete/incomplete semantics - complete=True: upserts, soft-deletes unseen children, marks fresh, advances last_refreshed_at - complete=False: upserts seen items only, never deletes unseen, preserves prior timestamp, warns - browse.py: replace unconditional update_folder_state('fresh') with apply_listing_and_finalize - cloud_tasks.py: replace unconditional fresh transition with apply_listing_and_finalize - Celery task returns structured warning status on incomplete (no credentials/provider item names) - Fix timezone naive/aware comparison in test (SQLite strips tzinfo) --- backend/api/cloud/browse.py | 14 ++--- backend/services/cloud_items.py | 92 +++++++++++++++++++++++++++++++ backend/tasks/cloud_tasks.py | 26 +++++---- backend/tests/test_cloud_items.py | 8 ++- 4 files changed, 118 insertions(+), 22 deletions(-) diff --git a/backend/api/cloud/browse.py b/backend/api/cloud/browse.py index bf76e8d..383c7b6 100644 --- a/backend/api/cloud/browse.py +++ b/backend/api/cloud/browse.py @@ -31,6 +31,7 @@ from deps.db import get_db from deps.utils import parse_uuid from services.cloud_items import ( ConnectionNotFound, + apply_listing_and_finalize, list_cloud_children, get_or_create_folder_state, reconcile_cloud_listing, @@ -259,7 +260,9 @@ async def browse_connection_items( or _is_stale ) if should_refresh and not cached_items: - # First visit: do a synchronous bounded fetch so user sees content + # First visit: do a synchronous bounded fetch so user sees content. + # apply_listing_and_finalize is the single shared gate — it handles + # complete vs. incomplete semantics without unconditional fresh state. try: credentials = _decrypt_connection(conn, current_user.id) adapter = build_cloud_resource_adapter(conn.provider, credentials) @@ -268,20 +271,13 @@ async def browse_connection_items( current_user.id, parent_ref=parent_ref, ) - await reconcile_cloud_listing( + await apply_listing_and_finalize( session, user_id=uid_str, connection_id=cid_str, parent_ref=parent_ref, listing=listing, ) - await update_folder_state( - session, - user_id=uid_str, - connection_id=cid_str, - parent_ref=parent_ref or "", - refresh_state="fresh", - ) await session.commit() # Re-read items after reconciliation cached_items = await list_cloud_children( diff --git a/backend/services/cloud_items.py b/backend/services/cloud_items.py index 4730d69..e9ef7b4 100644 --- a/backend/services/cloud_items.py +++ b/backend/services/cloud_items.py @@ -12,6 +12,7 @@ Domain exceptions: from __future__ import annotations import uuid +from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Optional, Sequence @@ -250,6 +251,97 @@ async def get_or_create_folder_state( return fs +# ── Listing application result ──────────────────────────────────────────────── + +@dataclass +class ListingResult: + """Structured result from apply_listing_and_finalize. + + is_fresh: True only when the provider supplied a complete authoritative listing. + warning_code: stable controlled code when is_fresh is False (never raw exception text). + warning_message: stable controlled message for API consumers (no credentials/URLs). + """ + is_fresh: bool + warning_code: Optional[str] = None + warning_message: Optional[str] = None + + +# ── Single listing application gate ────────────────────────────────────────── + +async def apply_listing_and_finalize( + session: AsyncSession, + *, + user_id: str, + connection_id: str, + parent_ref: Optional[str], + listing: CloudListing, +) -> ListingResult: + """Apply a provider listing and finalize CloudFolderState atomically. + + This is the SINGLE shared gate used by both the synchronous browse endpoint + and the Celery background worker. Neither caller may independently decide + freshness — they MUST use this function. + + Semantics: + - listing.complete=True (authoritative): + Upserts all items, soft-deletes unseen children, marks folder FRESH, + advances last_refreshed_at. Returns ListingResult(is_fresh=True). + + - listing.complete=False (partial/unknown): + Upserts only seen items (never deletes unseen children), does NOT + advance last_refreshed_at, sets folder state to WARNING with stable + code 'incomplete_listing'. Returns ListingResult(is_fresh=False, …). + + No exception is raised on incomplete results — callers receive a structured + result and must propagate the warning state to the API response or task result. + This function never raises HTTPException — domain exceptions only. + This function never calls the quota service. + """ + # Reconcile items from the provider listing + await reconcile_cloud_listing( + session, + user_id=user_id, + connection_id=connection_id, + parent_ref=parent_ref, + listing=listing, + ) + + if listing.complete: + # Authoritative: mark folder fresh and advance last_refreshed_at + await update_folder_state( + session, + user_id=user_id, + connection_id=connection_id, + parent_ref=parent_ref or "", + refresh_state="fresh", + ) + return ListingResult(is_fresh=True) + else: + # Incomplete/partial: retain cached state; do NOT advance last_refreshed_at + # We only update the warning fields, preserving last_refreshed_at from prior success. + fs = await get_or_create_folder_state( + session, + user_id=user_id, + connection_id=connection_id, + parent_ref=parent_ref or "", + ) + now = datetime.now(timezone.utc) + # Set warning state without touching last_refreshed_at + fs.refresh_state = "warning" + fs.error_code = "incomplete_listing" + fs.error_message = ( + "Provider returned an incomplete listing. Showing cached results." + ) + fs.updated_at = now + # Deliberately: do NOT update fs.last_refreshed_at + await session.flush() + return ListingResult( + is_fresh=False, + warning_code="incomplete_listing", + warning_message="Provider returned an incomplete listing. Showing cached results.", + ) + + async def update_folder_state( session: AsyncSession, *, diff --git a/backend/tasks/cloud_tasks.py b/backend/tasks/cloud_tasks.py index aed2faf..19d232b 100644 --- a/backend/tasks/cloud_tasks.py +++ b/backend/tasks/cloud_tasks.py @@ -59,6 +59,7 @@ async def _run(user_id: str, connection_id: str, parent_ref: str | None) -> dict from db.session import AsyncSessionLocal from services.cloud_items import ( ConnectionNotFound, + apply_listing_and_finalize, get_or_create_folder_state, reconcile_cloud_listing, resolve_owned_connection, @@ -137,25 +138,28 @@ async def _run(user_id: str, connection_id: str, parent_ref: str | None) -> dict # Transient failure — let Celery retry raise _TransientProviderError("Provider error. Will retry.") from exc - # Reconcile listing into durable cloud_items rows - await reconcile_cloud_listing( + # apply_listing_and_finalize is the single shared gate — it handles + # complete vs. incomplete semantics; only a complete authoritative + # listing transitions the folder to fresh state. + listing_result = await apply_listing_and_finalize( session, user_id=user_id, connection_id=connection_id, parent_ref=parent_ref, listing=listing, ) - - await update_folder_state( - session, - user_id=user_id, - connection_id=connection_id, - parent_ref=parent_ref or "", - refresh_state="fresh", - ) await session.commit() - return {"status": "ok", "items_fetched": len(listing.items)} + if listing_result.is_fresh: + return {"status": "ok", "items_fetched": len(listing.items)} + else: + # Incomplete listing: task succeeded but folder is not fresh. + # Return a controlled status — no credentials, no provider item names. + return { + "status": "warning", + "warning_code": listing_result.warning_code, + "items_fetched": len(listing.items), + } # ── Celery task ─────────────────────────────────────────────────────────────── diff --git a/backend/tests/test_cloud_items.py b/backend/tests/test_cloud_items.py index 24e82c3..4a7d029 100644 --- a/backend/tests/test_cloud_items.py +++ b/backend/tests/test_cloud_items.py @@ -730,8 +730,12 @@ async def test_incomplete_listing_retains_cached_rows_and_last_success(db_sessio ) )).scalars().first() assert fs is not None - assert fs.last_refreshed_at == prior_ts, ( - f"incomplete listing must not overwrite last_refreshed_at: {fs.last_refreshed_at} != {prior_ts}" + # SQLite stores naive datetimes; strip tzinfo for comparison + stored_ts = fs.last_refreshed_at + if stored_ts is not None and stored_ts.tzinfo is None: + stored_ts = stored_ts.replace(tzinfo=timezone.utc) + assert stored_ts == prior_ts, ( + f"incomplete listing must not overwrite last_refreshed_at: {fs.last_refreshed_at!r} != {prior_ts!r}" )