feat(CFTL-489): fail-on-missing-permissions option for Meta extractors (SUPPORT-15700)

docs/superpowers/specs/2026-06-10-fail-on-missing-permissions-design.md

@@ -0,0 +1,101 @@
+# Fail on missing permissions — Meta extractors (CFTL-489 / SUPPORT-15700)
+
+## Problem
+
+The Meta extractors (`keboola.ex-facebook-pages`, `keboola.ex-facebook-ads-v2`,
+`keboola.ex-instagram-v2`, all built from this one codebase) can finish a job with
+**exit 0 / Success** while extracting nothing beyond the `accounts` table.
+
+The `/me/accounts` endpoint does not require `ads_read` / `ads_management`, so it always
+succeeds. When the token has lost per-account permissions (or expired), every other query
+fails per-account with a Facebook OAuth error (codes 190 / 200 / 10, or 100/subcode 33
+"missing permissions"). Those errors are currently **swallowed** in two layers:
+
+1. `page_loader.FacebookErrorHandler.is_recoverable_error()` treats the 100/33
+   "missing permissions" case as recoverable and returns `{"data": []}`.
+2. The per-account `except Exception: ... continue` loops in `client.py` log-and-skip
+   non-recoverable errors (e.g. code 200).
+
+Result: the job reports Success and the missing data goes unnoticed.
+
+## Goal
+
+Add an opt-in option so the job **fails** (exit 1, `UserException`) when the token is
+missing permissions, instead of silently succeeding. Default OFF preserves today's
+behavior (most existing configs rely on it).
+
+## Decisions (confirmed with requester)
+
+- **Detection:** per-account authorization errors raised during extraction. No upfront
+  `debug_token()` pre-flight — the reported case is a valid token missing per-account
+  grants, which a token-validity check would not catch.
+- **Failure timing:** **collect all, fail at the end.** Authorization errors are accumulated
+  per account during extraction; after all queries run, the job fails once with a
+  `UserException` listing every affected account. Safe because a non-zero exit makes the
+  platform discard all output — no partial data is committed regardless of when we raise — so
+  this gives a complete error message at no data-integrity cost. (Initial design was fail-fast;
+  switched to collect-all on review.)
+- **UI scope:** show the checkbox for all three Meta extractors (shared codebase + shared
+  `ex-facebook` UI module).
+- **Config key:** `parameters.fail-on-missing-permissions` (boolean, default `false`).
+- **UI label:** "Fail the job on authorization errors".
+
+## Implementation
+
+### Python (`component-meta`)
+
+**`configuration.py`** — add
+`fail_on_missing_permissions: bool = Field(alias="fail-on-missing-permissions", default=False)`.
+
+**`page_loader.py`**
+- `_FB_AUTHORIZATION_ERROR_CODES = frozenset({10, 190, 200})`.
+- `class AuthorizationError(Exception)` — internal sentinel carrying `account_id`, `code`,
+  `message`. Deliberately **not** a `UserException` so it doesn't abort on the first hit.
+- `FacebookErrorHandler.is_authorization_error(http_error) -> bool` — True for the codes
+  above, or the existing `OBJECT_NOT_FOUND_ERROR` (100/33 "missing permissions") match.
+- `FacebookErrorHandler.authorization_error_details(http_error) -> (code, message)`.
+- `PageLoader.__init__` gains `fail_on_missing_permissions: bool = False`.
+- In each HTTP error boundary — `_load_regular_page`, `load_page_from_url`,
+  `start_async_insights_job`, and the async final-results fetch in `poll_async_job` — when
+  the flag is on and the error is an authorization error, raise `AuthorizationError`
+  **before** the recoverable/return-empty logic.
+
+**`client.py`**
+- `FacebookClient.__init__` gains the flag and a `permission_errors: list[dict]` collector;
+  passes the flag to all three `PageLoader(...)` constructions.
+- `_record_permission_error(error, query_name)` appends a record; `raise_for_permission_errors()`
+  raises one `UserException` summarizing every affected account (deduped by account id).
+- Per-account loops (`_start_async_jobs_for_query`, `_poll_and_process_async_jobs`,
+  `_process_single_sync_query` + its page-token/user-token fallback) catch `AuthorizationError`
+  and record-and-continue; `except UserException: raise` is kept for genuine immediate failures.
+- Batch path records when the error is an authorization error.
+
+**`component.py`** — pass `self.config.fail_on_missing_permissions` into `FacebookClient(...)`,
+and call `self.client.raise_for_permission_errors()` after `_process_queries` finishes the loop.
+
+### UI (`ui/apps/kbc-ui/src/scripts/modules/ex-facebook`)
+
+Mirror the existing config-level API-version pattern:
+- `constants.ts` — help/tooltip text constant.
+- `storeProvisioning.js` — read `parameters.get('fail-on-missing-permissions', false)`.
+- `Index.jsx` — render the checkbox in the config-level settings area.
+- `actionsProvisioning.js` — `saveFailOnMissingPermissions` writing
+  `parameters.fail-on-missing-permissions`.
+
+Shared module → the checkbox shows for facebook-pages, facebook-ads-v2, instagram-v2.
+
+## Tests
+
+- Unit tests for `is_authorization_error` / `authorization_error_details`
+  (codes 10/190/200 + 100/33 → True; unrelated codes → False).
+- PageLoader boundary: flag ON → code-200 / 100-33 raise `AuthorizationError`; flag OFF →
+  100/33 returns empty and code-200 re-raises `HTTPError` (current behavior).
+- Client: collects per-account `AuthorizationError`s without aborting mid-iteration, then
+  `raise_for_permission_errors()` raises one `UserException` listing every account (deduped).
+  Existing datadir/VCR tests unaffected (default off).
+
+## Deliverables
+
+- PR in `component-meta` (Python + tests).
+- PR in `ui` (checkbox).
+- Cross-link both PRs and attach to Linear CFTL-489.

src/client.py

@@ -16,6 +16,8 @@
     _FB_TRANSIENT_ERROR_BACKOFF_BASE,
     _FB_TRANSIENT_ERROR_MAX_RETRIES,
     AsyncInsightsJobTransientError,
+    AuthorizationError,
+    FacebookErrorHandler,
     PageLoader,
 )
 
@@ -121,9 +123,13 @@ def get_page_tokens(
 
 
 class FacebookClient:
-    def __init__(self, oauth: OauthCredentials, api_version: str):
+    def __init__(self, oauth: OauthCredentials, api_version: str, fail_on_missing_permissions: bool = False):
         self.oauth = oauth
         self.api_version = api_version
+        self.fail_on_missing_permissions = fail_on_missing_permissions
+        # Authorization errors collected across all accounts/queries when the option is enabled.
+        # The run fails at the end (raise_for_permission_errors) so every affected account is reported.
+        self.permission_errors: list[dict[str, Any]] = []
         self.page_tokens = None  # Cache for page tokens
         # Count of objects skipped due to contained API errors; surfaced as an
         # end-of-run warning so partial output is not silently read as complete.
@@ -148,6 +154,48 @@ def _with_token(self, params: dict[str, Any] | None, token: str | None = None) -
         params["access_token"] = token or self.oauth.data.get("access_token")
         return params
 
+    def _record_permission_error(self, error: AuthorizationError, query_name: str | None = None) -> None:
+        """Collect a per-account authorization error to report (and fail) at the end of the run."""
+        self.permission_errors.append(
+            {
+                "account_id": error.account_id,
+                "query": query_name,
+                "code": error.code,
+                "message": error.message,
+            }
+        )
+
+    def raise_for_permission_errors(self) -> None:
+        """Raise a single UserException summarizing every account that hit an authorization error.
+
+        Called after all queries have been processed so the user sees the full list at once.
+        A non-zero exit means the platform discards all output, so no partial data is committed.
+        """
+        if not self.permission_errors:
+            return
+
+        # One line per affected account (dedupe by account id; fall back to message when unknown)
+        seen: dict[Any, dict[str, Any]] = {}
+        for err in self.permission_errors:
+            key = err.get("account_id") or err.get("message")
+            seen.setdefault(key, err)
+
+        lines = []
+        for err in seen.values():
+            account = err.get("account_id") or "unknown account"
+            message = err.get("message") or ""
+            code = err.get("code")
+            detail = f"{message} (code {code})" if message else f"code {code}"
+            lines.append(f"  - account {account}: {detail}")
+
+        raise UserException(
+            f"Facebook authorization errors prevented data extraction for {len(seen)} account(s). "
+            "The access token is missing required permissions — re-authorize the extractor or grant "
+            "the necessary access (e.g. ads_read / ads_management) on the affected accounts:\n"
+            + "\n".join(lines)
+            + "\nDisable 'Fail the job on authorization errors' to skip inaccessible accounts and continue."
+        )
+
     def _extract_page_content(self, query_path: str | None, page_data: dict[str, Any]) -> list[dict[str, Any]]:
         """
         Extract content from page data response.
@@ -212,7 +260,12 @@ def _start_async_jobs_for_query(self, accounts: list, row_config) -> dict:
             page_id = str(page_id)
             try:
                 # Use the shared client and pass token in params
-                page_loader = PageLoader(self.client, row_config.type, self.api_version)
+                page_loader = PageLoader(
+                    self.client,
+                    row_config.type,
+                    self.api_version,
+                    fail_on_missing_permissions=self.fail_on_missing_permissions,
+                )
                 report_id = page_loader.start_async_insights_job(
                     row_config.query, page_id, params=self._with_token({}, token)
                 )
@@ -227,6 +280,8 @@ def _start_async_jobs_for_query(self, accounts: list, row_config) -> dict:
                         "row_config": row_config,
                         "start_params": self._with_token({}, token),
                     }
+            except AuthorizationError as e:
+                self._record_permission_error(e, row_config.name)
             except Exception as e:
                 logger.error(f"Failed to start async job for {page_id}: {e}")
         return job_details
@@ -239,6 +294,12 @@ def _poll_and_process_async_jobs(self, all_job_details: dict) -> Iterator[dict]:
                 if not page_data.get("data"):
                     continue
                 yield from details["output_parser"].iter_parsed_data(page_data, details["fb_graph_node"], page_id)
+            except UserException:
+                raise
+            except AuthorizationError as e:
+                if e.account_id is None:
+                    e.account_id = page_id
+                self._record_permission_error(e, details["row_config"].name)
             except _CONTAINED_OBJECT_ERRORS as e:
                 # Transient/API failure for this one report — contain it so the rest of the
                 # run completes; UserException and programming errors still propagate.
@@ -345,6 +406,14 @@ def _process_single_sync_query(self, accounts: list[Account], row_config: QueryR
                         logger.info("Batch request requires page token, falling back to individual requests.")
                         # Let the code fall through to individual processing below.
                     else:
+                        # Collect authorization errors (batch covers all accounts at once)
+                        if self.fail_on_missing_permissions and FacebookErrorHandler.is_authorization_error(e):
+                            code, message = FacebookErrorHandler.authorization_error_details(e)
+                            self._record_permission_error(
+                                AuthorizationError(account_id=None, code=code, message=message),
+                                row_config.name,
+                            )
+                            return
                         logger.error(f"Batch request failed with a non-token error: {error_text}")
                         return  # A definitive failure, stop processing.
 
@@ -375,7 +444,12 @@ def _process_single_sync_query(self, accounts: list[Account], row_config: QueryR
             try:
                 # Create new client with page token
                 # Use the shared client and pass token in params
-                page_loader = PageLoader(self.client, row_config.type, self.api_version)
+                page_loader = PageLoader(
+                    self.client,
+                    row_config.type,
+                    self.api_version,
+                    fail_on_missing_permissions=self.fail_on_missing_permissions,
+                )
                 output_parser = OutputParser(page_loader, page_id, row_config)
 
                 # Construct Facebook Graph node path
@@ -386,18 +460,32 @@ def _process_single_sync_query(self, accounts: list[Account], row_config: QueryR
                 page_content = self._extract_page_content(row_config.query.path, page_data)
 
             except Exception as e:
-                if is_page_token and str(e).startswith("400"):
-                    logger.debug(f"Page token failed for {page_id}, trying user token")
+                is_auth_error = isinstance(e, AuthorizationError)
+                # For page-token queries, the page token itself may lack access — retry with the
+                # user token (also the original recovery for "400 Page Access Token" errors).
+                if is_page_token and (is_auth_error or str(e).startswith("400")):
+                    logger.debug(f"Primary token failed for {page_id}, trying user token")
                     try:
                         # Fallback to user token
-                        page_loader = PageLoader(self.client, row_config.type, self.api_version)
+                        page_loader = PageLoader(
+                            self.client,
+                            row_config.type,
+                            self.api_version,
+                            fail_on_missing_permissions=self.fail_on_missing_permissions,
+                        )
                         output_parser = OutputParser(page_loader, page_id, row_config)
                         fb_graph_node = self._get_fb_graph_node(False, row_config)
                         page_data = page_loader.load_page(row_config.query, page_id, params=self._with_token({}))
                         page_content = self._extract_page_content(row_config.query.path, page_data)
+                    except AuthorizationError as fallback_error:
+                        self._record_permission_error(fallback_error, row_config.name)
+                        continue
                     except Exception as user_token_error:
                         logger.debug(f"User token also failed for {page_id}: {str(user_token_error)}")
                         continue
+                elif is_auth_error:
+                    self._record_permission_error(e, row_config.name)
+                    continue
                 else:
                     logger.error(f"Failed to load data for {page_id}: {str(e)}")
                     continue
@@ -412,6 +500,12 @@ def _process_single_sync_query(self, accounts: list[Account], row_config: QueryR
             query_name = getattr(row_config, "name", None) or getattr(getattr(row_config, "query", None), "path", "?")
             try:
                 yield from output_parser.iter_parsed_data(page_data, fb_graph_node, page_id)
+            except AuthorizationError as e:
+                # A permission error during lazy pagination is collected like any other.
+                if e.account_id is None:
+                    e.account_id = page_id
+                self._record_permission_error(e, row_config.name)
+                continue
             except _CONTAINED_OBJECT_ERRORS as e:
                 # Contain transient/API failures for this one object; UserException
                 # (user-actionable) and programming errors deliberately propagate.

src/component.py

@@ -144,7 +144,11 @@ def __init__(self):
         params = self.configuration.parameters
         params["accounts"] = params.get("accounts") or {}
         self.config = Configuration(**params)
-        self.client: FacebookClient = FacebookClient(self.configuration.oauth_credentials, self.config.api_version)
+        self.client: FacebookClient = FacebookClient(
+            self.configuration.oauth_credentials,
+            self.config.api_version,
+            fail_on_missing_permissions=self.config.fail_on_missing_permissions,
+        )
         self.bucket_id = self._retrieve_bucket_id()
 
     def run(self) -> None:
@@ -195,6 +199,10 @@ def _process_queries(self, config: Configuration) -> None:
                 f"output may be incomplete for this run."
             )
 
+        # If the option is enabled, fail the job once with every account that lacked permissions.
+        # Raising here (exit 1) means the platform discards all output, so no partial data lands.
+        self.client.raise_for_permission_errors()
+
     def _finalize_tables(self) -> None:
         for cache_record in self._writer_cache.values():
             cache_record.writer.writeheader()

src/configuration.py

@@ -39,3 +39,6 @@ class Configuration(BaseModel):
     queries: list[QueryRow] = Field(default_factory=list)
     api_version: str = Field(alias="api-version", default="v23.0")
     bucket_id: str | None = Field(alias="bucket-id", default=None)
+    # When True, a Facebook authorization error (expired token or missing per-account
+    # permissions) raised during extraction fails the job instead of being skipped.
+    fail_on_missing_permissions: bool = Field(alias="fail-on-missing-permissions", default=False)